"""Fanout planner — pick fans for each DSP to be dispatched to its SQS queue."""

import logging
import math
from dataclasses import dataclass, field
from datetime import timedelta
from typing import Literal

from fansifter_common.utils import timezone

from resonance_engine.config import settings
from resonance_engine.dsp.enums import DSPClientName, DSPClientStatus, DSPId
from resonance_engine.dsp.gateway import dsp_gateway
from resonance_engine.dsp.models import DSPClient
from resonance_engine.fandata.models import (
    FanConnection,
    FanConnectionFilter,
)
from resonance_engine.fandata.types import FanRecord
from resonance_engine.tasks.enums import FanoutSource
from resonance_engine.tasks.models import CollectTask, FanoutTask

logger = logging.getLogger(__name__)

# Target this fraction of the window, not 100% — the slack absorbs noise,
# stragglers and timeouts so the fleet doesn't chronically overrun.
FILL_SAFETY = 0.9  # warm: fraction of the window to fill
SAFETY_FACTOR = 0.85  # cold-start headroom against the DSP rate budget
REQUESTS_PER_FAN = 9  # DSP fact: refresh_token + profile + 7 data endpoints
COLD_REQUEST_LATENCY_S = 0.5  # default per-request RTT when no observed data
MIN_BATCH_SIZE = 10  # floor — prevent single-fan messages
MAX_BATCH_SIZE = 250
# Per-fan wall-clock assumed before any drain is observed (worst-case cold start).
COLD_TIME_PER_FAN_S = REQUESTS_PER_FAN * COLD_REQUEST_LATENCY_S


def get_batch_size_for(time_per_fan_s: float) -> int:
    """Fans per SQS message — `floor(batch_budget / time_per_fan)`, clamped to
    `[MIN_BATCH_SIZE, MAX_BATCH_SIZE]`. The budget is the collect Lambda's timeout
    minus its margin, so a batch finishes within the worker's runtime and **scales
    with the deployed timeout**. Uses the *observed* per-fan time, not the
    worst-case constant."""
    budget = settings.fan_collect_worker_timeout_s - settings.fan_collect_batch_margin_s
    raw = math.floor(budget / time_per_fan_s)
    return min(MAX_BATCH_SIZE, max(MIN_BATCH_SIZE, raw))


@dataclass(frozen=True, kw_only=True)
class FanoutPlan:
    fanout_count: int
    fanout_count_source: Literal["seed", "collected"]
    fanout_window_s: int
    rps: float | None = None  # observed request rate last fanout (None at cold start)
    reqs_per_fan: float | None = None  # observed requests per worked fan
    last_run_duration_s: float | None = None
    overran: bool = False  # last run exceeded the window
    pending: int = 0  # fans still pending from prior runs, subtracted as headroom


@dataclass(frozen=True, kw_only=True)
class FanoutPlanResult:
    client_name: DSPClientName
    plan: FanoutPlan
    batch_size: int
    fans: list[FanRecord]
    dispatch_pairs: list[tuple[str, DSPId]] = field(default_factory=list)
    filters: FanConnectionFilter


def compute_fanout_plan(
    nominal_rps: int,
    *,
    rps: float | None = None,
    reqs_per_fan: float | None = None,
    fans_per_s: float | None = None,
    last_fans: int | None = None,
    window_s: int | None = None,
    last_run_duration_s: float | None = None,
    pending: int = 0,
) -> FanoutPlan:
    """Size the next dispatch from the observed fan drain rate `fans_per_s`
    (fans / drain span, medianed over recent fanouts): dispatch what drains in
    `FILL_SAFETY * window`. It's a single per-fanout ratio, so it self-corrects
    on duration — an overrun lowers fans/span and shrinks the next run. Cold start
    (no signal) seeds from `nominal_rps`. `pending` is subtracted as headroom.
    `rps`/`reqs_per_fan` are carried for display only."""
    if window_s is None:
        window_s = settings.fan_fanout_window_s

    is_cold_start = fans_per_s is None or fans_per_s <= 0

    if is_cold_start:
        fanout_count = max(
            0, math.floor((nominal_rps * SAFETY_FACTOR * window_s) / REQUESTS_PER_FAN)
        )
        source = "seed"
    else:
        fanout_count = max(1, math.floor(fans_per_s * FILL_SAFETY * window_s))
        source = "collected"

    has_overran = last_run_duration_s is not None and last_run_duration_s > window_s

    # Overrun damper: while the fleet is still behind, never dispatch MORE than the
    # last run worked. fans_per_s already shrinks on overrun (span ↑ → rate ↓); this
    # only catches the pathological case where a stale/optimistic rate would still
    # grow dispatch into a backlog. A no-growth cap, not a second shrink.
    if source == "collected" and has_overran and last_fans and fanout_count > last_fans:
        fanout_count = last_fans

    # Headroom: subtract fans still pending from overlapping runs, so each tick
    # adds only free capacity instead of piling up concurrent fanouts.
    if pending > 0:
        fanout_count = max(0, fanout_count - pending)

    return FanoutPlan(
        fanout_count=fanout_count,
        fanout_count_source=source,
        fanout_window_s=window_s,
        rps=rps,
        reqs_per_fan=reqs_per_fan,
        last_run_duration_s=last_run_duration_s,
        overran=has_overran,
        pending=pending,
    )


def drain_budget_s(source: FanoutSource) -> int:
    """Seconds to drain this run. Scheduled runs get the full window; off-cadence
    runs are clamped to the time left until the next scheduled tick."""
    window_s = settings.fan_fanout_window_s
    if source is FanoutSource.scheduled:
        return window_s
    last_scheduled = FanoutTask.query.last_scheduled_started_at()
    if last_scheduled is None:
        return window_s
    elapsed = (timezone.now() - last_scheduled).total_seconds()
    return max(1, math.ceil(window_s - (elapsed % window_s)))


def get_default_filters() -> FanConnectionFilter:
    """Build the default fan selection filter for a scheduled fanout tick."""
    now = timezone.now()
    threshold = settings.fan_collect_token_error_revoke_threshold
    return FanConnectionFilter(
        # Nothing is due before the shortest resource interval — a more recently
        # collected fan would only skip, so gate on it. NULL (new/re-connected) passes.
        not_collected_since=now
        - timedelta(seconds=settings.fan_collect_min_interval_s),
        # Safety net for lost messages — must outlast queue drain + SQS retries.
        not_last_dispatched_since=now - timedelta(seconds=settings.fan_dispatch_lock_s),
        max_consecutive_failures=threshold if threshold > 0 else None,
    )


def plan_fanout(
    filters: FanConnectionFilter | None = None,
    *,
    source: FanoutSource = FanoutSource.scheduled,
) -> list[FanoutPlanResult]:
    """Per active client, size the fanout and pick the next batch of fans;
    callers handle SQS dispatch. `filters` overrides the default; `source`
    clamps the drain budget for off-cadence runs."""

    results: list[FanoutPlanResult] = []
    filters = filters or get_default_filters()
    window_s = drain_budget_s(source)

    clients = (
        DSPClient.query.where(
            DSPClient.status.in_(
                [
                    DSPClientStatus.active,
                    DSPClientStatus.paused,
                ]
            )
        )
        .order_by(DSPClient.name)
        .all()
    )

    for client in clients:
        if client.is_paused:
            logger.info(
                "Client %s is paused",
                client.name,
                extra={
                    "dsp_client_name": client.name,
                },
            )
            continue

        if client.nominal_rps is None:
            logger.warning(
                "Client %s has no nominal rps",
                client.name,
                extra={"dsp_client_name": client.name},
            )
            continue

        if not dsp_gateway.is_configured(client.name):
            logger.warning(
                "Client %s is not configured",
                client.name,
                extra={"dsp_client_name": client.name},
            )
            continue

        signals = CollectTask.query.fanout_signals(client.name)
        pending = CollectTask.query.pending_fans(client.name)
        plan = compute_fanout_plan(
            client.nominal_rps,
            rps=signals.rps if signals else None,
            reqs_per_fan=signals.reqs_per_fan if signals else None,
            fans_per_s=signals.fans_per_s if signals else None,
            last_fans=signals.last_fans if signals else None,
            window_s=window_s,
            last_run_duration_s=signals.duration_s if signals else None,
            pending=pending,
        )
        batch_size = get_batch_size_for(
            signals.time_per_fan_s if signals else COLD_TIME_PER_FAN_S
        )
        if plan.fanout_count == 0:
            results.append(
                FanoutPlanResult(
                    client_name=client.name,
                    plan=plan,
                    batch_size=batch_size,
                    fans=[],
                    dispatch_pairs=[],
                    filters=filters,
                )
            )
            continue

        connections = (
            FanConnection.query.filter(
                filters.model_copy(
                    update={
                        "dsp_client_id": client.id,
                        "limit": plan.fanout_count,
                    }
                )
            )
            .order_by_collection_priority()
            .all()
        )

        logger.info(
            "Fanout %s — selected %d fans (capacity=%d source=%s rps=%s reqs_per_fan=%s overran=%s pending=%d last_run_s=%s window=%ds batch_size=%d)",
            client.name,
            len(connections),
            plan.fanout_count,
            plan.fanout_count_source,
            plan.rps,
            plan.reqs_per_fan,
            plan.overran,
            plan.pending,
            plan.last_run_duration_s,
            plan.fanout_window_s,
            batch_size,
            extra={"dsp_client_name": client.name},
        )

        fans: list[FanRecord] = []
        dispatch_pairs: list[tuple[str, DSPId]] = []

        for connection in connections:
            fans.append(
                FanRecord(
                    fan_id=connection.fan_id,
                    token_encrypted=connection.token_encrypted,
                )
            )
            dispatch_pairs.append((connection.fan_id, connection.dsp_id))

        results.append(
            FanoutPlanResult(
                client_name=client.name,
                plan=plan,
                batch_size=batch_size,
                fans=fans,
                dispatch_pairs=dispatch_pairs,
                filters=filters,
            )
        )

    return results