"""Read-only, sanitized Phase 5E.2 production evidence collector.

This collector uses Core's read-only database connection and Cloud's
authenticated read APIs.  It never prints emails, hashes, tokens, secrets, or
raw metadata.  Cloud credential/event/session counts remain explicitly marked
unavailable when the deployed read surface does not expose them.
"""

from __future__ import annotations

import argparse
import asyncio
import hashlib
import json
from pathlib import Path
from typing import Any

import httpx
from dotenv import dotenv_values
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

ROOT = Path(__file__).parents[1]
INHERITED_SOURCES = {"workspace_owner", "workspace_member", "workspace_inherited"}


def identity_key(email: str) -> str:
    return hashlib.sha256(email.strip().lower().encode()).hexdigest()[:16]


def inherited(user: dict[str, Any]) -> bool:
    metadata = user.get("user_metadata") if isinstance(user.get("user_metadata"), dict) else {}
    authority = metadata.get("p2hs_authority")
    return str(metadata.get("source") or "").lower() in INHERITED_SOURCES or (
        isinstance(authority, dict) and authority.get("source") == "workspace"
    )


def evidence_unavailable(reason: str) -> dict[str, str]:
    return {"status": "UNAVAILABLE", "reason": reason}


async def collect(project_ids: set[str] | None) -> dict[str, Any]:
    env = dotenv_values(ROOT / ".env")
    engine = create_async_engine(str(env["DATABASE_URL"]), pool_pre_ping=True)
    async with engine.connect() as db:
        rows = (
            await db.execute(
                text(
                    """
                    SELECT p.id::text AS project_id, p.workspace_id::text AS workspace_id,
                           w.owner_id::text AS owner_id,
                           a.providers, a.auth_methods, a.audiences,
                           a.allow_signup, a.admin_panel_path, a.customer_login_path
                    FROM projects p
                    JOIN workspaces w ON w.id = p.workspace_id
                    JOIN project_auth_config a ON a.project_id = p.id
                    WHERE p.deleted_at IS NULL AND a.enabled = true
                    ORDER BY p.id
                    """
                )
            )
        ).mappings().all()
        memberships = (
            await db.execute(
                text(
                    """
                    SELECT wm.workspace_id::text AS workspace_id, wm.user_id::text AS user_id,
                           u.email, r.name AS role_name
                    FROM workspace_memberships wm
                    JOIN users u ON u.id = wm.user_id
                    LEFT JOIN roles r ON r.id = wm.role_id
                    """
                )
            )
        ).mappings().all()
    await engine.dispose()

    selected = [row for row in rows if not project_ids or row["project_id"] in project_ids]
    if project_ids and len(selected) != len(project_ids):
        missing = sorted(project_ids - {row["project_id"] for row in selected})
        raise SystemExit(f"requested project ids are not auth-enabled/live: {','.join(missing)}")

    by_workspace: dict[str, list[dict[str, Any]]] = {}
    for row in memberships:
        by_workspace.setdefault(row["workspace_id"], []).append(dict(row))

    cloud_url = str(env["CLOUD_CONTROL_API_URL"]).rstrip("/")
    headers = {"Authorization": f"Bearer {env['CLOUD_INTERNAL_TOKEN']}"}
    result: list[dict[str, Any]] = []
    async with httpx.AsyncClient(timeout=httpx.Timeout(20), headers=headers) as client:
        for row in selected:
            project_id = row["project_id"]
            config_response = await client.get(f"{cloud_url}/v1/cloud/control/projects/{project_id}/auth-config")
            config_response.raise_for_status()
            cloud_config = config_response.json()
            users_response = await client.get(f"{cloud_url}/v1/projects/{project_id}/auth/users?limit=100")
            users_response.raise_for_status()
            payload = users_response.json()
            envelope = payload.get("data") if isinstance(payload.get("data"), dict) else payload
            users = envelope.get("users", []) if isinstance(envelope, dict) else []
            if not isinstance(users, list):
                raise ValueError(f"invalid user response for {project_id}")

            expected = []
            for member in by_workspace.get(row["workspace_id"], []):
                role = str(member.get("role_name") or "member").lower()
                if member["user_id"] == row["owner_id"] or role == "admin":
                    expected.append(identity_key(member["email"]))
            expected_set = set(expected)
            cloud_inherited = {identity_key(str(user.get("email"))) for user in users if inherited(user) and user.get("email")}
            cloud_all = {identity_key(str(user.get("email"))) for user in users if user.get("email")}
            role_mismatches = 0
            for user in users:
                if not inherited(user) or not user.get("email"):
                    continue
                metadata = user.get("user_metadata") if isinstance(user.get("user_metadata"), dict) else {}
                expected_role = "super_admin" if identity_key(str(user["email"])) == identity_key(next((m["email"] for m in by_workspace.get(row["workspace_id"], []) if m["user_id"] == row["owner_id"]), "__none__")) else "admin"
                if str(user.get("role") or "") != expected_role and identity_key(str(user["email"])) in expected_set:
                    role_mismatches += 1

            result.append(
                {
                    "project_id": project_id,
                    "workspace_id": row["workspace_id"],
                    "evidence_schema_version": "phase5e2.v1",
                    "target_inputs": {
                        "providers": row["providers"] or [],
                        "auth_methods": row["auth_methods"] or [],
                        "audiences": row["audiences"] or [],
                        "allow_signup": row["allow_signup"],
                        "cloud_auth_methods": cloud_config.get("auth_methods") or [],
                        "cloud_audiences": cloud_config.get("audiences") or [],
                    },
                    "identities": {
                        "workspace_derived_count": len(cloud_inherited),
                        "app_local_count": max(0, len(cloud_all - cloud_inherited)),
                        "expected_workspace_derived_count": len(expected_set),
                        "missing_inherited_count": len(expected_set - cloud_inherited),
                        "stale_inherited_count": len(cloud_inherited - expected_set),
                        "role_mismatch_count": role_mismatches,
                        "app_local_collision_count": len((cloud_all - cloud_inherited) & expected_set),
                    },
                    "password_drift": evidence_unavailable("Cloud user read response omits password_hash"),
                    "auth_events": evidence_unavailable("No deployed read endpoint for cloud_auth_events"),
                    "active_sessions": evidence_unavailable("No deployed read endpoint for project session counts"),
                }
            )

    return {
        "evidence_schema_version": "phase5e2.v1",
        "mode": "read_only",
        "population": len(result),
        "projects": result,
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--project-id", action="append", dest="project_ids")
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()
    output = asyncio.run(collect(set(args.project_ids) if args.project_ids else None))
    rendered = json.dumps(output, sort_keys=True, indent=2) + "\n"
    if args.output:
        args.output.write_text(rendered, encoding="utf-8")
    else:
        print(rendered, end="")


if __name__ == "__main__":
    main()