"""Read-only Core↔repository source-authority scan for Phase 5E.5A.

ProjectRevision.git_sha is compared with the configured preview branch for
generated projects. The default branch is reported separately as publication
authority; it is never treated as the generated-source authority implicitly.
"""

from __future__ import annotations

import asyncio
import json
from collections import Counter
from pathlib import Path
from typing import Any

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

from src.domain.project_auth.source_authority import classify_source_authority

ROOT = Path(__file__).parents[1]
OUTPUT = ROOT / "phase5e5-source-authority-scan.json"


async def git_ls_remote(repo_url: str) -> tuple[dict[str, str], str | None]:
    proc = await asyncio.create_subprocess_exec(
        "git",
        "ls-remote",
        "--heads",
        repo_url,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    if proc.returncode != 0:
        return {}, (stderr or b"").decode(errors="replace").strip()[:500]
    refs: dict[str, str] = {}
    for line in (stdout or b"").decode(errors="replace").splitlines():
        sha, _, ref = line.partition("\t")
        if sha and ref.startswith("refs/heads/"):
            refs[ref.removeprefix("refs/heads/")] = sha
    return refs, None


def repo_url(row: dict[str, Any]) -> str | None:
    if row.get("source_kind") != "generated":
        return None
    full_name = row.get("source_repo_full_name") or (
        f"P2HS/{row['github_repo_name']}" if row.get("github_repo_name") else None
    )
    return f"https://github.com/{full_name}.git" if full_name else None


def classify(row: dict[str, Any], refs: dict[str, str], error: str | None) -> str:
    if row.get("source_kind") != "generated":
        return "NOT_APPLICABLE"
    default_branch = str(row.get("default_branch") or "main")
    preview_branch = str(row.get("preview_branch") or "preview")
    if error or not refs:
        return "INSUFFICIENT_EVIDENCE"
    return classify_source_authority(
        generated_ref=preview_branch,
        generated_sha=row.get("active_git_sha"),
        publication_ref=default_branch,
        publication_sha=refs.get(default_branch),
        deployment_sha=row.get("latest_deploy_sha") or row.get("production_target_sha"),
        reachable_refs=refs,
    ).classification


async def main() -> None:
    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,
                           p.status::text AS status, p.source_kind, p.github_repo_name,
                           p.source_repo_full_name, p.source_branch,
                           p.active_revision_id::text AS active_revision_id,
                           r.git_sha AS active_git_sha, r.source AS active_revision_source,
                           r.source_id::text AS active_revision_source_id,
                           a.enabled AS auth_enabled,
                           pr.default_branch, pr.preview_branch,
                           prod.latest_commit_sha AS production_target_sha,
                           prev.latest_commit_sha AS preview_target_sha,
                           pd.source_git_sha AS latest_deploy_sha,
                           pd.status::text AS latest_deploy_status,
                           pd.environment AS latest_deploy_environment
                    FROM projects p
                    LEFT JOIN project_revisions r ON r.id = p.active_revision_id
                    LEFT JOIN project_auth_config a ON a.project_id = p.id
                    LEFT JOIN project_repos pr ON pr.project_id = p.id
                    LEFT JOIN project_deployment_targets prod
                      ON prod.project_id = p.id AND prod.environment = 'production'
                    LEFT JOIN project_deployment_targets prev
                      ON prev.project_id = p.id AND prev.environment = 'preview'
                    LEFT JOIN LATERAL (
                      SELECT d.source_git_sha, d.status, d.environment
                      FROM project_deploys d
                      WHERE d.project_id = p.id
                      ORDER BY d.created_at DESC
                      LIMIT 1
                    ) pd ON TRUE
                    WHERE p.deleted_at IS NULL
                      AND p.github_repo_name IS NOT NULL
                    ORDER BY p.id
                    """
                )
            )
        ).mappings().all()
    await engine.dispose()

    async def inspect(row: Any) -> dict[str, Any]:
        item = dict(row)
        url = repo_url(item)
        refs: dict[str, str] = {}
        error = None
        if url:
            refs, error = await git_ls_remote(url)
        classification = classify(item, refs, error)
        default_branch = str(item.get("default_branch") or "main")
        preview_branch = str(item.get("preview_branch") or "preview")
        active = item.get("active_git_sha")
        default_sha = refs.get(default_branch)
        preview_sha = refs.get(preview_branch)
        deployment_sha = item.get("latest_deploy_sha") or item.get("production_target_sha")
        item.update(
            {
                "repository_url": url,
                "remote_refs": refs,
                "remote_error": error,
                "default_head_sha": default_sha,
                "preview_head_sha": preview_sha,
                "active_sha_reachable": bool(active and active in refs.values()),
                "active_matches_default": bool(active and active == default_sha),
                "active_matches_preview": bool(active and active == preview_sha),
                "deployment_disagrees_with_default_and_active": bool(
                    deployment_sha
                    and deployment_sha != default_sha
                    and deployment_sha != active
                ),
                "classification": classification,
            }
        )
        return item

    records = await asyncio.gather(*(inspect(row) for row in rows))
    counts = Counter(str(row["classification"]) for row in records)
    auth_records = [row for row in records if row.get("auth_enabled")]
    output = {
        "mode": "read_only",
        "population": len(records),
        "auth_enabled_population": len(auth_records),
        "classification_counts": dict(sorted(counts.items())),
        "auth_enabled_classification_counts": dict(
            sorted(Counter(str(row["classification"]) for row in auth_records).items())
        ),
        "deployment_disagreement_count": sum(
            bool(row["deployment_disagrees_with_default_and_active"]) for row in records
        ),
        "records": records,
    }
    assert sum(counts.values()) == len(records)
    OUTPUT.write_text(json.dumps(output, sort_keys=True, indent=2, default=str) + "\n", encoding="utf-8")
    print(json.dumps({k: output[k] for k in output if k != "records"}, sort_keys=True, indent=2))


if __name__ == "__main__":
    asyncio.run(main())