"""The connector registry — N identities at the door, from the durable store.

Membership is DATA, not configuration. The env layer knows exactly one connector
(`CL_CONNECTOR_ID` + `CL_DEMO_TOKEN`); that was fine for a demo host and dies at the
second member. This module puts members where the journal already lives — Postgres —
so admitting the 10,000th member is one INSERT, not a restart.

Three rules, all of them the same culture as the rest of the door:

  * **The plaintext token exists only at issuance.** The CLI generates it on the box,
    prints it exactly once, and stores sha256(token). A leaked database leaks no tokens.
  * **Revocation needs no restart.** `revoke` flips a status row; the very next request
    with that token fails closed. The attested trail of the refusal lives in the journal.
  * **The demo pair keeps working.** `store_token_verifier` composes: the static
    fallback (Arena's demo token) is consulted first, then the store. Adding members
    must never make an existing neighbour blink.

Admin is a CLI on the box (`python -m cl.connectors add|revoke|list`), deliberately NOT
a new HTTP endpoint: the door's public surface stays five routes, and issuing identity
remains an operator act on the box, like every other credential here.

Honest limit, named: the door does not yet enforce token↔connector_id BINDING — the
claims (sub/tenant) come from the token and are authoritative for identity, but a valid
registry token presented with a different (existing) connector_id gets that connector's
profile. Enforcing the binding is a door-law change that belongs to NEOfortress; noted,
not smuggled in.
"""

from __future__ import annotations

import argparse
import hashlib
import os
import secrets
import sys
from typing import Any, Mapping, Optional

from cl.agent_seam.bridge import ConnectorProfile
from cl.agent_seam.context import TokenVerifier
from cl.http.identity import make_claims

ACTIVE = "active"
REVOKED = "revoked"


def token_hash(token: str) -> str:
    """sha256 hex of the plaintext token — the only form the store ever sees."""
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


class ConnectorRegistry:
    """CRUD over the `connectors` table. One registry per service process; every lookup
    is a single indexed SELECT (partial index on active token hashes), so the door's
    latency does not grow with membership."""

    def __init__(self, connection: Any):
        self._connection = connection

    @classmethod
    def connect(cls, dsn: str) -> "ConnectorRegistry":
        import psycopg

        return cls(psycopg.connect(dsn, autocommit=False))

    # ── admin (CLI) ──────────────────────────────────────────────────────────────────

    def add(self, connector_id: str, *, tenant: str, sub: str,
            token: Optional[str] = None) -> str:
        """Admit a member. Returns the PLAINTEXT token — the caller prints it once and
        forgets it; only the hash is stored. A duplicate connector_id is refused loudly."""
        plaintext = token or secrets.token_urlsafe(32)
        try:
            with self._connection.cursor() as cursor:
                cursor.execute(
                    "INSERT INTO connectors (connector_id, token_hash, tenant, sub, status)"
                    " VALUES (%s, %s, %s, %s, 'active')",
                    (connector_id, token_hash(plaintext), tenant, sub),
                )
            self._connection.commit()
        except Exception:
            self._connection.rollback()
            raise
        return plaintext

    def revoke(self, connector_id: str) -> bool:
        """Close the door for one member, effective on the very next request. No restart.
        Returns False when there was nothing active to revoke."""
        try:
            with self._connection.cursor() as cursor:
                cursor.execute(
                    "UPDATE connectors SET status = 'revoked', revoked_at = now()"
                    " WHERE connector_id = %s AND status = 'active'",
                    (connector_id,),
                )
                changed = cursor.rowcount
            self._connection.commit()
        except Exception:
            self._connection.rollback()
            raise
        return changed > 0

    def rows(self) -> list[dict]:
        with self._connection.cursor() as cursor:
            cursor.execute(
                "SELECT connector_id, tenant, sub, status, created_at, revoked_at"
                " FROM connectors ORDER BY connector_id"
            )
            out = [
                {"connector_id": r[0], "tenant": r[1], "sub": r[2], "status": r[3],
                 "created_at": str(r[4]), "revoked_at": str(r[5]) if r[5] else None}
                for r in cursor.fetchall()
            ]
        self._connection.commit()   # release the read snapshot
        return out

    # ── the door (hot path) ──────────────────────────────────────────────────────────

    def lookup_active_by_token(self, token: str) -> Optional[dict]:
        """token → {connector_id, tenant, sub} for an ACTIVE member, else None. The one
        query the door runs; a revoked or unknown token is None, never an error."""
        try:
            with self._connection.cursor() as cursor:
                cursor.execute(
                    "SELECT connector_id, tenant, sub FROM connectors"
                    " WHERE token_hash = %s AND status = 'active'",
                    (token_hash(token),),
                )
                row = cursor.fetchone()
            self._connection.commit()
        except Exception:
            self._connection.rollback()
            raise
        if row is None:
            return None
        return {"connector_id": row[0], "tenant": row[1], "sub": row[2]}

    def has_active(self, connector_id: str) -> bool:
        try:
            with self._connection.cursor() as cursor:
                cursor.execute(
                    "SELECT 1 FROM connectors WHERE connector_id = %s AND status = 'active'",
                    (connector_id,),
                )
                found = cursor.fetchone() is not None
            self._connection.commit()
        except Exception:
            self._connection.rollback()
            raise
        return found


def store_token_verifier(registry: ConnectorRegistry,
                         fallback: Optional[TokenVerifier] = None) -> TokenVerifier:
    """A `TokenVerifier` over the registry, composed AFTER the static fallback.

    Order matters and is deliberate: the demo pair (Arena) is consulted first so that
    standing up the registry cannot change the behaviour of a single existing caller;
    only an unknown token reaches the store. Both misses → None → fail-closed upstream."""

    def verify(token: str) -> Optional[dict]:
        if fallback is not None:
            claims = fallback(token)
            if claims is not None:
                return claims
        row = registry.lookup_active_by_token(token)
        if row is None:
            return None
        return make_claims(sub=row["sub"], tenant=row["tenant"], jti=row["connector_id"])

    return verify


class StoreBackedProfiles:
    """`config.profiles` that answers for static profiles first, then the registry.

    The service only ever calls `.get(connector_id)`. A registry member's profile uses
    the door's OWN consequential-type set — one law at the door for everyone; per-member
    policy would be a silent privilege system and is refused by design."""

    def __init__(self, static: Mapping[str, ConnectorProfile], registry: ConnectorRegistry,
                 *, consequential_event_types: frozenset):
        self._static = dict(static)
        self._registry = registry
        self._consequential = consequential_event_types

    def get(self, connector_id: str) -> Optional[ConnectorProfile]:
        profile = self._static.get(connector_id)
        if profile is not None:
            return profile
        if not self._registry.has_active(connector_id):
            return None
        return ConnectorProfile(
            connector_id=connector_id,
            consequential_event_types=self._consequential,
        )


# ── the CLI — admin lives on the box, not on the network ─────────────────────────────


def main(argv: Optional[list] = None) -> int:
    parser = argparse.ArgumentParser(
        prog="python -m cl.connectors",
        description="Admit, revoke and list door identities. Runs on the box, against "
                    "CL_POSTGRES_DSN; there is deliberately no HTTP for this.",
    )
    parser.add_argument("--dsn", default=os.environ.get("CL_POSTGRES_DSN", "").strip(),
                        help="Postgres DSN (default: CL_POSTGRES_DSN)")
    sub = parser.add_subparsers(dest="command", required=True)

    add = sub.add_parser("add", help="admit a member; prints the token EXACTLY ONCE")
    add.add_argument("connector_id")
    add.add_argument("--tenant", required=True)
    add.add_argument("--sub", required=True)

    revoke = sub.add_parser("revoke", help="revoke a member, effective immediately")
    revoke.add_argument("connector_id")

    sub.add_parser("list", help="list members (no hashes, no tokens)")

    args = parser.parse_args(argv)
    if not args.dsn:
        print("no DSN: set CL_POSTGRES_DSN or pass --dsn", file=sys.stderr)
        return 2

    registry = ConnectorRegistry.connect(args.dsn)
    if args.command == "add":
        token = registry.add(args.connector_id, tenant=args.tenant, sub=args.sub)
        print(f"connector_id: {args.connector_id}")
        print(f"token (SHOWN ONCE, only the hash is stored):\n{token}")
        return 0
    if args.command == "revoke":
        if registry.revoke(args.connector_id):
            print(f"revoked: {args.connector_id} (effective immediately, no restart)")
            return 0
        print(f"nothing active under {args.connector_id!r}", file=sys.stderr)
        return 1
    for row in registry.rows():
        flag = "" if row["status"] == ACTIVE else f"  [{row['status']} {row['revoked_at']}]"
        print(f"{row['connector_id']:<24} {row['tenant']:<20} {row['sub']}{flag}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
