from __future__ import annotations

import base64
import gzip
import json
import os
import socket
import ssl
import tempfile
import threading
import uuid
from concurrent.futures import Future, ThreadPoolExecutor
import urllib.error
import urllib.request
from pathlib import Path

from .certificates import validate_pfx
from .db import MysqlCli, sql_quote
from .nfse_xml import parse_nfse_event_xml, parse_nfse_xml
from .security import decrypt_certificate_value
from .settings import MAX_SYNC_MAX_WORKERS, get_sync_max_workers
from .tenants import ensure_tenant_tables


SYNC_EXECUTOR = ThreadPoolExecutor(max_workers=MAX_SYNC_MAX_WORKERS)
_SYNC_FUTURES: set[Future[None]] = set()
_SYNC_FUTURES_LOCK = threading.Lock()
_SYNC_MAINTENANCE_LOCK = threading.Lock()
_SYNC_MAINTENANCE_STARTED = False
_ACTIVE_SYNC_WORKERS: set[str] = set()
_ACTIVE_SYNC_WORKERS_LOCK = threading.Lock()


DOCUMENT_FIELD_COLS = [
    "numero_nfse",
    "numero_dfse",
    "serie_dps",
    "numero_dps",
    "municipio_emissor_codigo",
    "municipio_incidencia_codigo",
    "municipio_prestacao_codigo",
    "prestador_documento",
    "prestador_nome",
    "tomador_documento",
    "tomador_nome",
    "intermediario_documento",
    "intermediario_nome",
    "data_emissao",
    "data_competencia",
    "data_processamento",
    "status_documento",
    "codigo_tributacao_nacional",
    "descricao_tributacao_nacional",
    "descricao_servico",
    "codigo_nbs",
    "descricao_nbs",
    "valor_servico",
    "valor_base_calculo",
    "aliquota",
    "valor_issqn",
    "valor_total_retido",
    "valor_liquido",
]


def _sql_quote_any(value: object) -> str:
    if value is None:
        return "NULL"
    return sql_quote(str(value))


def _sync_stale_lock_minutes() -> int:
    return int(os.environ.get("NFSE_SYNC_STALE_LOCK_MINUTES", "120"))


def _sync_progress_every() -> int:
    return max(1, int(os.environ.get("NFSE_SYNC_PROGRESS_EVERY", "100")))


def _sync_batch_size() -> int:
    return min(1000, max(1, int(os.environ.get("NFSE_SYNC_BATCH_SIZE", "200"))))


def _sync_batch_max_bytes() -> int:
    return min(8 * 1024 * 1024, max(64 * 1024, int(os.environ.get("NFSE_SYNC_BATCH_MAX_BYTES", str(2 * 1024 * 1024)))))


def _update_sync_progress(db: MysqlCli, municipio_id: int, last_nsu: int, max_nsu: int) -> None:
    db.execute(
        f"""
UPDATE nfse_sync_state
SET last_nsu = {last_nsu},
    max_nsu = {max_nsu},
    locked_at = NOW()
WHERE municipio_id = {municipio_id};

UPDATE nfse_sync_jobs
SET locked_at = NOW()
WHERE municipio_id = {municipio_id}
  AND status = 'running';
"""
    )


def _max_stored_nsu(db: MysqlCli, tabela_documentos: str, ambiente: str) -> int:
    rows = db.query_rows(
        f"SELECT COALESCE(MAX(nsu), 0) FROM {tabela_documentos} WHERE ambiente = {sql_quote(ambiente)};"
    )
    if not rows:
        return 0
    return int(rows[0][0] or "0")


def _write_pfx_to_pem(pfx_data: bytes, password: str, tmpdir: Path) -> tuple[Path, Path]:
    pfx_path = tmpdir / "certificado.pfx"
    cert_path = tmpdir / "cert.pem"
    key_path = tmpdir / "key.pem"
    pfx_path.write_bytes(pfx_data)
    import subprocess

    commands = [
        [
            "openssl",
            "pkcs12",
            "-in",
            str(pfx_path),
            "-passin",
            f"pass:{password}",
            "-nokeys",
            "-clcerts",
            "-out",
            str(cert_path),
        ],
        [
            "openssl",
            "pkcs12",
            "-in",
            str(pfx_path),
            "-passin",
            f"pass:{password}",
            "-nocerts",
            "-nodes",
            "-out",
            str(key_path),
        ],
    ]
    for command in commands:
        proc = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        if proc.returncode != 0:
            raise RuntimeError(proc.stderr.strip() or "Falha ao preparar certificado.")
    return cert_path, key_path


def _new_sync_worker_id() -> str:
    host = socket.gethostname().replace(":", "_")
    return f"{host}:{os.getpid()}:{threading.get_ident()}:{uuid.uuid4().hex[:8]}"


def _process_is_alive(pid: int) -> bool:
    if pid <= 0:
        return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True


def _worker_lock_is_abandoned(locked_by: str, stale: bool) -> bool:
    if stale or not locked_by:
        return True
    parts = locked_by.split(":", 3)
    if len(parts) != 4 or parts[0] != socket.gethostname().replace(":", "_"):
        return False
    try:
        pid = int(parts[1])
    except ValueError:
        return False
    if pid == os.getpid():
        with _ACTIVE_SYNC_WORKERS_LOCK:
            return locked_by not in _ACTIVE_SYNC_WORKERS
    return not _process_is_alive(pid)


def recover_abandoned_sync_jobs(db: MysqlCli) -> int:
    stale_minutes = max(1, _sync_stale_lock_minutes())
    with db.transaction():
        rows = db.query_rows(
            f"""
SELECT
  id,
  municipio_id,
  COALESCE(locked_by, ''),
  CASE
    WHEN locked_at IS NULL
      OR locked_at < DATE_SUB(NOW(), INTERVAL {stale_minutes} MINUTE)
    THEN 1 ELSE 0
  END AS stale
FROM nfse_sync_jobs
WHERE status = 'running'
FOR UPDATE;
"""
        )
        abandoned = [
            row
            for row in rows
            if _worker_lock_is_abandoned(row[2], row[3] == "1")
        ]
        if not abandoned:
            return 0

        job_ids = ", ".join(str(int(row[0])) for row in abandoned)
        municipio_ids = ", ".join(sorted({str(int(row[1])) for row in abandoned}, key=int))
        db.execute(
            f"""
UPDATE nfse_sync_jobs
SET status = 'pending',
    available_at = NOW(),
    locked_at = NULL,
    locked_by = NULL,
    finished_at = NULL,
    last_error_message = 'Job recuperado apos interrupcao do worker'
WHERE id IN ({job_ids})
  AND status = 'running';

UPDATE nfse_sync_state
SET status = 'pending',
    locked_at = NULL,
    locked_by = NULL,
    last_error_message = NULL
WHERE municipio_id IN ({municipio_ids})
  AND status IN ('pending', 'running');
"""
        )
    return len(abandoned)


def _run_next_sync_job(db: MysqlCli) -> bool:
    worker = _new_sync_worker_id()
    with _ACTIVE_SYNC_WORKERS_LOCK:
        _ACTIVE_SYNC_WORKERS.add(worker)
    try:
        return _run_next_sync_job_as_worker(db, worker)
    finally:
        with _ACTIVE_SYNC_WORKERS_LOCK:
            _ACTIVE_SYNC_WORKERS.discard(worker)


def _run_next_sync_job_as_worker(db: MysqlCli, worker: str) -> bool:
    db.execute(f"""
UPDATE nfse_sync_jobs
SET status = 'running', attempts = attempts + 1, locked_at = NOW(), locked_by = {sql_quote(worker)}
WHERE id = (
  SELECT id FROM (
    SELECT id FROM nfse_sync_jobs
    WHERE status = 'pending' AND available_at <= NOW()
    ORDER BY available_at, id LIMIT 1
  ) AS candidate
)
  AND status = 'pending'
  AND available_at <= NOW();
""")
    rows = db.query_rows(
        f"""
SELECT id, municipio_id
FROM nfse_sync_jobs
WHERE status = 'running'
  AND locked_by = {sql_quote(worker)}
LIMIT 1;
"""
    )
    if not rows:
        return False

    job_id, municipio_id = rows[0]
    run_sync_full(db.database, int(municipio_id), worker)
    state = db.query_rows(
        f"""
SELECT status, COALESCE(last_error_message, '')
FROM nfse_sync_state
WHERE municipio_id = {int(municipio_id)}
LIMIT 1;
"""
    )
    ok = bool(state and state[0][0] == "ready")
    final_status = "done" if ok else "error"
    error_message = "" if ok else (state[0][1] if state else "sync sem estado")
    db.execute(
        f"UPDATE nfse_sync_jobs SET status={sql_quote(final_status)}, finished_at=NOW(), "
        f"locked_at=NULL, locked_by=NULL, last_error_message={sql_quote(error_message)} "
        f"WHERE id={int(job_id)} AND locked_by={sql_quote(worker)};"
    )
    return True


def _drain_sync_jobs(database: str) -> None:
    db = MysqlCli(database=database)
    try:
        _run_next_sync_job(db)
    finally:
        db.close()


def _ensure_sync_workers(
    database: str,
    *,
    max_new_workers: int,
    worker_limit: int | None = None,
) -> int:
    if worker_limit is None:
        config_db = MysqlCli(database=database)
        try:
            worker_limit = get_sync_max_workers(config_db)
        finally:
            config_db.close()
    new_futures: list[Future[None]] = []
    with _SYNC_FUTURES_LOCK:
        _SYNC_FUTURES.difference_update(future for future in _SYNC_FUTURES if future.done())
        slots = max(0, worker_limit - len(_SYNC_FUTURES))
        to_start = min(max(0, max_new_workers), slots)
        for _ in range(to_start):
            future = SYNC_EXECUTOR.submit(_drain_sync_jobs, database)
            _SYNC_FUTURES.add(future)
            new_futures.append(future)
    for future in new_futures:
        future.add_done_callback(
            lambda completed, target_database=database: _continue_sync_queue(
                completed,
                target_database,
            )
        )
    return to_start


def _pending_sync_jobs(db: MysqlCli) -> int:
    rows = db.query_rows(
        """
SELECT COUNT(*)
FROM nfse_sync_jobs
WHERE status = 'pending'
  AND available_at <= NOW();
"""
    )
    return int(rows[0][0]) if rows else 0


def _continue_sync_queue(future: Future[None], database: str) -> None:
    db = MysqlCli(database=database)
    try:
        error = future.exception()
        if error is not None:
            print(f"Worker da fila encerrou com erro: {error}", flush=True)
        recover_abandoned_sync_jobs(db)
        pending = _pending_sync_jobs(db)
        worker_limit = get_sync_max_workers(db)
        if pending:
            _ensure_sync_workers(
                database,
                max_new_workers=min(pending, worker_limit),
                worker_limit=worker_limit,
            )
    except Exception as exc:
        print(f"Falha ao continuar a fila de sincronizacao: {exc}", flush=True)
    finally:
        db.close()


def _sync_queue_maintenance_loop(database: str) -> None:
    db = MysqlCli(database=database)
    poll_seconds = min(
        300,
        max(5, int(os.environ.get("NFSE_SYNC_QUEUE_POLL_SECONDS", "15"))),
    )
    stop = threading.Event()
    while True:
        try:
            recover_abandoned_sync_jobs(db)
            pending = _pending_sync_jobs(db)
            worker_limit = get_sync_max_workers(db)
            if pending:
                _ensure_sync_workers(
                    database,
                    max_new_workers=min(pending, worker_limit),
                    worker_limit=worker_limit,
                )
        except Exception as exc:
            print(f"Falha na manutencao da fila de sincronizacao: {exc}", flush=True)
        stop.wait(poll_seconds)


def start_sync_queue_dispatcher(database: str) -> bool:
    global _SYNC_MAINTENANCE_STARTED
    with _SYNC_MAINTENANCE_LOCK:
        if _SYNC_MAINTENANCE_STARTED:
            return False
        _SYNC_MAINTENANCE_STARTED = True
        thread = threading.Thread(
            target=_sync_queue_maintenance_loop,
            args=(database,),
            name="nfse-sync-queue-maintenance",
            daemon=True,
        )
        thread.start()
        return True


def enqueue_sync_jobs(db: MysqlCli, jobs: list[tuple[int, str, str]]) -> int:
    unique_jobs = sorted(
        {(int(municipio_id), ambiente, perfil) for municipio_id, ambiente, perfil in jobs},
        key=lambda item: item[0],
    )
    if not unique_jobs:
        return 0

    values = ", ".join(
        f"({municipio_id}, {sql_quote(ambiente)}, {sql_quote(perfil)}, 'pending', NOW())"
        for municipio_id, ambiente, perfil in unique_jobs
    )
    municipio_ids = ", ".join(str(municipio_id) for municipio_id, _ambiente, _perfil in unique_jobs)
    with db.transaction():
        db.execute(
            f"""
INSERT INTO nfse_sync_jobs (municipio_id, ambiente, perfil, status, available_at)
VALUES {values}
ON DUPLICATE KEY UPDATE
  available_at = IF(status IN ('done','error'), NOW(), available_at),
  status = IF(status IN ('done','error'), 'pending', status),
  last_error_message = IF(status = 'running', last_error_message, NULL);

UPDATE nfse_sync_state s
JOIN nfse_sync_jobs j
  ON j.municipio_id = s.municipio_id
 AND j.ambiente = s.ambiente
 AND j.perfil = s.perfil
SET s.status = 'pending',
    s.locked_at = NULL,
    s.locked_by = NULL
WHERE s.municipio_id IN ({municipio_ids})
  AND j.status = 'pending';
"""
        )
    return len(unique_jobs)


def trigger_sync(db: MysqlCli, municipio_id: int) -> None:
    rows = db.query_rows(
        f"SELECT ambiente, perfil FROM nfse_sync_state WHERE municipio_id={municipio_id} LIMIT 1;"
    )
    if not rows:
        raise ValueError("Estado de sincronizacao nao encontrado.")
    ambiente, perfil = rows[0]
    enqueue_sync_jobs(db, [(municipio_id, ambiente, perfil)])
    _ensure_sync_workers(db.database, max_new_workers=1)

def _fetch_dfe_lote(base_url: str, next_nsu: int, context: ssl.SSLContext) -> tuple[int, dict[str, object]]:
    req = urllib.request.Request(
        f"{base_url}/DFe/{next_nsu}?tipoNSU=DISTRIBUICAO&lote=true",
        headers={"Accept": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, context=context, timeout=30) as response:
            body = response.read().decode("utf-8")
            http_status = response.status
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8")
        http_status = exc.code
    return http_status, json.loads(body)


def _prepare_nfse_document(
    *,
    ambiente: str,
    nsu: int,
    item: dict[str, object],
    xml_text: str,
) -> dict[str, object]:
    normalized: dict[str, object] = {"fields": {}, "parties": []}
    if xml_text:
        normalized = parse_nfse_xml(xml_text)
    fields = normalized["fields"]  # type: ignore[index]
    parties = normalized["parties"]  # type: ignore[index]

    insert_cols = [
        "ambiente",
        "nsu",
        "chave_acesso",
        "tipo_documento",
        *DOCUMENT_FIELD_COLS,
        "xml_bruto",
        "dados_principais",
        "recebido_at",
    ]
    access_key = str(item.get("ChaveAcesso") or "")
    insert_values = [
        sql_quote(ambiente),
        str(nsu),
        sql_quote(access_key or None),
        sql_quote(str(item.get("TipoDocumento") or "")),
        *[_sql_quote_any(fields.get(col)) for col in DOCUMENT_FIELD_COLS],  # type: ignore[union-attr]
        sql_quote(xml_text),
        sql_quote(json.dumps(item, ensure_ascii=False)),
        "NOW()",
    ]
    return {
        "nsu": nsu,
        "chave_acesso": access_key,
        "insert_cols": insert_cols,
        "insert_values": insert_values,
        "parties": parties,
        "sql_bytes": sum(len(str(value).encode("utf-8")) for value in insert_values),
    }


def _store_nfse_documents_batch(
    db: MysqlCli,
    *,
    tabela_documentos: str,
    tabela_partes: str,
    ambiente: str,
    documents: list[dict[str, object]],
) -> None:
    if not documents:
        return

    insert_cols = documents[0]["insert_cols"]
    document_values = [
        "(" + ", ".join(document["insert_values"]) + ")"  # type: ignore[arg-type]
        for document in documents
    ]
    update_cols = [*DOCUMENT_FIELD_COLS, "xml_bruto", "dados_principais"]
    db.execute(
        f"""
INSERT INTO {tabela_documentos} ({", ".join(insert_cols)})
VALUES {", ".join(document_values)}
ON DUPLICATE KEY UPDATE
  {", ".join(f"{col} = VALUES({col})" for col in update_cols)},
  updated_at = NOW();
"""
    )

    nsus = ", ".join(str(int(document["nsu"])) for document in documents)
    access_keys = [str(document["chave_acesso"]) for document in documents if document["chave_acesso"]]
    key_predicate = ""
    if access_keys:
        key_predicate = " OR chave_acesso IN (" + ", ".join(sql_quote(key) for key in access_keys) + ")"
    rows = db.query_rows(
        f"""
SELECT id, nsu, COALESCE(chave_acesso, '')
FROM {tabela_documentos}
WHERE ambiente = {sql_quote(ambiente)}
  AND (nsu IN ({nsus}){key_predicate});
"""
    )
    ids_by_nsu = {int(row[1]): int(row[0]) for row in rows}
    ids_by_key = {row[2]: int(row[0]) for row in rows if len(row) >= 3 and row[2]}

    party_values: list[str] = []
    for document in documents:
        documento_id = ids_by_nsu.get(int(document["nsu"])) or ids_by_key.get(str(document["chave_acesso"]))
        if documento_id is None:
            raise RuntimeError(f"Documento sem id apos insert em lote: NSU {document['nsu']}")
        for party in document["parties"]:  # type: ignore[union-attr]
            if not party.get("documento_federal"):
                continue
            party_values.append(
                "("
                f"{documento_id}, "
                f"{_sql_quote_any(party.get('papel'))}, "
                f"{_sql_quote_any(party.get('documento_federal'))}, "
                f"{_sql_quote_any(party.get('nome_razao'))}, "
                f"{_sql_quote_any(party.get('municipio_codigo'))}, "
                f"{_sql_quote_any(party.get('uf'))}, "
                f"{sql_quote(json.dumps(party, ensure_ascii=False))}"
                ")"
            )
    if not party_values:
        return
    db.execute(
        f"""
INSERT INTO {tabela_partes} (
  documento_id, papel, documento_federal, nome_razao, municipio_codigo, uf, dados_principais
) VALUES {", ".join(party_values)}
ON DUPLICATE KEY UPDATE
  documento_federal = VALUES(documento_federal),
  nome_razao = VALUES(nome_razao),
  municipio_codigo = VALUES(municipio_codigo),
  uf = VALUES(uf),
  dados_principais = VALUES(dados_principais),
  updated_at = NOW();
"""
    )

def _event_indicates_cancel(event_fields: dict[str, str | None]) -> bool:
    haystack = " ".join(
        str(value or "").lower()
        for value in (
            event_fields.get("tipo_evento"),
            event_fields.get("descricao_evento"),
        )
    )
    return "cancel" in haystack


def _store_nfse_event(
    db: MysqlCli,
    *,
    tabela_documentos: str,
    tabela_eventos: str,
    ambiente: str,
    nsu: int,
    item: dict[str, object],
    xml_text: str,
) -> None:
    event_fields: dict[str, str | None] = {}
    if xml_text:
        event_fields = parse_nfse_event_xml(xml_text)

    chave_acesso_nfse = (
        event_fields.get("chave_acesso_nfse")
        or str(item.get("ChaveAcessoNFSe") or "")
        or str(item.get("ChaveAcesso") or "")
        or f"SEM_CHAVE_NSU_{nsu}"
    )
    tipo_evento = (
        event_fields.get("tipo_evento")
        or str(item.get("TipoEvento") or "")
        or str(item.get("TipoDocumento") or "")
        or "EVENTO"
    )
    try:
        numero_sequencial = int(event_fields.get("numero_sequencial") or item.get("NumeroSequencial") or 1)
    except (TypeError, ValueError):
        numero_sequencial = 1

    doc_rows = db.query_rows(
        f"""
SELECT id
FROM {tabela_documentos}
WHERE chave_acesso = {sql_quote(chave_acesso_nfse)}
LIMIT 1;
"""
    )
    documento_id = int(doc_rows[0][0]) if doc_rows else None
    dados = {
        "adn": item,
        "evento": event_fields,
    }

    db.execute(
        f"""
INSERT INTO {tabela_eventos} (
  documento_id, ambiente, nsu, chave_acesso_nfse, tipo_evento, numero_sequencial,
  data_evento, xml_bruto, dados_principais, recebido_at
) VALUES (
  {str(documento_id) if documento_id is not None else 'NULL'},
  {sql_quote(ambiente)},
  {nsu},
  {sql_quote(chave_acesso_nfse)},
  {sql_quote(tipo_evento)},
  {numero_sequencial},
  {sql_quote(event_fields.get('data_evento'))},
  {sql_quote(xml_text)},
  {sql_quote(json.dumps(dados, ensure_ascii=False))},
  NOW()
)
ON DUPLICATE KEY UPDATE
  documento_id = VALUES(documento_id),
  nsu = VALUES(nsu),
  data_evento = VALUES(data_evento),
  xml_bruto = VALUES(xml_bruto),
  dados_principais = VALUES(dados_principais),
  updated_at = NOW();
"""
    )

    if documento_id is not None and _event_indicates_cancel(event_fields):
        db.execute(
            f"""
UPDATE {tabela_documentos}
SET status_documento = 'cancelada',
    updated_at = NOW()
WHERE id = {documento_id};
"""
        )


def run_sync_full(database: str, municipio_id: int, locked_by: str = "manual") -> None:
    db = MysqlCli(database=database)
    try:
        db.require_persistent_connection()
        db.execute(
            f"""
UPDATE nfse_sync_state
SET status = 'running', locked_at = NOW(), locked_by = {sql_quote(locked_by)}
WHERE municipio_id = {municipio_id};

UPDATE nfse_sync_jobs
SET locked_at = NOW()
WHERE municipio_id = {municipio_id}
  AND status = 'running';
"""
        )
        rows = db.query_rows(
            f"""
SELECT
  m.codigo_siafi,
  m.perfil,
  m.ambiente,
  m.tabela_documentos,
  m.tabela_eventos,
  m.tabela_partes,
  COALESCE(s.last_nsu, 0),
  c.certificado_criptografado,
  c.senha_criptografada,
  c.certificado_key_id,
  c.senha_key_id
FROM nfse_municipios m
JOIN nfse_sync_state s ON s.municipio_id = m.id
JOIN nfse_certificados c ON c.municipio_id = m.id AND c.ativo = 1
WHERE m.id = {municipio_id}
LIMIT 1;
"""
        )
        if not rows:
            raise RuntimeError("Município/certificado não encontrado.")
        codigo_siafi, perfil, ambiente, tabela_documentos, tabela_eventos, tabela_partes, last_nsu_s, cert_enc, senha_enc, cert_key_id, senha_key_id = rows[0]
        ensure_tenant_tables(db, codigo_siafi, ambiente, perfil)

        last_nsu = max(int(last_nsu_s or "0"), _max_stored_nsu(db, tabela_documentos, ambiente))
        next_nsu = last_nsu + 1
        pfx_data = decrypt_certificate_value(
            cert_enc.encode("ascii"),
            codigo_siafi=codigo_siafi,
            key_id=cert_key_id,
        )
        password = decrypt_certificate_value(
            senha_enc,
            codigo_siafi=codigo_siafi,
            key_id=senha_key_id,
        ).decode("utf-8")
        validate_pfx(pfx_data, password)
        base_url = "https://adn.nfse.gov.br/municipios"
        if ambiente == "producao_restrita":
            base_url = "https://adn.producaorestrita.nfse.gov.br/municipios"

        with tempfile.TemporaryDirectory() as tmp:
            cert_path, key_path = _write_pfx_to_pem(pfx_data, password, Path(tmp))
            context = ssl.create_default_context()
            context.load_cert_chain(str(cert_path), str(key_path))
            max_loops = int(os.environ.get("NFSE_SYNC_MAX_LOOPS", "500"))
            max_nsu = next_nsu - 1
            documentos = 0
            eventos = 0
            chamadas = 0
            ultimo_http_status = 0
            ultimo_status_processamento = ""
            progress_every = _sync_progress_every()
            batch_size = _sync_batch_size()
            batch_max_bytes = _sync_batch_max_bytes()
            itens_desde_progresso = 0

            for _ in range(max_loops):
                chamadas += 1
                http_status, data = _fetch_dfe_lote(base_url, next_nsu, context)
                ultimo_http_status = http_status
                ultimo_status_processamento = str(data.get("StatusProcessamento") or "")
                lote = data.get("LoteDFe") or []
                if not lote:
                    break

                itens_no_batch = 0
                document_batch: list[dict[str, object]] = []
                document_batch_bytes = 0
                db.execute("START TRANSACTION;")
                try:
                    for item in lote:
                        if not isinstance(item, dict):
                            continue
                        nsu = int(item.get("NSU") or 0)
                        max_nsu = max(max_nsu, nsu)
                        tipo = str(item.get("TipoDocumento") or "")
                        xml_text = ""
                        if item.get("ArquivoXml"):
                            xml_text = gzip.decompress(base64.b64decode(str(item["ArquivoXml"]))).decode("utf-8", errors="replace")
                        if tipo.upper() == "EVENTO":
                            _store_nfse_documents_batch(
                                db,
                                tabela_documentos=tabela_documentos,
                                tabela_partes=tabela_partes,
                                ambiente=ambiente,
                                documents=document_batch,
                            )
                            document_batch.clear()
                            document_batch_bytes = 0
                            eventos += 1
                            _store_nfse_event(
                                db,
                                tabela_documentos=tabela_documentos,
                                tabela_eventos=tabela_eventos,
                                ambiente=ambiente,
                                nsu=nsu,
                                item=item,
                                xml_text=xml_text,
                            )
                        else:
                            documentos += 1
                            prepared_document = _prepare_nfse_document(
                                ambiente=ambiente,
                                nsu=nsu,
                                item=item,
                                xml_text=xml_text,
                            )
                            prepared_bytes = int(prepared_document["sql_bytes"])
                            if document_batch and document_batch_bytes + prepared_bytes > batch_max_bytes:
                                _store_nfse_documents_batch(
                                    db,
                                    tabela_documentos=tabela_documentos,
                                    tabela_partes=tabela_partes,
                                    ambiente=ambiente,
                                    documents=document_batch,
                                )
                                document_batch.clear()
                                document_batch_bytes = 0
                            document_batch.append(prepared_document)
                            document_batch_bytes += prepared_bytes
                        if nsu > last_nsu:
                            last_nsu = nsu
                        itens_desde_progresso += 1
                        itens_no_batch += 1
                        if itens_desde_progresso >= progress_every:
                            _update_sync_progress(db, municipio_id, last_nsu, max_nsu)
                            itens_desde_progresso = 0
                        if itens_no_batch >= batch_size:
                            _store_nfse_documents_batch(
                                db,
                                tabela_documentos=tabela_documentos,
                                tabela_partes=tabela_partes,
                                ambiente=ambiente,
                                documents=document_batch,
                            )
                            document_batch.clear()
                            document_batch_bytes = 0
                            db.execute("COMMIT; START TRANSACTION;")
                            itens_no_batch = 0

                    _store_nfse_documents_batch(
                        db,
                        tabela_documentos=tabela_documentos,
                        tabela_partes=tabela_partes,
                        ambiente=ambiente,
                        documents=document_batch,
                    )
                    db.execute("COMMIT;")
                except Exception:
                    db.execute("ROLLBACK;")
                    raise

                if max_nsu < next_nsu:
                    break
                last_nsu = max_nsu
                next_nsu = max_nsu + 1
                _update_sync_progress(db, municipio_id, last_nsu, max_nsu)
                itens_desde_progresso = 0
            else:
                raise RuntimeError(f"Sincronização interrompida por limite de segurança ({max_loops} lotes).")

        with db.transaction():
            db.execute(
                f"""
UPDATE nfse_sync_state
SET status = 'ready',
    last_nsu = {last_nsu},
    max_nsu = {max_nsu if max_nsu >= 0 else 'NULL'},
    locked_at = NULL,
    locked_by = NULL,
    last_success_at = NOW(),
    last_error_message = NULL
WHERE municipio_id = {municipio_id};

INSERT INTO nfse_sync_logs (municipio_id, ambiente, nivel, evento, mensagem, contexto)
VALUES (
  {municipio_id},
  {sql_quote(ambiente)},
  'info',
  'sync.full.finalizado',
  {sql_quote('Sincronização completa finalizada')},
  {sql_quote(json.dumps({'http_status': ultimo_http_status, 'documentos': documentos, 'eventos': eventos, 'status': ultimo_status_processamento, 'chamadas': chamadas, 'last_nsu': last_nsu}, ensure_ascii=False))}
);
"""
            )
    except Exception as exc:
        with db.transaction():
            db.execute(
                f"""
UPDATE nfse_sync_state
SET status = 'error',
    locked_at = NULL,
    locked_by = NULL,
    last_error_at = NOW(),
    last_error_message = {sql_quote(str(exc))}
WHERE municipio_id = {municipio_id};

INSERT INTO nfse_sync_logs (municipio_id, nivel, evento, mensagem)
VALUES ({municipio_id}, 'error', 'sync.full.erro', {sql_quote(str(exc))});
"""
            )