import asyncio
import io
import json
import zipfile
from urllib.parse import urlsplit

from fastapi.testclient import TestClient
from sqlalchemy import select

from memrelay.memory_service import build_context, rebuild_memory_curation_states
from memrelay.models import CuratedDocument, MemoryReference, UsageProfile
from tests.conftest import csrf_headers
from tests.fakes import FakeBasicMemory


def create_project(client: TestClient) -> str:
    response = client.post(
        "/api/v1/projects",
        json={"name": "Memory Project"},
        headers=csrf_headers(client),
    )
    return response.json()["id"]


def save_memory(client: TestClient, payload: dict) -> dict:
    response = client.post("/api/v1/memories", json=payload, headers=csrf_headers(client))
    assert response.status_code == 201, response.text
    return response.json()


def test_api_timestamps_carry_utc_offset(initialized_client: TestClient) -> None:
    # SQLite 丢弃时区信息，ORM 读回 naive 值直接 isoformat 会输出无偏移的时间串，
    # 前端 new Date() 与 MCP 客户端会把它误当本地时间（显示偏差 8 小时）。
    # UTCDateTime 类型必须保证 DB 往返后的序列化仍带 +00:00。
    client = initialized_client
    client.app.state.basic_memory = FakeBasicMemory()
    save_memory(
        client,
        {
            "request_id": "request-tz",
            "scope": "global",
            "memory_type": "fact",
            "title": "Timezone check",
            "content": "timestamps must stay UTC-aware",
        },
    )
    listed = client.get("/api/v1/memories?scope=global").json()
    record = next(item for item in listed if item["title"] == "Timezone check")
    for field in ("created_at", "updated_at"):
        value = record[field]
        assert value.endswith("+00:00") or value.endswith("Z"), (field, value)


def test_memory_revision_idempotency_search_and_archive(initialized_client: TestClient) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    project_id = create_project(client)
    payload = {
        "request_id": "request-1",
        "scope": "project",
        "project_id": project_id,
        "memory_type": "decision",
        "title": "Database choice",
        "content": "Use SQLite for system metadata.",
        "tags": ["database"],
    }
    created = save_memory(client, payload)
    assert created["revision"] == 1
    assert fake.write_calls == 1
    assert fake.notes[created["path"]]["metadata"]["stable_id"] == created["id"]
    created_at = fake.notes[created["path"]]["metadata"]["created_at"]

    repeated = save_memory(client, {**payload, "content": "ignored by idempotency"})
    assert repeated == created
    assert fake.write_calls == 1

    conflict = client.post(
        "/api/v1/memories",
        json={
            **payload,
            "id": created["id"],
            "request_id": "request-2",
            "content": "Changed",
            "expected_revision": 2,
        },
        headers=csrf_headers(client),
    )
    assert conflict.status_code == 409
    assert conflict.json()["error"]["code"] == "REVISION_CONFLICT"

    updated = save_memory(
        client,
        {
            **payload,
            "id": created["id"],
            "request_id": "request-3",
            "content": "Use SQLite with WAL for metadata.",
            "expected_revision": 1,
        },
    )
    assert updated["revision"] == 2
    assert fake.notes[updated["path"]]["metadata"]["created_at"] == created_at
    assert (
        client.get(f"/api/v1/memories/{created['id']}").json()["content"].startswith("Use SQLite")
    )

    search = client.get("/api/v1/memories/search", params={"query": "SQLite"}).json()
    assert search["results"][0]["id"] == created["id"]
    assert search["results"][0]["project_id"] == project_id
    assert search["results"][0]["project_name"] == "Memory Project"

    archived = client.post(
        f"/api/v1/memories/{created['id']}/archive",
        params={"expected_revision": 2},
        headers=csrf_headers(client),
    )
    assert archived.status_code == 200
    assert archived.json()["status"] == "archived"
    deleted = client.delete(f"/api/v1/memories/{created['id']}", headers=csrf_headers(client))
    assert deleted.status_code == 204
    assert client.get(f"/api/v1/memories/{created['id']}").status_code == 404


def test_context_order_limit_and_semantic_fallback(initialized_client: TestClient) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    project_id = create_project(client)
    global_rule = save_memory(
        client,
        {
            "request_id": "global-rule",
            "scope": "global",
            "memory_type": "rule",
            "title": "Language",
            "content": "Always reply in Chinese.",
        },
    )
    project_fact = save_memory(
        client,
        {
            "request_id": "project-fact",
            "scope": "project",
            "project_id": project_id,
            "memory_type": "fact",
            "title": "Stack",
            "content": "The backend uses FastAPI.",
        },
    )
    fake.fail_semantic = True
    context = client.post(
        "/api/v1/memories/context",
        json={"project_id": project_id, "query": "FastAPI", "max_chars": 4000},
    )
    assert context.status_code == 200
    body = context.json()
    assert body["semantic_degraded"] is True
    assert body["included_memory_ids"][:2] == [global_rule["id"], project_fact["id"]]
    assert body["content"].index("Always reply") < body["content"].index("FastAPI")


def test_project_search_includes_global_memory_unless_scope_is_explicit(
    initialized_client: TestClient,
) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    project_id = create_project(client)
    global_rule = save_memory(
        client,
        {
            "request_id": "project-search-global-rule",
            "scope": "global",
            "memory_type": "rule",
            "title": "Shared retrieval rule",
            "content": "Shared project search marker for every workspace.",
        },
    )
    with client.app.state.session_factory() as db:
        document = CuratedDocument(
            scope="global",
            usage_profile_id=global_rule["usage_profile_id"],
            document_type="profile",
            title="Shared retrieval profile",
            path="global/curated/shared-retrieval-profile",
            source_hash="c" * 64,
        )
        db.add(document)
        db.commit()
    fake.notes[document.path] = {
        "title": document.title,
        "content": "Shared project search marker in the consolidated profile.",
        "permalink": document.path,
    }

    inherited = client.get(
        "/api/v1/memories/search",
        params={"query": "shared project search marker", "project_id": project_id},
    )
    assert inherited.status_code == 200
    inherited_results = inherited.json()["results"]
    assert {item["id"] for item in inherited_results} == {global_rule["id"], document.id}
    assert {item["scope"] for item in inherited_results} == {"global"}

    strict_project = client.get(
        "/api/v1/memories/search",
        params={
            "query": "shared project search marker",
            "project_id": project_id,
            "scope": "project",
        },
    )
    assert strict_project.status_code == 200
    assert strict_project.json()["results"] == []

    strict_global = client.get(
        "/api/v1/memories/search",
        params={
            "query": "shared project search marker",
            "project_id": project_id,
            "scope": "global",
        },
    )
    assert strict_global.status_code == 200
    assert {item["id"] for item in strict_global.json()["results"]} == {
        global_rule["id"],
        document.id,
    }


def test_context_includes_every_durable_memory_type_without_curation(
    initialized_client: TestClient,
) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    project_id = create_project(client)
    memory_types = [
        "rule",
        "preference",
        "fact",
        "decision",
        "experience",
        "prd",
        "task_list",
    ]
    expected_ids: list[str] = []
    for scope in ("global", "project"):
        for memory_type in memory_types:
            marker = f"{scope}-{memory_type}"
            payload = {
                "request_id": marker,
                "scope": scope,
                "memory_type": memory_type,
                "title": marker,
                "content": f"Durable context for {marker}.",
            }
            if scope == "project":
                payload["project_id"] = project_id
            expected_ids.append(save_memory(client, payload)["id"])

    context = client.post(
        "/api/v1/memories/context",
        json={"project_id": project_id, "max_chars": 100_000},
    )

    assert context.status_code == 200
    body = context.json()
    assert body["included_memory_ids"] == expected_ids
    for scope in ("global", "project"):
        positions = [
            body["content"].index(f"Durable context for {scope}-{item}") for item in memory_types
        ]
        assert positions == sorted(positions)


def test_context_keeps_recent_progress_when_curated_documents_are_large(
    initialized_client: TestClient,
) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    project_id = create_project(client)
    checkpoint = save_memory(
        client,
        {
            "request_id": "large-context-checkpoint",
            "scope": "project",
            "project_id": project_id,
            "memory_type": "checkpoint",
            "title": "Latest progress",
            "content": "The latest checkpoint must remain visible.",
        },
    )
    with client.app.state.session_factory() as db:
        documents = [
            CuratedDocument(
                scope="project",
                project_id=project_id,
                document_type=document_type,
                title=f"Large {document_type}",
                path=f"projects/{project_id}/curated/{document_type}",
                source_hash=str(index) * 64,
            )
            for index, document_type in enumerate(("architecture", "decisions"), start=1)
        ]
        db.add_all(documents)
        db.commit()
    for document in documents:
        fake.notes[document.path] = {
            "title": document.title,
            "content": f"{document.title}\n" + ("Long stable knowledge. " * 200),
            "permalink": document.path,
        }

    context = client.post(
        "/api/v1/memories/context",
        json={"project_id": project_id, "max_chars": 1000},
    )

    assert context.status_code == 200
    body = context.json()
    assert body["truncated"] is True
    assert body["included_curated_document_ids"] == [item.id for item in documents]
    assert checkpoint["id"] in body["included_memory_ids"]
    assert "The latest checkpoint must remain visible" in body["content"]


def test_memory_lifecycle_hides_curated_sources_without_deleting_history(
    initialized_client: TestClient,
) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    memory = save_memory(
        client,
        {
            "request_id": "lifecycle-source",
            "scope": "global",
            "memory_type": "rule",
            "title": "Lifecycle source",
            "content": "Keep the original Markdown as historical evidence.",
        },
    )
    with client.app.state.session_factory() as db:
        document = CuratedDocument(
            scope="global",
            usage_profile_id=memory["usage_profile_id"],
            document_type="profile",
            title="Stable profile",
            path="global/curated/profile",
            source_hash="a" * 64,
            metadata_json=json.dumps(
                {
                    "cited_source_ids": [f"memory:{memory['id']}"],
                    "source_revisions": {f"memory:{memory['id']}": str(memory["revision"])},
                    "supersedes": [],
                }
            ),
        )
        db.add(document)
        db.commit()
        rebuild_memory_curation_states(db)
    fake.notes[document.path] = {
        "title": document.title,
        "content": "Stable profile preserves historical evidence.",
        "permalink": document.path,
    }

    assert client.get("/api/v1/memories", params={"scope": "global"}).json() == []
    covered = client.get(
        "/api/v1/memories",
        params={"scope": "global", "lifecycle": "covered"},
    ).json()
    assert covered[0]["id"] == memory["id"]
    assert covered[0]["curation_status"] == "covered"
    assert covered[0]["curated_revision"] == memory["revision"]
    assert covered[0]["covered_by"] == [document.id]
    assert memory["path"] in fake.notes

    current_search = client.get(
        "/api/v1/memories/search",
        params={"query": "historical evidence"},
    ).json()["results"]
    assert len(current_search) == 1
    assert current_search[0]["id"] == document.id
    assert current_search[0]["result_kind"] == "curated_document"
    assert current_search[0]["read_tool"] == "curated_document_get"
    assert current_search[0]["document_type"] == "profile"
    pending_search = client.get(
        "/api/v1/memories/search",
        params={"query": "historical evidence", "lifecycle": "pending"},
    ).json()["results"]
    assert pending_search == []
    history_search = client.get(
        "/api/v1/memories/search",
        params={"query": "historical evidence", "lifecycle": "all"},
    ).json()
    assert history_search["results"][0]["curation_status"] == "covered"

    reset = client.post(
        f"/api/v1/memories/{memory['id']}/pending",
        params={"expected_revision": memory["revision"]},
        headers=csrf_headers(client),
    )
    assert reset.status_code == 200
    assert reset.json()["curation_status"] == "pending"
    assert (
        client.get("/api/v1/memories", params={"scope": "global"}).json()[0]["id"] == memory["id"]
    )


def test_legacy_memory_without_profile_is_covered_by_profile_scoped_document(
    initialized_client: TestClient,
) -> None:
    # 生产回归：早于记忆空间功能创建的全局记忆 usage_profile_id 为 NULL，对所有
    # 空间可见并会被整理引用，但旧覆盖匹配不认 NULL，被引用后仍停留在待整理。
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    memory = save_memory(
        client,
        {
            "request_id": "legacy-profile-source",
            "scope": "global",
            "memory_type": "rule",
            "title": "Legacy rule",
            "content": "Created before usage profiles existed.",
        },
    )
    with client.app.state.session_factory() as db:
        reference = db.get(MemoryReference, memory["id"])
        reference.usage_profile_id = None
        document = CuratedDocument(
            scope="global",
            usage_profile_id=memory["usage_profile_id"],
            document_type="profile",
            title="Stable profile",
            path="global/curated/legacy-profile",
            source_hash="c" * 64,
            metadata_json=json.dumps(
                {
                    "cited_source_ids": [f"memory:{memory['id']}"],
                    "source_revisions": {f"memory:{memory['id']}": str(memory["revision"])},
                    "supersedes": [],
                }
            ),
        )
        db.add(document)
        db.commit()
        counts = rebuild_memory_curation_states(db)
        assert counts["covered"] == 1

    covered = client.get(
        "/api/v1/memories",
        params={"scope": "global", "lifecycle": "covered"},
    ).json()
    assert covered[0]["id"] == memory["id"]
    assert covered[0]["covered_reason"] == "cited"


def test_editing_a_curated_memory_returns_it_to_pending(initialized_client: TestClient) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    memory = save_memory(
        client,
        {
            "request_id": "curated-edit-source",
            "scope": "global",
            "memory_type": "fact",
            "title": "Curated fact",
            "content": "Original fact.",
        },
    )
    with client.app.state.session_factory() as db:
        document = CuratedDocument(
            scope="global",
            usage_profile_id=memory["usage_profile_id"],
            document_type="profile",
            title="Stable profile",
            path="global/curated/edit-profile",
            source_hash="b" * 64,
            metadata_json=json.dumps(
                {
                    "cited_source_ids": [f"memory:{memory['id']}"],
                    "source_revisions": {f"memory:{memory['id']}": "1"},
                    "supersedes": [],
                }
            ),
        )
        db.add(document)
        db.commit()
        rebuild_memory_curation_states(db)

    updated = save_memory(
        client,
        {
            "id": memory["id"],
            "request_id": "curated-edit-update",
            "scope": "global",
            "memory_type": "fact",
            "title": "Curated fact",
            "content": "Updated fact.",
            "expected_revision": 1,
        },
    )
    assert updated["revision"] == 2
    assert updated["curation_status"] == "pending"
    assert updated["curated_revision"] is None
    assert updated["covered_by"] == []


def test_context_keeps_personal_preferences_in_the_current_usage_profile(
    initialized_client: TestClient,
) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    project_id = create_project(client)
    alice = client.post(
        "/api/v1/curation/profiles",
        json={"name": "alice", "description": "Alice preferences"},
        headers=csrf_headers(client),
    ).json()
    bob = client.post(
        "/api/v1/curation/profiles",
        json={"name": "bob", "description": "Bob preferences"},
        headers=csrf_headers(client),
    ).json()
    alice_memory = save_memory(
        client,
        {
            "request_id": "alice-preference",
            "scope": "global",
            "usage_profile_id": alice["id"],
            "memory_type": "preference",
            "title": "Alice editor",
            "content": "Alice prefers compact editor layouts.",
        },
    )
    bob_memory = save_memory(
        client,
        {
            "request_id": "bob-preference",
            "scope": "global",
            "usage_profile_id": bob["id"],
            "memory_type": "preference",
            "title": "Bob editor",
            "content": "Bob prefers spacious editor layouts.",
        },
    )
    project_fact = save_memory(
        client,
        {
            "request_id": "shared-project-fact",
            "scope": "project",
            "project_id": project_id,
            "usage_profile_id": alice["id"],
            "memory_type": "fact",
            "title": "Shared stack",
            "content": "The project uses FastAPI.",
        },
    )

    with client.app.state.session_factory() as db:
        alice_context = asyncio.run(build_context(db, fake, project_id, None, 4000, alice["id"]))
    with client.app.state.session_factory() as db:
        bob_context = asyncio.run(build_context(db, fake, project_id, None, 4000, bob["id"]))

    assert alice_memory["id"] in alice_context.included_memory_ids
    assert bob_memory["id"] not in alice_context.included_memory_ids
    assert bob_memory["id"] in bob_context.included_memory_ids
    assert alice_memory["id"] not in bob_context.included_memory_ids
    assert project_fact["id"] in alice_context.included_memory_ids
    assert project_fact["id"] in bob_context.included_memory_ids
    assert fake.notes[alice_memory["path"]]["metadata"]["usage_profile_id"] == alice["id"]


def test_context_includes_only_the_current_workspace_scenario_documents(
    initialized_client: TestClient,
) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    project = client.post(
        "/api/v1/projects",
        json={"name": "Development context", "workspace_type": "development"},
        headers=csrf_headers(client),
    ).json()
    alice = client.post(
        "/api/v1/curation/profiles",
        json={"name": "context-alice", "description": "Alice context"},
        headers=csrf_headers(client),
    ).json()
    bob = client.post(
        "/api/v1/curation/profiles",
        json={"name": "context-bob", "description": "Bob context"},
        headers=csrf_headers(client),
    ).json()
    with client.app.state.session_factory() as db:
        shared = db.scalar(select(UsageProfile).where(UsageProfile.is_shared.is_(True)))
        assert shared is not None
        documents = [
            CuratedDocument(
                scope="global",
                usage_profile_id=shared.id,
                document_type="preferences",
                title="Shared preferences",
                path="global/curated/shared/preferences",
                source_hash="1" * 64,
            ),
            CuratedDocument(
                scope="global",
                usage_profile_id=alice["id"],
                document_type="context_development",
                title="Alice development",
                path="global/curated/alice/context-development",
                source_hash="2" * 64,
            ),
            CuratedDocument(
                scope="global",
                usage_profile_id=alice["id"],
                document_type="context_office",
                title="Alice office",
                path="global/curated/alice/context-office",
                source_hash="3" * 64,
            ),
            CuratedDocument(
                scope="global",
                usage_profile_id=bob["id"],
                document_type="context_development",
                title="Bob development",
                path="global/curated/bob/context-development",
                source_hash="4" * 64,
            ),
            CuratedDocument(
                scope="project",
                project_id=project["id"],
                document_type="overview",
                title="Workspace overview",
                path=f"projects/{project['id']}/curated/overview",
                source_hash="5" * 64,
            ),
        ]
        db.add_all(documents)
        db.commit()
        document_ids = {item.title: item.id for item in documents}
    for title, path in {
        "Shared preferences": "global/curated/shared/preferences",
        "Alice development": "global/curated/alice/context-development",
        "Alice office": "global/curated/alice/context-office",
        "Bob development": "global/curated/bob/context-development",
        "Workspace overview": f"projects/{project['id']}/curated/overview",
    }.items():
        fake.notes[path] = {
            "title": title,
            "content": f"# {title}\n\n{title} body",
            "permalink": path,
        }

    with client.app.state.session_factory() as db:
        context = asyncio.run(build_context(db, fake, project["id"], None, 10_000, alice["id"]))

    assert document_ids["Shared preferences"] in context.included_curated_document_ids
    assert document_ids["Alice development"] in context.included_curated_document_ids
    assert document_ids["Workspace overview"] in context.included_curated_document_ids
    assert document_ids["Alice office"] not in context.included_curated_document_ids
    assert document_ids["Bob development"] not in context.included_curated_document_ids


def test_project_scope_validation(initialized_client: TestClient) -> None:
    client = initialized_client
    client.app.state.basic_memory = FakeBasicMemory()
    response = client.post(
        "/api/v1/memories",
        json={
            "request_id": "invalid-scope",
            "scope": "project",
            "memory_type": "fact",
            "title": "Invalid",
            "content": "Missing project.",
        },
        headers=csrf_headers(client),
    )
    assert response.status_code == 422
    assert response.json()["error"]["code"] == "MEMORY_SCOPE_INVALID"


def test_memory_title_can_be_renamed(initialized_client: TestClient) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    created = save_memory(
        client,
        {
            "request_id": "rename-create",
            "scope": "global",
            "memory_type": "fact",
            "title": "Old title",
            "content": "Original content",
        },
    )
    old_path = created["path"]
    renamed = save_memory(
        client,
        {
            "id": created["id"],
            "request_id": "rename-update",
            "scope": "global",
            "memory_type": "fact",
            "title": "New title",
            "content": "Original content",
            "expected_revision": 1,
        },
    )
    assert renamed["title"] == "New title"
    assert renamed["revision"] == 2
    assert renamed["path"] != old_path
    assert old_path not in fake.notes
    assert renamed["path"] in fake.notes


def test_project_export_contains_timeline_and_markdown(initialized_client: TestClient) -> None:
    client = initialized_client
    fake = FakeBasicMemory()
    client.app.state.basic_memory = fake
    project_id = create_project(client)
    save_memory(
        client,
        {
            "request_id": "export-global",
            "scope": "global",
            "memory_type": "rule",
            "title": "Language",
            "content": "Always reply in Chinese.",
        },
    )
    checkpoint = save_memory(
        client,
        {
            "request_id": "export-checkpoint",
            "scope": "project",
            "project_id": project_id,
            "memory_type": "checkpoint",
            "title": "Implemented export",
            "content": "## 完成内容\n\nAdded direct ZIP export.",
        },
    )
    with client.app.state.session_factory() as db:
        curated = CuratedDocument(
            scope="project",
            project_id=project_id,
            document_type="overview",
            title="Current project overview",
            path=f"projects/{project_id}/curated/overview",
            revision=2,
            source_hash="e" * 64,
        )
        db.add(curated)
        db.commit()
    fake.notes[curated.path] = {
        "title": curated.title,
        "content": "The current consolidated project overview.",
        "permalink": curated.path,
    }

    prepared = client.get(f"/api/v1/projects/{project_id}/export")
    assert prepared.status_code == 200
    path = urlsplit(prepared.json()["url"])
    downloaded = client.get(f"{path.path}?{path.query}")
    assert downloaded.status_code == 200
    assert downloaded.headers["cache-control"] == "no-store"

    with zipfile.ZipFile(io.BytesIO(downloaded.content)) as archive:
        names = archive.namelist()
        assert "aidocs/project_context.md" in names
        assert "aidocs/manifest.json" in names
        assert "aidocs/curated/project/overview.md" in names
        assert any(checkpoint["id"] in name for name in names)
        timeline = archive.read("aidocs/project_context.md").decode()
        assert "Implemented export" in timeline
        assert "Added direct ZIP export" in timeline
        assert "curated/project/overview.md" in timeline
        curated_content = archive.read("aidocs/curated/project/overview.md").decode()
        assert "The current consolidated project overview" in curated_content
        assert "revision: 2" in curated_content
        manifest = json.loads(archive.read("aidocs/manifest.json"))
        assert manifest["project"]["id"] == project_id
        assert manifest["project_curated_documents"][0]["id"] == curated.id