"""Le planificateur : cadences pures, puis le sweep du worker (spec Imports T8).

Deux promesses de la spec sont testées ici, et elles ne sont pas la même :

1. **jamais de doublon** — deux passes du sweep sur le même rendez-vous n'enfilent qu'un
   import, parce que la `dedup_key` porte la période ;
2. **un tick raté se rattrape** — un worker arrêté trois semaines retrouve le rendez-vous en
   retard, l'honore **une** fois, et repointe l'aiguille dans le futur : il ne rejoue pas les
   trois semaines manquées.
"""

from datetime import UTC, datetime, timedelta, timezone

import pytest
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import crypto
from app.core.errors import AppError
from app.integrations import schedule as cadence
from app.integrations import service
from app.integrations.models import ImportSchedule, Integration
from app.integrations.pipeline import JOB_TYPE
from app.jobs.models import Job
from app.keywords.models import Import
from tests.conftest import Fixture

# --- cadences pures -------------------------------------------------------------------------


def test_period_key_is_the_iso_week() -> None:
    monday = datetime(2026, 8, 31, 6, 0, tzinfo=UTC)
    sunday = datetime(2026, 9, 6, 23, 0, tzinfo=UTC)
    assert cadence.period_key("weekly", monday) == cadence.period_key("weekly", sunday)
    assert cadence.period_key("weekly", monday) == "2026-W36"
    #  Le lundi suivant est une autre période : la synchro repart.
    assert cadence.period_key("weekly", monday + timedelta(days=7)) == "2026-W37"


def test_period_key_is_the_calendar_month() -> None:
    assert cadence.period_key("monthly", datetime(2026, 9, 1, tzinfo=UTC)) == "2026-09"
    assert cadence.period_key("monthly", datetime(2026, 9, 30, 23, tzinfo=UTC)) == "2026-09"
    assert cadence.period_key("monthly", datetime(2026, 10, 1, tzinfo=UTC)) == "2026-10"


def test_period_key_is_computed_in_utc() -> None:
    """Deux serveurs de fuseaux différents doivent nommer la même période, sinon la dedup saute."""
    instant = datetime(2026, 9, 1, 0, 30, tzinfo=timezone(timedelta(hours=2)))
    assert cadence.period_key("monthly", instant) == "2026-08"


def test_advance_skips_every_missed_tick() -> None:
    """Trois semaines d'arrêt = **une** échéance future, pas trois rendez-vous empilés."""
    due = datetime(2026, 8, 3, 9, 0, tzinfo=UTC)
    now = datetime(2026, 8, 25, 10, 0, tzinfo=UTC)
    following = cadence.advance("weekly", due, now)
    assert following == datetime(2026, 8, 31, 9, 0, tzinfo=UTC)
    assert following is not None and following > now


def test_advance_monthly_keeps_to_the_calendar() -> None:
    due = datetime(2026, 1, 31, 8, 0, tzinfo=UTC)
    now = datetime(2026, 2, 1, 8, 0, tzinfo=UTC)
    #  Le 31 janvier n'a pas d'équivalent en février : le dernier jour du mois fait office,
    #  et surtout aucun mois n'est sauté.
    assert cadence.advance("monthly", due, now) == datetime(2026, 2, 28, 8, 0, tzinfo=UTC)


def test_manual_has_no_appointment() -> None:
    now = datetime(2026, 9, 2, tzinfo=UTC)
    assert cadence.first_run_at("manual", now) is None
    assert cadence.advance("manual", now, now) is None
    assert cadence.period_key("manual", now) == "manual"


def test_first_run_at_is_one_full_cadence_later() -> None:
    """Choisir « hebdomadaire » ne doit pas coûter un import immédiat sur le quota du client."""
    now = datetime(2026, 9, 2, 12, tzinfo=UTC)
    assert cadence.first_run_at("weekly", now) == now + timedelta(days=7)
    assert cadence.first_run_at("monthly", now) == datetime(2026, 10, 2, 12, tzinfo=UTC)


# --- le sweep --------------------------------------------------------------------------------


async def _integration(db: AsyncSession, org_id: str) -> Integration:
    record = Integration(
        org_id=org_id,
        provider="mangools",
        credentials=crypto.encrypt("clef-de-test-1234"),
        credentials_hint=crypto.hint("clef-de-test-1234"),
    )
    db.add(record)
    await db.commit()
    return record


async def _schedule(
    db: AsyncSession,
    sandbox: Fixture,
    integration: Integration,
    *,
    mode: str,
    due: datetime | None,
) -> ImportSchedule:
    record = ImportSchedule(
        org_id=sandbox.org_id,
        integration_id=integration.id,
        site_target_id=sandbox.target_id,
        sync_mode=mode,
        next_run_at=due,
    )
    db.add(record)
    await db.commit()
    return record


async def _count_jobs(db: AsyncSession, org_id: str) -> int:
    return (
        await db.scalar(
            select(func.count()).select_from(Job).where(Job.org_id == org_id, Job.type == JOB_TYPE)
        )
    ) or 0


async def test_sweep_enqueues_a_due_schedule(db: AsyncSession, sandbox: Fixture) -> None:
    integration = await _integration(db, sandbox.org_id)
    now = datetime.now(UTC)
    record = await _schedule(
        db, sandbox, integration, mode="weekly", due=now - timedelta(minutes=5)
    )

    assert await service.sweep_due_schedules(db, now=now) == 1

    assert await _count_jobs(db, sandbox.org_id) == 1
    entry = await db.scalar(select(Import).where(Import.org_id == sandbox.org_id))
    assert entry is not None
    assert entry.kind == "api"
    assert entry.source == "mangools"
    assert entry.status == "queued"
    #  L'aiguille a bougé dans le futur : le rendez-vous n'est plus dû.
    await db.refresh(record)
    assert record.next_run_at is not None and record.next_run_at > now


async def test_sweep_is_idempotent_within_a_period(db: AsyncSession, sandbox: Fixture) -> None:
    """Deux passes, un seul import : la `dedup_key` porte (intégration, cible, période)."""
    integration = await _integration(db, sandbox.org_id)
    now = datetime.now(UTC)
    await _schedule(db, sandbox, integration, mode="weekly", due=now - timedelta(minutes=5))

    assert await service.sweep_due_schedules(db, now=now) == 1
    assert await service.sweep_due_schedules(db, now=now) == 0
    assert await _count_jobs(db, sandbox.org_id) == 1


async def test_a_missed_tick_runs_once_and_catches_up(db: AsyncSession, sandbox: Fixture) -> None:
    """Worker arrêté trois semaines : **un** import, et l'échéance repart dans le futur."""
    integration = await _integration(db, sandbox.org_id)
    now = datetime.now(UTC)
    record = await _schedule(db, sandbox, integration, mode="weekly", due=now - timedelta(days=22))

    assert await service.sweep_due_schedules(db, now=now) == 1
    assert await _count_jobs(db, sandbox.org_id) == 1
    await db.refresh(record)
    assert record.next_run_at is not None
    assert record.next_run_at > now
    assert record.next_run_at <= now + timedelta(days=7)


async def test_sweep_ignores_disabled_and_manual(db: AsyncSession, sandbox: Fixture) -> None:
    integration = await _integration(db, sandbox.org_id)
    now = datetime.now(UTC)
    suspended = await _schedule(
        db, sandbox, integration, mode="weekly", due=now - timedelta(hours=1)
    )
    suspended.enabled = False
    await db.commit()

    assert await service.sweep_due_schedules(db, now=now) == 0
    assert await _count_jobs(db, sandbox.org_id) == 0

    #  Le mode manuel n'a par construction aucune échéance : la contrainte de base l'impose.
    suspended.enabled = True
    suspended.sync_mode = "manual"
    suspended.next_run_at = None
    await db.commit()
    assert await service.sweep_due_schedules(db, now=now) == 0


async def test_a_future_appointment_is_not_due(db: AsyncSession, sandbox: Fixture) -> None:
    integration = await _integration(db, sandbox.org_id)
    now = datetime.now(UTC)
    await _schedule(db, sandbox, integration, mode="monthly", due=now + timedelta(days=1))
    assert await service.sweep_due_schedules(db, now=now) == 0


async def test_next_period_enqueues_again(db: AsyncSession, sandbox: Fixture) -> None:
    """La semaine suivante est une autre période : la synchro repart, elle n'est pas bloquée."""
    integration = await _integration(db, sandbox.org_id)
    now = datetime.now(UTC)
    await _schedule(db, sandbox, integration, mode="weekly", due=now - timedelta(minutes=1))
    assert await service.sweep_due_schedules(db, now=now) == 1

    #  Le premier job doit avoir fini pour relâcher sa clé `'active'` — c'est le comportement
    #  réel du socle, reproduit ici sans worker.
    job = await db.scalar(select(Job).where(Job.org_id == sandbox.org_id, Job.type == JOB_TYPE))
    assert job is not None
    job.status = "done"
    job.dedup_key = None
    await db.commit()

    assert await service.sweep_due_schedules(db, now=now + timedelta(days=8)) == 1
    assert await _count_jobs(db, sandbox.org_id) == 2


async def test_manual_trigger_does_not_stack(db: AsyncSession, sandbox: Fixture) -> None:
    """Deux clics sur « Importer maintenant » ne lancent qu'un import."""
    integration = await _integration(db, sandbox.org_id)
    [(first, job, created, *_)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id
    )
    assert created is True
    [(second, same_job, again, *_)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id
    )
    assert again is False
    assert second.id == first.id
    assert same_job.id == job.id
    assert await _count_jobs(db, sandbox.org_id) == 1


async def test_job_payload_never_carries_the_key(db: AsyncSession, sandbox: Fixture) -> None:
    """`jobs.payload` est du jsonb en clair, relu par `GET /jobs/{id}` : rien de secret dedans."""
    integration = await _integration(db, sandbox.org_id)
    [(_, job, *_rest)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id
    )
    payload = dict(job.payload or {})
    flat = str(payload)
    assert "clef-de-test-1234" not in flat
    assert integration.credentials not in flat
    assert payload["integration_id"] == str(integration.id)
    assert payload["source"] == "mangools"
    assert payload["locale"] == "fr-FR"


async def test_a_poisoned_schedule_does_not_starve_the_others(
    db: AsyncSession, sandbox: Fixture, other_sandbox: Fixture
) -> None:
    """Un rendez-vous qui lève n'empêche pas les suivants — **d'aucune organisation**.

    Le piège : `next_run_at` gouverne l'ordre du sweep. Un planning dont l'enfilage lève garde
    son échéance échue, reste donc *premier* du tri à chaque passe, et gèle le planificateur
    pour tout le monde. Le `config` vérolé n'est ici qu'un moyen d'y arriver — c'est
    l'isolement de l'itération qu'on protège.
    """
    now = datetime.now(UTC)
    poisoned = await _integration(db, sandbox.org_id)
    #  Écrit directement en base : le `PUT` le refuse désormais (test dédié plus bas), mais une
    #  ligne écrite avant la validation, ou restaurée, existe toujours.
    bad = await _schedule(db, sandbox, poisoned, mode="weekly", due=now - timedelta(hours=2))
    bad.config = {"location_id": "pas-un-entier"}
    await db.commit()

    healthy_integration = await _integration(db, other_sandbox.org_id)
    healthy = await _schedule(
        db, other_sandbox, healthy_integration, mode="weekly", due=now - timedelta(hours=1)
    )

    #  Le vérolé est le plus ancien, donc le premier servi : c'est bien lui qui bloquerait.
    assert await service.sweep_due_schedules(db, now=now) == 1

    #  L'organisation saine a été servie malgré le voisin cassé.
    assert await _count_jobs(db, other_sandbox.org_id) == 1
    assert await _count_jobs(db, sandbox.org_id) == 0

    #  Et le rendez-vous fautif a vu son aiguille avancer : il ne se represente pas en boucle.
    await db.refresh(bad)
    assert bad.next_run_at is not None and bad.next_run_at > now
    await db.refresh(healthy)
    assert healthy.next_run_at is not None and healthy.next_run_at > now

    #  La passe suivante ne retrouve plus rien à faire : la file n'est pas affamée.
    assert await service.sweep_due_schedules(db, now=now) == 0


async def test_a_poisoned_config_refuses_the_manual_import_too(
    db: AsyncSession, sandbox: Fixture
) -> None:
    """Le même `config` vérolé rend un 422 lisible sur « Importer maintenant », pas un 500."""
    integration = await _integration(db, sandbox.org_id)
    record = await _schedule(db, sandbox, integration, mode="manual", due=None)
    record.config = {"location_id": []}
    await db.commit()

    with pytest.raises(AppError) as caught:
        await service.trigger_import(db, sandbox.org_id, integration.id, sandbox.target_id)
    assert caught.value.status == 422
    assert "location_id" in caught.value.detail


async def test_the_enqueued_job_belongs_to_the_schedule_org(
    db: AsyncSession, sandbox: Fixture
) -> None:
    """Le job et la ligne `imports` héritent de l'organisation du rendez-vous, pas d'une autre."""
    integration = await _integration(db, sandbox.org_id)
    now = datetime.now(UTC)
    await _schedule(db, sandbox, integration, mode="weekly", due=now - timedelta(minutes=1))
    await service.sweep_due_schedules(db, now=now)
    job = await db.scalar(select(Job).where(Job.type == JOB_TYPE, Job.org_id == sandbox.org_id))
    assert job is not None
    entry = await db.scalar(select(Import).where(Import.job_id == job.id))
    assert entry is not None
    assert entry.org_id == job.org_id == sandbox.org_id
    assert entry.site_target_id == sandbox.target_id


async def test_a_deleted_integration_takes_its_schedules(
    db: AsyncSession, sandbox: Fixture
) -> None:
    integration = await _integration(db, sandbox.org_id)
    now = datetime.now(UTC)
    await _schedule(db, sandbox, integration, mode="weekly", due=now + timedelta(days=1))
    await service.delete_integration(db, sandbox.org_id, integration.id)
    left = await db.scalar(
        select(func.count())
        .select_from(ImportSchedule)
        .where(ImportSchedule.org_id == sandbox.org_id)
    )
    assert left == 0