"""
Cross-node duplicate detection for call recordings.

Two edge nodes within range of the same trunked system both decode and upload
the same transmission. That is the normal case for a distributed network, not
an error — but without this, one transmission is transcribed twice, billed
twice, and correlated twice, and the resulting incident shows two "units"
where there was one.

CANONICAL SELECTION IS DELIBERATELY NOT "FIRST UPLOAD WINS". Upload order
depends on encode time and network latency, so it varies run to run; picking
by it would make which recording is authoritative non-deterministic. The call
document is created from the MQTT call_start event *before* the upload
arrives, so by upload time every node's document for the transmission already
exists and can be ranked. Canonical is the earliest ``started_at``, breaking
ties on ``call_id`` so both nodes independently reach the same verdict.

The loser keeps its audio — it is ~60 KB and may be the cleaner capture if the
winner's node had a weak signal — but is excluded from the AI pipeline.
"""
from datetime import datetime, timedelta, timezone
from typing import Awaitable, Callable, Optional
from app.config import settings
from app.internal.logger import logger

# Firestore is reached through an injected callable rather than a module-level
# import. app.internal.firestore initialises firebase-admin at import time,
# which needs credentials and the SDK present — so importing it here would make
# this module unimportable in a unit test. Same reasoning as the deferred
# import in app/internal/auth.py.
QueryFn = Callable[[str, list], Awaitable[list[dict]]]


def _parse_dt(value) -> Optional[datetime]:
    """Firestore hands back Timestamp, datetime, or ISO string depending on writer."""
    if not value:
        return None
    if isinstance(value, datetime):
        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
    try:
        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError:
        return None
    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)


def _is_canonical(call: dict, others: list[dict]) -> bool:
    """True if `call` is the one recording of this transmission that should be processed."""
    started = _parse_dt(call.get("started_at"))
    call_id = call.get("call_id") or ""
    for other in others:
        other_started = _parse_dt(other.get("started_at"))
        if not other_started or not started:
            continue
        if other_started < started:
            return False
        if other_started == started and (other.get("call_id") or "") < call_id:
            return False
    return True


async def find_duplicate_of(call: dict, query: Optional[QueryFn] = None) -> Optional[str]:
    """Return the canonical call_id if `call` duplicates another node's recording.

    Returns None when this call is the canonical one, or when there is nothing
    to compare against (single node in range, or the call lacks the talkgroup
    and system identifiers the match is keyed on).
    """
    system_id = call.get("system_id")
    talkgroup_id = call.get("talkgroup_id")
    call_id = call.get("call_id")
    started = _parse_dt(call.get("started_at"))
    if not (system_id and talkgroup_id is not None and call_id and started):
        return None

    if query is None:
        from app.internal import firestore as fstore
        query = fstore.collection_where

    window = timedelta(seconds=settings.duplicate_window_seconds)
    try:
        # Range-scan on started_at, then filter the rest in Python — Firestore
        # allows a range on only one field per query.
        nearby = await query("calls", [
            ("system_id", "==", system_id),
            ("started_at", ">=", started - window),
            ("started_at", "<=", started + window),
        ])
    except Exception as e:
        # Never block an upload on dedup — worst case is the pre-existing
        # behaviour of processing both copies.
        logger.warning(f"Duplicate check failed for call {call_id}: {e}")
        return None

    matches = [
        c for c in nearby
        if c.get("call_id") != call_id
        and c.get("talkgroup_id") == talkgroup_id
        and c.get("node_id") != call.get("node_id")   # same node twice is a real repeat
        and not c.get("duplicate_of")                 # never point at another duplicate
    ]
    if not matches:
        return None

    if _is_canonical(call, matches):
        return None

    canonical = min(
        matches,
        key=lambda c: (_parse_dt(c.get("started_at")) or started, c.get("call_id") or ""),
    )
    return canonical.get("call_id")