"""Daily read-only Google Workspace license alert for Slack.

Reads purchased and assigned seats per SKU for every connected workspace from
the customer usage report (the same read the weekly digest makes), compares
them with the last run's snapshot in GCS, and posts to the digest channel only
when something changed:

  - a workspace's free seats DROPPED and are now at or below the threshold
    (default 3): @-mentions the alert list;
  - a workspace that was at or below the threshold is above it again: posted,
    no mention.

Nothing repeats: a workspace sitting at 1 free seat is announced once, on the
day it got there. The first run posts one baseline of everyone already at or
below the threshold. A day with no drop and no recovery posts nothing.

Google publishes these numbers once a day, 2-4 days behind, so the threshold is
also the buffer: by the time a workspace reads 1 free here, the Admin console
may already read 0. Live counts would need the License Manager API, whose only
scope is not read-only, and this assistant never holds a write scope.
"""

from __future__ import annotations

import json
import logging
import os
import sys
from dataclasses import dataclass
from typing import Any

from googleapiclient.errors import HttpError

import digest_job
import workspace

logger = logging.getLogger("gw_license_alerts")

STATE_VERSION = 1
TOTAL_SUFFIX = "_total_licenses"
USED_SUFFIX = "_used_licenses"
PARAM_PREFIX = "accounts:"


@dataclass(frozen=True)
class Seats:
    """One SKU in one workspace, as of `day` (the usage report's date)."""

    workspace: str
    label: str
    sku: str
    total: int
    used: int
    day: str

    @property
    def free(self) -> int:
        return max(self.total - self.used, 0)


@dataclass(frozen=True)
class Change:
    kind: str  # "baseline" | "drop" | "recovered"
    seats: Seats
    previous_free: int | None = None


def seats_from_params(route: workspace.WorkspaceRoute, day: str, params: dict[str, Any]) -> list[Seats]:
    """One Seats per SKU that has purchased seats. Each SKU stands alone: summing
    them would let one SKU's spare seats hide another that has run out. A SKU
    with no purchased seats is not a license anyone can run out of."""
    seats = []
    for name, value in params.items():
        if not (name.startswith(PARAM_PREFIX) and name.endswith(TOTAL_SUFFIX)):
            continue
        sku = name[len(PARAM_PREFIX):-len(TOTAL_SUFFIX)]
        used_name = PARAM_PREFIX + sku + USED_SUFFIX
        total = digest_job.as_int(value)
        if total <= 0 or used_name not in params:
            continue
        seats.append(Seats(route.key, route.label, sku, total, digest_job.as_int(params[used_name]), day))
    return sorted(seats, key=lambda s: s.sku)


def collect(routes: list[workspace.WorkspaceRoute]) -> tuple[list[Seats], list[str]]:
    """Seats for every workspace that answered, and the labels of those that did not."""
    seats: list[Seats] = []
    failed: list[str] = []
    for route in routes:
        try:
            day, params = digest_job.usage_report(route)
        except Exception:  # noqa: BLE001 - one bad tenant must not kill the run
            logger.exception("License read failed for %s", route.key)
            failed.append(route.label)
            continue
        found = seats_from_params(route, day, params)
        if not found:
            # No report in nine days, or no paid SKU: nothing to compare. Logged,
            # not posted - an alert channel is for seats, not for plumbing.
            logger.info("No license data for %s", route.key)
            continue
        seats.extend(found)
    return seats, failed


def state_key(seats: Seats) -> str:
    return seats.workspace + "|" + seats.sku


def load_state(bucket: str, name: str) -> dict[str, Any]:
    """The last run's snapshot, or {} only when there is none yet. digest_job's
    loader treats every error as "no snapshot", which here would turn a storage
    hiccup into a first run: the whole baseline re-posted, with mentions. So
    anything but a 404 raises."""
    if not bucket:
        return {}
    try:
        raw = digest_job.storage_client().objects().get_media(bucket=bucket, object=name).execute()
    except HttpError as exc:
        if getattr(exc.resp, "status", None) == 404:
            logger.info("No previous snapshot at gs://%s/%s; this is a first run.", bucket, name)
            return {}
        raise
    if isinstance(raw, bytes):
        raw = raw.decode("utf-8")
    parsed = json.loads(raw)
    if not isinstance(parsed, dict):
        raise ValueError("the license alert snapshot is not a JSON object")
    return parsed


def stored_entries(previous: dict[str, Any]) -> dict[str, dict[str, Any]]:
    entries = previous.get("seats") if previous.get("version") == STATE_VERSION else None
    if not isinstance(entries, dict):
        return {}
    return {key: entry for key, entry in entries.items() if isinstance(entry, dict)}


def previous_free(previous: dict[str, Any]) -> dict[str, int]:
    return {key: digest_job.as_int(entry.get("free")) for key, entry in stored_entries(previous).items()}


def skip_older_days(current: list[Seats], previous: dict[str, Any]) -> list[Seats]:
    """A SKU whose report day is older than the one already stored sits this run
    out and keeps its stored entry. If Google's newest day for a tenant ever
    stepped back, comparing it would invent a drop today and a recovery tomorrow."""
    stored = stored_entries(previous)
    kept = []
    for seats in current:
        stored_day = stored.get(state_key(seats), {}).get("day")
        if isinstance(stored_day, str) and seats.day and seats.day < stored_day:
            logger.info("Skipping %s: report day %s is older than the stored %s",
                        state_key(seats), seats.day, stored_day)
            continue
        kept.append(seats)
    return kept


def find_changes(current: list[Seats], previous: dict[str, Any], warn_free: int) -> list[Change]:
    before = previous_free(previous)
    first_run = not before
    changes = []
    for seats in current:
        was = before.get(state_key(seats))
        if first_run:
            if seats.free <= warn_free:
                changes.append(Change("baseline", seats))
        elif was is None:
            # Newly onboarded (or newly reporting) workspace: no history to drop
            # from, but a new tenant already out of seats still needs a person.
            if seats.free <= warn_free:
                changes.append(Change("drop", seats))
        elif seats.free < was and seats.free <= warn_free:
            changes.append(Change("drop", seats, was))
        elif was <= warn_free < seats.free:
            changes.append(Change("recovered", seats, was))
    return changes


def snapshot_payload(current: list[Seats], previous: dict[str, Any]) -> dict[str, Any]:
    """A workspace that failed this run keeps its last good entry, so a one-day
    outage is not followed by every one of its seats being "new" again."""
    entries: dict[str, Any] = {}
    if previous.get("version") == STATE_VERSION and isinstance(previous.get("seats"), dict):
        entries.update(previous["seats"])
    for seats in current:
        entries[state_key(seats)] = {"free": seats.free, "total": seats.total, "used": seats.used, "day": seats.day}
    return {"version": STATE_VERSION, "seats": entries}


def slack_escape(text: str) -> str:
    return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


def describe(seats: Seats, multi_sku: set[str]) -> str:
    name = slack_escape(seats.label)
    if seats.workspace in multi_sku:
        name += " (" + slack_escape(seats.sku) + ")"
    return name


def seat_line(change: Change, multi_sku: set[str]) -> str:
    seats = change.seats
    line = f"- {describe(seats, multi_sku)}: {seats.free} free ({seats.used}/{seats.total} assigned)"
    if change.previous_free is not None:
        line += f", was {change.previous_free}"
    return line


def render(changes: list[Change], warn_free: int, mention_ids: list[str], failed: list[str]) -> str:
    multi: dict[str, int] = {}
    for change in changes:
        multi[change.seats.workspace] = multi.get(change.seats.workspace, 0) + 1
    multi_sku = {key for key, count in multi.items() if count > 1}
    order = lambda change: (change.seats.free, change.seats.label.lower())  # noqa: E731

    alerting = [c for c in changes if c.kind in ("baseline", "drop")]
    recovered = sorted((c for c in changes if c.kind == "recovered"), key=order)
    out = sorted((c for c in alerting if c.seats.free == 0), key=order)
    low = sorted((c for c in alerting if c.seats.free > 0), key=order)
    baseline = any(c.kind == "baseline" for c in changes)

    mentions = " ".join("<@" + uid + ">" for uid in mention_ids)
    lines = []
    if baseline:
        lines.append(
            f":credit_card: *License alerts are on.* Each weekday I'll flag a workspace whose free "
            f"Google Workspace seats drop to {warn_free} or fewer. Where things stand today:"
            + (" " + mentions if mentions else "")
        )
    elif alerting:
        lines.append(":rotating_light: *Google Workspace seats running low*" + (" " + mentions if mentions else ""))
    else:
        lines.append(":white_check_mark: *Google Workspace seats freed up*")

    if out:
        lines += ["", f"*Out of licenses ({len(out)})*"] + [seat_line(c, multi_sku) for c in out]
    if low:
        lines += ["", f"*{warn_free} or fewer free ({len(low)})*"] + [seat_line(c, multi_sku) for c in low]
    if recovered:
        lines += ["", f"*Back above {warn_free} free ({len(recovered)})*"] + [seat_line(c, multi_sku) for c in recovered]

    days = sorted({c.seats.day for c in changes if c.seats.day})
    if days:
        as_of = days[0] if len(days) == 1 else days[0] + " to " + days[-1]
        lines += ["", f"_Google's license numbers as of {as_of}. They run 2-4 days behind, so the "
                      "Admin console may already show fewer free seats._"]
    if failed:
        lines.append("_Couldn't read: " + ", ".join(slack_escape(name) for name in sorted(failed)) + "._")
    return "\n".join(lines)


def main() -> int:
    logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO").upper())

    channel = os.getenv("LICENSE_ALERT_SLACK_CHANNEL", "").strip()
    mention_ids = [
        item.strip()
        for item in os.getenv("LICENSE_ALERT_MENTION_USER_IDS", "").replace(",", " ").split()
        if item.strip()
    ]
    bucket = os.getenv("LICENSE_ALERT_STATE_BUCKET", "").strip()
    state_object = os.getenv("LICENSE_ALERT_STATE_OBJECT", "gw-license-alerts/latest.json").strip()
    dry_run = digest_job.env_flag("LICENSE_ALERT_DRY_RUN")
    warn_free = digest_job.env_int("LICENSE_ALERT_WARN_FREE", 3)

    if not dry_run and not (channel and bucket):
        # Without state every run is a first run, which would re-post the baseline
        # daily - exactly the repetition this job exists to avoid.
        logger.error("LICENSE_ALERT_SLACK_CHANNEL and LICENSE_ALERT_STATE_BUCKET are required unless dry run.")
        return 2

    routes = workspace.workspace_routes()
    if not routes:
        logger.error("No workspace routes configured.")
        return 2

    try:
        previous = load_state(bucket, state_object)
    except Exception:  # noqa: BLE001 - unreadable state must never look like a first run
        logger.exception("Couldn't read the last snapshot; not posting, so nothing is re-announced.")
        return 1
    current, failed = collect(routes)
    if not current:
        logger.error("No workspace returned license data; leaving the snapshot as it was.")
        return 1
    current = skip_older_days(current, previous)
    changes = find_changes(current, previous, warn_free)
    logger.info("License check: workspaces=%s skus=%s changes=%s failed=%s",
                len(routes), len(current), len(changes), len(failed))

    text = render(changes, warn_free, mention_ids, failed) if changes else ""
    if dry_run:
        print(text or "(no changes; nothing would be posted)")
        logger.info("LICENSE_ALERT_DRY_RUN set; not posting and not saving a snapshot.")
        return 0

    if text and not digest_job.post_to_slack(channel, text):
        # The snapshot stays put, so tomorrow's run sees the same drop and posts it.
        return 1
    digest_job.save_snapshot(bucket, state_object, snapshot_payload(current, previous))
    return 0


if __name__ == "__main__":
    sys.exit(main())