#!/usr/bin/env python3
"""Daily metrics snapshot for the Inference Cooperative portal.

Writes one row per day into the portal's SQLite store (`metrics_daily`, the same
`members.db` that holds the members table) so the admin panel and the member
dashboard can chart trends from one source of truth.

    metrics_snapshot.py             record / refresh TODAY's row
    metrics_snapshot.py --backfill  rebuild history (idempotent)

Rules this file follows deliberately:

- **Fail closed per metric.** A source that cannot be read yields NULL for its
  columns, never a zero. A zero reads as "nobody", which is a lie that would
  survive into a chart; NULL reads as "not measured". Each failure is printed,
  and the task's stdout lands in the app log (`cloudron logs --app <id>`).
- **Cumulative vs daily.** `members_active`, `members_total`, `newsletter_subs`
  and `matrix_space_members` are counts AS OF that day; `spend_usd` and
  `requests` are THAT day's traffic. Charts must treat them accordingly.
- **Runs in Cloudron's scheduler container**, which gets the platform env
  (declared app env + addon vars) but NOT variables the app's own start.sh
  sourced. Everything needed here is a declared app env var: LITELLM_DB_*,
  LISTMONK_*, MATRIX_BOT_TOKEN, LISTMONK_LIST_ID.
- `members_new` uses `welcome_sent_at`, the provisioning timestamp — the closest
  thing the portal stores to a join date. Manual members have no welcome email,
  so their joins are absent from the growth curve (noted, not guessed).
"""

import datetime as dt
import json
import os
import sqlite3
import sys
import urllib.parse
import urllib.request

DB_PATH = os.environ.get("DB_PATH", "/app/data/members.db")
LISTMONK_BASE = os.environ.get("LISTMONK_BASE", "https://newsletter.inference.coop")
LISTMONK_LIST_ID = int(os.environ.get("LISTMONK_LIST_ID") or 3)
MATRIX_BASE = os.environ.get("MATRIX_BASE", "https://matrix.inference.coop")
SPACE = os.environ.get("MATRIX_SPACE_ID", "!FKOD2sSqeYvEq9jzr8jDL9ogf3TdJCEI1k4IsnBZsCM")


def today() -> str:
    return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d")


def connect() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH, timeout=20)
    conn.execute(
        """CREATE TABLE IF NOT EXISTS metrics_daily (
               day TEXT PRIMARY KEY,
               members_active INTEGER,
               members_total INTEGER,
               members_new INTEGER,
               newsletter_subs INTEGER,
               matrix_space_members INTEGER,
               spend_usd REAL,
               requests INTEGER,
               recorded_at TEXT)"""
    )
    return conn


def _get_json(url: str, headers: dict | None = None):
    req = urllib.request.Request(url, headers=headers or {})
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read() or b"{}")


# ---------------------------------------------------------------- sources

def from_sqlite() -> dict:
    """Members, plus the provisioning dates that give the growth curve."""
    conn = connect()
    try:
        active = conn.execute("SELECT COUNT(*) FROM members WHERE active=1").fetchone()[0]
        total = conn.execute("SELECT COUNT(*) FROM members").fetchone()[0]
        joins = [
            row[0]
            for row in conn.execute(
                "SELECT welcome_sent_at FROM members WHERE welcome_sent_at IS NOT NULL AND welcome_sent_at != ''"
            )
        ]
    finally:
        conn.close()
    return {"active": active, "total": total, "joins": joins}


def from_listmonk() -> int | None:
    """Confirmed subscribers on the members list, counted rather than trusted.

    Listmonk's cached `subscriber_count` on the list object is stale and
    double-counted (it read 142 for a list that truly holds 71), so we page the
    subscribers endpoint and count confirmed memberships ourselves.
    """
    token, user = os.environ.get("LISTMONK_API_TOKEN"), os.environ.get("LISTMONK_API_USER") or "portal-sync"
    if not token:
        print("metrics: LISTMONK_API_TOKEN not set — newsletter_subs left NULL")
        return None
    try:
        total, page, seen = 0, 1, 0
        while True:
            data = _get_json(
                f"{LISTMONK_BASE}/api/subscribers?list_id={LISTMONK_LIST_ID}&per_page=100&page={page}",
                {"Authorization": f"token {user}:{token}"},
            ) or {}
            payload = data.get("data") or {}
            rows = payload.get("results") or []
            # The subscriber's own `status` is enabled/blocked; the LIST
            # membership status is what means "on this list" — counting the
            # former returned a silent, wrong 0.
            for s in rows:
                for membership in s.get("lists") or []:
                    if membership.get("id") == LISTMONK_LIST_ID and (
                        membership.get("subscription_status") or ""
                    ) == "confirmed":
                        total += 1
            seen += len(rows)
            if not rows or len(rows) < 100 or page >= 20:
                break
            page += 1
        if seen == 0:
            # Nothing came back at all: the endpoint or payload shape changed.
            # Record "not measured" (NULL) rather than a zero that reads as
            # "nobody is subscribed".
            print("metrics: listmonk returned no rows — newsletter_subs left NULL")
            return None
        return total
    except Exception as exc:  # noqa: BLE001 - report, never guess
        print(f"metrics: listmonk read failed ({exc}) — newsletter_subs left NULL")
        return None


def from_matrix() -> int | None:
    token = os.environ.get("MATRIX_BOT_TOKEN")
    if not token:
        print("metrics: MATRIX_BOT_TOKEN not set — matrix_space_members left NULL")
        return None
    try:
        data = _get_json(
            f"{MATRIX_BASE}/_matrix/client/v3/rooms/{urllib.parse.quote(SPACE)}/joined_members",
            {"Authorization": f"Bearer {token}"},
        ) or {}
        return len(data.get("joined") or {})
    except Exception as exc:  # noqa: BLE001
        print(f"metrics: matrix read failed ({exc}) — matrix_space_members left NULL")
        return None


def spend_by_day(since: dt.date | None = None) -> dict[str, tuple[float, int]]:
    """{day: (spend_usd, requests)} straight from LiteLLM's spend logs.

    Prefers `model_group` the way aggregate_model_usage does, but here we only
    need volume, so every row in the window counts.
    """
    try:
        import psycopg2
    except Exception as exc:  # noqa: BLE001
        print(f"metrics: psycopg2 unavailable ({exc})")
        return {}
    try:
        conn = psycopg2.connect(
            host=os.environ.get("LITELLM_DB_HOST"),
            port=os.environ.get("LITELLM_DB_PORT"),
            user=os.environ.get("LITELLM_DB_USER"),
            password=os.environ.get("LITELLM_DB_PASSWORD"),
            dbname=os.environ.get("LITELLM_DB_NAME"),
            connect_timeout=10,
            sslmode="prefer",
        )
    except Exception as exc:  # noqa: BLE001
        print(f"metrics: LiteLLM DB connect failed ({exc}) — spend left NULL")
        return {}
    try:
        where, params = "", []
        if since:
            where, params = 'WHERE "startTime" >= %s', [since.isoformat()]
        with conn.cursor() as cur:
            cur.execute(
                f"""SELECT to_char("startTime", 'YYYY-MM-DD') AS day,
                           COALESCE(SUM(spend), 0), COUNT(*)
                    FROM "LiteLLM_SpendLogs" {where}
                    GROUP BY 1 ORDER BY 1""",
                params,
            )
            out = {row[0]: (round(float(row[1]), 6), int(row[2])) for row in cur.fetchall()}
    finally:
        conn.close()
    return out


# ---------------------------------------------------------------- writes

def upsert(day: str, row: dict) -> None:
    conn = connect()
    try:
        conn.execute(
            """INSERT INTO metrics_daily
                   (day, members_active, members_total, members_new, newsletter_subs,
                    matrix_space_members, spend_usd, requests, recorded_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
               ON CONFLICT(day) DO UPDATE SET
                   members_active=excluded.members_active,
                   members_total=excluded.members_total,
                   members_new=excluded.members_new,
                   newsletter_subs=excluded.newsletter_subs,
                   matrix_space_members=excluded.matrix_space_members,
                   spend_usd=excluded.spend_usd,
                   requests=excluded.requests,
                   recorded_at=excluded.recorded_at""",
            (
                day,
                row.get("members_active"),
                row.get("members_total"),
                row.get("members_new"),
                row.get("newsletter_subs"),
                row.get("matrix_space_members"),
                row.get("spend_usd"),
                row.get("requests"),
                dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
            ),
        )
        conn.commit()
    finally:
        conn.close()


def snapshot(day: str | None = None) -> dict:
    day = day or today()
    m = from_sqlite()
    subs = from_listmonk()
    space = from_matrix()
    spend = spend_by_day(since=dt.date.fromisoformat(day) - dt.timedelta(days=1))
    today_spend = spend.get(day, (None, None))
    row = {
        "members_active": m["active"],
        "members_total": m["total"],
        "members_new": sum(1 for j in m["joins"] if str(j)[:10] == day),
        "newsletter_subs": subs,
        "matrix_space_members": space,
        "spend_usd": today_spend[0],
        "requests": today_spend[1],
    }
    upsert(day, row)
    print(f"metrics: recorded {day} -> {json.dumps(row)}")
    return row


def backfill() -> None:
    """Rebuild history: cumulative members from provisioning dates, spend by day."""
    m = from_sqlite()
    joins = sorted(str(j)[:10] for j in m["joins"])
    spend = spend_by_day()
    if not joins and not spend:
        print("metrics: backfill found no history")
        return

    first = min([d for d in joins] + ([min(spend)] if spend else []))
    last = today()
    print(
        f"metrics: backfilling {first}..{last} "
        f"({len(joins)} provisioning dates, {len(spend)} spend days)"
    )
    day = dt.date.fromisoformat(first)
    end = dt.date.fromisoformat(last)
    written = 0
    while day <= end:
        key = day.isoformat()
        # Cumulative members AS OF this day, from provisioning dates. Members
        # provisioned without a welcome email (manual/founder adds) are not in
        # this history, so backfilled days are a LOWER BOUND — today's row below
        # carries the exact count.
        members_total = sum(1 for j in joins if j <= key) if joins else None
        row = {
            "members_total": members_total,
            "members_new": sum(1 for j in joins if j == key),
            "spend_usd": spend.get(key, (None, None))[0],
            "requests": spend.get(key, (None, None))[1],
        }
        if key == today():
            # today's row carries the exact live metrics
            row.update(
                members_active=m["active"],
                members_total=m["total"],
                newsletter_subs=from_listmonk(),
                matrix_space_members=from_matrix(),
            )
        upsert(key, row)
        written += 1
        day += dt.timedelta(days=1)
    print(f"metrics: backfill wrote {written} day(s)")


if __name__ == "__main__":
    if "--backfill" in sys.argv:
        backfill()
    else:
        snapshot()