"""Worker RAR — proces propriu (NU task asyncio in uvicorn).

Bucla: heartbeat -> recupereaza orfane -> claim atomic -> login -> postPrezentare -> update.
Ruleaza ca proces separat sub `restart: always` (docker compose).

  - claim atomic anti-race (BEGIN IMMEDIATE), respecta next_attempt_at (backoff).
  - reconciliere anti-duplicat pe raspuns pierdut: pe eroare tranzitorie/timeout SAU pe
    rand 'sending' orfan (worker mort mid-POST), interogheaza finalizate si match pe
    vin+dataPrestatie+odometruFinal; daca exista -> 'sent' (NU re-trimite).
  - retry/backoff exponential pe erori tranzitorii; peste worker_max_retries -> 'error'.
  - lease/timeout pe randuri 'sending' orfane.
  - re-login la token expirat (401 mid-sesiune) — JWT 30h, retry NU plafonat la 30h.

Creds per-cerere: fiecare submission poarta creds RAR CRIPTATE (submissions.rar_creds_enc).
Worker-ul face login per CONT cu acele creds, cache-uieste JWT (30h) in memorie si
STERGE creds-urile efemere dupa primul login reusit. Token-ul in memorie acopera
restul trimiterilor; la restart token-ul se pierde si contul re-logheaza la urmatorul
submission care aduce creds proaspete (degradare acceptata). Fallback durabil: slotul
per-env al contului (accounts.rar_creds_{env}_enc, US-013; coloana legacy dropata).
Dev: `worker_use_test_creds` foloseste creds <test> cand submission-ul nu are enc.

Pornire:    python -m app.worker
"""

from __future__ import annotations

import json
import signal
import sys
import time
from datetime import datetime, timedelta, timezone

import httpx

from .. import errors
from ..config import Settings, get_settings, load_test_credentials
from ..crypto import decrypt_creds
from ..db import get_connection, init_db, read_heartbeat, write_heartbeat
from ..observ import log_event, set_source
from ..mapping import DEFAULT_ACCOUNT_ID, upsert_nomenclator
from ..payload import build_rar_payload
from ..reconcile import match_finalizata
from ..rar_client import RarAuthError, RarClient, RarError, base_url_pentru_env

_running = True


def _stop(signum: int, frame: object) -> None:
    global _running
    _running = False


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _iso(dt: datetime) -> str:
    return dt.isoformat(timespec="seconds")


def _wlog(conn, tip: str, mesaj: str, *, nivel: str = "INFO", account_id=None, cod=None, context=None) -> None:
    """Emite evenimentul (sursa=worker, dublu canal DB+fisier) SI pastreaza linia in
    stdout (operatorul tailuieste .run/worker.log)."""
    print(f"[worker] {mesaj}", flush=True)
    log_event(tip, nivel=nivel, account_id=account_id, cod=cod, mesaj=mesaj, context=context,
              conn=conn, sursa="worker")


def _backoff_seconds(settings: Settings, retry_count: int) -> int:
    return min(settings.worker_retry_base_s * (2 ** max(0, retry_count - 1)), settings.worker_retry_max_s)


def _is_transient(exc: Exception) -> bool:
    """Eroare in care POST-ul poate sa fi ajuns/sa nu fi ajuns la RAR -> reconciliere + retry."""
    if isinstance(exc, (httpx.TimeoutException, httpx.TransportError)):
        return True
    if isinstance(exc, RarError) and not isinstance(exc, RarAuthError):
        code = exc.status_code
        return code is None or code >= 500 or code in (408, 429)
    return False


# --- Operatii pe submissions ---

# Stari blocate ne-sent care primesc retentie proprie. Mai scurta decat cele 90z
# ale `sent`: un blocat n-are valoare de audit ca o trimitere reusita.
_BLOCKED_STATES = ("error", "needs_data", "needs_mapping")


def mark(conn, submission_id: int, status: str, *, rar_status_code=None, rar_error=None, id_prezentare=None) -> None:
    if status == "sent":
        # purge_after = sent + 90 zile (GDPR/L.142 retentie maxima).
        purge_expr = "datetime('now', '+90 days')"
    elif status in _BLOCKED_STATES:
        # Randurile blocate primesc si ele purge_after (altfel raman permanent).
        days = int(get_settings().blocked_retention_days)
        purge_expr = f"datetime('now', '+{days} days')"
    else:
        purge_expr = None

    if purge_expr is not None:
        conn.execute(
            "UPDATE submissions SET status=?, rar_status_code=?, rar_error=?, id_prezentare=?, "
            f"sending_since=NULL, updated_at=datetime('now'), purge_after={purge_expr} WHERE id=?",
            (status, rar_status_code, rar_error, id_prezentare, submission_id),
        )
    else:
        conn.execute(
            "UPDATE submissions SET status=?, rar_status_code=?, rar_error=?, id_prezentare=?, "
            "sending_since=NULL, updated_at=datetime('now') WHERE id=?",
            (status, rar_status_code, rar_error, id_prezentare, submission_id),
        )


# Purge interval in secunde (odata pe ora, nu prea agresiv)
_PURGE_INTERVAL_S = 3600


def purge_expired(conn) -> dict[str, int]:
    """Sterge randurile expirate (purge_after < now).

    Submissions `sent` SI blocate (error/needs_data/needs_mapping) expirate;
    import_batches expirate (import_rows via CASCADE); app_events expirate (jurnal).
    EXCLUDE explicit `queued`/`sending` (randuri active — nu se purjeaza niciodata, chiar
    daca ar avea un purge_after rezidual; reactivarea il curata oricum).
    Intoarce {submissions_purged, batches_purged, events_purged}.
    """
    cur_sub = conn.execute(
        "DELETE FROM submissions WHERE purge_after IS NOT NULL AND purge_after < datetime('now') "
        "AND status IN ('sent','error','needs_data','needs_mapping')"
    )
    cur_batch = conn.execute(
        "DELETE FROM import_batches WHERE purge_after IS NOT NULL AND purge_after < datetime('now')"
    )
    cur_events = conn.execute(
        "DELETE FROM app_events WHERE purge_after IS NOT NULL AND purge_after < datetime('now')"
    )
    return {
        "submissions_purged": cur_sub.rowcount,
        "batches_purged": cur_batch.rowcount,
        "events_purged": cur_events.rowcount,
    }


def expire_held(conn, settings: Settings) -> int:
    """Expira randurile tinute manual (`queued AND held=1`) mai vechi de `held_retention_days`.

    PRD 5.19 US-008 (inchide gaura GDPR/L.142): un `queued held=1` nu e nici `sent` nici
    blocat -> altfel NU primeste `purge_after` -> PII criptat ar sta la nesfarsit.
    Le trece la `error`/`TINUT_EXPIRAT` (terminal) si seteaza `purge_after` DIRECT la
    momentul expirarii (NU lasa retentia de 30z a randurilor blocate sa se adauge; altfel
    viata reala = 90 + 30 = 120 zile, nu 90). `purge_after` e setat cu o secunda IN TRECUT
    (`datetime('now','-1 second')`): `purge_expired` foloseste comparatie STRICTA
    (`purge_after < datetime('now')`), iar `datetime('now')` are rezolutie de o secunda —
    cu `purge_after=now` cele doua ar fi egale in acelasi ciclu si randul NU s-ar purja
    decat la ciclul urmator (~1h de PII in plus). Cu `-1 second` randul e purjabil IMEDIAT
    in acelasi ciclu de purjare (US-008 inchide gaura strict).

    Intoarce numarul de randuri expirate.
    """
    # Numar de zile validat (int) — construim string-ul '-N days' fara input extern.
    days = int(settings.held_retention_days)
    cutoff = f"-{days} days"
    cur = conn.execute(
        "UPDATE submissions SET status='error', rar_error='TINUT_EXPIRAT', "
        "purge_after=datetime('now', '-1 second'), sending_since=NULL, updated_at=datetime('now') "
        "WHERE status='queued' AND held=1 AND created_at < datetime('now', ?)",
        (cutoff,),
    )
    return cur.rowcount


def requeue_with_backoff(conn, settings: Settings, submission_id: int, *, reason: str) -> None:
    """Re-pune randul in coada cu retry++ si next_attempt_at = now + backoff."""
    row = conn.execute("SELECT retry_count FROM submissions WHERE id=?", (submission_id,)).fetchone()
    new_retry = int(row["retry_count"]) + 1 if row else 1
    if new_retry > settings.worker_max_retries:
        mark(conn, submission_id, "error", rar_error=f"esuat dupa {new_retry-1} reincercari: {reason}")
        print(f"[worker] submission {submission_id} -> error (max retries): {reason}", flush=True)
        return
    next_at = _iso(_now() + timedelta(seconds=_backoff_seconds(settings, new_retry)))
    conn.execute(
        "UPDATE submissions SET status='queued', retry_count=?, next_attempt_at=?, "
        "rar_error=?, sending_since=NULL, updated_at=datetime('now') WHERE id=?",
        (new_retry, next_at, reason, submission_id),
    )
    print(f"[worker] submission {submission_id} -> requeue (retry {new_retry}, next {next_at}): {reason}", flush=True)


def claim_one(conn) -> dict | None:
    """Claim atomic 'queued' -> 'sending', respectand next_attempt_at. Intoarce randul sau None.

    Randul include `account_id` si `submissions.rar_creds_enc` (creds RAR criptate efemere)
    pentru login-ul per-cont din `run`.
    """
    conn.execute("BEGIN IMMEDIATE")
    try:
        row = conn.execute(
            "SELECT s.id, s.account_id, s.payload_json, s.rar_creds_enc, s.rar_env "
            "FROM submissions s LEFT JOIN accounts a ON a.id = s.account_id "
            "WHERE s.status='queued' "
            # PRD 5.19 US-003: randurile tinute manual (held=1) NU se revendica —
            # asteapta eliberarea umana (manual/auto). Se aplica DOAR la claim din
            # 'queued'; randurile deja 'sending' (orfani/reconciliere) sunt neafectate.
            "AND s.held = 0 "
            "AND (s.next_attempt_at IS NULL OR s.next_attempt_at <= ?) "
            # Gate pe stare de cont: doar 'active' trimite. Derivam defensiv din `active`
            # cand `status` lipseste (DB veche pre-migrare), pastrand active=1 <=> 'active'.
            "AND COALESCE(a.status, CASE WHEN COALESCE(a.active,1)=1 THEN 'active' ELSE 'pending' END) = 'active' "
            "ORDER BY s.id LIMIT 1",
            (_iso(_now()),),
        ).fetchone()
        if not row:
            conn.execute("COMMIT")
            return None
        conn.execute(
            "UPDATE submissions SET status='sending', sending_since=datetime('now'), "
            "updated_at=datetime('now') WHERE id=?",
            (row["id"],),
        )
        conn.execute("COMMIT")
        return {
            "id": row["id"],
            "account_id": row["account_id"] if row["account_id"] is not None else DEFAULT_ACCOUNT_ID,
            "rar_env": row["rar_env"],
            "creds_enc": row["rar_creds_enc"],
            "content": json.loads(row["payload_json"]),
        }
    except Exception:
        conn.execute("ROLLBACK")
        raise


def reconcile_and_mark(conn, rar: RarClient, token: str, submission_id: int, content: dict) -> bool:
    """Cauta la RAR o prezentare deja inregistrata pentru acest continut. Marcheaza 'sent' daca exista.

    Intoarce True daca a gasit (si a marcat sent), False altfel.
    """
    finalizate = rar.get_finalizate(token)
    found_id = match_finalizata(
        finalizate,
        vin=content.get("vin", ""),
        data_prestatie=content.get("data_prestatie", ""),
        odometru_final=content.get("odometru_final"),
    )
    if found_id is not None:
        mark(conn, submission_id, "sent", rar_status_code=200, id_prezentare=found_id,
             rar_error="reconciliat (raspuns pierdut)")
        _wlog(conn, "submission_reconciliat",
              f"submission {submission_id} -> sent prin reconciliere (idPrezentare={found_id})",
              context={"submission_id": submission_id, "id_prezentare": found_id})
        return True
    return False


def process_one(conn, settings: Settings, rar: RarClient, token: str, claimed: dict) -> str:
    """Trimite o prezentare claimed. Intoarce starea finala (pentru teste/loguri)."""
    sid = claimed["id"]
    account_id = claimed.get("account_id")
    content = claimed["content"]
    payload = build_rar_payload(content)
    try:
        data = rar.post_prezentare(token, payload)
        mark(conn, sid, "sent", rar_status_code=200, id_prezentare=data.get("id"))
        _wlog(conn, "submission_sent", f"submission {sid} -> sent (idPrezentare={data.get('id')})",
              account_id=account_id, context={"submission_id": sid, "id_prezentare": data.get("id")})
        return "sent"
    except RarError as exc:
        if exc.status_code == 400:
            if exc.field_errors:
                enriched = [
                    errors.eroare("RAR_VALIDARE", field=fe.get("field"), cauza=fe.get("message"))
                    for fe in exc.field_errors
                ]
            else:
                enriched = [errors.eroare("RAR_VALIDARE", cauza=str(exc))]
            detail = json.dumps(enriched, ensure_ascii=False)
            mark(conn, sid, "needs_data", rar_status_code=400, rar_error=detail)
            _wlog(conn, "submission_needs_data", f"submission {sid} -> needs_data (RAR 400)",
                  nivel="WARNING", account_id=account_id, cod="RAR_VALIDARE",
                  context={"submission_id": sid})
            return "needs_data"
        if exc.status_code == 500 and exc.rar_message:
            # RAR a raspuns DEFINITIV cu o eroare de procesare (ex. ORA-12899). NU e o
            # pierdere de raspuns ambigua -> NU reconcilia (recordul, daca exista la RAR,
            # e PARTIAL/rupt si nu trebuie marcat fals 'sent') si NU reincerca (acelasi
            # input va esua iar). Marcam 'error' cu mesajul real RAR.
            detail = json.dumps(errors.eroare("RAR_EROARE_SERVER", cauza=exc.rar_message), ensure_ascii=False)
            mark(conn, sid, "error", rar_status_code=500, rar_error=detail)
            _wlog(conn, "submission_error", f"submission {sid} -> error (RAR 500): {exc.rar_message}",
                  nivel="ERROR", account_id=account_id, cod="RAR_EROARE_SERVER",
                  context={"submission_id": sid, "http": 500})
            return "error"
        if _is_transient(exc):
            return _handle_transient(conn, settings, rar, token, sid, content, str(exc))
        # 4xx nerecuperabil (nu 400/401/408/429) -> error.
        mark(conn, sid, "error", rar_status_code=exc.status_code, rar_error=str(exc))
        _wlog(conn, "submission_error", f"submission {sid} -> error: {exc}",
              nivel="ERROR", account_id=account_id,
              context={"submission_id": sid, "http": exc.status_code})
        return "error"
    except (httpx.TimeoutException, httpx.TransportError) as exc:
        return _handle_transient(conn, settings, rar, token, sid, content, f"retea: {exc}")


def _handle_transient(conn, settings: Settings, rar: RarClient, token: str, sid: int, content: dict, reason: str) -> str:
    """Eroare ambigua: POST-ul poate sa fi ajuns la RAR. Reconciliaza intai; altfel retry/backoff."""
    try:
        if reconcile_and_mark(conn, rar, token, sid, content):
            return "sent"
    except RarError as exc:
        # Reconcilierea insasi a esuat -> nu putem confirma; tratam ca tranzitoriu si retry.
        reason = f"{reason}; reconciliere esuata: {exc}"
    requeue_with_backoff(conn, settings, sid, reason=reason)
    return "requeued"


def recover_orphans(conn, settings: Settings, rar: RarClient, token: str,
                    account_id: int | None = None, rar_env: str | None = None) -> int:
    """Randuri 'sending' mai vechi de lease (worker mort mid-POST). Reconciliaza; altfel requeue.

    `account_id` filtreaza la orfanii unui cont (login-ul e per-cont); None = toti.
    `rar_env` filtreaza la orfanii unui mediu (1b/E6): orfanii prod contra endpoint prod,
    NU contra test — altfel no-match -> re-POST prod = DUPLICAT real ireversibil.
    """
    # Cutoff calculat SQLite-side, in ACELASI format ca sending_since (scris cu
    # datetime('now') in claim_one -> 'YYYY-MM-DD HH:MM:SS', cu spatiu). Daca am
    # compara cu _iso() (format ISO cu 'T'), spatiul (0x20) < 'T' (0x54) ar face
    # orice rand 'sending' sa para mereu <= cutoff -> lease-ul de 120s ignorat,
    # iar fiecare rand proaspat revendicat ar fi tratat instant ca orfan.
    lease = f"-{int(settings.worker_sending_lease_s)} seconds"
    base_sql = (
        "SELECT id, payload_json FROM submissions WHERE status='sending' "
        "AND (sending_since IS NULL OR sending_since <= datetime('now', ?))"
    )
    params: list = [lease]
    if account_id is not None:
        base_sql += " AND account_id=?"
        params.append(account_id)
    if rar_env is not None:
        base_sql += " AND rar_env=?"
        params.append(rar_env)
    orphans = conn.execute(base_sql, params).fetchall()
    recovered = 0
    for row in orphans:
        sid = row["id"]
        content = json.loads(row["payload_json"])
        try:
            if reconcile_and_mark(conn, rar, token, sid, content):
                recovered += 1
            else:
                # Nu e la RAR -> sigur de re-trimis.
                requeue_with_backoff(conn, settings, sid, reason="orfan recuperat (nu exista la RAR)")
                recovered += 1
        except RarError as exc:
            print(f"[worker] orfan {sid}: reconciliere esuata ({exc}); il las pentru data viitoare", flush=True)
    return recovered


def _queue_depth(conn) -> int:
    row = conn.execute("SELECT COUNT(*) AS n FROM submissions WHERE status='queued'").fetchone()
    return int(row["n"]) if row else 0


def _refresh_nomenclator(conn, rar: RarClient, token: str) -> None:
    """Ia nomenclatorul live din RAR si il upsert-eaza local. Erorile NU opresc worker-ul."""
    try:
        items = rar.get_nomenclator(token)
        n = upsert_nomenclator(conn, items)
        print(f"[worker] nomenclator refresh: {n} coduri", flush=True)
    except Exception as exc:  # noqa: BLE001 — refresh best-effort, nu blocheaza trimiterea
        print(f"[worker] nomenclator refresh esuat (continui): {exc}", flush=True)


class AccountSessions:
    """Sesiuni RAR per (cont, env): login lazy cu creds din submission + cache JWT (30h).

    Cheia = (account_id, rar_env): test si prod sunt sisteme RAR separate cu JWT separate.
    La primul login reusit pentru (cont, env) sterge creds-urile efemere scoped pe acel env.
    Pe 401 mid-sesiune se invalideaza sesiunea -> re-login la urmatorul submission cu creds.
    """

    def __init__(self, settings: Settings):
        self.settings = settings
        self._sessions: dict[tuple[int, str], tuple[RarClient, str]] = {}

    def get_token(self, conn, account_id: int, creds: dict | None, rar_env: str = "test") -> str | None:
        """Token valid pentru (cont, env). Login daca lipseste din cache si avem creds; altfel None."""
        key = (account_id, rar_env)
        sess = self._sessions.get(key)
        if sess is not None:
            return sess[1]
        if not creds or not creds.get("email") or not creds.get("password"):
            return None
        rar = RarClient(self.settings, base_url=base_url_pentru_env(self.settings, rar_env))
        try:
            token = rar.login(creds["email"], creds["password"])
        except RarAuthError as exc:
            rar.close()
            # Login esuat (401) — FARA email/parola (doar codul HTTP + contul).
            log_event("rar_login", nivel="WARNING", account_id=account_id,
                      cod="RAR_CREDS_INVALIDE",
                      mesaj=f"login RAR esuat (cont {account_id}, env={rar_env}): {exc.status_code or 401}",
                      context={"rezultat": "esuat", "http": exc.status_code or 401},
                      conn=conn, sursa="worker")
            raise
        except Exception:
            rar.close()
            raise
        self._sessions[key] = (rar, token)
        write_heartbeat(conn, rar_login_ok=True, detail=f"login RAR ok (cont {account_id}, env={rar_env})")
        # Login reusit (fara email/parola in clar — context curat).
        log_event("rar_login", account_id=account_id,
                  mesaj=f"login RAR ok (cont {account_id}, env={rar_env})",
                  context={"rezultat": "ok", "http": 200}, conn=conn, sursa="worker")
        # Creds efemere pe submissions nu mai sunt necesare: JWT acopera retry-urile -> sterge.
        # GATE PURJARE: sterge DOAR submissions.rar_creds_enc (WHERE account_id=? AND rar_env=?),
        # NU accounts.rar_creds_{env}_enc. Scoped pe env: login test NU sterge creds prod (E1/1a).
        conn.execute(
            "UPDATE submissions SET rar_creds_enc=NULL "
            "WHERE account_id=? AND rar_env=? AND rar_creds_enc IS NOT NULL",
            (account_id, rar_env),
        )
        # Nomenclator live (autoritativ) la fiecare login proaspat.
        # Nota (1e): nomenclatorul e presupus identic intre medii (aceleasi 18 coduri RAR);
        # daca diverge in viitor, scoparea per-env a tabelei nomenclator_rar e out of scope acum.
        _refresh_nomenclator(conn, rar, token)
        return token

    def rar(self, account_id: int, rar_env: str = "test") -> RarClient:
        return self._sessions[(account_id, rar_env)][0]

    def active(self) -> list[tuple[int, str, RarClient, str]]:
        """Sesiunile active: lista de (account_id, rar_env, RarClient, token)."""
        return [(acct, env, rar, tok) for (acct, env), (rar, tok) in self._sessions.items()]

    def invalidate(self, account_id: int, rar_env: str | None = None) -> None:
        """Invalideaza sesiunea (cont, env). rar_env=None invalideaza TOATE sesiunile contului."""
        if rar_env is not None:
            sess = self._sessions.pop((account_id, rar_env), None)
            if sess is not None:
                sess[0].close()
        else:
            # Invalideaza toate mediile pentru acest cont.
            keys_to_remove = [k for k in self._sessions if k[0] == account_id]
            for k in keys_to_remove:
                sess = self._sessions.pop(k)
                sess[0].close()

    def close_all(self) -> None:
        for rar, _tok in self._sessions.values():
            rar.close()
        self._sessions.clear()


def _creds_for(claimed: dict, settings: Settings) -> dict | None:
    """Creds pentru un submission: decripteaza enc-ul lipit; altfel creds <test> (dev)."""
    creds = decrypt_creds(claimed.get("creds_enc"))
    if creds:
        return creds
    if settings.worker_use_test_creds:
        return load_test_credentials()
    return None


def _creds_from_account(conn, account_id: int, rar_env: str = "test") -> dict | None:
    """Creds RAR durabile per-cont din slotul per-env (US-013 — coloana legacy dropata).

    Canal web: creds in accounts.rar_creds_{rar_env}_enc (per-env, singurul slot valid).
    """
    env_slot = f"rar_creds_{rar_env}_enc"
    row = conn.execute(
        f"SELECT {env_slot} FROM accounts WHERE id=?", (account_id,)
    ).fetchone()
    if not row:
        return None
    enc = row[env_slot]
    return decrypt_creds(enc) if enc else None


def _keepalive_target(conn, settings: Settings) -> tuple[int | None, dict | None]:
    """Un cont cu creds durabile pentru login-ul de proba (sau creds <test> in dev).

    Ancora M2: cauta in slotul per-env al mediului `settings.rar_env` (ancora globala).
    US-013: coloana legacy accounts.rar_creds_enc a fost dropata — se foloseste EXCLUSIV
    slotul per-env. Sare conturile ale caror creds NU se decripteaza sub cheia curenta
    (in dev `start.sh both` genereaza o cheie efemera noua la fiecare pornire).
    """
    env_slot = f"rar_creds_{settings.rar_env}_enc"
    rows = conn.execute(
        f"SELECT id, {env_slot} FROM accounts ORDER BY id"
    ).fetchall()
    for row in rows:
        enc = row[env_slot]
        if not enc:
            continue
        creds = decrypt_creds(enc)
        if creds and creds.get("email") and creds.get("password"):
            return row["id"], creds
    if settings.worker_use_test_creds:
        return DEFAULT_ACCOUNT_ID, load_test_credentials()
    return None, None


def _maybe_keepalive(conn, settings: Settings, sessions: "AccountSessions", state: dict) -> None:
    """Login de proba periodic cand coada e goala — verifica reachability RAR si
    pastreaza last_rar_login_ok proaspat ca dashboard-ul sa nu afiseze fals
    'RAR inaccesibil' doar din lipsa de trafic.

    Sondeaza la cel mult o data pe interval (si pe succes, si pe esec): pe succes
    heartbeat-ul se reimprospateaza singur; pe esec real (RAR jos) last_rar_login_ok
    ramane vechi -> dashboard-ul degradeaza corect. Forteaza login real (invalideaza
    sesiunea cache-uita) ca proba sa fie autentica, nu un token vechi din cache.
    """
    interval = settings.worker_rar_keepalive_interval_s
    if interval <= 0:
        return
    hb = read_heartbeat(conn)
    last = hb["last_rar_login_ok"] if hb else None
    if last:
        try:
            age = (datetime.now(timezone.utc) - datetime.fromisoformat(last)).total_seconds()
            if age < interval:
                return  # login inca proaspat — nimic de facut
        except (ValueError, TypeError):
            pass
    now_ts = time.time()
    if now_ts - state["last_attempt"] < interval:
        return  # deja am incercat recent (nu hartui RAR daca e jos)
    state["last_attempt"] = now_ts

    account_id, creds = _keepalive_target(conn, settings)
    if account_id is None or not creds:
        return  # niciun cont cu creds durabile — nimic de sondat
    sessions.invalidate(account_id, settings.rar_env)  # forteaza login real pt env-ul global
    try:
        sessions.get_token(conn, account_id, creds, settings.rar_env)  # reimprospateaza last_rar_login_ok la succes
    except RarAuthError:
        pass  # creds invalide — deja logat in get_token (WARNING)
    except Exception as exc:
        # RAR indisponibil: last_rar_login_ok ramane vechi (corect). Nu propaga.
        log_event("rar_keepalive", nivel="WARNING", account_id=account_id,
                  mesaj=f"keepalive RAR esuat (cont {account_id}): {type(exc).__name__}",
                  context={"rezultat": "esuat"}, conn=conn, sursa="worker")


def run() -> int:
    signal.signal(signal.SIGTERM, _stop)
    signal.signal(signal.SIGINT, _stop)

    settings = get_settings()
    set_source("worker")  # evenimentele worker-ului au sursa=worker (fisier app-worker.log)
    init_db()
    conn = get_connection()
    print(f"[worker] pornit (send_enabled={settings.worker_send_enabled}, env={settings.rar_env})", flush=True)

    sessions = AccountSessions(settings)
    _last_purge_time: float = 0.0
    _keepalive_state = {"last_attempt": 0.0}

    while _running:
        try:
            write_heartbeat(conn, detail=f"poll (queue={_queue_depth(conn)})")

            # Purjare periodica (odata pe ora) — NU mai frecvent.
            now_ts = time.time()
            if now_ts - _last_purge_time >= _PURGE_INTERVAL_S:
                # Expira intai randurile tinute imbatranite (US-008), apoi purjeaza:
                # expire_held seteaza purge_after=now-1s -> randul e purjabil IMEDIAT in
                # acelasi ciclu (purge_expired foloseste comparatie stricta purge_after<now).
                expired = expire_held(conn, settings)
                if expired:
                    _wlog(conn, "held_expirat",
                          f"{expired} randuri tinute expirate (>{settings.held_retention_days}z) -> error/TINUT_EXPIRAT",
                          nivel="WARNING", context={"expirate": expired})
                stats = purge_expired(conn)
                if stats["submissions_purged"] or stats["batches_purged"] or stats["events_purged"]:
                    print(
                        f"[worker] purjare: {stats['submissions_purged']} submissions, "
                        f"{stats['batches_purged']} batches, {stats['events_purged']} evenimente sterse",
                        flush=True,
                    )
                _last_purge_time = now_ts

            if not settings.worker_send_enabled:
                time.sleep(settings.worker_poll_interval_s)
                continue

            claimed = claim_one(conn)
            if claimed is None:
                # Nimic de trimis: recupereaza orfanii conturilor deja logate.
                for acct, env, rar, tok in sessions.active():
                    recover_orphans(conn, settings, rar, tok, account_id=acct, rar_env=env)
                # Login de proba periodic ca dashboard-ul sa nu afiseze fals
                # "RAR inaccesibil" din lipsa de trafic (vezi _maybe_keepalive).
                _maybe_keepalive(conn, settings, sessions, _keepalive_state)
                time.sleep(settings.worker_poll_interval_s)
                continue

            sid = claimed["id"]
            account_id = claimed["account_id"]
            # Mediul tinta al trimiterii (per-submission, cu fallback la ancora globala).
            rar_env = claimed.get("rar_env") or settings.rar_env
            # Randul poarta creds proaspete (rar_creds_enc != NULL) — fie prima trimitere
            # a contului, fie o REACTIVARE dupa creds gresite. Invalidam sesiunea RAR
            # cache-uita (per env) ca un JWT vechi (30h) din parola GRESITA sa nu trimita
            # cu ea, ignorand corectia. Re-login imediat cu creds-urile noi.
            if claimed.get("creds_enc"):
                sessions.invalidate(account_id, rar_env)
            # Incearca creds din submission (canal API efemer), cu fallback la
            # accounts.rar_creds_{rar_env}_enc (canal web durabil, per-env).
            creds = _creds_for(claimed, settings) or _creds_from_account(conn, account_id, rar_env)

            try:
                token = sessions.get_token(conn, account_id, creds, rar_env)
            except RarAuthError as exc:
                # Creds gresite (login 401): NU se face retry.
                mark(conn, sid, "error", rar_status_code=401,
                     rar_error=json.dumps(errors.eroare("RAR_CREDS_INVALIDE", cauza="credentiale RAR invalide"), ensure_ascii=False))
                # rar_login esuat e deja logat in get_token; aici doar tranzitia submission-ului.
                _wlog(conn, "submission_error", f"submission {sid} (cont {account_id}) -> error: creds RAR invalide",
                      nivel="ERROR", account_id=account_id, cod="RAR_CREDS_INVALIDE",
                      context={"submission_id": sid, "http": 401})
                continue

            if token is None:
                # Fara creds disponibile (token pierdut la restart + creds sterse).
                # Re-pune in coada cu backoff; ROAAUTO re-trimite creds proaspete.
                requeue_with_backoff(conn, settings, sid, reason="creds RAR indisponibile (astept re-trimitere)")
                continue

            rar = sessions.rar(account_id, rar_env)
            # Recupereaza orfanii contului + env-ului inainte de trimitere (acelasi token).
            recover_orphans(conn, settings, rar, token, account_id=account_id, rar_env=rar_env)
            # Guard: recover_orphans putea atinge chiar randul tocmai revendicat
            # (reconciliat 'sent' sau requeue 'queued'). Daca nu mai e 'sending',
            # NU mai face POST — altfel s-ar crea un duplicat la RAR.
            still_sending = conn.execute(
                "SELECT 1 FROM submissions WHERE id=? AND status='sending'", (sid,)
            ).fetchone()
            if still_sending is None:
                continue
            try:
                process_one(conn, settings, rar, token, claimed)
            except RarAuthError as exc:
                # Token expirat mid-sesiune: invalideaza sesiunea (per env), re-pune randul.
                print(f"[worker] cont {account_id} env={rar_env} token expirat: {exc}; re-login data viitoare", flush=True)
                sessions.invalidate(account_id, rar_env)
                requeue_with_backoff(conn, settings, sid, reason="token RAR expirat")

        except Exception as exc:  # noqa: BLE001 — loop top-level: o eroare punctuala nu opreste worker-ul
            print(f"[worker] eroare neasteptata: {exc}", flush=True)
            time.sleep(settings.worker_poll_interval_s)

    sessions.close_all()
    conn.close()
    print("[worker] oprit curat", flush=True)
    return 0


if __name__ == "__main__":
    sys.exit(run())
