"""Pipeline integration tests against real Postgres (task 5.7).

Four properties, each of which is a way the system could message 800 people wrongly. None
of them can be tested without a real database: `SKIP LOCKED`, unique-constraint races and
transactional claiming are Postgres behaviours, not application logic, and a fake would
assert only that the mock behaves like the mock.

Run with `make test` (the api container has DATABASE_URL); they skip elsewhere.
"""

import asyncio
from datetime import UTC, datetime, timedelta
from typing import Any

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

from app.models import Event, Invitation, MessageJob
from app.models.enums import InvitationStatus, MessageJobStatus
from app.services import messaging, planner
from tests.conftest import requires_db

pytestmark = requires_db


async def _job_count(session: AsyncSession, event_id: Any) -> int:
    return (
        await session.scalar(
            select(func.count(MessageJob.id))
            .join(Invitation, Invitation.id == MessageJob.invitation_id)
            .where(Invitation.event_id == event_id)
        )
    ) or 0


async def _event_job_ids(session: AsyncSession, event_id: Any) -> set[Any]:
    rows = await session.scalars(
        select(MessageJob.id)
        .join(Invitation, Invitation.id == MessageJob.invitation_id)
        .where(Invitation.event_id == event_id)
    )
    return set(rows)


async def _make_due(session: AsyncSession, event_id: Any) -> set[Any]:
    """Backdate this event's jobs so they are claimable, and return their ids.

    Scoped to the fixture's event on purpose. An unfiltered `UPDATE message_job SET
    scheduled_for = now()` would make every seeded and real job in the database due at
    once, and the running worker would immediately try to send all of them.
    """
    ids = await _event_job_ids(session, event_id)
    if ids:
        await session.execute(
            update(MessageJob)
            .where(MessageJob.id.in_(ids))
            .values(scheduled_for=datetime.now(UTC) - timedelta(minutes=1))
        )
    await session.commit()
    assert ids, "the fixture produced no jobs, so this test would pass vacuously"
    return ids


async def _still_queued(session: AsyncSession, ids: set[Any]) -> set[Any]:
    """Which of these jobs nobody has claimed. Read on a fresh transaction so it sees the
    concurrent claimers' commits rather than this session's stale snapshot."""
    await session.rollback()
    rows = await session.scalars(
        select(MessageJob.id).where(
            MessageJob.id.in_(ids), MessageJob.status == MessageJobStatus.QUEUED
        )
    )
    return set(rows)


async def _load_event(session: AsyncSession, event_id: Any) -> Event:
    from sqlalchemy.orm import selectinload

    event = await session.scalar(
        select(Event).where(Event.id == event_id).options(selectinload(Event.wedding))
    )
    assert event is not None
    return event


# ------------------------------------------------------------------ 1. double planning


async def test_running_the_planner_twice_creates_no_duplicates(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """The planner is on an hourly cron and is also called at worker startup. A deploy at
    10:59 runs it twice within a minute; that must not double every guest's reminder."""
    event = await _load_event(session, fixture_event["event_id"])

    first = await planner.plan_event_waves(session, event)
    await session.commit()
    after_first = await _job_count(session, event.id)

    second = await planner.plan_event_waves(session, event)
    await session.commit()
    after_second = await _job_count(session, event.id)

    assert sum(p.created for p in first) == fixture_event["guest_count"]
    assert sum(p.created for p in second) == 0, "second run must create nothing"
    assert after_first == after_second == fixture_event["guest_count"]


async def test_the_duplicate_is_stopped_by_the_constraint_not_by_a_lookup(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """Design D5. A check-then-insert would still race; the unique index is the guarantee.

    Enqueuing the identical triple directly — bypassing the planner — must still be
    refused, which proves the defence is in the schema.
    """
    invitation_id = fixture_event["invitation_ids"][0]
    schedule_id = fixture_event["schedule_id"]
    when = datetime.now(UTC) + timedelta(days=5)

    first = await messaging.enqueue(
        session,
        invitation_id=invitation_id,
        template_id=None,
        schedule_id=schedule_id,
        scheduled_for=when,
    )
    second = await messaging.enqueue(
        session,
        invitation_id=invitation_id,
        template_id=None,
        schedule_id=schedule_id,
        scheduled_for=when + timedelta(hours=3),  # different time, same triple
    )
    await session.commit()

    # `enqueue` returns the new row's id, or None when the unique key rejected the insert.
    assert first is not None
    assert second is None, "the same invitation:schedule:channel must insert once"


# ------------------------------------------------------------------ 2. restart mid-batch


async def test_a_restart_mid_batch_resends_nothing(
    session: AsyncSession,
    sessions: async_sessionmaker[AsyncSession],
    fixture_event: dict[str, Any],
) -> None:
    """Claiming commits `sending` before the provider call. A worker killed at that moment
    must not have its jobs picked up again by the next process while they may be in flight
    — and once sent, they must never be claimed again at all.
    """
    event = await _load_event(session, fixture_event["event_id"])
    await planner.plan_event_waves(session, event)
    mine = await _make_due(session, event.id)

    async with sessions() as claimer:
        claimed = await messaging.claim_due_jobs(claimer, 100)
        claimed_ids = {j.id for j in claimed} & mine
        await claimer.commit()  # the "crash" happens right here, after the claim commits

    # A fresh process starts and polls immediately.
    async with sessions() as restarted:
        second = await messaging.claim_due_jobs(restarted, 100)
        second_ids = {j.id for j in second} & mine
        await restarted.commit()

    assert not (second_ids & claimed_ids), (
        "a restart re-claimed jobs already marked sending; every one would be a duplicate email"
    )
    assert not await _still_queued(session, mine), "a due job was left unclaimed by everyone"


async def test_stuck_jobs_are_recovered_only_after_the_grace_window(
    session: AsyncSession,
    sessions: async_sessionmaker[AsyncSession],
    fixture_event: dict[str, Any],
) -> None:
    """The flip side: a job abandoned in `sending` must eventually come back, or the guest
    silently gets nothing.

    `requeue_stuck_jobs` is the only path that recovers it, and the grace window is what
    stops it stealing a job from a sender that is merely slow — requeueing an in-flight
    send is how one guest gets the same email twice.
    """
    from app.worker.main import STUCK_AFTER

    event = await _load_event(session, fixture_event["event_id"])
    await planner.plan_event_waves(session, event)
    mine = sorted(await _event_job_ids(session, event.id))
    stale_job, fresh_job = mine[0], mine[1]

    # `updated_at` carries an onupdate default, so it is set through a direct UPDATE
    # rather than an ORM assignment that the flush would immediately overwrite.
    await session.execute(
        update(MessageJob)
        .where(MessageJob.id == stale_job)
        .values(
            status=MessageJobStatus.SENDING,
            updated_at=datetime.now(UTC) - STUCK_AFTER - timedelta(minutes=1),
        )
    )
    await session.execute(
        update(MessageJob)
        .where(MessageJob.id == fresh_job)
        .values(status=MessageJobStatus.SENDING, updated_at=datetime.now(UTC))
    )
    await session.commit()

    # Calling the service directly rather than the worker's scheduler entry point: that
    # one opens its own session from a process-wide cached engine, which under pytest
    # belongs to an event loop this test does not own.
    requeued = await messaging.requeue_stuck(session, STUCK_AFTER)
    await session.commit()
    assert requeued >= 1

    await session.rollback()
    statuses = {
        row.id: row.status
        for row in await session.execute(
            select(MessageJob.id, MessageJob.status).where(
                MessageJob.id.in_([stale_job, fresh_job])
            )
        )
    }
    assert statuses[stale_job] is MessageJobStatus.QUEUED, "an abandoned job was never recovered"
    assert statuses[fresh_job] is MessageJobStatus.SENDING, (
        "a send still inside the grace window was requeued; that guest would be emailed twice"
    )


# ------------------------------------------------------------------ 3. cancel wins


async def test_a_cancellation_between_planning_and_sending_is_skipped(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """The whole reason the sender re-reads live status. A guest who cancels on Tuesday
    must not receive Wednesday's "see you in 2 days" from a job queued last week.
    """
    event = await _load_event(session, fixture_event["event_id"])
    await planner.plan_event_waves(session, event)
    await session.commit()

    invitation_id = fixture_event["invitation_ids"][0]
    job = await session.scalar(
        select(MessageJob).where(MessageJob.invitation_id == invitation_id).limit(1)
    )
    assert job is not None

    invitation = await session.get(Invitation, invitation_id)
    assert invitation is not None
    invitation.status = InvitationStatus.CANCELLED
    await session.commit()

    result = await messaging.process_job(session, job)
    await session.commit()

    assert result is MessageJobStatus.SKIPPED
    assert job.skip_reason is not None


# ------------------------------------------------------------------ 4. concurrent senders


async def test_two_concurrent_senders_never_claim_the_same_job(
    session: AsyncSession,
    sessions: async_sessionmaker[AsyncSession],
    fixture_event: dict[str, Any],
) -> None:
    """`FOR UPDATE SKIP LOCKED` is what makes a second worker safe to start. Without it the
    two would either block on each other or both send the same batch.
    """
    event = await _load_event(session, fixture_event["event_id"])
    await planner.plan_event_waves(session, event)
    mine = await _make_due(session, event.id)

    async def claim() -> set[Any]:
        async with sessions() as s:
            jobs = await messaging.claim_due_jobs(s, 100)
            ids = {j.id for j in jobs} & mine
            await s.commit()
            return ids

    first, second = await asyncio.gather(claim(), claim())

    assert not (first & second), f"both senders claimed {first & second}"
    # Completeness is asserted through the table rather than through the two return values,
    # because the dev stack's own worker is a third sender polling the same queue — see
    # `_still_queued`. Every job leaving `queued` is the real guarantee; which of the three
    # took it is not.
    assert not await _still_queued(session, mine), "a due job was left unclaimed by everyone"


# ------------------------------------------------------------------ date-change replan


async def test_moving_the_date_moves_the_queued_reminders(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """Task 5.5. Leaving jobs pointed at the old date would send "15 days to go" on a day
    that is no longer 15 days out."""
    event = await _load_event(session, fixture_event["event_id"])
    await planner.plan_event_waves(session, event)
    await session.commit()

    before = {
        j.id: j.scheduled_for
        for j in await session.scalars(
            select(MessageJob)
            .join(Invitation, Invitation.id == MessageJob.invitation_id)
            .where(Invitation.event_id == event.id)
        )
    }
    assert before

    event.starts_at = event.starts_at + timedelta(days=7)
    await session.flush()
    result = await planner.replan_event(session, event)
    await session.commit()

    after = {
        j.id: j.scheduled_for
        for j in await session.scalars(
            select(MessageJob)
            .join(Invitation, Invitation.id == MessageJob.invitation_id)
            .where(Invitation.event_id == event.id)
        )
    }

    assert result.rescheduled == len(before)
    assert len(after) == len(before), "re-planning must move rows, not create new ones"
    for job_id, old in before.items():
        assert after[job_id] == old + timedelta(days=7)


async def test_moving_the_date_closer_drops_waves_that_are_now_past(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """A wedding brought forward past a wave's send date must skip it, not fire it late."""
    event = await _load_event(session, fixture_event["event_id"])
    await planner.plan_event_waves(session, event)
    await session.commit()

    # T-15 for an event 3 days away is 12 days in the past.
    event.starts_at = datetime.now(UTC) + timedelta(days=3)
    await session.flush()
    result = await planner.replan_event(session, event)
    await session.commit()

    assert result.cancelled_past == fixture_event["guest_count"]
    assert result.rescheduled == 0

    remaining = await session.scalars(
        select(MessageJob)
        .join(Invitation, Invitation.id == MessageJob.invitation_id)
        .where(Invitation.event_id == event.id)
    )
    statuses = {j.status for j in remaining}
    assert statuses == {MessageJobStatus.SKIPPED}


@pytest.mark.parametrize("attempts", [1, 2])
async def test_replanning_repeatedly_is_stable(
    session: AsyncSession, fixture_event: dict[str, Any], attempts: int
) -> None:
    """An admin who fat-fingers the date three times must not end up with three copies."""
    event = await _load_event(session, fixture_event["event_id"])
    await planner.plan_event_waves(session, event)
    await session.commit()

    for _ in range(attempts):
        event.starts_at = event.starts_at + timedelta(days=1)
        await session.flush()
        await planner.replan_event(session, event)
        await session.commit()

    assert await _job_count(session, event.id) == fixture_event["guest_count"]


# --------------------------------------------- 5. manual sends (add-guest-invitation-send)
#
# The manual path shares this queue with every wave, and each of these guards a way that
# sharing could go wrong: content that cannot be re-rendered, a quiet-hours rule that must
# bend for a human and not for a planner, and duplicate protection that must apply to one
# path and not the other.


async def _process(sessions: async_sessionmaker[AsyncSession], job_id: Any) -> tuple[str, Any]:
    """Run one job through the sender in its own session, returning its status and row."""
    async with sessions() as s:
        job = await s.get(MessageJob, job_id)
        assert job is not None
        status = await messaging.process_job(s, job, audience_accepted=False)
        await s.commit()
        return status, job


def _capture_sends() -> Any:
    """The dry-run provider, emptied, so the next send can be read back.

    Asserting on what actually reached the provider is the only way to tell the stored-body
    path from the template path apart from the outside: both end in `sent`, and the whole
    point of the distinction is which words the guest receives.
    """
    from app.models.enums import Channel
    from app.services import providers

    provider = providers.get_provider(Channel.EMAIL)
    assert isinstance(provider, providers.DryRunProvider), (
        "these tests read what was sent, which requires DRY_RUN"
    )
    provider.sent.clear()
    return provider


async def test_a_stored_body_is_sent_instead_of_a_template(
    session: AsyncSession,
    sessions: async_sessionmaker[AsyncSession],
    fixture_event: dict[str, Any],
) -> None:
    """D3: a hand-edited message has no template to re-render from, so the job carries it."""
    provider = _capture_sends()
    job_id = await messaging.enqueue(
        session,
        invitation_id=fixture_event["invitation_ids"][0],
        template_id=None,
        schedule_id=None,
        scheduled_for=datetime.now(UTC),
        subject="Please come",
        body_text="Dear Guest 0\n\nWe would love to see you.",
        override_quiet_hours=True,
        key=messaging.manual_idempotency_key(fixture_event["invitation_ids"][0]),
    )
    await session.commit()
    assert job_id is not None

    status, job = await _process(sessions, job_id)

    assert status == MessageJobStatus.SENT
    assert job.sent_at is not None
    assert job.error_code is None

    # What the guest receives is the admin's words, not a rendered template.
    assert len(provider.sent) == 1
    assert provider.sent[0].subject == "Please come"
    assert "We would love to see you." in provider.sent[0].text
    # …and the opt-out the admin never typed is there anyway (D12).
    assert "/u/" in provider.sent[0].text


async def test_a_job_without_stored_content_still_renders_from_its_template(
    session: AsyncSession,
    sessions: async_sessionmaker[AsyncSession],
    fixture_event: dict[str, Any],
) -> None:
    """The path every wave uses must be untouched by the manual one."""
    provider = _capture_sends()
    job_id = await messaging.enqueue(
        session,
        invitation_id=fixture_event["invitation_ids"][1],
        template_id=None,
        schedule_id=None,
        scheduled_for=datetime.now(UTC),
        override_quiet_hours=True,
    )
    await session.commit()
    assert job_id is not None

    status, job = await _process(sessions, job_id)

    # The job stores no content of its own, so every word that reached the provider came
    # from a template — the branch added for manual sends was not taken.
    assert job.body_text is None and job.subject is None
    assert status == MessageJobStatus.SENT
    assert len(provider.sent) == 1
    assert provider.sent[0].text.strip()


async def test_quiet_hours_defer_a_wave_and_release_a_confirmed_manual_send(
    session: AsyncSession,
    sessions: async_sessionmaker[AsyncSession],
    fixture_event: dict[str, Any],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """D5: the override exists for a human who confirmed, and for nothing else.

    Quiet hours are patched to always apply rather than waiting for 22:00 — what is under
    test is the branch, not the clock arithmetic, which `test_reminders` already covers.
    """
    monkeypatch.setattr(
        messaging,
        "defer_past_quiet_hours",
        lambda moment, tz, start, end: moment + timedelta(hours=8),
    )

    deferred_id = await messaging.enqueue(
        session,
        invitation_id=fixture_event["invitation_ids"][2],
        template_id=None,
        schedule_id=None,
        scheduled_for=datetime.now(UTC),
        subject="Wave",
        body_text="A planned message.",
    )
    confirmed_id = await messaging.enqueue(
        session,
        invitation_id=fixture_event["invitation_ids"][3],
        template_id=None,
        schedule_id=None,
        scheduled_for=datetime.now(UTC),
        subject="Manual",
        body_text="An admin pressed send at 23:30.",
        override_quiet_hours=True,
        key=messaging.manual_idempotency_key(fixture_event["invitation_ids"][3]),
    )
    await session.commit()
    assert deferred_id is not None and confirmed_id is not None

    deferred_status, deferred_job = await _process(sessions, deferred_id)
    confirmed_status, _ = await _process(sessions, confirmed_id)

    assert deferred_status == "deferred"
    assert deferred_job.status == MessageJobStatus.QUEUED
    assert confirmed_status == MessageJobStatus.SENT


async def test_two_manual_sends_to_one_invitation_both_insert(
    session: AsyncSession, fixture_event: dict[str, Any]
) -> None:
    """D2, against the database that enforces it — the unique key must not collapse these."""
    invitation_id = fixture_event["invitation_ids"][4]

    first = await messaging.enqueue(
        session,
        invitation_id=invitation_id,
        template_id=None,
        schedule_id=None,
        scheduled_for=datetime.now(UTC),
        subject="First",
        body_text="Please come.",
        key=messaging.manual_idempotency_key(invitation_id),
    )
    second = await messaging.enqueue(
        session,
        invitation_id=invitation_id,
        template_id=None,
        schedule_id=None,
        scheduled_for=datetime.now(UTC),
        subject="Second",
        body_text="Resending — you said it never arrived.",
        key=messaging.manual_idempotency_key(invitation_id),
    )
    await session.commit()

    assert first is not None
    assert second is not None, "an admin resending an invitation must not be silently ignored"
    assert first != second