"""Regression tests: delta+zlib push when the base object is stored zlib-compressed.

Root cause:
    Objects pushed via the old wire path were stored zlib-compressed in R2.
    When a subsequent push sends a new object as delta+zlib against such a base,
    wire_push_object_pack fetches the base via backend.get() — which returns the
    raw R2 bytes (zlib-compressed), not the plain content.  apply_delta() then
    operates on compressed bytes and produces garbage.  Hash verification is
    skipped for non-"sha256:"-prefixed IDs, so the garbage passes silently.

Real-world failure:
    staging.musehub.ai/gabriel/muse-zsh showed "0!" in the README after a one-line
    README edit was pushed as delta+zlib against the old zlib-compressed README
    object already in R2.

Fix:
    1. musehub/muse_contracts/compression.py — expose decompress_if_needed().
    2. musehub/services/musehub_wire.py — apply decompress_if_needed() to base_raw
       before calling apply_delta() so the delta is always applied against plain bytes.
    3. musehub/api/routes/wire.py + musehub/services/musehub_wire.py — add
       POST /{owner}/{slug}/repair-object endpoint so operators can correct an object
       already stored with wrong bytes without requiring direct DB/R2 access.
"""
from __future__ import annotations

import hashlib
import struct
import zlib

import msgpack
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from tests.factories import create_repo as factory_create_repo
from musehub.db import musehub_models as db
from musehub.types.json_types import JSONValue, StrDict


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _sha256_oid(raw: bytes) -> str:
    """Canonical sha256:-prefixed object_id — required by WireObject validator."""
    return "sha256:" + hashlib.sha256(raw).hexdigest()


def _compress_zlib(raw: bytes) -> bytes:
    return zlib.compress(raw)


def _make_delta_stream(target: bytes) -> bytes:
    """Minimal DATA-only delta (no COPY), zlib-compressed.

    Format: b'\\x01' + struct.pack('>I', len) + data
    This is a valid encoding that always works regardless of the base content.
    """
    stream = b"\x01" + struct.pack(">I", len(target)) + target
    return zlib.compress(stream, level=1)


def _mp(data: JSONValue) -> bytes:
    return msgpack.packb(data, use_bin_type=True)


# ---------------------------------------------------------------------------
# Unit tests — decompress_if_needed in compression module
# ---------------------------------------------------------------------------

def test_decompress_if_needed_importable_from_compression_module() -> None:
    """decompress_if_needed must live in muse_contracts.compression, not ui_tree."""
    from musehub.types.compression import decompress_if_needed
    assert callable(decompress_if_needed)


def test_decompress_if_needed_zlib_level_1_magic_78_01() -> None:
    from musehub.types.compression import decompress_if_needed
    data = zlib.compress(b"hello world\n", level=1)
    assert data[:2] == b"\x78\x01"
    assert decompress_if_needed(data) == b"hello world\n"


def test_decompress_if_needed_zlib_level_6_magic_78_9c() -> None:
    from musehub.types.compression import decompress_if_needed
    data = zlib.compress(b"hello world\n", level=6)
    assert data[:2] == b"\x78\x9c"
    assert decompress_if_needed(data) == b"hello world\n"


def test_decompress_if_needed_zlib_level_9_magic_78_da() -> None:
    from musehub.types.compression import decompress_if_needed
    data = zlib.compress(b"hello world\n", level=9)
    assert data[:2] == b"\x78\xda"
    assert decompress_if_needed(data) == b"hello world\n"


def test_decompress_if_needed_plain_text_passthrough() -> None:
    from musehub.types.compression import decompress_if_needed
    data = b"# plain README\n\nSome content.\n"
    assert decompress_if_needed(data) == data


def test_decompress_if_needed_binary_non_zlib_passthrough() -> None:
    from musehub.types.compression import decompress_if_needed
    # PNG magic — must pass through unchanged
    data = b"\x89PNG\r\n\x1a\n" + b"\x00" * 50
    assert decompress_if_needed(data) == data


def test_decompress_if_needed_empty_bytes_passthrough() -> None:
    from musehub.types.compression import decompress_if_needed
    assert decompress_if_needed(b"") == b""


def test_decompress_if_needed_short_data_passthrough() -> None:
    from musehub.types.compression import decompress_if_needed
    assert decompress_if_needed(b"\x78") == b"\x78"  # only 1 byte — no magic match


def test_decompress_if_needed_truncated_zlib_returns_original() -> None:
    from musehub.types.compression import decompress_if_needed
    # Valid header, invalid body — must not raise; returns original bytes.
    data = b"\x78\x9c\x00"
    result = decompress_if_needed(data)
    assert isinstance(result, bytes)
    assert result == data  # original returned on decompression failure


def test_decompress_if_needed_full_readme_roundtrip() -> None:
    from musehub.types.compression import decompress_if_needed
    readme = b"# muse-zsh\n\nOh My ZSH plugin for Muse version control.\n" * 10
    compressed = zlib.compress(readme)
    assert decompress_if_needed(compressed) == readme


# ---------------------------------------------------------------------------
# Integration — baseline: delta against plain base works (before + after fix)
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_push_delta_plain_base_baseline(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """Baseline: delta+zlib against a plain-text base must work before and after fix."""
    repo = await factory_create_repo(
        db_session, slug="delta-plain-base-baseline", owner="test-user-wire"
    )
    path = "README.md"
    base_raw = b"# Repo\n\nInitial content.\n" * 30
    base_oid = _sha256_oid(base_raw)

    r_base = await client.post(
        f"/{repo.owner}/{repo.slug}/push/object-pack",
        content=_mp({"objects": [
            {"object_id": base_oid, "content": base_raw, "path": path, "encoding": "raw"}
        ]}),
        headers=wire_headers,
    )
    assert r_base.status_code == 200

    target_raw = b"# Repo\n\nUpdated content.\n" * 30
    target_oid = _sha256_oid(target_raw)

    r_target = await client.post(
        f"/{repo.owner}/{repo.slug}/push/object-pack",
        content=_mp({"objects": [{
            "object_id": target_oid,
            "content": _make_delta_stream(target_raw),
            "path": path,
            "encoding": "delta+zlib",
            "base_id": base_oid,
        }]}),
        headers=wire_headers,
    )
    assert r_target.status_code == 200
    assert msgpack.unpackb(r_target.content, raw=False)["stored"] == 1


# ---------------------------------------------------------------------------
# Integration — regression: delta against ZLIB-COMPRESSED base
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_push_delta_zlib_base_in_content_cache_succeeds(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """delta+zlib push must succeed when the base is in content_cache as zlib bytes.

    Simulates the old wire path: base object arrived zlib-compressed and was stored
    in content_cache without decompression.  The server must decompress before
    apply_delta(); otherwise the delta is applied against the compressed bytes and
    the result is garbage.
    """
    repo = await factory_create_repo(
        db_session, slug="delta-zlib-cache-base", owner="test-user-wire"
    )
    path = "README.md"

    base_raw = b"# muse-zsh\n\nOh My ZSH plugin.\n" * 30
    base_oid = _sha256_oid(base_raw)
    base_zlib = _compress_zlib(base_raw)

    # Inject base row with ZLIB bytes in content_cache — simulating old wire path.
    await db_session.execute(
        db.MusehubObject.__table__.insert().values(
            object_id=base_oid,
            path=path,
            size_bytes=len(base_raw),
            disk_path="",
            storage_uri="pending",
            content_cache=base_zlib,  # ← zlib bytes, not plain
        )
    )
    db_session.add(db.MusehubObjectRef(repo_id=repo.repo_id, object_id=base_oid))
    await db_session.commit()

    target_raw = b"# muse-zsh\n\nUpdated plugin.\n" * 30
    target_oid = _sha256_oid(target_raw)

    r = await client.post(
        f"/{repo.owner}/{repo.slug}/push/object-pack",
        content=_mp({"objects": [{
            "object_id": target_oid,
            "content": _make_delta_stream(target_raw),
            "path": path,
            "encoding": "delta+zlib",
            "base_id": base_oid,
        }]}),
        headers=wire_headers,
    )
    assert r.status_code == 200, r.text
    assert msgpack.unpackb(r.content, raw=False)["stored"] == 1


@pytest.mark.asyncio
async def test_push_delta_zlib_base_in_storage_succeeds(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """delta+zlib push must succeed when the base is in storage as zlib bytes.

    Simulates R2: base object was uploaded as zlib-compressed (old wire path),
    content_cache is NULL.  backend.get() returns zlib bytes; server must
    decompress before apply_delta().
    """
    import musehub.services.musehub_wire as _wire_svc

    repo = await factory_create_repo(
        db_session, slug="delta-zlib-storage-base", owner="test-user-wire"
    )
    path = "README.md"

    base_raw = b"# muse-zsh\n\nOld content.\n" * 30
    base_oid = _sha256_oid(base_raw)
    base_zlib = _compress_zlib(base_raw)

    backend = _wire_svc.get_backend()
    # Store ZLIB-COMPRESSED bytes in storage — simulating what old push path did.
    await backend.put(base_oid, base_zlib)

    await db_session.execute(
        db.MusehubObject.__table__.insert().values(
            object_id=base_oid,
            path=path,
            size_bytes=len(base_raw),
            disk_path="",
            storage_uri=backend.uri_for(base_oid),
            content_cache=None,  # ← forces fetch from storage backend
        )
    )
    db_session.add(db.MusehubObjectRef(repo_id=repo.repo_id, object_id=base_oid))
    await db_session.commit()

    target_raw = b"# muse-zsh\n\nNew content.\n" * 30
    target_oid = _sha256_oid(target_raw)

    r = await client.post(
        f"/{repo.owner}/{repo.slug}/push/object-pack",
        content=_mp({"objects": [{
            "object_id": target_oid,
            "content": _make_delta_stream(target_raw),
            "path": path,
            "encoding": "delta+zlib",
            "base_id": base_oid,
        }]}),
        headers=wire_headers,
    )
    assert r.status_code == 200, r.text
    assert msgpack.unpackb(r.content, raw=False)["stored"] == 1


@pytest.mark.asyncio
async def test_push_delta_reconstructed_bytes_match_target(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """The bytes written to storage after delta reconstruction must equal the original target.

    The critical assertion: stored != garbage, stored == target_raw.
    """
    import musehub.services.musehub_wire as _wire_svc

    repo = await factory_create_repo(
        db_session, slug="delta-content-exact", owner="test-user-wire"
    )
    path = "README.md"

    base_raw = b"# muse-zsh\n\nOld.\n" * 50
    base_oid = _sha256_oid(base_raw)
    base_zlib = _compress_zlib(base_raw)

    backend = _wire_svc.get_backend()
    await backend.put(base_oid, base_zlib)
    await db_session.execute(
        db.MusehubObject.__table__.insert().values(
            object_id=base_oid, path=path,
            size_bytes=len(base_raw), disk_path="",
            storage_uri=backend.uri_for(base_oid),
            content_cache=None,
        )
    )
    db_session.add(db.MusehubObjectRef(repo_id=repo.repo_id, object_id=base_oid))
    await db_session.commit()

    target_raw = b"# muse-zsh\n\nNew.\n" * 50
    target_oid = _sha256_oid(target_raw)

    r = await client.post(
        f"/{repo.owner}/{repo.slug}/push/object-pack",
        content=_mp({"objects": [{
            "object_id": target_oid,
            "content": _make_delta_stream(target_raw),
            "path": path,
            "encoding": "delta+zlib",
            "base_id": base_oid,
        }]}),
        headers=wire_headers,
    )
    assert r.status_code == 200

    stored = await backend.get(target_oid)
    assert stored is not None, "target object must be written to storage"
    assert stored == target_raw, (
        f"stored bytes do not match target:\n"
        f"  first 80 stored:  {stored[:80]!r}\n"
        f"  first 80 target:  {target_raw[:80]!r}"
    )


# ---------------------------------------------------------------------------
# repair-object endpoint
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_repair_object_corrects_corrupted_stored_bytes(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """POST /repair-object replaces corrupt stored bytes with the verified correct content.

    Simulates the staging scenario: an object has garbage bytes in storage (from a
    failed delta reconstruction).  repair-object accepts the correct raw bytes,
    verifies SHA-256, overwrites storage, and updates the DB row.
    """
    import musehub.services.musehub_wire as _wire_svc

    repo = await factory_create_repo(
        db_session, slug="repair-corrupted-object", owner="test-user-wire"
    )
    correct_raw = b"# README\n\nCorrect content.\n" * 20
    object_id = _sha256_oid(correct_raw)
    garbage = b"\x00\x01\x02\x03garbage\xff"

    backend = _wire_svc.get_backend()
    await backend.put(object_id, garbage)
    await db_session.execute(
        db.MusehubObject.__table__.insert().values(
            object_id=object_id,
            path="README.md",
            size_bytes=len(garbage),
            disk_path="",
            storage_uri=backend.uri_for(object_id),
            content_cache=None,
        )
    )
    db_session.add(db.MusehubObjectRef(repo_id=repo.repo_id, object_id=object_id))
    await db_session.commit()

    r = await client.post(
        f"/{repo.owner}/{repo.slug}/repair-object",
        content=_mp({"object_id": object_id, "content": correct_raw}),
        headers=wire_headers,
    )
    assert r.status_code == 200, r.text
    data = msgpack.unpackb(r.content, raw=False)
    assert data["repaired"] is True

    stored = await backend.get(object_id)
    assert stored == correct_raw


@pytest.mark.asyncio
async def test_repair_object_rejects_hash_mismatch(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """repair-object rejects content whose SHA-256 does not match the declared object_id."""
    repo = await factory_create_repo(
        db_session, slug="repair-hash-mismatch", owner="test-user-wire"
    )
    correct_raw = b"correct content bytes for hash test"
    wrong_content = b"completely different content"
    object_id = _sha256_oid(correct_raw)

    r = await client.post(
        f"/{repo.owner}/{repo.slug}/repair-object",
        content=_mp({"object_id": object_id, "content": wrong_content}),
        headers=wire_headers,
    )
    assert r.status_code == 422, r.text


@pytest.mark.asyncio
async def test_repair_object_accepts_sha256_prefixed_id(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """repair-object works with sha256:-prefixed IDs."""
    import musehub.services.musehub_wire as _wire_svc

    repo = await factory_create_repo(
        db_session, slug="repair-prefixed-id", owner="test-user-wire"
    )
    correct_raw = b"content with prefixed object_id\n" * 10
    object_id_prefixed = _sha256_oid(correct_raw)  # "sha256:<hex>"

    backend = _wire_svc.get_backend()
    await backend.put(object_id_prefixed, b"garbage")
    await db_session.execute(
        db.MusehubObject.__table__.insert().values(
            object_id=object_id_prefixed,
            path="src/file.py",
            size_bytes=7,
            disk_path="",
            storage_uri=backend.uri_for(object_id_prefixed),
            content_cache=None,
        )
    )
    db_session.add(db.MusehubObjectRef(repo_id=repo.repo_id, object_id=object_id_prefixed))
    await db_session.commit()

    r = await client.post(
        f"/{repo.owner}/{repo.slug}/repair-object",
        content=_mp({"object_id": object_id_prefixed, "content": correct_raw}),
        headers=wire_headers,
    )
    assert r.status_code == 200, r.text
    assert msgpack.unpackb(r.content, raw=False)["repaired"] is True


@pytest.mark.asyncio
async def test_repair_object_is_idempotent(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """Calling repair-object twice with the same correct content is safe."""
    import musehub.services.musehub_wire as _wire_svc

    repo = await factory_create_repo(
        db_session, slug="repair-idempotent-test", owner="test-user-wire"
    )
    correct_raw = b"idempotent repair content\n" * 15
    object_id = _sha256_oid(correct_raw)

    backend = _wire_svc.get_backend()
    await backend.put(object_id, b"garbage bytes")
    await db_session.execute(
        db.MusehubObject.__table__.insert().values(
            object_id=object_id,
            path="README.md",
            size_bytes=13,
            disk_path="",
            storage_uri=backend.uri_for(object_id),
            content_cache=None,
        )
    )
    db_session.add(db.MusehubObjectRef(repo_id=repo.repo_id, object_id=object_id))
    await db_session.commit()

    url = f"/{repo.owner}/{repo.slug}/repair-object"
    payload = _mp({"object_id": object_id, "content": correct_raw})

    r1 = await client.post(url, content=payload, headers=wire_headers)
    assert r1.status_code == 200
    r2 = await client.post(url, content=payload, headers=wire_headers)
    assert r2.status_code == 200

    stored = await backend.get(object_id)
    assert stored == correct_raw