"""MuseHub garbage collection — prune commits, snapshots, object refs, and orphaned objects.

After a force push, the old commit chain remains in the database but is no longer
reachable from any branch.  This service identifies those orphaned objects and
removes them, keeping the DB consistent with the actual repo state.

Architecture:
    Objects are content-addressed and globally shared.  A single ``musehub_objects``
    row may be referenced by many repos via ``musehub_object_refs``.  GC must never
    delete a storage object that is still referenced by another repo.

    Phase 1 — Commits:  BFS from branch heads; delete unreachable commits.
    Phase 2 — Snapshots: delete snapshots not referenced by any live commit.
    Phase 3 — Refs:     delete ref rows for objects no longer in any live snapshot
                        of *this* repo.
    Phase 4 — Objects:  delete ``musehub_objects`` rows (and storage bytes) for
                        objects with zero remaining refs across *all* repos.

Usage:

    from musehub.services.musehub_gc import run_gc
    result = await run_gc(session, repo_id)

The returned ``GCResult`` contains counts of what was deleted.
"""

import logging
from dataclasses import dataclass, field

import msgpack
from sqlalchemy import delete, select
from sqlalchemy.engine import CursorResult
from sqlalchemy.ext.asyncio import AsyncSession

from musehub.db.musehub_repo_models import (
    MusehubBranch,
    MusehubCommit,
    MusehubCommitRef,
    MusehubFetchMPackCache,
    MusehubObject,
    MusehubObjectRef,
    MusehubSnapshot,
)
from musehub.storage import get_backend

type _CommitGraph = dict[str, list[str]]

logger = logging.getLogger(__name__)

@dataclass
class GCResult:
    """Counts of what was pruned during a single GC run.

    Attributes:
        repo_id: The repo this GC run targeted.
        commits_deleted: Commits unreachable from all branch heads.
        snapshots_deleted: Snapshots no longer referenced by any live commit.
        object_refs_deleted: Rows removed from ``musehub_object_refs`` for this repo.
        objects_deleted: Globally orphaned ``musehub_objects`` rows removed from DB.
        storage_bytes_freed: Bytes removed from R2/local storage (best-effort; 0 if
            the backend does not report sizes).
        reachable_commit_count: How many commits survived (for observability).
        errors: Non-fatal errors (e.g. storage delete failed for one object).
    """

    repo_id: str
    commits_deleted: int = 0
    snapshots_deleted: int = 0
    object_refs_deleted: int = 0
    objects_deleted: int = 0
    storage_bytes_freed: int = 0
    reachable_commit_count: int = 0
    errors: list[str] = field(default_factory=list)

def _decode_manifest(blob: bytes | None) -> set[str]:
    """Decode a msgpack manifest blob and return the set of object_ids it contains.

    Manifests are stored as ``{path: object_id}`` msgpack maps.  Returns an empty
    set if ``blob`` is None or malformed.
    """
    if not blob:
        return set()
    try:
        mapping = msgpack.unpackb(blob, raw=False)
        return {v for v in mapping.values() if isinstance(v, str)}
    except Exception:
        return set()

async def run_gc(session: AsyncSession, repo_id: str) -> GCResult:
    """Prune all commits, snapshots, object refs, and orphaned objects for a repo.

    Algorithm:
    1. Load all branch heads for the repo from ``musehub_branches``.
    2. Walk from every head via walk_dag to collect reachable commit IDs.
    3. Collect snapshot_ids from orphaned commits (capture manifest blobs first).
    4. Collect live snapshot object_ids from surviving commits.
    5. DELETE orphaned commits.
    6. DELETE orphaned snapshots (those not referenced by any live commit).
    7. DELETE ref rows in ``musehub_object_refs`` for objects no longer in any
       live snapshot of this repo.
    8. Find objects globally unreferenced (zero ref rows remaining across all repos).
    9. DELETE those rows from ``musehub_objects``; delete bytes from storage.

    Steps 7-9 implement the object-level GC that makes the ref table the single
    source of truth for object reachability.
    """
    result = GCResult(repo_id=repo_id)
    backend = get_backend()

    # ── 1. Collect branch heads ──────────────────────────────────────────────
    branches_result = await session.execute(
        select(MusehubBranch.head_commit_id).where(
            MusehubBranch.repo_id == repo_id
        )
    )
    heads = [row[0] for row in branches_result.fetchall() if row[0]]

    if not heads:
        logger.warning("GC: no branch heads found for repo %s — skipping", repo_id)
        return result

    # ── 2. Load all commits for this repo, then BFS from every head ──────────
    all_result = await session.execute(
        select(MusehubCommit.commit_id, MusehubCommit.parent_ids)
        .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id)
        .where(MusehubCommitRef.repo_id == repo_id)
    )
    all_commits: _CommitGraph = {
        row[0]: (row[1] or []) for row in all_result.fetchall()
    }

    from muse.core.graph import walk_dag
    reachable: set[str] = {
        cid for cid in walk_dag(heads, lambda cid: all_commits.get(cid, []))
        if cid in all_commits
    }

    result.reachable_commit_count = len(reachable)
    orphaned_commit_ids = [cid for cid in all_commits if cid not in reachable]

    if not orphaned_commit_ids:
        logger.info("GC: repo %s is clean — no orphaned commits", repo_id)
        return result

    logger.info(
        "GC: repo %s — %d reachable, %d orphaned commits to prune",
        repo_id, len(reachable), len(orphaned_commit_ids),
    )

    # ── 3. Collect orphaned snapshot IDs + manifest blobs (before deletion) ──
    orphaned_snap_result = await session.execute(
        select(MusehubSnapshot.snapshot_id, MusehubSnapshot.manifest_blob).where(
            MusehubCommit.commit_id.in_(orphaned_commit_ids),
            MusehubCommit.snapshot_id == MusehubSnapshot.snapshot_id,
            MusehubCommit.snapshot_id.isnot(None),
        )
    )
    orphaned_snap_rows = orphaned_snap_result.fetchall()
    orphaned_snapshot_ids = [row[0] for row in orphaned_snap_rows if row[0]]

    # Decode object_ids referenced by orphaned snapshots.
    orphaned_object_ids: set[str] = set()
    for _, blob in orphaned_snap_rows:
        orphaned_object_ids.update(_decode_manifest(blob))

    # ── 4. Collect live snapshot object_ids (snapshots reachable from heads) ─
    if reachable:
        live_snap_result = await session.execute(
            select(MusehubSnapshot.snapshot_id, MusehubSnapshot.manifest_blob)
            .join(MusehubCommit, MusehubCommit.snapshot_id == MusehubSnapshot.snapshot_id)
            .where(
                MusehubCommit.commit_id.in_(list(reachable)),
                MusehubCommit.snapshot_id.isnot(None),
            )
        )
        live_snap_rows = live_snap_result.fetchall()
        live_snapshot_ids = {row[0] for row in live_snap_rows if row[0]}
        live_object_ids: set[str] = set()
        for _, blob in live_snap_rows:
            live_object_ids.update(_decode_manifest(blob))
    else:
        live_snapshot_ids = set()
        live_object_ids = set()

    truly_orphaned_snapshots = [
        sid for sid in orphaned_snapshot_ids if sid not in live_snapshot_ids
    ]

    # ── 5. Delete orphaned commits ────────────────────────────────────────────
    del_commits = await session.execute(
        delete(MusehubCommit).where(
            MusehubCommit.commit_id.in_(orphaned_commit_ids)
        )
    )
    assert isinstance(del_commits, CursorResult)
    result.commits_deleted = del_commits.rowcount

    # ── 6. Delete orphaned snapshots ─────────────────────────────────────────
    if truly_orphaned_snapshots:
        del_snaps = await session.execute(
            delete(MusehubSnapshot).where(
                MusehubSnapshot.snapshot_id.in_(truly_orphaned_snapshots)
            )
        )
        assert isinstance(del_snaps, CursorResult)
        result.snapshots_deleted = del_snaps.rowcount

    await session.commit()

    # ── 7. Prune stale object refs for this repo ─────────────────────────────
    # An object_id that appeared only in orphaned snapshots (not in any live
    # snapshot) no longer has a logical reference from this repo.
    stale_oids = orphaned_object_ids - live_object_ids
    if stale_oids:
        del_refs = await session.execute(
            delete(MusehubObjectRef).where(
                MusehubObjectRef.repo_id == repo_id,
                MusehubObjectRef.object_id.in_(list(stale_oids)),
            )
        )
        assert isinstance(del_refs, CursorResult)
        result.object_refs_deleted = del_refs.rowcount
        await session.commit()

        logger.info(
            "GC: repo %s — pruned %d stale object refs",
            repo_id, result.object_refs_deleted,
        )

    # ── 8. Find globally orphaned objects (zero remaining refs) ──────────────
    # Only bother checking the stale candidates — objects still referenced by
    # other repos will have surviving ref rows and will not appear here.
    if stale_oids:
        still_referenced_result = await session.execute(
            select(MusehubObjectRef.object_id).where(
                MusehubObjectRef.object_id.in_(list(stale_oids))
            ).distinct()
        )
        still_referenced = {row[0] for row in still_referenced_result.fetchall()}
        globally_orphaned = stale_oids - still_referenced
    else:
        globally_orphaned = set()

    # ── 9. Delete globally orphaned objects from DB + storage ─────────────────
    if globally_orphaned:
        logger.info(
            "GC: repo %s — %d globally orphaned objects to delete from storage",
            repo_id, len(globally_orphaned),
        )
        for oid in globally_orphaned:
            try:
                await backend.delete(oid)
            except Exception as exc:
                msg = f"storage delete failed for {oid}: {exc}"
                logger.warning("GC: %s", msg)
                result.errors.append(msg)

        del_objs = await session.execute(
            delete(MusehubObject).where(
                MusehubObject.object_id.in_(list(globally_orphaned))
            )
        )
        assert isinstance(del_objs, CursorResult)
        result.objects_deleted = del_objs.rowcount
        await session.commit()

        logger.info(
            "GC: repo %s — deleted %d objects from DB",
            repo_id, result.objects_deleted,
        )

    logger.info(
        "GC complete: repo %s — commits=%d snapshots=%d refs=%d objects=%d errors=%d",
        repo_id,
        result.commits_deleted,
        result.snapshots_deleted,
        result.object_refs_deleted,
        result.objects_deleted,
        len(result.errors),
    )
    return result


async def gc_fetch_mpack_cache(session: AsyncSession, repo_id: str) -> int:
    """Delete expired fetch mpack cache rows and their R2 objects for a repo.

    Called by the gc worker handler after every normal GC run.  Only touches
    rows whose ``expires_at`` is in the past — fresh entries are never removed.

    Returns the number of cache rows deleted.
    """
    from datetime import datetime, timezone as _tz
    now = datetime.now(tz=_tz.utc)

    expired_q = await session.execute(
        select(MusehubFetchMPackCache.cache_id, MusehubFetchMPackCache.mpack_id)
        .where(MusehubFetchMPackCache.repo_id == repo_id)
        .where(MusehubFetchMPackCache.expires_at <= now)
    )
    expired_rows = expired_q.all()
    if not expired_rows:
        return 0

    backend = get_backend()
    for _, mpack_id in expired_rows:
        try:
            await backend.delete(mpack_id)
        except Exception as exc:
            logger.warning(
                "gc_fetch_mpack_cache: R2 delete failed mpack_id=%s: %s",
                mpack_id[:20] if mpack_id else "?", exc,
            )

    expired_cache_ids = [row[0] for row in expired_rows]
    await session.execute(
        delete(MusehubFetchMPackCache)
        .where(MusehubFetchMPackCache.cache_id.in_(expired_cache_ids))
    )
    logger.info(
        "gc_fetch_mpack_cache: repo=%s deleted=%d expired cache rows",
        repo_id, len(expired_cache_ids),
    )
    return len(expired_cache_ids)


async def invalidate_fetch_mpack_cache(
    session: AsyncSession,
    repo_id: str,
    tips: list[str] | None = None,
) -> int:
    """Actively delete fetch-mpack cache rows (and best-effort their R2 mpacks).

    Unlike gc_fetch_mpack_cache (expired-only), this deletes rows REGARDLESS of
    expires_at — the content behind the cached mpack is no longer trustworthy.

    tips=None  → delete ALL rows for repo_id (repair: a repaired object/snapshot/
                 commit may be reachable from any tip; repo-wide is the only safe
                 scope because there is no reverse object→tip index).
    tips=[...] → delete only the rows for those specific tip_commit_ids (force-push:
                 scope to the branch's old + new tip so sibling caches survive).

    Does NOT commit — the caller owns the transaction so invalidation is atomic
    with the repair/branch-advance it accompanies (Design Decision D2).  A caller
    that rolls back will also roll back the deletion.

    Returns the number of cache rows deleted.
    """
    if tips is not None:
        q = (
            select(MusehubFetchMPackCache.cache_id, MusehubFetchMPackCache.mpack_id)
            .where(MusehubFetchMPackCache.repo_id == repo_id)
            .where(MusehubFetchMPackCache.tip_commit_id.in_(tips))
        )
    else:
        q = (
            select(MusehubFetchMPackCache.cache_id, MusehubFetchMPackCache.mpack_id)
            .where(MusehubFetchMPackCache.repo_id == repo_id)
        )

    rows = (await session.execute(q)).all()
    if not rows:
        return 0

    backend = get_backend()
    for _, mpack_id in rows:
        try:
            await backend.delete(mpack_id)
        except Exception as exc:
            logger.warning(
                "invalidate_fetch_mpack_cache: R2 delete failed mpack_id=%s: %s",
                mpack_id[:20] if mpack_id else "?", exc,
            )

    cache_ids = [row[0] for row in rows]
    await session.execute(
        delete(MusehubFetchMPackCache)
        .where(MusehubFetchMPackCache.cache_id.in_(cache_ids))
    )
    logger.info(
        "invalidate_fetch_mpack_cache: repo=%s tips=%s deleted=%d rows",
        repo_id,
        "all" if tips is None else len(tips),
        len(cache_ids),
    )
    return len(cache_ids)