"""TDD — musehub#195 narrow fix: committed_at_raw preserves commit identity
across the Postgres timestamptz UTC-normalization boundary.

Root cause: a commit's content-addressed id is a hash whose preimage
includes the exact committed_at ISO string the client used
(muse/core/ids.py::hash_commit). musehub_commits.timestamp is a Postgres
timestamptz, which normalizes any non-UTC offset to UTC on write. The wire-
serve path (_to_wire_commit) previously reconstructed committed_at via
`timestamp.isoformat()` — for a commit originally authored with a non-UTC
offset, this produces a DIFFERENT string than the one baked into commit_id,
so the client's own hash check correctly rejects the downloaded commit.

Fix: an additive `committed_at_raw` column stores the exact original string.
The serve path prefers it verbatim; when absent (legacy rows), it falls
back to `timestamp.isoformat()` unchanged from prior behavior.

verify_commit_identity_or_raise (a standalone, explicitly-callable helper --
NOT wired automatically into the serve path) recomputes a commit's id from
its wire representation and raises ObjectHashMismatch on mismatch. It is
deliberately NOT called automatically by _commit_to_wire_s3/wire_fetch_mpack:
confirmed experimentally that at least two legitimate write sites
(musehub_repository.py's repo-init synthetic commit, musehub_sync.py's
synthetic content commit) never used hash_commit-derived ids in the first
place, and wire_repair_commit's entire purpose is repairing a commit whose
identity fields don't currently reproduce its id -- an automatic check on
every serve broke that flow outright. Available for explicit/diagnostic use
the same way wire_repair_commit already uses the same recompute-and-compare
pattern.

Tests:
  U1  _to_wire_commit uses committed_at_raw verbatim when present -- the
      actual regression this closes.
  U2  _to_wire_commit falls back to timestamp.isoformat() when raw is
      absent -- unchanged behavior for legacy rows, no regression.
  U3  verify_commit_identity_or_raise passes for a self-consistent commit.
  U4  verify_commit_identity_or_raise raises ObjectHashMismatch for a
      mismatched commit (available for explicit/diagnostic use).
  E1  wire_fetch_mpack serves the exact original non-UTC string when
      committed_at_raw is present.
  E3  wire_fetch_mpack still serves fine for a legacy row with
      committed_at_raw=NULL when the original was already UTC (the common
      case — reconstruction matches, no regression for existing data).
"""
from __future__ import annotations

from datetime import datetime, timezone

import pytest
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from muse.core.mpack import parse_wire_mpack
from muse.core.types import blob_id
from musehub.db import musehub_repo_models as db
from musehub.muse_cli.snapshot import compute_commit_id as hub_compute_commit_id
from musehub.services.musehub_wire_shared import ObjectHashMismatch, _to_wire_commit
from tests.factories import create_repo

_NON_UTC_TS = "2026-03-18T12:00:00-07:00"
_UTC_TS = "2026-03-18T19:00:00+00:00"  # same instant as _NON_UTC_TS


def _commit_row(
    *,
    commit_id: str,
    committed_at_iso: str,
    committed_at_raw: str | None,
    snapshot_id: str = "",
) -> db.MusehubCommit:
    return db.MusehubCommit(
        commit_id=commit_id,
        branch="main",
        parent_ids=[],
        message="m",
        author="gabriel",
        timestamp=datetime.fromisoformat(committed_at_iso),
        committed_at_raw=committed_at_raw,
        snapshot_id=snapshot_id or None,
    )


# ---------------------------------------------------------------------------
# U1/U2 — _to_wire_commit
# ---------------------------------------------------------------------------


class TestToWireCommitPrefersRaw:
    def test_uses_committed_at_raw_verbatim_when_present(self) -> None:
        # timestamp simulates what Postgres normalized the row to (UTC) --
        # deliberately DIFFERENT from committed_at_raw, so this test actually
        # discriminates "prefers raw" from "always reconstructs from timestamp".
        row = _commit_row(
            commit_id="sha256:" + "a" * 64,
            committed_at_iso=_UTC_TS,
            committed_at_raw=_NON_UTC_TS,
        )
        wire = _to_wire_commit(row)
        assert wire.committed_at == _NON_UTC_TS
        assert wire.committed_at != row.timestamp.isoformat()

    def test_falls_back_to_timestamp_isoformat_when_raw_absent(self) -> None:
        row = _commit_row(
            commit_id="sha256:" + "b" * 64,
            committed_at_iso=_UTC_TS,
            committed_at_raw=None,
        )
        wire = _to_wire_commit(row)
        assert wire.committed_at == row.timestamp.isoformat()


# ---------------------------------------------------------------------------
# U3/U4 — fail-closed verification helper
# ---------------------------------------------------------------------------


class TestVerifyCommitIdentityOrRaise:
    def test_passes_for_self_consistent_commit(self) -> None:
        from musehub.services.musehub_wire_shared import verify_commit_identity_or_raise

        snap_id = "sha256:" + "0" * 64
        commit_id = hub_compute_commit_id(
            parent_ids=[], snapshot_id=snap_id, message="m",
            committed_at_iso=_NON_UTC_TS, author="gabriel", signer_public_key="",
        )
        row = _commit_row(
            commit_id=commit_id,
            committed_at_iso=_NON_UTC_TS,
            committed_at_raw=_NON_UTC_TS,
            snapshot_id=snap_id,
        )
        wire = _to_wire_commit(row)
        verify_commit_identity_or_raise(row, wire)  # must not raise

    def test_raises_for_mismatched_commit(self) -> None:
        from musehub.services.musehub_wire_shared import verify_commit_identity_or_raise

        snap_id = "sha256:" + "0" * 64
        commit_id = hub_compute_commit_id(
            parent_ids=[], snapshot_id=snap_id, message="m",
            committed_at_iso=_NON_UTC_TS, author="gabriel", signer_public_key="",
        )
        # Row's timestamp reconstructs to the UTC-normalized string, not the
        # original non-UTC one baked into commit_id -- and raw is absent.
        row = _commit_row(
            commit_id=commit_id,
            committed_at_iso=_UTC_TS,
            committed_at_raw=None,
            snapshot_id=snap_id,
        )
        wire = _to_wire_commit(row)
        with pytest.raises(ObjectHashMismatch):
            verify_commit_identity_or_raise(row, wire)


# ---------------------------------------------------------------------------
# E1-E3 — wire_fetch_mpack end-to-end
# ---------------------------------------------------------------------------


def _stub_backend(monkeypatch: pytest.MonkeyPatch) -> dict[str, bytes]:
    from unittest.mock import AsyncMock

    store: dict[str, bytes] = {}

    async def _put(oid: str, data: bytes, **_: object) -> str:
        store[oid] = data
        return f"mem://{oid}"

    async def _get(oid: str) -> bytes | None:
        return store.get(oid)

    async def _exists(oid: str, **_: object) -> bool:
        return oid in store

    async def _presign_get(oid: str, ttl: int) -> str:
        return f"https://minio.test/{oid}?ttl={ttl}"

    async def _get_mpack(mpack_id: str) -> bytes | None:
        return store.get(mpack_id)

    async def _put_mpack(mpack_id: str, data: bytes) -> str:
        store[mpack_id] = data
        return f"mem://mpacks/{mpack_id}"

    async def _presign_mpack_get(mpack_id: str, ttl: int) -> str:
        return f"https://minio.test/mpacks/{mpack_id}?ttl={ttl}"

    backend = AsyncMock()
    backend.put = _put
    backend.get = _get
    backend.exists = _exists
    backend.presign_get = _presign_get
    backend.get_mpack = _get_mpack
    backend.put_mpack = _put_mpack
    backend.presign_mpack_get = _presign_mpack_get
    backend.supports_presign = True
    monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend)
    monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend)
    monkeypatch.setattr("musehub.services.musehub_wire_shared.get_backend", lambda: backend)
    return store


async def _seed_commit(
    session: AsyncSession,
    repo_id: str,
    *,
    committed_at_iso: str,
    committed_at_raw: str | None,
    seed: str,
) -> str:
    import msgpack

    from muse.core.ids import hash_snapshot as _hash_snapshot

    manifest: dict[str, str] = {}
    snap_id = _hash_snapshot(manifest, [])
    await session.execute(
        pg_insert(db.MusehubSnapshot)
        .values(
            snapshot_id=snap_id,
            directories=[],
            entry_count=0,
            manifest_blob=msgpack.packb(manifest, use_bin_type=True),
            created_at=datetime.now(timezone.utc),
        )
        .on_conflict_do_nothing(index_elements=["snapshot_id"])
    )
    await session.execute(
        pg_insert(db.MusehubSnapshotRef)
        .values(repo_id=repo_id, snapshot_id=snap_id)
        .on_conflict_do_nothing()
    )

    message = f"commit {seed}"
    commit_id = hub_compute_commit_id(
        parent_ids=[], snapshot_id=snap_id, message=message,
        committed_at_iso=committed_at_iso, author="gabriel", signer_public_key="",
    )
    session.add(db.MusehubCommit(
        commit_id=commit_id,
        branch="main",
        parent_ids=[],
        message=message,
        author="gabriel",
        timestamp=datetime.fromisoformat(committed_at_iso),
        committed_at_raw=committed_at_raw,
        snapshot_id=snap_id,
    ))
    await session.execute(
        pg_insert(db.MusehubCommitRef)
        .values(repo_id=repo_id, commit_id=commit_id)
        .on_conflict_do_nothing()
    )
    await session.execute(
        pg_insert(db.MusehubCommitGraph)
        .values(commit_id=commit_id, parent_ids=[], generation=0, snapshot_id=snap_id)
        .on_conflict_do_nothing()
    )
    await session.commit()
    # Force a real re-read from Postgres -- without this, SQLAlchemy's
    # identity map keeps serving the in-memory datetime we constructed (with
    # its original, un-normalized offset) instead of what the DB actually
    # stored, masking the exact bug this test exists to catch.
    from sqlalchemy import select as _select
    await session.execute(
        _select(db.MusehubCommit)
        .where(db.MusehubCommit.commit_id == commit_id)
        .execution_options(populate_existing=True)
    )
    return commit_id


@pytest.mark.asyncio
async def test_e1_serves_exact_non_utc_string_when_raw_present(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    from musehub.services.musehub_wire import wire_fetch_mpack

    store = _stub_backend(monkeypatch)
    repo = await create_repo(db_session, owner="gabriel", visibility="public")
    commit_id = await _seed_commit(
        db_session, repo.repo_id,
        committed_at_iso=_NON_UTC_TS, committed_at_raw=_NON_UTC_TS, seed="e1",
    )

    result = await wire_fetch_mpack(
        db_session, repo.repo_id, want=[commit_id], have=[], force_build=True
    )
    mpack = parse_wire_mpack(store[result["mpack_id"]])
    served = next(c for c in mpack["commits"] if c["commit_id"] == commit_id)
    assert served["committed_at"] == _NON_UTC_TS


@pytest.mark.asyncio
async def test_e2_legacy_non_utc_row_missing_raw_is_unchanged_not_worsened(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Explicit scope boundary: a pre-migration row authored non-UTC, with
    no committed_at_raw to fall back on, still serves a mismatched
    committed_at exactly as it did before this fix -- NOT newly rejected.

    This is deliberate, not an oversight: an automatic fail-closed check
    was tried and reverted after it broke wire_repair_commit (whose entire
    purpose is repairing an already-mismatched commit) and at least two
    legitimate write sites that never used hash_commit-derived ids at all
    (repo-init, sync's synthetic content commit). verify_commit_identity_or_raise
    still exists and is correct (see TestVerifyCommitIdentityOrRaise) for
    explicit/diagnostic use -- it is just not an automatic gate on every serve.
    Fixing already-affected legacy data is out of scope for this narrow fix.
    """
    from musehub.services.musehub_wire import wire_fetch_mpack

    store = _stub_backend(monkeypatch)
    repo = await create_repo(db_session, owner="gabriel", visibility="public")
    commit_id = await _seed_commit(
        db_session, repo.repo_id,
        committed_at_iso=_NON_UTC_TS, committed_at_raw=None, seed="e2",
    )

    result = await wire_fetch_mpack(
        db_session, repo.repo_id, want=[commit_id], have=[], force_build=True
    )
    mpack = parse_wire_mpack(store[result["mpack_id"]])
    served = next(c for c in mpack["commits"] if c["commit_id"] == commit_id)
    # Reconstructed from the UTC-normalized `timestamp`, not the original --
    # same lossy behavior as before this fix, for pre-existing data only.
    assert served["committed_at"] != _NON_UTC_TS


@pytest.mark.asyncio
async def test_e3_legacy_utc_row_still_serves_fine(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """No regression for the common case: a legacy row with no
    committed_at_raw, originally authored in UTC, still serves normally --
    reconstruction matches, nothing to fail closed on."""
    from musehub.services.musehub_wire import wire_fetch_mpack

    store = _stub_backend(monkeypatch)
    repo = await create_repo(db_session, owner="gabriel", visibility="public")
    commit_id = await _seed_commit(
        db_session, repo.repo_id,
        committed_at_iso=_UTC_TS, committed_at_raw=None, seed="e3",
    )

    result = await wire_fetch_mpack(
        db_session, repo.repo_id, want=[commit_id], have=[], force_build=True
    )
    mpack = parse_wire_mpack(store[result["mpack_id"]])
    served = next(c for c in mpack["commits"] if c["commit_id"] == commit_id)
    assert served["committed_at"] == _UTC_TS