"""Vendor-agnostic "when did this person last actually use it?" signals.

Why this exists
───────────────
``software_seat.idle_reclaim`` recovers paid seats nobody uses. Until
now it could only do that for **Atlassian**, because Atlassian was the
only vendor whose activity Helix knew how to read
(:mod:`app.integrations.atlassian_activity`). Every other paid seat —
Google Workspace, Zoom — was reclaimable only by a human noticing.

This module is the seam: one **persisted** store of
``(person × provider × product) → last active``, written by per-vendor
collectors and read by the reclaim job, so adding a vendor is adding a
collector rather than rewriting the state machine.

The four answers, and why ``unknown`` is not ``idle``
────────────────────────────────────────────────────
* ``active``      — we hold a real last-active timestamp.
* ``never``       — the vendor **affirmatively** says the account has
  never signed in. Reclaimable, but the idle clock then runs from the
  SEAT GRANT, not from a login that never happened: a seat handed out
  yesterday and not yet used is not idle, it is new.
* ``unknown``     — the vendor answered, but gave nothing usable
  (blank field, unparseable value, account not found in the roster).
* ``unsupported`` — this vendor **cannot** answer at all with the
  credentials/APIs Helix has. Declared per provider in
  :data:`NO_USAGE_SIGNAL` with the reason, never inferred.

``unknown`` and ``unsupported`` must never collapse into "idle". The
2026-07-17 identity-sync assurance run produced 36 false findings out of
48 by treating "the snapshot cannot testify" as "the account is open";
the pre-2026-06-11 Atlassian attempt flagged daily users as two-years-
idle by trusting a sandbox's frozen timestamp. Same failure, twice, in
two domains — so here the state is carried explicitly all the way to the
UI instead of being flattened into a number.

Freshness is part of the answer
───────────────────────────────
Every row carries ``collected_at``. A row older than
:data:`STALE_AFTER_HOURS` cannot testify about today, so
:func:`resolve_signal` downgrades it to ``unknown`` — which means a
broken collector makes the reclaim job stop reclaiming rather than start
reclaiming on stale evidence. Failing closed is the only acceptable
direction for a job that revokes access.
"""
from __future__ import annotations

import logging
import re
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger(__name__)

SIGNAL_ACTIVE = "active"
SIGNAL_NEVER = "never"
SIGNAL_UNKNOWN = "unknown"
SIGNAL_UNSUPPORTED = "unsupported"

#: A collected row older than this cannot testify about current usage.
#: 36h, so one missed daily run still resolves but two do not.
STALE_AFTER_HOURS = 36

#: Timestamps at or before this are vendor sentinels for "never", not
#: real logins. Google Directory returns ``1970-01-01T00:00:00.000Z``
#: for an account that has never signed in; parsed naively that reads as
#: fifty-odd years idle — the single most expensive misreading available
#: here, since it would reclaim the seat of every brand-new hire who has
#: not logged in yet. Anything before 2000 is a sentinel, not a login.
_SENTINEL_BEFORE = datetime(2000, 1, 1, tzinfo=UTC)

#: ``<date>[T<time>][.frac][±hh:mm]`` — the offset is captured
#: separately so truncating the fraction cannot corrupt it.
_TS_RE = re.compile(
    r"^(\d{4}-\d{2}-\d{2}(?:[T ]\d{2}:\d{2}(?::\d{2})?)?)"
    r"(\.\d+)?"
    r"([+-]\d{2}:?\d{2})?$"
)


@dataclass(slots=True)
class UsageObservation:
    """One collector's answer about one person on one product."""

    person_email: str
    product_key: str
    state: str
    last_active_at: datetime | None = None
    evidence: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class CollectorResult:
    observations: list[UsageObservation] = field(default_factory=list)
    #: Set when the vendor could not be read at all. The refresh job
    #: records the failure and leaves existing rows alone rather than
    #: overwriting them with a fabricated "unknown" — a stale-but-real
    #: row is better evidence than a fresh lie, and the freshness gate
    #: already downgrades it on its own schedule.
    error: str | None = None


Collector = Callable[[AsyncSession, Any], Awaitable[CollectorResult]]


def parse_vendor_timestamp(value: Any) -> tuple[str, datetime | None]:
    """``(state, timestamp)`` for one raw vendor last-login value.

    Handles the three shapes every vendor here produces: a real ISO
    timestamp, a blank/absent value, and an epoch sentinel meaning
    "never". Nanosecond precision (Atlassian) and a trailing ``Z``
    (Google, Zoom) are both accepted.
    """
    if value is None:
        return SIGNAL_UNKNOWN, None
    raw = str(value).strip()
    if not raw:
        # Blank is the vendor declining to answer, NOT a claim of "never".
        # Admin / API-only / mobile-only accounts and telemetry gaps all
        # legitimately produce blanks.
        return SIGNAL_UNKNOWN, None
    text = raw.replace("Z", "+00:00").replace("z", "+00:00")
    # Atlassian is nanosecond-precision, which fromisoformat rejects.
    # Split the UTC OFFSET off first: scraping digits out of the whole
    # tail also eats the offset's digits, which silently turned an
    # ordinary ``...:11.000+00:00`` into an unparseable string — caught
    # by probing the parser with each vendor's real shape rather than
    # only the awkward one it was written for.
    m = _TS_RE.match(text)
    if m is None:
        logger.debug("usage_signals: unparseable vendor timestamp %r", raw)
        return SIGNAL_UNKNOWN, None
    base, fraction, offset = m.group(1), m.group(2) or "", m.group(3) or ""
    if fraction:
        base = f"{base}.{fraction[1:7].ljust(6, '0')}"
    text = f"{base}{offset}"
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        logger.debug("usage_signals: unparseable vendor timestamp %r", raw)
        return SIGNAL_UNKNOWN, None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=UTC)
    parsed = parsed.astimezone(UTC)
    if parsed <= _SENTINEL_BEFORE:
        return SIGNAL_NEVER, None
    return SIGNAL_ACTIVE, parsed


def resolve_signal(
    state: str, collected_at: datetime | None, *, now: datetime,
) -> str:
    """Apply the freshness gate to a stored row's state.

    A row nobody has refreshed recently cannot testify about current
    usage, so it resolves to ``unknown`` regardless of what it says.
    ``unsupported`` is a statement about the VENDOR, not about a moment,
    so it never goes stale.
    """
    if state == SIGNAL_UNSUPPORTED:
        return SIGNAL_UNSUPPORTED
    if collected_at is None:
        return SIGNAL_UNKNOWN
    if collected_at.tzinfo is None:
        collected_at = collected_at.replace(tzinfo=UTC)
    if now - collected_at > timedelta(hours=STALE_AFTER_HOURS):
        return SIGNAL_UNKNOWN
    return state


#: Providers that cannot supply a usage signal, and why. Declared, never
#: inferred — an absent provider is a gap somebody has to decide about,
#: which is what ``test_every_seat_provider_has_a_usage_decision`` pins.
#: Each entry was checked against the driver in this repo, not assumed.
NO_USAGE_SIGNAL: dict[str, str] = {
    "slack": (
        "``users.list`` carries no last-activity field and the SCIM schema "
        "has none either; per-member activity needs the Enterprise Grid "
        "analytics / admin.users.session APIs, which this workspace's token "
        "does not grant. Checked against drivers/slack_software.py — every "
        "method there is roster or SCIM."
    ),
    "feishu_group": (
        "The Feishu software driver exposes seat grant/revoke and roster "
        "reads only; the open platform gives no per-user last-login for the "
        "app-level seats Helix tracks."
    ),
    "github": (
        "Per-member last activity needs the org audit log (GitHub "
        "Enterprise) or per-user event scraping, neither of which this "
        "GitHub App installation is scoped for. Note also that the App JWT "
        "cannot read /users/* at all — that path needs an installation "
        "token (see the 2026 connector note)."
    ),
    "microsoft_graph": (
        "Graph DOES expose ``signInActivity`` on /users, but it requires "
        "Entra ID P1/P2 plus the AuditLog.Read.All application permission, "
        "and this driver requests neither. This is the one entry here that "
        "is a licensing/scope gap rather than a missing API — if the tenant "
        "is ever on P1+, this becomes a collector, not an exemption."
    ),
    "nexuscrm": (
        "The NexusCRM driver models seats and roles only; no login or "
        "last-activity field is exposed on its user objects."
    ),
    "manual": (
        "``manual`` is the no-vendor provider — a catalog entry Helix "
        "tracks with nobody to ask. There is no API behind it by "
        "definition, so a usage signal cannot exist; these seats stay "
        "human-reviewed."
    ),
}


def usage_collectors() -> dict[str, Collector]:
    """provider slug → collector. Imported lazily to keep drivers out of
    module import (they pull heavy SDKs)."""
    from app.integrations.usage_collectors import (
        collect_atlassian,
        collect_google_workspace,
        collect_zoom,
    )

    return {
        "jira": collect_atlassian,
        "google_workspace": collect_google_workspace,
        "zoom": collect_zoom,
    }


__all__ = [
    "NO_USAGE_SIGNAL",
    "SIGNAL_ACTIVE",
    "SIGNAL_NEVER",
    "SIGNAL_UNKNOWN",
    "SIGNAL_UNSUPPORTED",
    "STALE_AFTER_HOURS",
    "CollectorResult",
    "UsageObservation",
    "parse_vendor_timestamp",
    "resolve_signal",
    "usage_collectors",
]