import uuid
from datetime import timedelta
from itertools import count
from unittest import mock

import pytest
from fansifter_common.utils import timezone
from pytest_mock import MockerFixture

from app.dsp.enums import DSPClientStatus
from app.dsp.exceptions import (
    DSPError,
    RateLimitError,
    TokenRefreshError,
    TokenRevokedError,
)
from app.dsp.models import DSPClient
from app.dsp.types import (
    FollowedArtistsResult,
    PlaylistsResult,
    ProfileResult,
    RecentlyPlayedResult,
    SavedAlbumsResult,
    SavedTracksResult,
    TokenResult,
    TopArtistsResult,
    TopTracksResult,
)
from app.fandata.enums import FanCollectionStatus
from app.fandata.models import (
    FanCollectionState,
    FanCollectionStateQuery,
    FanCredentials,
)
from app.fandata.sinks.base import DataSink
from app.fandata.types import FanBatch, FanRecord
from app.pipeline.enums import RunFinishedReason, RunStatus
from app.pipeline.exceptions import PipelineWaitTimeoutError
from app.pipeline.fan_collect import (
    CollectResult,
    _call_fan_in_transaction,
    _collect_batch,
    _collect_fans,
    _FanOutcome,
    collect,
)
from app.pipeline.models import PipelineRun
from app.pipeline.timings import FanTimings

from tests.unit.helpers import create_model, override_settings


class TestFanCollect:
    @pytest.fixture(autouse=True)
    def gateway_mock(self, mocker: MockerFixture) -> mock.MagicMock:
        m = mock.MagicMock()
        m.refresh_token.return_value = TokenResult(
            access_token="acc", refresh_token="ref"
        )
        m.get_profile.return_value = ProfileResult(email="fan@example.com")
        m.get_top_artists.return_value = TopArtistsResult()
        m.get_top_tracks.return_value = TopTracksResult()
        m.get_recently_played.return_value = RecentlyPlayedResult()
        m.get_playlists.return_value = PlaylistsResult()
        m.get_saved_albums.return_value = SavedAlbumsResult()
        m.get_saved_tracks.return_value = SavedTracksResult()
        m.get_followed_artists.return_value = FollowedArtistsResult()
        mocker.patch("app.pipeline.fan_collect.dsp_gateway", new=m)
        return m

    @pytest.fixture(autouse=True)
    def decrypt_mock(self, mocker: MockerFixture) -> mock.MagicMock:
        return mocker.patch(
            "app.pipeline.fan_collect.decrypt", return_value="plain_token"
        )

    @pytest.fixture(autouse=True)
    def encrypt_mock(self, mocker: MockerFixture) -> mock.MagicMock:
        return mocker.patch(
            "app.pipeline.fan_collect.encrypt", side_effect=lambda t: f"enc:{t}"
        )

    @pytest.fixture(autouse=True)
    def sink_mock(self, mocker: MockerFixture) -> mock.MagicMock:
        return mocker.patch("app.pipeline.fan_collect.data_sink", spec=DataSink)

    @pytest.fixture
    def call_fan_mock(self, mocker: MockerFixture) -> mock.MagicMock:
        return mocker.patch(
            "app.pipeline.fan_collect._call_fan_in_transaction",
            return_value=_FanOutcome(processed=1, api_requests=4),
        )

    @pytest.fixture
    def collect_fans_mock(self, mocker: MockerFixture) -> mock.MagicMock:
        return mocker.patch(
            "app.pipeline.fan_collect._collect_fans",
            return_value=CollectResult(
                processed=5, errors=1, rate_limited=0, stale_tokens=0, timed_out=False
            ),
        )

    @pytest.fixture
    def collect_batch_mock(self, mocker: MockerFixture) -> mock.MagicMock:
        return mocker.patch(
            "app.pipeline.fan_collect._collect_batch",
            return_value=CollectResult(
                processed=1, errors=0, rate_limited=0, stale_tokens=0, timed_out=False
            ),
        )

    # ── _call_fan ─────────────────────────────────────────────────────────────

    @pytest.mark.db
    def test_call_fan_success_returns_processed_one(self) -> None:
        client = create_model(DSPClient)
        run_started_at = timezone.now() - timedelta(hours=1)

        outcome = _call_fan_in_transaction(
            FanRecord(dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"),
            client,
            FanTimings(),
            run_started_at,
        )

        assert outcome.processed == 1
        assert outcome.errors == 0
        assert outcome.api_requests == 9  # token + 8 steps

    @pytest.mark.db
    def test_call_fan_success_records_collection_state(self) -> None:
        client = create_model(DSPClient)
        run_started_at = timezone.now() - timedelta(hours=1)

        _call_fan_in_transaction(
            FanRecord(dsp_user_id="u99", refresh_token_encrypted="enc:abc123"),
            client,
            FanTimings(),
            run_started_at,
        )

        state = FanCollectionState.query.where(
            FanCollectionState.dsp_user_id == "u99"
        ).one_or_none()
        assert state is not None
        assert state.last_status == FanCollectionStatus.success

    @pytest.mark.db
    def test_call_fan_saves_new_refresh_token_to_credentials(
        self, gateway_mock: mock.MagicMock
    ) -> None:
        client = create_model(DSPClient)
        create_model(
            FanCredentials,
            dsp_id=client.dsp_id,
            client_id=client.client_id,
            dsp_user_id="user_1",
        )
        gateway_mock.refresh_token.return_value = TokenResult(
            access_token="acc", refresh_token="new_token"
        )

        _call_fan_in_transaction(
            FanRecord(dsp_user_id="user_1", refresh_token_encrypted="enc:old"),
            client,
            FanTimings(),
            timezone.now() - timedelta(hours=1),
        )

        cred = FanCredentials.query.where(
            FanCredentials.dsp_user_id == "user_1",
            FanCredentials.dsp_id == client.dsp_id,
        ).one()
        assert cred.refresh_token_encrypted == "enc:new_token"

    @pytest.mark.db
    def test_call_fan_no_refresh_token_returns_error(self) -> None:
        client = create_model(DSPClient)
        run_started_at = timezone.now() - timedelta(hours=1)

        outcome = _call_fan_in_transaction(
            FanRecord(dsp_user_id="user_1", refresh_token_encrypted=""),
            client,
            FanTimings(),
            run_started_at,
        )

        assert outcome.processed == 0
        assert outcome.errors == 1
        assert outcome.api_requests == 0

    @pytest.mark.db
    def test_call_fan_token_revoked_increments_stale_tokens(
        self, gateway_mock: mock.MagicMock
    ) -> None:
        client = create_model(DSPClient)
        create_model(
            FanCredentials,
            dsp_id=client.dsp_id,
            client_id=client.client_id,
            dsp_user_id="user_1",
        )
        gateway_mock.refresh_token.side_effect = TokenRevokedError()

        outcome = _call_fan_in_transaction(
            FanRecord(dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"),
            client,
            FanTimings(),
            timezone.now() - timedelta(hours=1),
        )

        assert outcome.stale_tokens == 1
        assert outcome.processed == 0
        assert outcome.api_requests == 1

    @pytest.mark.db
    def test_call_fan_token_refresh_error_increments_errors(
        self, gateway_mock: mock.MagicMock
    ) -> None:
        client = create_model(DSPClient)
        gateway_mock.refresh_token.side_effect = TokenRefreshError()

        outcome = _call_fan_in_transaction(
            FanRecord(dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"),
            client,
            FanTimings(),
            timezone.now() - timedelta(hours=1),
        )

        assert outcome.errors == 1
        assert outcome.api_requests == 1

    @pytest.mark.db
    def test_call_fan_rate_limit_sets_reset_consecutive_false(
        self, gateway_mock: mock.MagicMock
    ) -> None:
        client = create_model(DSPClient)
        gateway_mock.get_profile.side_effect = RateLimitError(
            "rate limited", status_code=429, retry_after=60
        )

        outcome = _call_fan_in_transaction(
            FanRecord(dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"),
            client,
            FanTimings(),
            timezone.now() - timedelta(hours=1),
        )

        assert outcome.rate_limited == 1
        assert outcome.processed == 0
        assert not outcome.reset_consecutive

    @pytest.mark.db
    def test_call_fan_dsp_error_on_profile_returns_api_error(
        self, gateway_mock: mock.MagicMock
    ) -> None:
        client = create_model(DSPClient)
        gateway_mock.get_profile.side_effect = DSPError("server error")

        outcome = _call_fan_in_transaction(
            FanRecord(dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"),
            client,
            FanTimings(),
            timezone.now() - timedelta(hours=1),
        )

        assert outcome.errors == 1
        assert outcome.processed == 0
        assert outcome.api_requests == 2  # token + failed profile attempt

    # ── _collect_fans ─────────────────────────────────────────────────────────

    @pytest.mark.db
    def test_collect_fans_returns_zero_when_client_not_found(
        self, call_fan_mock: mock.MagicMock
    ) -> None:
        result = _collect_fans(
            FanBatch(
                run_id=uuid.uuid4(),
                fans=[
                    FanRecord(
                        dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"
                    )
                ],
            ),
            dsp_client_id=999999,
            run_started_at=timezone.now(),
        )

        assert result.processed == 0
        assert result.errors == 0

    @pytest.mark.db
    def test_collect_fans_accumulates_processed_across_fans(
        self, call_fan_mock: mock.MagicMock
    ) -> None:
        client = create_model(DSPClient)
        batch = FanBatch(
            run_id=uuid.uuid4(),
            fans=[
                FanRecord(dsp_user_id="u1", refresh_token_encrypted="enc:abc123"),
                FanRecord(dsp_user_id="u2", refresh_token_encrypted="enc:abc123"),
                FanRecord(dsp_user_id="u3", refresh_token_encrypted="enc:abc123"),
            ],
        )

        result = _collect_fans(batch, client.id, timezone.now() - timedelta(hours=1))

        assert result.processed == 3
        assert call_fan_mock.call_count == 3

    @pytest.mark.db
    def test_collect_fans_circuit_breaker_stops_loop_at_threshold(
        self, call_fan_mock: mock.MagicMock
    ) -> None:
        client = create_model(DSPClient)
        call_fan_mock.return_value = _FanOutcome(
            rate_limited=1, api_requests=1, reset_consecutive=False
        )

        with override_settings(fan_collect_rate_limit_circuit_breaker_threshold=3):
            result = _collect_fans(
                FanBatch(
                    run_id=uuid.uuid4(),
                    fans=[
                        FanRecord(
                            dsp_user_id=f"u{i}", refresh_token_encrypted="enc:abc123"
                        )
                        for i in range(10)
                    ],
                ),
                client.id,
                timezone.now() - timedelta(hours=1),
            )

        assert result.circuit_broken
        assert call_fan_mock.call_count == 3

    @pytest.mark.db
    def test_collect_fans_timeout_stops_loop_early(
        self, call_fan_mock: mock.MagicMock, mocker: MockerFixture
    ) -> None:
        client = create_model(DSPClient)
        tick = count(0)
        mocker.patch(
            "app.pipeline.fan_collect.time.monotonic",
            side_effect=lambda: next(tick) * 10.0,
        )

        with override_settings(
            fan_collect_time_limit_s=5, fan_collect_timeout_buffer_s=0
        ):
            result = _collect_fans(
                FanBatch(
                    run_id=uuid.uuid4(),
                    fans=[
                        FanRecord(
                            dsp_user_id=f"u{i}", refresh_token_encrypted="enc:abc123"
                        )
                        for i in range(10)
                    ],
                ),
                client.id,
                timezone.now() - timedelta(hours=1),
            )

        assert result.timed_out
        assert call_fan_mock.call_count < 10

    @pytest.mark.db
    def test_collect_fans_cooldown_skips_recently_collected(
        self, call_fan_mock: mock.MagicMock, mocker: MockerFixture
    ) -> None:
        client = create_model(DSPClient)
        mocker.patch.object(
            FanCollectionStateQuery, "recently_collected_ids", return_value={"u1"}
        )
        batch = FanBatch(
            run_id=uuid.uuid4(),
            fans=[
                FanRecord(dsp_user_id="u1", refresh_token_encrypted="enc:abc123"),
                FanRecord(dsp_user_id="u2", refresh_token_encrypted="enc:abc123"),
            ],
        )

        with override_settings(fan_collect_cooldown_s=3600):
            _collect_fans(batch, client.id, timezone.now() - timedelta(hours=1))

        assert call_fan_mock.call_count == 1
        assert call_fan_mock.call_args[0][0].dsp_user_id == "u2"

    # ── _collect_batch ────────────────────────────────────────────────────────

    @pytest.mark.db
    def test_collect_batch_returns_skip_when_run_not_found(
        self, collect_fans_mock: mock.MagicMock
    ) -> None:
        result = _collect_batch(
            FanBatch(
                run_id=uuid.uuid4(),
                fans=[
                    FanRecord(
                        dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"
                    )
                ],
            )
        )

        assert result.processed == 0
        assert result.errors == 0

    @pytest.mark.db
    def test_collect_batch_returns_skip_when_run_not_running(
        self, collect_fans_mock: mock.MagicMock
    ) -> None:
        run = create_model(PipelineRun, status=RunStatus.done)

        result = _collect_batch(
            FanBatch(
                run_id=run.id,
                fans=[
                    FanRecord(
                        dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"
                    )
                ],
            )
        )

        assert result.processed == 0

    @pytest.mark.db
    def test_collect_batch_returns_skip_when_run_cancelled_after_collection(
        self, collect_fans_mock: mock.MagicMock
    ) -> None:
        run = create_model(
            PipelineRun,
            status=RunStatus.running,
            total_batches=2,
            started_at=timezone.now(),
        )

        def cancel_mid(*_args: object, **_kwargs: object) -> CollectResult:
            run.status = RunStatus.cancelled
            run.save()
            return CollectResult(
                processed=3, errors=0, rate_limited=0, stale_tokens=0, timed_out=False
            )

        collect_fans_mock.side_effect = cancel_mid

        result = _collect_batch(
            FanBatch(
                run_id=run.id,
                fans=[
                    FanRecord(
                        dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"
                    )
                ],
            )
        )

        assert result.processed == 0

    @pytest.mark.db
    def test_collect_batch_commits_result_and_returns_it(
        self, collect_fans_mock: mock.MagicMock
    ) -> None:
        run = create_model(
            PipelineRun,
            status=RunStatus.running,
            total_batches=2,
            started_at=timezone.now(),
        )

        result = _collect_batch(
            FanBatch(
                run_id=run.id,
                fans=[
                    FanRecord(
                        dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"
                    )
                ],
            )
        )

        assert result.processed == 5
        assert result.errors == 1

    @pytest.mark.db
    def test_collect_batch_finalizes_run_when_last_batch_completes(
        self, collect_fans_mock: mock.MagicMock, mocker: MockerFixture
    ) -> None:
        run = create_model(
            PipelineRun,
            status=RunStatus.running,
            total_batches=1,
            started_at=timezone.now(),
        )
        finalize_run_mock = mocker.patch(
            "app.pipeline.fan_collect.services.finalize_run"
        )

        _collect_batch(
            FanBatch(
                run_id=run.id,
                fans=[
                    FanRecord(
                        dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"
                    )
                ],
            )
        )

        finalize_run_mock.assert_called_once_with(
            run.id, finished_reason=RunFinishedReason.done
        )

    # ── collect ───────────────────────────────────────────────────────────────

    @pytest.mark.db
    def test_collect_happy_path_delegates_to_collect_batch(
        self, collect_batch_mock: mock.MagicMock, mocker: MockerFixture
    ) -> None:
        mocker.patch("app.pipeline.fan_collect._wait_while_pipeline_paused")
        mocker.patch("app.pipeline.fan_collect._wait_while_client_paused")
        client = create_model(DSPClient)
        run = create_model(
            PipelineRun,
            status=RunStatus.running,
            dsp_client_id=client.id,
            started_at=timezone.now(),
        )
        batch = FanBatch(
            run_id=run.id,
            fans=[
                FanRecord(dsp_user_id="user_1", refresh_token_encrypted="enc:abc123")
            ],
        )

        result = collect(batch)

        collect_batch_mock.assert_called_once_with(batch)
        assert result.processed == 1

    @pytest.mark.db
    def test_collect_raises_pipeline_wait_timeout_when_run_stays_paused(
        self, mocker: MockerFixture
    ) -> None:
        run = create_model(PipelineRun, status=RunStatus.paused)
        mocker.patch("app.pipeline.fan_collect.time.sleep")
        tick = count(9999)
        mocker.patch(
            "app.pipeline.fan_collect.time.monotonic", side_effect=lambda: next(tick)
        )

        with (
            pytest.raises(PipelineWaitTimeoutError),
            override_settings(
                fan_collect_time_limit_s=1, fan_collect_timeout_buffer_s=0
            ),
        ):
            collect(
                FanBatch(
                    run_id=run.id,
                    fans=[
                        FanRecord(
                            dsp_user_id="user_1",
                            refresh_token_encrypted="enc:abc123",
                        )
                    ],
                )
            )

    @pytest.mark.db
    def test_collect_skips_client_wait_when_run_not_found(
        self, collect_batch_mock: mock.MagicMock, mocker: MockerFixture
    ) -> None:
        mocker.patch("app.pipeline.fan_collect._wait_while_pipeline_paused")
        wait_client = mocker.patch("app.pipeline.fan_collect._wait_while_client_paused")

        collect(
            FanBatch(
                run_id=uuid.uuid4(),
                fans=[
                    FanRecord(
                        dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"
                    )
                ],
            )
        )

        wait_client.assert_not_called()

    @pytest.mark.db
    def test_collect_waits_while_client_paused_before_collecting(
        self, collect_batch_mock: mock.MagicMock, mocker: MockerFixture
    ) -> None:
        client = create_model(DSPClient, status=DSPClientStatus.active)
        run = create_model(
            PipelineRun,
            status=RunStatus.running,
            dsp_client_id=client.id,
            started_at=timezone.now(),
        )
        mocker.patch("app.pipeline.fan_collect._wait_while_pipeline_paused")
        wait_client = mocker.patch("app.pipeline.fan_collect._wait_while_client_paused")

        collect(
            FanBatch(
                run_id=run.id,
                fans=[
                    FanRecord(
                        dsp_user_id="user_1", refresh_token_encrypted="enc:abc123"
                    )
                ],
            )
        )

        wait_client.assert_called_once()
        assert wait_client.call_args[0][0].id == client.id