"""Fan collection — pipeline orchestration, DSP API calls, run finalization."""

import logging
import random
import time
import uuid
from collections.abc import Callable
from datetime import datetime, timedelta
from typing import NamedTuple

from fansifter_common.utils import timezone
from psycopg.errors import SerializationFailure
from sqlalchemy.exc import OperationalError

from app.adapters.db import db
from app.config import settings
from app.core.encrypter import decrypt, encrypt
from app.dsp.enums import DSPClientName, DSPClientStatus
from app.dsp.exceptions import (
    DSPError,
    RateLimitError,
    TokenRefreshError,
    TokenRevokedError,
)
from app.dsp.gateway import dsp_gateway
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, FanCredentials
from app.fandata.sinks import data_sink
from app.fandata.types import FanBatch, FanRecord, FanStepState
from app.pipeline import services
from app.pipeline.enums import RunFinishedReason, RunStatus
from app.pipeline.exceptions import PipelineWaitTimeoutError
from app.pipeline.models import PipelineRun
from app.pipeline.timings import FanTimings

logger = logging.getLogger(__name__)

_PIPELINE_WAIT_POLL_INTERVAL_S = 2
_CLIENT_WAIT_POLL_INTERVAL_S = 2
_OCC_MAX_RETRIES = 5


def _occ_retry[T](fn: Callable[[], T]) -> T:
    """Retry fn on DSQL OCC serialization conflicts (OC000) with exponential backoff."""
    for attempt in range(_OCC_MAX_RETRIES):
        try:
            return fn()
        except OperationalError as exc:
            if (
                not isinstance(exc.__cause__, SerializationFailure)
                or attempt == _OCC_MAX_RETRIES - 1
            ):
                raise
            delay = 0.05 * (2**attempt) + random.uniform(0, 0.01)
            logger.warning(
                "OCC conflict (attempt %d/%d) — retrying in %.0fms",
                attempt + 1,
                _OCC_MAX_RETRIES,
                delay * 1000,
            )
            time.sleep(delay)
    raise RuntimeError("unreachable")


class CollectResult(NamedTuple):
    processed: int
    errors: int
    rate_limited: int
    stale_tokens: int
    timed_out: bool
    circuit_broken: bool = False
    steps_skipped: int = 0
    api_requests: int = 0


_SKIP_RESULT = CollectResult(
    processed=0,
    errors=0,
    rate_limited=0,
    stale_tokens=0,
    timed_out=False,
)


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------


def collect(batch: FanBatch) -> CollectResult:
    """Wait if the pipeline run or DSP client is paused, then collect fans per-fan transactions."""
    logger.info(
        "Batch received: run_id=%s fans=%d",
        batch.run_id,
        len(batch.fans),
        extra={"run_id": str(batch.run_id), "fans": len(batch.fans)},
    )
    _wait_while_pipeline_paused(batch.run_id)
    with db.autocommit():
        run = PipelineRun.query.where(PipelineRun.id == batch.run_id).one_or_none()
    if run is not None:
        _wait_while_client_paused(run.dsp_client)
    dsp_gateway.warm_up()  # force lazy-proxy init before entering transaction
    try:
        return _collect_batch(batch)
    except Exception:
        logger.exception(
            "Batch failed unexpectedly: run_id=%s fans=%d — counting as errors",
            batch.run_id,
            len(batch.fans),
            extra={"run_id": str(batch.run_id), "errors": len(batch.fans)},
        )
        try:
            with db.transaction():
                PipelineRun.query.increment_batch_result(
                    batch.run_id,
                    processed=0,
                    errors=len(batch.fans),
                    rate_limited=0,
                    stale_tokens=0,
                )
        except Exception:
            logger.exception(
                "Failed to record batch error result for run_id=%s",
                batch.run_id,
                extra={"run_id": str(batch.run_id)},
            )
        return CollectResult(
            processed=0,
            errors=len(batch.fans),
            rate_limited=0,
            stale_tokens=0,
            timed_out=False,
        )


# ---------------------------------------------------------------------------
# Pipeline orchestration
# ---------------------------------------------------------------------------


def _wait_while_client_paused(client: DSPClient) -> None:
    """Wait until the DSP client is active.

    Must be called outside a transaction — uses autocommit to always see the
    latest committed status regardless of any surrounding session state.
    """
    while True:
        with db.autocommit():
            paused = (
                DSPClient.query.where(
                    DSPClient.id == client.id,
                    DSPClient.status == DSPClientStatus.paused,
                ).first()
                is not None
            )
        if not paused:
            return
        logger.info(
            "DSP client %s — paused, sleeping %ds",
            client.name,
            _CLIENT_WAIT_POLL_INTERVAL_S,
            extra={"dsp_client": client.name},
        )
        time.sleep(_CLIENT_WAIT_POLL_INTERVAL_S)


def _wait_while_pipeline_paused(run_id: uuid.UUID) -> None:
    deadline_ms = (
        settings.fan_collect_time_limit_s - settings.fan_collect_timeout_buffer_s
    ) * 1000
    started_at = time.monotonic()

    while True:
        with db.autocommit():
            run = PipelineRun.query.where(PipelineRun.id == run_id).one_or_none()
        if run is None or run.status != RunStatus.paused:
            return
        elapsed_ms = (time.monotonic() - started_at) * 1000
        if elapsed_ms + _PIPELINE_WAIT_POLL_INTERVAL_S * 1000 >= deadline_ms:
            logger.warning(
                "Run %s — timed out waiting to resume",
                run_id,
                extra={"run_id": str(run_id)},
            )
            raise PipelineWaitTimeoutError(run_id)
        logger.debug(
            "Run %s — paused, sleeping %ds",
            run_id,
            _PIPELINE_WAIT_POLL_INTERVAL_S,
            extra={"run_id": str(run_id)},
        )
        time.sleep(_PIPELINE_WAIT_POLL_INTERVAL_S)


def _collect_batch(batch: FanBatch) -> CollectResult:
    with db.autocommit():
        run = PipelineRun.query.where(PipelineRun.id == batch.run_id).first()
    if run is None or run.status != RunStatus.running:
        logger.info(
            "Batch skipped — run %s is %s",
            batch.run_id,
            run.status if run else "not found",
            extra={"run_id": str(batch.run_id)},
        )
        return _SKIP_RESULT

    run_started_at = run.started_at or run.created_at
    result = _collect_fans(batch, run.dsp_client_id, run_started_at)
    data_sink.flush()

    # Re-read status — run may have been cancelled while fans were being collected
    with db.autocommit():
        run = PipelineRun.query.where(PipelineRun.id == batch.run_id).first()
    if run is None or run.status not in (RunStatus.running, RunStatus.paused):
        logger.info(
            "Batch result discarded — run %s is now %s",
            batch.run_id,
            run.status if run else "not found",
            extra={"run_id": str(batch.run_id)},
        )
        return _SKIP_RESULT

    completed, total = _occ_retry(lambda: _commit_batch_result(batch.run_id, result))

    logger.info(
        "Batch done: run_id=%s processed=%d errors=%d rate_limited=%d steps_skipped=%d api_requests=%d timed_out=%s circuit_broken=%s completed=%d/%d",
        batch.run_id,
        result.processed,
        result.errors,
        result.rate_limited,
        result.steps_skipped,
        result.api_requests,
        result.timed_out,
        result.circuit_broken,
        completed,
        total,
        extra={
            "run_id": str(batch.run_id),
            "processed": result.processed,
            "errors": result.errors,
            "rate_limited": result.rate_limited,
            "steps_skipped": result.steps_skipped,
            "api_requests": result.api_requests,
        },
    )
    return result


def _commit_batch_result(run_id: uuid.UUID, result: CollectResult) -> tuple[int, int]:
    with db.transaction():
        completed, total = PipelineRun.query.increment_batch_result(
            run_id,
            processed=result.processed,
            errors=result.errors,
            rate_limited=result.rate_limited,
            stale_tokens=result.stale_tokens,
            steps_skipped=result.steps_skipped,
            api_requests=result.api_requests,
        )
        all_batches_done = total > 0 and completed >= total

        if result.circuit_broken or all_batches_done:
            if result.circuit_broken:
                finished_reason = RunFinishedReason.circuit_breaker
            elif result.timed_out:
                finished_reason = RunFinishedReason.timed_out
            else:
                finished_reason = RunFinishedReason.done
            services.finalize_run(run_id, finished_reason=finished_reason)
    return completed, total


# ---------------------------------------------------------------------------
# Fan collection
# ---------------------------------------------------------------------------


def _collect_fans(
    batch: FanBatch, dsp_client_id: int, run_started_at: datetime
) -> CollectResult:
    with db.autocommit():
        client = DSPClient.query.where(DSPClient.id == dsp_client_id).one_or_none()
    if client is None:
        logger.error("DSP client not found — skipping batch")
        return CollectResult(
            processed=0, errors=0, rate_limited=0, stale_tokens=0, timed_out=False
        )

    fans = batch.fans
    if settings.fan_collect_cooldown_s > 0:
        cutoff = timezone.now() - timedelta(seconds=settings.fan_collect_cooldown_s)
        with db.autocommit():
            skip_ids = FanCollectionState.query.recently_collected_ids(
                dsp_user_ids=[f.dsp_user_id for f in fans],
                dsp_id=client.dsp_id,
                since=cutoff,
            )
        if skip_ids:
            fans = [f for f in fans if f.dsp_user_id not in skip_ids]
            logger.info(
                "Cooldown: skipping %d/%d recently collected fans",
                len(skip_ids),
                len(batch.fans),
                extra={
                    "dsp_client": client.name,
                    "skipped": len(skip_ids),
                    "total": len(batch.fans),
                },
            )

    deadline_ms = (
        settings.fan_collect_time_limit_s - settings.fan_collect_timeout_buffer_s
    ) * 1000
    started_at = time.monotonic()
    processed = errors = rate_limited = stale_tokens = steps_skipped = api_requests = 0
    timed_out = False
    consecutive_rate_limits = 0
    timings = FanTimings()

    logger.debug(
        "Fan loop start: %d fans, client=%s, deadline=%.0fs",
        len(fans),
        client.name,
        deadline_ms / 1000,
        extra={"dsp_client": client.name, "fans": len(fans)},
    )

    for i, fan in enumerate(fans):
        elapsed_ms = (time.monotonic() - started_at) * 1000
        if elapsed_ms >= deadline_ms:
            logger.warning(
                "Timeout approaching (%.0fms elapsed) — stopping batch early",
                elapsed_ms,
                extra={"dsp_client": client.name, "elapsed_ms": round(elapsed_ms)},
            )
            timed_out = True
            break
        outcome = _occ_retry(
            lambda f=fan: _call_fan_in_transaction(f, client, timings, run_started_at)
        )
        processed += outcome.processed
        errors += outcome.errors
        stale_tokens += outcome.stale_tokens
        rate_limited += outcome.rate_limited
        steps_skipped += outcome.steps_skipped
        api_requests += outcome.api_requests
        logger.debug(
            "Fan %s [%d/%d] — processed=%d errors=%d rate_limited=%d skipped=%d api=%d",
            fan.dsp_user_id,
            i + 1,
            len(fans),
            outcome.processed,
            outcome.errors,
            outcome.rate_limited,
            outcome.steps_skipped,
            outcome.api_requests,
            extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
        )
        if outcome.reset_consecutive:
            consecutive_rate_limits = 0
        else:
            consecutive_rate_limits += 1
        if (
            outcome.stop
            or consecutive_rate_limits
            >= settings.fan_collect_rate_limit_circuit_breaker_threshold
        ):
            timed_out = outcome.timed_out
            if (
                consecutive_rate_limits
                >= settings.fan_collect_rate_limit_circuit_breaker_threshold
            ):
                logger.warning(
                    "Circuit breaker: %d consecutive rate limits — stopping batch at fan %d/%d",
                    consecutive_rate_limits,
                    i + 1,
                    len(fans),
                    extra={
                        "dsp_client": client.name,
                        "consecutive_rate_limits": consecutive_rate_limits,
                    },
                )
            break

    circuit_broken = (
        consecutive_rate_limits
        >= settings.fan_collect_rate_limit_circuit_breaker_threshold
    )
    timings.log_summary(processed, errors, dsp_client=client.name)
    return CollectResult(
        processed=processed,
        errors=errors,
        rate_limited=rate_limited,
        stale_tokens=stale_tokens,
        timed_out=timed_out,
        circuit_broken=circuit_broken,
        steps_skipped=steps_skipped,
        api_requests=api_requests,
    )


class _FanOutcome(NamedTuple):
    processed: int = 0
    errors: int = 0
    stale_tokens: int = 0
    rate_limited: int = 0
    steps_skipped: int = 0
    api_requests: int = 0
    reset_consecutive: bool = True
    stop: bool = False
    timed_out: bool = False


def _call_fan_in_transaction(
    fan: FanRecord,
    client: DSPClient,
    timings: FanTimings,
    run_started_at: datetime,
) -> _FanOutcome:
    with db.transaction():
        return _call_fan(fan, client, timings, run_started_at)


def _call_fan(
    fan: FanRecord,
    client: DSPClient,
    timings: FanTimings,
    run_started_at: datetime,
) -> _FanOutcome:
    collected_at = timezone.now()
    try:
        success, steps_skipped, api_requests = _process_fan(
            fan=fan,
            client=client,
            timings=timings,
            collected_at=collected_at,
            run_started_at=run_started_at,
        )
        status = (
            FanCollectionStatus.success if success else FanCollectionStatus.api_error
        )
        FanCollectionState.query.record(
            dsp_user_id=fan.dsp_user_id,
            dsp_id=client.dsp_id,
            dsp_client_id=client.id,
            status=status,
            collected_at=collected_at,
        )
        if success:
            logger.debug(
                "Fan %s — ok: steps_skipped=%d api=%d",
                fan.dsp_user_id,
                steps_skipped,
                api_requests,
                extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
            )
        else:
            logger.debug(
                "Fan %s — api_error: steps_skipped=%d api=%d",
                fan.dsp_user_id,
                steps_skipped,
                api_requests,
                extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
            )
        return _FanOutcome(
            processed=int(success),
            errors=int(not success),
            steps_skipped=steps_skipped,
            api_requests=api_requests,
        )
    except TokenRevokedError:
        FanCredentials.query.mark_revoked(
            fan.dsp_user_id, client.dsp_id, client.client_id
        )
        FanCollectionState.query.record(
            dsp_user_id=fan.dsp_user_id,
            dsp_id=client.dsp_id,
            dsp_client_id=client.id,
            status=FanCollectionStatus.token_error,
            collected_at=collected_at,
        )
        logger.warning(
            "Fan %s — token revoked on %s",
            fan.dsp_user_id,
            client.name,
            extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
        )
        return _FanOutcome(stale_tokens=1, api_requests=1)
    except TokenRefreshError as exc:
        FanCollectionState.query.record(
            dsp_user_id=fan.dsp_user_id,
            dsp_id=client.dsp_id,
            dsp_client_id=client.id,
            status=FanCollectionStatus.token_error,
            collected_at=collected_at,
        )
        logger.warning(
            "Fan %s — token refresh failed on %s: %s",
            fan.dsp_user_id,
            client.name,
            exc,
            extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
        )
        return _FanOutcome(errors=1, api_requests=1)
    except RateLimitError as exc:
        FanCollectionState.query.record(
            dsp_user_id=fan.dsp_user_id,
            dsp_id=client.dsp_id,
            dsp_client_id=client.id,
            status=FanCollectionStatus.rate_limited,
            collected_at=collected_at,
        )
        logger.warning(
            "Fan %s — rate limited on %s: %s",
            fan.dsp_user_id,
            client.name,
            exc,
            extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
        )
        return _FanOutcome(rate_limited=1, api_requests=1, reset_consecutive=False)


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


def _execute_step[T](
    *,
    fan: FanRecord,
    client: DSPClient,
    step_name: str,
    collected_at: datetime,
    needed: bool,
    fetch: Callable[[], T | None],
    write: Callable[[T], None],
) -> tuple[bool, int, int]:
    """Run or skip one collection step. Returns (ok, steps_skipped, api_requests)."""
    if not needed:
        logger.debug(
            "Fan %s — %s: skip",
            fan.dsp_user_id,
            step_name,
            extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
        )
        return True, 1, 0
    logger.debug(
        "Fan %s — %s: run",
        fan.dsp_user_id,
        step_name,
        extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
    )
    result = fetch()
    if result is None:
        logger.warning(
            "Fan %s — %s fetch failed",
            fan.dsp_user_id,
            step_name,
            extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
        )
        return False, 0, 1
    write(result)
    FanCollectionState.query.record_step(
        dsp_user_id=fan.dsp_user_id,
        dsp_id=client.dsp_id,
        dsp_client_id=client.id,
        step=step_name,
        collected_at=collected_at,
    )
    return True, 0, 1


def _process_fan_steps(
    *,
    fan: FanRecord,
    client: DSPClient,
    timings: FanTimings,
    collected_at: datetime,
    run_started_at: datetime,
    step_state: FanStepState,
    access_token: str,
) -> tuple[bool, int, int]:
    """Run all data-collection steps. Returns (success, steps_skipped, api_requests)."""
    steps_skipped = api_requests = 0

    ok, skipped, reqs = _execute_step(
        fan=fan,
        client=client,
        step_name="profile",
        collected_at=collected_at,
        needed=_step_needed(
            step_state.profile_collected_at,
            run_started_at,
            settings.fan_collect_profile_interval_s,
        ),
        fetch=lambda: _fetch_profile(
            fan=fan, client_name=client.name, access_token=access_token, timings=timings
        ),
        write=lambda r: data_sink.write_fans(
            fan=fan, dsp_id=client.dsp_id, profile=r, collected_at=collected_at
        ),
    )
    steps_skipped += skipped
    api_requests += reqs
    if not ok:
        return False, steps_skipped, api_requests

    ok, skipped, reqs = _execute_step(
        fan=fan,
        client=client,
        step_name="top_artists",
        collected_at=collected_at,
        needed=_step_needed(
            step_state.top_artists_collected_at,
            run_started_at,
            settings.fan_collect_top_artists_interval_s,
        ),
        fetch=lambda: _fetch_top_artists(
            fan=fan, client_name=client.name, access_token=access_token, timings=timings
        ),
        write=lambda r: data_sink.write_fan_top_artists(
            fan=fan, dsp_id=client.dsp_id, result=r, collected_at=collected_at
        ),
    )
    steps_skipped += skipped
    api_requests += reqs
    if not ok:
        return False, steps_skipped, api_requests

    ok, skipped, reqs = _execute_step(
        fan=fan,
        client=client,
        step_name="top_tracks",
        collected_at=collected_at,
        needed=_step_needed(
            step_state.top_tracks_collected_at,
            run_started_at,
            settings.fan_collect_top_tracks_interval_s,
        ),
        fetch=lambda: _fetch_top_tracks(
            fan=fan, client_name=client.name, access_token=access_token, timings=timings
        ),
        write=lambda r: data_sink.write_fan_top_tracks(
            fan=fan, dsp_id=client.dsp_id, result=r, collected_at=collected_at
        ),
    )
    steps_skipped += skipped
    api_requests += reqs
    if not ok:
        return False, steps_skipped, api_requests

    after_ms = (
        int(step_state.recently_played_collected_at.timestamp() * 1000)
        if step_state.recently_played_collected_at is not None
        else None
    )
    ok, skipped, reqs = _execute_step(
        fan=fan,
        client=client,
        step_name="recently_played",
        collected_at=collected_at,
        needed=_step_needed(
            step_state.recently_played_collected_at,
            run_started_at,
            settings.fan_collect_recently_played_interval_s,
        ),
        fetch=lambda: _fetch_recently_played(
            fan=fan,
            client_name=client.name,
            access_token=access_token,
            timings=timings,
            after=after_ms,
        ),
        write=lambda r: data_sink.write_fan_recently_played(
            fan=fan, dsp_id=client.dsp_id, result=r, collected_at=collected_at
        ),
    )
    steps_skipped += skipped
    api_requests += reqs
    if not ok:
        return False, steps_skipped, api_requests

    ok, skipped, reqs = _execute_step(
        fan=fan,
        client=client,
        step_name="playlists",
        collected_at=collected_at,
        needed=_step_needed(
            step_state.playlists_collected_at,
            run_started_at,
            settings.fan_collect_playlists_interval_s,
        ),
        fetch=lambda: _fetch_playlists(
            fan=fan, client_name=client.name, access_token=access_token, timings=timings
        ),
        write=lambda r: data_sink.write_fan_playlists(
            fan=fan, dsp_id=client.dsp_id, result=r, collected_at=collected_at
        ),
    )
    steps_skipped += skipped
    api_requests += reqs
    if not ok:
        return False, steps_skipped, api_requests

    ok, skipped, reqs = _execute_step(
        fan=fan,
        client=client,
        step_name="saved_albums",
        collected_at=collected_at,
        needed=_step_needed(
            step_state.saved_albums_collected_at,
            run_started_at,
            settings.fan_collect_saved_albums_interval_s,
        ),
        fetch=lambda: _fetch_saved_albums(
            fan=fan, client_name=client.name, access_token=access_token, timings=timings
        ),
        write=lambda r: data_sink.write_fan_saved_albums(
            fan=fan, dsp_id=client.dsp_id, result=r, collected_at=collected_at
        ),
    )
    steps_skipped += skipped
    api_requests += reqs
    if not ok:
        return False, steps_skipped, api_requests

    ok, skipped, reqs = _execute_step(
        fan=fan,
        client=client,
        step_name="saved_tracks",
        collected_at=collected_at,
        needed=_step_needed(
            step_state.saved_tracks_collected_at,
            run_started_at,
            settings.fan_collect_saved_tracks_interval_s,
        ),
        fetch=lambda: _fetch_saved_tracks(
            fan=fan, client_name=client.name, access_token=access_token, timings=timings
        ),
        write=lambda r: data_sink.write_fan_saved_tracks(
            fan=fan, dsp_id=client.dsp_id, result=r, collected_at=collected_at
        ),
    )
    steps_skipped += skipped
    api_requests += reqs
    if not ok:
        return False, steps_skipped, api_requests

    ok, skipped, reqs = _execute_step(
        fan=fan,
        client=client,
        step_name="followed_artists",
        collected_at=collected_at,
        needed=_step_needed(
            step_state.followed_artists_collected_at,
            run_started_at,
            settings.fan_collect_followed_artists_interval_s,
        ),
        fetch=lambda: _fetch_followed_artists(
            fan=fan, client_name=client.name, access_token=access_token, timings=timings
        ),
        write=lambda r: data_sink.write_fan_followed_artists(
            fan=fan, dsp_id=client.dsp_id, result=r, collected_at=collected_at
        ),
    )
    steps_skipped += skipped
    api_requests += reqs
    if not ok:
        return False, steps_skipped, api_requests

    return True, steps_skipped, api_requests


def _process_fan(
    fan: FanRecord,
    client: DSPClient,
    timings: FanTimings,
    collected_at: datetime,
    run_started_at: datetime,
) -> tuple[bool, int, int]:
    """Return (success, steps_skipped, api_requests)."""
    if not fan.refresh_token_encrypted:
        logger.warning(
            "Fan %s — no refresh token, skipping",
            fan.dsp_user_id,
            extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
        )
        return False, 0, 0

    step_state = FanCollectionState.query.get_step_state(
        dsp_user_id=fan.dsp_user_id, dsp_id=client.dsp_id
    )

    token = _fetch_token(
        fan=fan,
        refresh_token=decrypt(fan.refresh_token_encrypted),
        client_name=client.name,
        timings=timings,
    )
    api_requests = 1  # token refresh
    if token is None:
        logger.warning(
            "Fan %s — token fetch failed",
            fan.dsp_user_id,
            extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
        )
        return False, 0, api_requests

    if token.refresh_token:
        FanCredentials.query.update_refresh_token(
            dsp_user_id=fan.dsp_user_id,
            dsp_id=client.dsp_id,
            client_id=client.client_id,
            refresh_token_encrypted=encrypt(token.refresh_token),
        )

    logger.debug(
        "Fan %s — token ok",
        fan.dsp_user_id,
        extra={"dsp_client": client.name, "fan_id": fan.dsp_user_id},
    )

    success, steps_skipped, step_requests = _process_fan_steps(
        fan=fan,
        client=client,
        timings=timings,
        collected_at=collected_at,
        run_started_at=run_started_at,
        step_state=step_state,
        access_token=token.access_token,
    )
    return success, steps_skipped, api_requests + step_requests


# ---------------------------------------------------------------------------
# DSP API calls
# ---------------------------------------------------------------------------


def _fetch_token(
    *,
    fan: FanRecord,
    refresh_token: str,
    client_name: DSPClientName,
    timings: FanTimings,
) -> TokenResult | None:
    t0 = time.monotonic()
    try:
        token = dsp_gateway.refresh_token(client_name, refresh_token=refresh_token)
    except TokenRefreshError, RateLimitError:
        raise
    except DSPError as exc:
        logger.warning(
            "Token refresh failed for fan %s",
            fan.dsp_user_id,
            exc_info=exc,
            extra={"dsp_client": client_name, "fan_id": fan.dsp_user_id},
        )
        return None
    timings.token_ms.append(round((time.monotonic() - t0) * 1000))
    return token


def _fetch_profile(
    *,
    fan: FanRecord,
    client_name: DSPClientName,
    access_token: str,
    timings: FanTimings,
) -> ProfileResult | None:
    t0 = time.monotonic()
    try:
        result = dsp_gateway.get_profile(client_name, access_token=access_token)
    except RateLimitError:
        raise
    except DSPError:
        logger.warning(
            "Profile failed for fan %s",
            fan.dsp_user_id,
            extra={"dsp_client": client_name, "fan_id": fan.dsp_user_id},
        )
        return None
    timings.profile_ms.append(round((time.monotonic() - t0) * 1000))
    return result


def _fetch_top_artists(
    *,
    fan: FanRecord,
    client_name: DSPClientName,
    access_token: str,
    timings: FanTimings,
) -> TopArtistsResult | None:
    t0 = time.monotonic()
    try:
        result = dsp_gateway.get_top_artists(client_name, access_token=access_token)
    except RateLimitError:
        raise
    except DSPError:
        logger.warning(
            "Top_artists failed for fan %s",
            fan.dsp_user_id,
            extra={"dsp_client": client_name, "fan_id": fan.dsp_user_id},
        )
        return None
    timings.top_artists_ms.append(round((time.monotonic() - t0) * 1000))
    return result


def _fetch_top_tracks(
    *,
    fan: FanRecord,
    client_name: DSPClientName,
    access_token: str,
    timings: FanTimings,
) -> TopTracksResult | None:
    t0 = time.monotonic()
    try:
        result = dsp_gateway.get_top_tracks(client_name, access_token)
    except RateLimitError:
        raise
    except DSPError:
        logger.warning(
            "Top_tracks failed for fan %s",
            fan.dsp_user_id,
            extra={"dsp_client": client_name, "fan_id": fan.dsp_user_id},
        )
        return None
    timings.top_tracks_ms.append(round((time.monotonic() - t0) * 1000))
    return result


def _fetch_recently_played(
    *,
    fan: FanRecord,
    client_name: DSPClientName,
    access_token: str,
    timings: FanTimings,
    after: int | None = None,
) -> RecentlyPlayedResult | None:
    t0 = time.monotonic()
    try:
        result = dsp_gateway.get_recently_played(client_name, access_token, after=after)
    except RateLimitError:
        raise
    except DSPError:
        logger.warning(
            "Recently_played failed for fan %s",
            fan.dsp_user_id,
            extra={"dsp_client": client_name, "fan_id": fan.dsp_user_id},
        )
        return None
    timings.recently_played_ms.append(round((time.monotonic() - t0) * 1000))
    return result


def _fetch_playlists(
    *,
    fan: FanRecord,
    client_name: DSPClientName,
    access_token: str,
    timings: FanTimings,
) -> PlaylistsResult | None:
    t0 = time.monotonic()
    try:
        result = dsp_gateway.get_playlists(client_name, access_token)
    except RateLimitError:
        raise
    except DSPError:
        logger.warning(
            "Playlists failed for fan %s",
            fan.dsp_user_id,
            extra={"dsp_client": client_name, "fan_id": fan.dsp_user_id},
        )
        return None
    timings.playlists_ms.append(round((time.monotonic() - t0) * 1000))
    return result


def _fetch_saved_albums(
    *,
    fan: FanRecord,
    client_name: DSPClientName,
    access_token: str,
    timings: FanTimings,
) -> SavedAlbumsResult | None:
    t0 = time.monotonic()
    try:
        result = dsp_gateway.get_saved_albums(client_name, access_token)
    except RateLimitError:
        raise
    except DSPError:
        logger.warning(
            "Saved_albums failed for fan %s",
            fan.dsp_user_id,
            extra={"dsp_client": client_name, "fan_id": fan.dsp_user_id},
        )
        return None
    timings.saved_albums_ms.append(round((time.monotonic() - t0) * 1000))
    return result


def _fetch_saved_tracks(
    *,
    fan: FanRecord,
    client_name: DSPClientName,
    access_token: str,
    timings: FanTimings,
) -> SavedTracksResult | None:
    t0 = time.monotonic()
    try:
        result = dsp_gateway.get_saved_tracks(client_name, access_token)
    except RateLimitError:
        raise
    except DSPError:
        logger.warning(
            "Saved_tracks failed for fan %s",
            fan.dsp_user_id,
            extra={"dsp_client": client_name, "fan_id": fan.dsp_user_id},
        )
        return None
    timings.saved_tracks_ms.append(round((time.monotonic() - t0) * 1000))
    return result


def _fetch_followed_artists(
    *,
    fan: FanRecord,
    client_name: DSPClientName,
    access_token: str,
    timings: FanTimings,
) -> FollowedArtistsResult | None:
    t0 = time.monotonic()
    try:
        result = dsp_gateway.get_followed_artists(client_name, access_token)
    except RateLimitError:
        raise
    except DSPError:
        logger.warning(
            "Followed_artists failed for fan %s",
            fan.dsp_user_id,
            extra={"dsp_client": client_name, "fan_id": fan.dsp_user_id},
        )
        return None
    timings.followed_artists_ms.append(round((time.monotonic() - t0) * 1000))
    return result