"""Cancellation regressions for weekly reminder provider admission."""

from __future__ import annotations

import fcntl
import os
import multiprocessing
import signal
from datetime import datetime
from pathlib import Path

import anyio
import pytest

from checkin_cli.weekly_operations_lifecycle import (
    ReminderDependencies,
    ProviderDelivered,
    ProviderRejected,
    ReminderOutcome,
    run_due_reminder,
)
from checkin_cli.weekly_operations import CustomerKey
from checkin_cli.weekly_operations_customer_authority import (
    CanonicalCheckinCustomerAuthority,
)
from checkin_cli.weekly_operations_store import WeeklyOperationsStore
from checkin_cli.weekly_reminder_authority import (
    BoundWeeklyReminderCustomer,
    WeeklyReminderAuthorizationFacts,
    WeeklyReminderBindingInput,
    WeeklyReminderRequest,
    bind_weekly_reminder_customer,
    seal_weekly_reminder_authorization,
)
from checkin_cli.weekly_reminder_ledger_open import (
    open_registered_weekly_reminder_ledger_authority,
)
from tests._weekly_operations_lifecycle_support import (
    KST,
    FakeReminderProvider,
    ProviderMode,
    fixture_at,
)


@pytest.mark.parametrize("variant", ("task_group", "cancel_scope"))
def test_provider_cancellation_releases_exact_flock_and_replays_without_send(
    tmp_path: Path, variant: str
) -> None:
    fixture = fixture_at(tmp_path, datetime(2026, 8, 17, 20, tzinfo=KST))
    provider = FakeReminderProvider(ProviderMode.BLOCKED)
    cancelled = anyio.Event()
    scope_ready = anyio.Event()
    scopes: list[anyio.CancelScope] = []

    async def run_attempt() -> None:
        _ = await run_due_reminder(
            fixture.request,
            ReminderDependencies(provider, lambda: fixture.request.bound_customer),
        )

    async def attempt() -> None:
        try:
            if variant == "cancel_scope":
                with anyio.CancelScope() as scope:
                    scopes.append(scope)
                    scope_ready.set()
                    await run_attempt()
            else:
                scope_ready.set()
                await run_attempt()
        finally:
            cancelled.set()

    async def exercise() -> None:
        async with anyio.create_task_group() as tasks:
            tasks.start_soon(attempt)
            await scope_ready.wait()
            await provider.entered.wait()
            if variant == "cancel_scope":
                scopes[0].cancel()
            else:
                tasks.cancel_scope.cancel()
        assert cancelled.is_set()

    try:
        anyio.run(exercise)
        receipts = fixture.request.bound_customer.ledger.receipts()
        assert receipts[-1].state == "unknown"
        assert receipts[-1].reason == "provider_interrupted"
        lock_path = next(tmp_path.rglob(".scheduled-deliveries.lock"))
        descriptor = os.open(lock_path, os.O_RDWR | os.O_CLOEXEC)
        try:
            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
            fcntl.flock(descriptor, fcntl.LOCK_UN)
        finally:
            os.close(descriptor)
        replay_provider = FakeReminderProvider()
        replay = anyio.run(
            run_due_reminder,
            fixture.request,
            ReminderDependencies(
                replay_provider, lambda: fixture.request.bound_customer
            ),
        )
        assert replay.outcome is ReminderOutcome.UNKNOWN
        assert replay_provider.calls == 0
    finally:
        fixture.close()


def _run_signal_attempt(
    request: WeeklyReminderRequest,
    source: CanonicalCheckinCustomerAuthority,
    store: WeeklyOperationsStore,
    entered_fd: int,
) -> None:
    inherited = request.bound_customer
    inherited.ledger.close()
    ledger = open_registered_weekly_reminder_ledger_authority(source, store)
    proof = seal_weekly_reminder_authorization(
        WeeklyReminderAuthorizationFacts(
            CustomerKey("pilot_customer_001"),
            inherited.candidate_digest,
            inherited.config_digest,
            inherited.runtime_registry_digest,
            inherited.owner_digest,
            inherited.consent_digest,
            inherited.feature_epoch,
            inherited.route,
        )
    )
    bound = bind_weekly_reminder_customer(
        WeeklyReminderBindingInput(proof, ledger, inherited.source, inherited.store)
    )
    child_request = WeeklyReminderRequest(bound, request.kst_day, request.now)

    class SignalProvider:
        async def send(
            self, bound_customer: BoundWeeklyReminderCustomer
        ) -> ProviderDelivered | ProviderRejected:
            del bound_customer
            _ = os.write(entered_fd, b"1")
            await anyio.sleep_forever()
            raise AssertionError("sleep_forever returned")

    _ = anyio.run(
        run_due_reminder,
        child_request,
        ReminderDependencies(SignalProvider(), lambda: child_request.bound_customer),
    )


def test_sigterm_releases_flock_and_restart_terminalizes_without_send(
    tmp_path: Path,
) -> None:
    fixture = fixture_at(tmp_path, datetime(2026, 8, 17, 20, tzinfo=KST))
    read_fd, write_fd = os.pipe()
    process = multiprocessing.get_context("fork").Process(
        target=_run_signal_attempt,
        args=(fixture.request, fixture.canonical.source, fixture.stores.store, write_fd),
    )
    try:
        process.start()
        os.close(write_fd)
        assert os.read(read_fd, 1) == b"1"
        pid = process.pid
        assert pid is not None
        os.kill(pid, signal.SIGTERM)
        process.join(timeout=5)
        assert process.exitcode == -signal.SIGTERM
        lock_path = next(tmp_path.rglob(".scheduled-deliveries.lock"))
        descriptor = os.open(lock_path, os.O_RDWR | os.O_CLOEXEC)
        try:
            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
            fcntl.flock(descriptor, fcntl.LOCK_UN)
        finally:
            os.close(descriptor)
        replay_provider = FakeReminderProvider()
        result = anyio.run(
            run_due_reminder,
            fixture.request,
            ReminderDependencies(
                replay_provider, lambda: fixture.request.bound_customer
            ),
        )
        assert result.outcome is ReminderOutcome.UNKNOWN
        assert result.receipt is not None
        assert result.receipt.reason == "provider_unknown_after_restart"
        assert replay_provider.calls == 0
    finally:
        os.close(read_fd)
        if process.is_alive():
            process.kill()
            process.join(timeout=5)
        fixture.close()