"""Fan collection — per-fan DSP API calls and resource execution."""

import logging
from collections.abc import Callable
from datetime import datetime, timedelta
from typing import Any, NamedTuple

from fansifter_common.utils import timezone
from pydantic import SecretStr

from resonance_engine.adapters.db import db, is_conflict
from resonance_engine.config import settings
from resonance_engine.dsp.enums import DSPResource
from resonance_engine.dsp.exceptions import (
    DSPError,
    DSPForbiddenError,
    DSPResourceUnsupportedError,
    RateLimitError,
    TokenRefreshError,
    TokenRevokedError,
)
from resonance_engine.dsp.gateway import dsp_gateway
from resonance_engine.dsp.models import DSPClient
from resonance_engine.fandata.enums import FanCollectionError
from resonance_engine.fandata.models import FanCollectionState, FanConnection
from resonance_engine.fandata.sink import data_sink
from resonance_engine.fandata.timings import CollectTimings
from resonance_engine.fandata.types import FanRecord
from resonance_engine.utils.encrypter import decrypt, encrypt

logger = logging.getLogger(__name__)


class FanOutcome(NamedTuple):
    processed: int
    errors: int
    stale_tokens: int
    rate_limited: int
    requests: int
    requests_skipped: int = 0
    reset_consecutive: bool = True
    stop: bool = False
    timed_out: bool = False


class FanBatchOutcome(NamedTuple):
    fans_processed: int
    fans_errors: int
    fans_stale_tokens: int
    fans_skipped: int
    requests: int
    requests_rate_limited: int
    requests_skipped: int


class _ResourceCall(NamedTuple):
    resource: DSPResource
    ms: list[int]
    interval_s: int
    fetch: Callable[[], Any]
    write: Callable[[Any], None]


class _Progress(NamedTuple):
    success: bool
    requests: int
    requests_skipped: int


def collect_fans(
    *, fans: list[FanRecord], client: DSPClient, force: bool = False
) -> FanBatchOutcome:
    """Collect a batch of fans sequentially, returning aggregated outcome counters."""
    started_at = timezone.now()

    with db.autocommit():
        states = FanCollectionState.query.bulk_get(
            fan_ids=[f.fan_id for f in fans],
            dsp_id=client.dsp_id,
        )

    due: list[tuple[FanRecord, FanCollectionState | None]] = []
    skipped = 0
    for fan in fans:
        state = states.get(fan.fan_id)
        if not force and not _any_resource_due(state, started_at):
            skipped += 1
            continue
        due.append((fan, state))

    if skipped:
        logger.debug(
            "collect_fans: %d/%d fans skipped (nothing due)",
            skipped,
            len(fans),
        )

    fans_processed = fans_errors = fans_stale_tokens = requests_rate_limited = 0
    requests = requests_skipped = 0
    budget_s = (
        settings.fan_collect_worker_timeout_s - settings.fan_collect_batch_margin_s
    )
    budget = timedelta(seconds=budget_s)
    for i, (fan, state) in enumerate(due):
        if timezone.now() - started_at >= budget:
            logger.warning(
                "collect_fans: batch budget %ds reached — deferring %d/%d fans",
                budget_s,
                len(due) - i,
                len(due),
                extra={
                    "dsp_client_name": client.name,
                },
            )
            break
        try:
            outcome = collect_fan(
                fan=fan,
                client=client,
                state=state,
                started_at=started_at,
                force=force,
            )
        except Exception as exc:
            # One fan must never sink the whole batch. Count it and move on.
            if is_conflict(exc):
                # Transient OCC conflict, not a real failure — warn, don't page.
                logger.warning(
                    "collect_fans: fan %s — serialization conflict on %s, deferring",
                    fan.fan_id,
                    client.name,
                    extra={"dsp_client_name": client.name},
                )
            else:
                logger.exception(
                    "collect_fans: fan %s failed on %s — counting as error",
                    fan.fan_id,
                    client.name,
                    extra={"dsp_client_name": client.name},
                )
            fans_errors += 1
            continue
        fans_processed += outcome.processed
        fans_errors += outcome.errors
        fans_stale_tokens += outcome.stale_tokens
        requests_rate_limited += outcome.rate_limited
        requests += outcome.requests
        requests_skipped += outcome.requests_skipped
    return FanBatchOutcome(
        fans_processed=fans_processed,
        fans_errors=fans_errors,
        fans_stale_tokens=fans_stale_tokens,
        requests_rate_limited=requests_rate_limited,
        requests=requests,
        fans_skipped=skipped,
        requests_skipped=requests_skipped,
    )


def collect_fan(
    *,
    fan: FanRecord,
    client: DSPClient,
    state: FanCollectionState | None = None,
    timings: CollectTimings | None = None,
    started_at: datetime | None = None,
    force: bool = False,
) -> FanOutcome:
    """Collect one fan: HTTP + sink flush outside any DB transaction (so a slow
    fan can't hold a tx open over network I/O and trip DSQL's tx time limit),
    then a short transactional state/token write.

    ``state`` skips the per-fan DB read; ``force`` re-runs every resource regardless
    of its interval (manual one-off collects)."""
    collector = _FanCollector(
        fan=fan,
        client=client,
        state=state,
        timings=timings or CollectTimings(),
        started_at=started_at or timezone.now(),
        force=force,
    )
    outcome = collector.run()
    data_sink.flush()

    @db.transaction
    def _persist() -> None:
        collector.persist()

    _persist()
    return outcome


class _FanCollector:
    def __init__(
        self,
        *,
        fan: FanRecord,
        client: DSPClient,
        state: FanCollectionState | None = None,
        timings: CollectTimings,
        started_at: datetime,
        force: bool = False,
    ) -> None:
        self.fan = fan
        self.client = client
        self._state = state  # pre-fetched by caller; None means fetch lazily
        self.timings = timings
        self.started_at = started_at
        self.force = force
        self.collected_at: datetime
        self.resources_done: dict[DSPResource, datetime] = {}
        # Deferred DB writes — recorded during run() (HTTP phase), applied in
        # persist() inside a short transaction so no HTTP is held open over a tx.
        self._pending_token_encrypted: str | None = None
        self._revoke = False
        self._error: FanCollectionError | None = None
        self.log_extra = {
            "dsp_client_name": client.name,
            "dsp_id": client.dsp_id,
            "fan_id": fan.fan_id,
        }

    def run(self) -> FanOutcome:
        """Collect one fan over HTTP, buffering sink writes and recording the
        intended DB mutations. Makes no DB writes — call persist() for those."""
        self.collected_at = timezone.now()
        try:
            progress = self._process()
            self._error = None if progress.success else FanCollectionError.api_error
            logger.debug(
                "Fan %s — %s: requests_skipped=%d api=%d",
                self.fan.fan_id,
                "ok" if progress.success else "api_error",
                progress.requests_skipped,
                progress.requests,
                extra=self.log_extra,
            )
            return FanOutcome(
                processed=int(progress.success),
                errors=int(not progress.success),
                stale_tokens=0,
                rate_limited=0,
                requests_skipped=progress.requests_skipped,
                requests=progress.requests,
            )
        except TokenRevokedError:
            self._revoke = True
            self._error = FanCollectionError.token_error
            logger.warning(
                "Fan %s — token revoked on %s",
                self.fan.fan_id,
                self.client.name,
                extra=self.log_extra,
            )
            return FanOutcome(
                processed=0,
                errors=0,
                stale_tokens=1,
                rate_limited=0,
                requests=1,
            )
        except TokenRefreshError as exc:
            self._error = FanCollectionError.token_error
            logger.warning(
                "Fan %s — token refresh failed on %s: %s",
                self.fan.fan_id,
                self.client.name,
                exc,
                extra=self.log_extra,
            )
            return FanOutcome(
                processed=0,
                errors=1,
                stale_tokens=0,
                rate_limited=0,
                requests=1,
            )
        except RateLimitError as exc:
            self._error = FanCollectionError.rate_limited
            logger.warning(
                "Fan %s — rate limited on %s: %s",
                self.fan.fan_id,
                self.client.name,
                exc,
                extra=self.log_extra,
            )
            return FanOutcome(
                processed=0,
                errors=0,
                stale_tokens=0,
                rate_limited=1,
                requests=1,
                reset_consecutive=False,
            )

    def persist(self) -> None:
        """Apply the DB writes recorded by run(). Call inside a transaction —
        it holds no HTTP, so the transaction stays short."""
        if self._revoke:
            FanConnection.query.mark_revoked(
                fan_id=self.fan.fan_id,
                dsp_id=self.client.dsp_id,
                dsp_client_id=self.client.id,
            )
        elif self._pending_token_encrypted is not None:
            FanConnection.query.update_token(
                fan_id=self.fan.fan_id,
                dsp_id=self.client.dsp_id,
                dsp_client_id=self.client.id,
                token_encrypted=self._pending_token_encrypted,
            )
        self._record_state(error=self._error)

    def _record_state(self, *, error: FanCollectionError | None) -> None:
        FanCollectionState.query.record(
            fan_id=self.fan.fan_id,
            dsp_id=self.client.dsp_id,
            dsp_client_id=self.client.id,
            collected_at=self.collected_at,
            error=error,
            resources=self.resources_done,
        )

    def _process(self) -> _Progress:
        if not self.fan.token_encrypted:
            logger.warning(
                "Fan %s — no token, skipping",
                self.fan.fan_id,
                extra=self.log_extra,
            )
            return _Progress(success=False, requests_skipped=0, requests=0)

        # Use the pre-fetched state when available; otherwise fetch inside the
        # transaction (e.g. when collect_fan is called directly, not via collect_fans).
        state = self._state
        if state is None:
            with db.autocommit():
                state = FanCollectionState.query.where(
                    FanCollectionState.fan_id == self.fan.fan_id,
                    FanCollectionState.dsp_id == self.client.dsp_id,
                ).one_or_none()

        if not self.force and not _any_resource_due(state, self.started_at):
            logger.debug(
                "Fan %s — nothing due, skipping token refresh",
                self.fan.fan_id,
                extra=self.log_extra,
            )
            return _Progress(
                success=True, requests_skipped=len(DSPResource), requests=0
            )

        try:
            with self.timings.measure(self.timings.token_ms):
                token = dsp_gateway.refresh_token(
                    client_name=self.client.name,
                    refresh_token=SecretStr(decrypt(self.fan.token_encrypted)),
                )
        except TokenRefreshError, RateLimitError:
            raise
        except DSPError as exc:
            logger.warning(
                "Fan %s — token refresh failed: %s",
                self.fan.fan_id,
                exc,
                extra=self.log_extra,
            )
            return _Progress(success=False, requests_skipped=0, requests=1)
        except Exception as exc:
            # A non-DSP failure (e.g. raw network/decrypt error) still made a
            # request; count it so it isn't lost in the batch catch-all.
            logger.warning(
                "Fan %s — token refresh errored: %s",
                self.fan.fan_id,
                exc,
                exc_info=exc,
                extra=self.log_extra,
            )
            return _Progress(success=False, requests_skipped=0, requests=1)

        if "refresh_token" in token:
            self._pending_token_encrypted = encrypt(
                token["refresh_token"].get_secret_value()
            )

        logger.debug("Fan %s — token ok", self.fan.fan_id, extra=self.log_extra)

        resources_progress = self._process_resource_calls(
            state,
            access_token=token["access_token"],
            scope=token.get("scope"),
        )
        return _Progress(
            success=resources_progress.success,
            requests_skipped=resources_progress.requests_skipped,
            requests=1 + resources_progress.requests,
        )

    def _process_resource_calls(
        self,
        state: FanCollectionState | None,
        access_token: SecretStr,
        scope: str | None = None,
    ) -> _Progress:
        after_ms = (
            int(state.recently_played_collected_at.timestamp() * 1000)
            if state and state.recently_played_collected_at is not None
            else None
        )
        resource_calls = [
            _ResourceCall(
                resource=DSPResource.profile,
                ms=self.timings.profile_ms,
                interval_s=settings.fan_collect_profile_interval_s,
                fetch=lambda: dsp_gateway.get_profile(
                    self.client.name, access_token, scope=scope
                ),
                write=lambda data: data_sink.write_fans(
                    fan=self.fan,
                    dsp_id=self.client.dsp_id,
                    data=data,
                ),
            ),
            _ResourceCall(
                resource=DSPResource.top_artists,
                ms=self.timings.top_artists_ms,
                interval_s=settings.fan_collect_top_artists_interval_s,
                fetch=lambda: dsp_gateway.get_top_artists(
                    self.client.name, access_token, scope=scope
                ),
                write=lambda data: data_sink.write_fan_top_artists(
                    fan=self.fan,
                    dsp_id=self.client.dsp_id,
                    items=data,
                ),
            ),
            _ResourceCall(
                resource=DSPResource.top_tracks,
                ms=self.timings.top_tracks_ms,
                interval_s=settings.fan_collect_top_tracks_interval_s,
                fetch=lambda: dsp_gateway.get_top_tracks(
                    self.client.name, access_token, scope=scope
                ),
                write=lambda data: data_sink.write_fan_top_tracks(
                    fan=self.fan,
                    dsp_id=self.client.dsp_id,
                    items=data,
                ),
            ),
            _ResourceCall(
                resource=DSPResource.recently_played,
                ms=self.timings.recently_played_ms,
                interval_s=settings.fan_collect_recently_played_interval_s,
                fetch=lambda: dsp_gateway.get_recently_played(
                    self.client.name, access_token, after=after_ms, scope=scope
                ),
                write=lambda items: data_sink.write_fan_recently_played(
                    fan=self.fan,
                    dsp_id=self.client.dsp_id,
                    items=items,
                ),
            ),
            _ResourceCall(
                resource=DSPResource.playlists,
                ms=self.timings.playlists_ms,
                interval_s=settings.fan_collect_playlists_interval_s,
                fetch=lambda: dsp_gateway.get_playlists(
                    self.client.name, access_token, scope=scope
                ),
                write=lambda items: data_sink.write_fan_playlists(
                    fan=self.fan,
                    dsp_id=self.client.dsp_id,
                    items=items,
                ),
            ),
            _ResourceCall(
                resource=DSPResource.saved_albums,
                ms=self.timings.saved_albums_ms,
                interval_s=settings.fan_collect_saved_albums_interval_s,
                fetch=lambda: dsp_gateway.get_saved_albums(
                    self.client.name, access_token, scope=scope
                ),
                write=lambda items: data_sink.write_fan_saved_albums(
                    fan=self.fan,
                    dsp_id=self.client.dsp_id,
                    items=items,
                ),
            ),
            _ResourceCall(
                resource=DSPResource.saved_tracks,
                ms=self.timings.saved_tracks_ms,
                interval_s=settings.fan_collect_saved_tracks_interval_s,
                fetch=lambda: dsp_gateway.get_saved_tracks(
                    self.client.name, access_token, scope=scope
                ),
                write=lambda items: data_sink.write_fan_saved_tracks(
                    fan=self.fan,
                    dsp_id=self.client.dsp_id,
                    items=items,
                ),
            ),
            _ResourceCall(
                resource=DSPResource.followed_artists,
                ms=self.timings.followed_artists_ms,
                interval_s=settings.fan_collect_followed_artists_interval_s,
                fetch=lambda: dsp_gateway.get_followed_artists(
                    self.client.name, access_token, scope=scope
                ),
                write=lambda items: data_sink.write_fan_followed_artists(
                    fan=self.fan,
                    dsp_id=self.client.dsp_id,
                    items=items,
                ),
            ),
        ]

        requests_skipped = requests = 0
        for call in resource_calls:
            last_at = _resource_last_at(state, call.resource)
            if not self.force and not _resource_needed(
                last_at, self.started_at, call.interval_s
            ):
                logger.debug(
                    "Fan %s — %s: skip",
                    self.fan.fan_id,
                    call.resource,
                    extra=self.log_extra,
                )
                requests_skipped += 1
                continue
            logger.debug(
                "Fan %s — %s: run",
                self.fan.fan_id,
                call.resource,
                extra=self.log_extra,
            )
            try:
                with self.timings.measure(call.ms):
                    result = call.fetch()
            except RateLimitError:
                raise
            except DSPForbiddenError as exc:
                level = logging.WARNING if exc.request_made else logging.DEBUG
                # Resource forbidden for this token — skip it, keep the fan going.
                # A real 403 made a request; a proactive scope pre-skip did not.
                logger.log(
                    level,
                    "Fan %s — %s forbidden, skipping resource: %s (granted scopes: %s)",
                    self.fan.fan_id,
                    call.resource,
                    exc,
                    scope or "unknown",
                    extra={
                        **self.log_extra,
                        "request_made": exc.request_made,
                    },
                )
                if exc.request_made:
                    requests += 1
                else:
                    requests_skipped += 1
                # Stamp it so a forbidden resource isn't perpetually due — else the
                # fan is re-selected and re-attempted every tick (a real 403 even
                # wastes a request each time). Retries once its interval elapses, in
                # case the token re-grants the scope. Not a token failure, so no
                # consecutive_failures / revoke.
                self.resources_done[call.resource] = self.collected_at
                continue
            except DSPResourceUnsupportedError:
                # DSP has no endpoint for this resource — no HTTP was made. Stamp it
                # so it isn't re-attempted every tick.
                logger.debug(
                    "Fan %s — %s unsupported by %s, skipping resource",
                    self.fan.fan_id,
                    call.resource,
                    self.client.name,
                    extra=self.log_extra,
                )
                requests_skipped += 1
                self.resources_done[call.resource] = self.collected_at
                continue
            except TokenRevokedError:
                # Token revoked mid-collection (e.g. Deezer surfaces it on a resource
                # call, not refresh) — propagate so run() marks the fan stale.
                raise
            except DSPError as exc:
                requests += 1
                logger.warning(
                    "Fan %s — %s fetch failed",
                    self.fan.fan_id,
                    call.resource,
                    extra=self.log_extra,
                    exc_info=exc,
                )
                return _Progress(
                    success=False,
                    requests_skipped=requests_skipped,
                    requests=requests,
                )
            requests += 1
            call.write(result)
            self.resources_done[call.resource] = self.collected_at
        return _Progress(
            success=True, requests_skipped=requests_skipped, requests=requests
        )


def _resource_last_at(
    state: FanCollectionState | None, resource: DSPResource
) -> datetime | None:
    if state is None:
        return None
    return getattr(state, f"{resource.value}_collected_at")


def _resource_needed(
    last_at: datetime | None, started_at: datetime, interval_s: int
) -> bool:
    """Return True if the resource should be (re-)fetched."""
    if last_at is None:
        return True
    if last_at >= started_at:
        return False  # already done in this run
    if interval_s > 0:
        return last_at < timezone.now() - timedelta(seconds=interval_s)
    return True


def _any_resource_due(state: FanCollectionState | None, started_at: datetime) -> bool:
    """True if at least one collection resource is due. Used to skip token refresh
    when there's nothing to do for this fan in the current tick."""
    if state is None:
        return True
    intervals: list[tuple[DSPResource, int]] = [
        (DSPResource.profile, settings.fan_collect_profile_interval_s),
        (DSPResource.top_artists, settings.fan_collect_top_artists_interval_s),
        (DSPResource.top_tracks, settings.fan_collect_top_tracks_interval_s),
        (
            DSPResource.recently_played,
            settings.fan_collect_recently_played_interval_s,
        ),
        (DSPResource.playlists, settings.fan_collect_playlists_interval_s),
        (DSPResource.saved_albums, settings.fan_collect_saved_albums_interval_s),
        (DSPResource.saved_tracks, settings.fan_collect_saved_tracks_interval_s),
        (
            DSPResource.followed_artists,
            settings.fan_collect_followed_artists_interval_s,
        ),
    ]
    for resource, interval_s in intervals:
        if _resource_needed(_resource_last_at(state, resource), started_at, interval_s):
            return True
    return False