from __future__ import annotations

import threading
import time
from pathlib import Path
from uuid import uuid4

import numpy as np

from iai_mcp.store import MemoryStore
from iai_mcp.types import MemoryRecord

_N_SEED = 200
_HOLD_SEC = 1.5
_BLOCK_THRESHOLD = _HOLD_SEC * 0.5

def _make_record(vec, community_id, idx: int) -> MemoryRecord:
    import datetime

    now = datetime.datetime.now(datetime.timezone.utc)
    return MemoryRecord(
        id=uuid4(),
        tier="semantic",
        literal_surface=f"record {idx} content",
        aaak_index="",
        embedding=vec.tolist(),
        community_id=community_id,
        centrality=float(idx % 17),
        detail_level=3,
        pinned=False,
        stability=0.0,
        difficulty=0.0,
        last_reviewed=None,
        never_decay=False,
        never_merge=False,
        provenance=[],
        created_at=now,
        updated_at=now,
        tags=["kind:note"],
        language="en",
        profile_modulation_gain={},
    )

def _seed_store(store: MemoryStore, n: int) -> None:
    dim = store._embed_dim
    cid = uuid4()
    rng = np.random.default_rng(11)
    for i in range(n):
        vec = rng.standard_normal(dim).astype(np.float32)
        vec /= np.linalg.norm(vec) + 1e-9
        store.insert(_make_record(vec, cid, i))

def _assert_hermetic(store: MemoryStore, tmp_path: Path) -> None:
    root = Path(store.root).resolve()
    assert str(root).startswith(str(tmp_path.resolve())), (
        f"store root {root} escaped tmp_path {tmp_path}"
    )
    real_home_store = (Path.home() / ".iai-mcp").resolve()
    assert real_home_store not in root.parents and root != real_home_store, (
        f"store root {root} resolved under the real ~/.iai-mcp"
    )

def _hold_conn_lock(store: MemoryStore, hold_sec: float,
                    started: threading.Event, done: threading.Event) -> None:
    with store.db._conn_lock:
        started.set()
        time.sleep(hold_sec)
    done.set()

def test_to_batches_fetch_holds_conn_lock(tmp_path):
    store_root = tmp_path / ".iai-mcp"
    store_root.mkdir(parents=True, exist_ok=True)
    store = MemoryStore(path=store_root)
    _assert_hermetic(store, tmp_path)
    _seed_store(store, _N_SEED)

    query = store.db.open_table("records").search()
    assert query._db is not None, "to_batches query lost its HippoDB reference"

    started = threading.Event()
    done = threading.Event()
    worker = threading.Thread(
        target=_hold_conn_lock,
        args=(store, _HOLD_SEC, started, done),
        daemon=True,
    )

    reader_elapsed = 0.0
    try:
        worker.start()
        assert started.wait(timeout=10.0), "worker never acquired _conn_lock"
        t0 = time.monotonic()
        rows = 0
        for batch in query.to_batches(batch_size=1):
            rows += batch.num_rows
        reader_elapsed = time.monotonic() - t0
        assert rows == _N_SEED, f"to_batches read {rows} rows, expected {_N_SEED}"
    finally:
        done.wait(timeout=_HOLD_SEC + 10.0)
        worker.join(timeout=5.0)
        store.close()

    assert reader_elapsed >= _BLOCK_THRESHOLD, (
        "to_batches drained without waiting for the held _conn_lock "
        f"(elapsed={reader_elapsed:.3f}s, hold={_HOLD_SEC}s) — the fetch did not "
        "hold the connection lock"
    )
