from datetime import UTC, datetime, timedelta

import pytest
import sqlalchemy as sa
from fansifter_common.utils import timezone
from freezegun import freeze_time

from resonance_engine.adapters.db import db
from resonance_engine.dsp.enums import DSPClientName, DSPId, DSPResource
from resonance_engine.dsp.models import DSPClient
from resonance_engine.fandata.enums import (
    FanCollectionError,
    FanConnectionStatus,
)
from resonance_engine.fandata.models import (
    FanCollectionState,
    FanConnection,
    FanConnectionFilter,
)

from tests.unit.helpers import create_model, create_model_batch


@pytest.mark.db
class TestFanConnectionQuery:
    def test_filter_by_status(self) -> None:
        create_model(FanConnection, status=FanConnectionStatus.active)
        create_model(FanConnection, status=FanConnectionStatus.revoked)

        result = FanConnection.query.filter(
            FanConnectionFilter(status=FanConnectionStatus.active)
        ).count()

        assert result == 1

    def test_approx_stats_estimates_total_active_revoked(self) -> None:
        for i in range(5):
            create_model(FanConnection, fan_id=f"active-{i}")
        for i in range(2):
            create_model(
                FanConnection,
                fan_id=f"revoked-{i}",
                status=FanConnectionStatus.revoked,
            )
        db.session.execute(sa.text("ANALYZE fan_connection"))

        stats = FanConnection.query.approx_stats()

        assert stats.total == 7
        assert stats.active == 5
        assert stats.revoked == 2

    def test_approx_stats_filters_by_client(self) -> None:
        dsp_client_1 = create_model(DSPClient, name=DSPClientName.spotify_songwhip)
        dsp_client_2 = create_model(DSPClient, name=DSPClientName.spotify_smf_sme)
        create_model_batch(FanConnection, size=4, dsp_client_id=dsp_client_1.id)
        create_model(FanConnection, fan_id="b-0", dsp_client_id=dsp_client_2.id)
        db.session.execute(sa.text("ANALYZE fan_connection"))

        stats = FanConnection.query.where(
            FanConnection.dsp_client_id == dsp_client_1.id
        ).approx_stats()

        assert stats.total == 4

    def test_filter_by_not_collected_since(self) -> None:
        cutoff = datetime(2026, 5, 1, tzinfo=UTC)
        create_model(
            FanConnection,
            fan_id="due",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 4, 1, tzinfo=UTC),
        )
        create_model(
            FanConnection,
            fan_id="not_due",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 10, tzinfo=UTC),
        )
        # last_collected_at IS NULL → collect ASAP.
        create_model(FanConnection, fan_id="asap", dsp_id=DSPId.spotify)

        result = FanConnection.query.filter(
            FanConnectionFilter(status=None, not_collected_since=cutoff)
        ).count()

        assert result == 2

    def test_filter_by_dsp_client_name(self) -> None:
        client_a = create_model(DSPClient, name=DSPClientName.spotify_songwhip)
        client_b = create_model(DSPClient, name=DSPClientName.spotify_smf_sme)
        create_model(FanConnection, fan_id="a", dsp_client_id=client_a.id)
        create_model(FanConnection, fan_id="b", dsp_client_id=client_b.id)

        result = FanConnection.query.filter(
            FanConnectionFilter(dsp_client_id=client_a.id)
        ).count()

        assert result == 1

    def test_filter_combines_status_and_state_predicates(self) -> None:
        cutoff = datetime(2026, 5, 1, tzinfo=UTC)
        # Matches: active + due + failures<=5
        create_model(
            FanConnection,
            fan_id="match",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 4, 1, tzinfo=UTC),
            consecutive_failures=2,
        )
        # Excluded by status (revoked)
        create_model(
            FanConnection,
            fan_id="revoked",
            dsp_id=DSPId.spotify,
            status=FanConnectionStatus.revoked,
            last_collected_at=datetime(2026, 4, 1, tzinfo=UTC),
            consecutive_failures=2,
        )
        # Excluded by failures
        create_model(
            FanConnection,
            fan_id="too_many_fails",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 4, 1, tzinfo=UTC),
            consecutive_failures=10,
        )
        # Excluded: not yet due
        create_model(
            FanConnection,
            fan_id="not_due",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 10, tzinfo=UTC),
            consecutive_failures=0,
        )

        result = FanConnection.query.filter(
            FanConnectionFilter(
                status=FanConnectionStatus.active,
                not_collected_since=cutoff,
                max_consecutive_failures=5,
            )
        ).count()

        assert result == 1

    def test_filter_by_not_last_dispatched_since_includes_never_dispatched(
        self,
    ) -> None:
        cutoff = datetime(2026, 5, 29, 9, 0, 0, tzinfo=UTC)
        # Never dispatched — last_dispatched_at IS NULL → always eligible.
        create_model(
            FanConnection,
            fan_id="eligible",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 29, 8, 0, 0, tzinfo=UTC),
            last_dispatched_at=None,
        )
        # In-flight control: dispatched recently, not yet collected after dispatch.
        create_model(
            FanConnection,
            fan_id="in_flight",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 29, 8, 0, 0, tzinfo=UTC),
            last_dispatched_at=datetime(2026, 5, 29, 11, 0, 0, tzinfo=UTC),
        )

        result = FanConnection.query.filter(
            FanConnectionFilter(status=None, not_last_dispatched_since=cutoff)
        ).count()

        assert result == 1

    def test_filter_by_not_last_dispatched_since_includes_collected_after_dispatch(
        self,
    ) -> None:
        cutoff = datetime(2026, 5, 29, 9, 0, 0, tzinfo=UTC)
        # Lambda completed: last_collected_at > last_dispatched_at → re-eligible immediately.
        create_model(
            FanConnection,
            fan_id="eligible",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 29, 11, 0, 0, tzinfo=UTC),
            last_dispatched_at=datetime(2026, 5, 29, 10, 0, 0, tzinfo=UTC),
        )
        # In-flight control: same dispatched_at but collection is older than dispatch.
        create_model(
            FanConnection,
            fan_id="in_flight",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 29, 8, 0, 0, tzinfo=UTC),
            last_dispatched_at=datetime(2026, 5, 29, 11, 0, 0, tzinfo=UTC),
        )

        result = FanConnection.query.filter(
            FanConnectionFilter(status=None, not_last_dispatched_since=cutoff)
        ).count()

        assert result == 1

    def test_filter_by_not_last_dispatched_since_includes_expired_lock(self) -> None:
        cutoff = datetime(2026, 5, 29, 9, 0, 0, tzinfo=UTC)
        # Dispatch lock expired (safety net for DLQ / Lambda crash):
        # last_dispatched_at < cutoff, even though last_collected_at < last_dispatched_at.
        create_model(
            FanConnection,
            fan_id="eligible",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 29, 7, 0, 0, tzinfo=UTC),
            last_dispatched_at=datetime(2026, 5, 29, 8, 0, 0, tzinfo=UTC),
        )
        # In-flight control: recently dispatched, collection not advanced.
        create_model(
            FanConnection,
            fan_id="in_flight",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 29, 8, 0, 0, tzinfo=UTC),
            last_dispatched_at=datetime(2026, 5, 29, 11, 0, 0, tzinfo=UTC),
        )

        result = FanConnection.query.filter(
            FanConnectionFilter(status=None, not_last_dispatched_since=cutoff)
        ).count()

        assert result == 1

    def test_filter_by_not_last_dispatched_since_excludes_in_flight(self) -> None:
        cutoff = datetime(2026, 5, 29, 9, 0, 0, tzinfo=UTC)
        # In-flight: dispatched recently, collection predates dispatch, lock not yet expired.
        create_model(
            FanConnection,
            fan_id="in_flight",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 29, 8, 0, 0, tzinfo=UTC),
            last_dispatched_at=datetime(2026, 5, 29, 11, 0, 0, tzinfo=UTC),
        )

        result = FanConnection.query.filter(
            FanConnectionFilter(status=None, not_last_dispatched_since=cutoff)
        ).count()

        assert result == 0

    def test_filter_by_max_consecutive_failures(self) -> None:
        create_model(
            FanConnection,
            fan_id="ok",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            consecutive_failures=2,
        )
        create_model(
            FanConnection,
            fan_id="bad",
            dsp_id=DSPId.spotify,
            last_collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            consecutive_failures=10,
        )

        result = FanConnection.query.filter(
            FanConnectionFilter(status=None, max_consecutive_failures=5)
        ).count()

        assert result == 1

    def test_connect_inserts_new_row(self) -> None:
        FanConnection.query.connect(
            [
                {
                    "fan_id": "u1",
                    "dsp_id": DSPId.spotify,
                    "dsp_client_id": 1,
                    "token_encrypted": "tok1",
                }
            ]
        )

        stored = FanConnection.query.where(FanConnection.fan_id == "u1").one()

        assert stored.token_encrypted == "tok1"
        assert stored.status == FanConnectionStatus.active

    def test_connect_updates_existing_row(self) -> None:
        create_model(
            FanConnection,
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            token_encrypted="old",
            status=FanConnectionStatus.revoked,
        )

        FanConnection.query.connect(
            [
                {
                    "fan_id": "u1",
                    "dsp_id": DSPId.spotify,
                    "dsp_client_id": 1,
                    "token_encrypted": "new",
                }
            ]
        )

        stored = FanConnection.query.where(FanConnection.fan_id == "u1").one()

        assert stored.token_encrypted == "new"
        assert stored.status == FanConnectionStatus.active
        assert stored.token_refreshed_at is not None

    @freeze_time("2026-06-25T12:00:00Z")
    def test_connect_requeues_collected_fan_for_next_collect(self) -> None:
        create_model(
            FanConnection,
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            token_encrypted="old",
            last_collected_at=datetime(2026, 6, 1, tzinfo=UTC),
            last_dispatched_at=datetime(2026, 5, 31, tzinfo=UTC),
            consecutive_failures=7,
        )

        FanConnection.query.connect(
            [
                {
                    "fan_id": "u1",
                    "dsp_id": DSPId.spotify,
                    "dsp_client_id": 1,
                    "token_encrypted": "new",
                }
            ]
        )

        stored = FanConnection.query.where(FanConnection.fan_id == "u1").one()

        # Re-queued to the front (collect ASAP), dispatch lock and failure gate cleared,
        # connect_rank stamped (= -epoch of now) so the fan sorts newest-first.
        assert stored.last_collected_at is None
        assert stored.last_dispatched_at is None
        assert stored.consecutive_failures is None
        assert stored.connect_rank == -int(
            datetime(2026, 6, 25, 12, 0, tzinfo=UTC).timestamp()
        )
        assert stored.last_connected_at == datetime(2026, 6, 25, 12, 0, tzinfo=UTC)

    def test_update_token(self) -> None:
        create_model(
            FanConnection,
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            token_encrypted="old",
        )

        FanConnection.query.update_token(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            token_encrypted="rotated",
        )

        stored = FanConnection.query.where(FanConnection.fan_id == "u1").one()

        assert stored.token_encrypted == "rotated"
        assert stored.token_refreshed_at is not None

    def test_mark_revoked(self) -> None:
        create_model(
            FanConnection,
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
        )

        FanConnection.query.mark_revoked(
            fan_id="u1", dsp_id=DSPId.spotify, dsp_client_id=1
        )

        stored = FanConnection.query.where(FanConnection.fan_id == "u1").one()

        assert stored.status == FanConnectionStatus.revoked


@pytest.mark.db
class TestFanCollectionStateQuery:
    def test_approx_stats_estimates_total_healthy_with_errors(self) -> None:
        now = timezone.now()
        for i in range(4):
            create_model(
                FanCollectionState,
                fan_id=f"ok-{i}",
                dsp_id=DSPId.spotify,
                last_dsp_client_id=7,
                last_collected_at=now,
                last_collection_error=None,
            )
        for i in range(3):
            create_model(
                FanCollectionState,
                fan_id=f"err-{i}",
                dsp_id=DSPId.spotify,
                last_dsp_client_id=7,
                last_collected_at=now,
                last_collection_error=FanCollectionError.api_error,
            )
        db.session.execute(sa.text("ANALYZE fan_collection_state"))

        stats = FanCollectionState.query.approx_stats()

        assert stats.total == 7
        assert stats.healthy == 4
        assert stats.with_errors == 3

    def test_record_inserts_new_row(self) -> None:
        FanCollectionState.query.record(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            error=None,
        )

        stored = FanCollectionState.query.where(FanCollectionState.fan_id == "u1").one()

        assert stored.last_collected_at == datetime(2026, 5, 1, tzinfo=UTC)
        assert stored.last_collection_error is None
        assert stored.consecutive_failures == 0

    def test_record_mirrors_onto_dispatched_connection(self) -> None:
        create_model(
            FanConnection,
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            last_collected_at=None,
            last_dispatched_at=datetime(2026, 4, 1, tzinfo=UTC),
        )

        FanCollectionState.query.record(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            error=None,
        )

        stored = FanConnection.query.where(FanConnection.fan_id == "u1").one()

        assert stored.last_collected_at == datetime(2026, 5, 1, tzinfo=UTC)

    def test_record_skips_mirror_for_reconnected_fan(self) -> None:
        # Re-connected (last_dispatched_at NULL): a collect finishing after the
        # re-connect must not overwrite the fresh re-queue.
        create_model(
            FanConnection,
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            last_collected_at=None,
            last_dispatched_at=None,
        )

        FanCollectionState.query.record(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            error=None,
        )

        stored = FanConnection.query.where(FanConnection.fan_id == "u1").one()

        assert stored.last_collected_at is None

    def test_record_success_resets_consecutive_failures(self) -> None:
        create_model(
            FanCollectionState,
            fan_id="u1",
            dsp_id=DSPId.spotify,
            last_dsp_client_id=1,
            last_collected_at=datetime(2026, 4, 1, tzinfo=UTC),
            last_collection_error=FanCollectionError.api_error,
            consecutive_failures=5,
        )

        FanCollectionState.query.record(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            error=None,
        )

        stored = FanCollectionState.query.where(FanCollectionState.fan_id == "u1").one()

        assert stored.last_collection_error is None
        assert stored.consecutive_failures == 0

    def test_record_failure_increments_consecutive_failures(self) -> None:
        create_model(
            FanCollectionState,
            fan_id="u1",
            dsp_id=DSPId.spotify,
            last_dsp_client_id=1,
            last_collected_at=datetime(2026, 4, 1, tzinfo=UTC),
            last_collection_error=FanCollectionError.api_error,
            consecutive_failures=2,
        )

        FanCollectionState.query.record(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            error=FanCollectionError.api_error,
        )

        stored = FanCollectionState.query.where(FanCollectionState.fan_id == "u1").one()

        assert stored.last_collection_error == FanCollectionError.api_error
        assert stored.consecutive_failures == 3

    def test_record_rate_limited_initial_insert_starts_at_zero(self) -> None:
        FanCollectionState.query.record(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            error=FanCollectionError.rate_limited,
        )

        stored = FanCollectionState.query.where(FanCollectionState.fan_id == "u1").one()

        assert stored.last_collection_error == FanCollectionError.rate_limited
        assert stored.consecutive_failures == 0

    def test_record_rate_limited_preserves_consecutive_failures(self) -> None:
        create_model(
            FanCollectionState,
            fan_id="u1",
            dsp_id=DSPId.spotify,
            last_dsp_client_id=1,
            last_collected_at=datetime(2026, 4, 1, tzinfo=UTC),
            consecutive_failures=2,
        )

        FanCollectionState.query.record(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            error=FanCollectionError.rate_limited,
        )

        stored = FanCollectionState.query.where(FanCollectionState.fan_id == "u1").one()

        assert stored.last_collection_error == FanCollectionError.rate_limited
        assert stored.consecutive_failures == 2

    def test_record_writes_resource_columns(self) -> None:
        FanCollectionState.query.record(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            collected_at=datetime(2026, 5, 1, 12, tzinfo=UTC),
            error=None,
            resources={
                DSPResource.profile: datetime(2026, 5, 1, 11, tzinfo=UTC),
                DSPResource.top_artists: datetime(2026, 5, 1, 11, 5, tzinfo=UTC),
            },
        )

        stored = FanCollectionState.query.where(FanCollectionState.fan_id == "u1").one()

        assert stored.profile_collected_at == datetime(2026, 5, 1, 11, tzinfo=UTC)
        assert stored.top_artists_collected_at == datetime(
            2026, 5, 1, 11, 5, tzinfo=UTC
        )
        assert stored.top_tracks_collected_at is None

    def test_record_updates_resource_columns_on_conflict(self) -> None:
        create_model(
            FanCollectionState,
            fan_id="u1",
            dsp_id=DSPId.spotify,
            last_dsp_client_id=1,
            last_collected_at=datetime(2026, 4, 1, tzinfo=UTC),
            profile_collected_at=datetime(2026, 4, 1, tzinfo=UTC),
            top_tracks_collected_at=datetime(2026, 4, 1, tzinfo=UTC),
        )

        FanCollectionState.query.record(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            error=None,
            resources={
                DSPResource.profile: datetime(2026, 5, 1, tzinfo=UTC),
            },
        )

        stored = FanCollectionState.query.where(FanCollectionState.fan_id == "u1").one()

        assert stored.profile_collected_at == datetime(2026, 5, 1, tzinfo=UTC)
        # Untouched resource retains its previous value.
        assert stored.top_tracks_collected_at == datetime(2026, 4, 1, tzinfo=UTC)

    def test_record_partial_failure_writes_completed_resources(self) -> None:
        FanCollectionState.query.record(
            fan_id="u1",
            dsp_id=DSPId.spotify,
            dsp_client_id=1,
            collected_at=datetime(2026, 5, 1, tzinfo=UTC),
            error=FanCollectionError.api_error,
            resources={
                DSPResource.profile: datetime(2026, 5, 1, tzinfo=UTC),
            },
        )

        stored = FanCollectionState.query.where(FanCollectionState.fan_id == "u1").one()

        assert stored.last_collection_error == FanCollectionError.api_error
        assert stored.consecutive_failures == 1
        assert stored.profile_collected_at == datetime(2026, 5, 1, tzinfo=UTC)
        assert stored.top_artists_collected_at is None

    def test_stamp_dispatched_updates_existing_rows(self) -> None:
        dispatched_at = datetime(2026, 5, 29, 12, 0, 0, tzinfo=UTC)
        create_model(FanConnection, fan_id="a", dsp_id=DSPId.spotify)
        create_model(FanConnection, fan_id="b", dsp_id=DSPId.spotify)

        FanCollectionState.query.stamp_dispatched(
            [("a", DSPId.spotify), ("b", DSPId.spotify)], now=dispatched_at
        )

        conn_a = FanConnection.query.where(FanConnection.fan_id == "a").one()
        conn_b = FanConnection.query.where(FanConnection.fan_id == "b").one()

        assert conn_a.last_dispatched_at == dispatched_at
        assert conn_b.last_dispatched_at == dispatched_at

    def test_stamp_dispatched_skips_fans_without_connection_row(self) -> None:
        # New fan — no FanConnection row yet. stamp_dispatched must not create one.
        FanCollectionState.query.stamp_dispatched(
            [("new_fan", DSPId.spotify)],
            now=datetime(2026, 5, 29, 12, 0, 0, tzinfo=UTC),
        )

        count = FanConnection.query.where(FanConnection.fan_id == "new_fan").count()
        assert count == 0

    def test_stamp_dispatched_noop_on_empty_list(self) -> None:
        # Must not raise even when there is nothing to stamp.
        FanCollectionState.query.stamp_dispatched(
            [], now=datetime(2026, 5, 29, 12, 0, 0, tzinfo=UTC)
        )

    def test_activity_counts_buckets_by_status(self) -> None:
        now = datetime.now(UTC)
        create_model(
            FanCollectionState,
            fan_id="ok",
            dsp_id=DSPId.spotify,
            last_dsp_client_id=1,
            last_collected_at=now - timedelta(days=1),
            last_collection_error=None,
        )
        create_model(
            FanCollectionState,
            fan_id="bad",
            dsp_id=DSPId.spotify,
            last_dsp_client_id=1,
            last_collected_at=now - timedelta(days=1),
            last_collection_error=FanCollectionError.api_error,
        )
        create_model(
            FanCollectionState,
            fan_id="throttled",
            dsp_id=DSPId.spotify,
            last_dsp_client_id=1,
            last_collected_at=now - timedelta(days=1),
            last_collection_error=FanCollectionError.rate_limited,
        )

        buckets = FanCollectionState.query.activity(days=7, granularity="daily")

        assert len(buckets) == 1

        bucket = buckets[0]

        assert bucket.success == 1
        assert bucket.errors == 1
        assert bucket.throttled == 1