"""
Élection d'un worker « leader » pour les tâches de fond.

Avec plusieurs workers uvicorn, chacun exécuterait le nettoyage RGPD, le backup
et l'organizer en double. Un verrou en base (bail de 60 s renouvelé toutes les
20 s) garantit qu'un seul worker les exécute ; si le leader tombe, un autre
reprend automatiquement.
"""
import asyncio
import logging
import os
import time
import uuid
from typing import Awaitable, Callable

from core.db import get_db

log = logging.getLogger("meeconic.leader")

_OWNER = f"{os.getpid()}-{uuid.uuid4().hex[:6]}"
_LEASE = 60.0
_RENEW = 20.0


def _try_acquire(name: str) -> bool:
    now = time.time()
    # Two single statements, each atomic in InnoDB: the INSERT wins only if nobody holds the row yet,
    # the UPDATE only if the row is ours or its lease has run out. No read-then-write window.
    with get_db() as conn:
        created = conn.execute(
            "INSERT IGNORE INTO leader_locks (name, owner, expires_at) VALUES (?,?,?)",
            (name, _OWNER, now + _LEASE)).rowcount
        if created:
            return True
        taken = conn.execute(
            "UPDATE leader_locks SET owner=?, expires_at=? WHERE name=? AND (owner=? OR expires_at<?)",
            (_OWNER, now + _LEASE, name, _OWNER, now)).rowcount
        return bool(taken)


def _release(name: str) -> None:
    try:
        with get_db() as conn:
            conn.execute("DELETE FROM leader_locks WHERE name=? AND owner=?", (name, _OWNER))
            conn.commit()
    except Exception:
        pass


def is_leader(name: str) -> bool:
    try:
        with get_db() as conn:
            row = conn.execute("SELECT owner, expires_at FROM leader_locks WHERE name=?", (name,)).fetchone()
        return bool(row and row["owner"] == _OWNER and row["expires_at"] > time.time())
    except Exception:
        return False


async def run_as_leader(name: str, factory: Callable[[], Awaitable[None]]) -> None:
    """Exécute `factory()` tant que ce worker détient le verrou `name`."""
    task: asyncio.Task | None = None
    try:
        while True:
            try:
                held = await asyncio.to_thread(_try_acquire, name)
            except Exception as e:
                log.error("Election '%s' failed: %s", name, e)
                held = False
            if held and (task is None or task.done()):
                if task is not None and task.done() and not task.cancelled() and task.exception():
                    log.error("Task %s stopped by an error: %s — restarting", name, task.exception())
                task = asyncio.create_task(factory(), name=f"bg-{name}")
                log.info("Ce worker est leader de « %s »", name)
            elif not held and task is not None and not task.done():
                task.cancel()
                log.warning("Lock '%s' lost — task stopped", name)
            await asyncio.sleep(_RENEW)
    except asyncio.CancelledError:
        if task is not None and not task.done():
            task.cancel()
            await asyncio.gather(task, return_exceptions=True)
        await asyncio.to_thread(_release, name)
        raise