"""FlowDeck — v6.9.0 semantic search + Ask AI.

Covers migration 25, chunking/hashing/cosine units, indexing (idempotent,
excluded/deleted skipped, orphans purged), vector recall on partial overlap,
hybrid RRF + ACL/workspace isolation + pagination headers, ask (offline
extractive with citations, auth, cache, rate limit, ACL) and index-status.
"""
from __future__ import annotations

import json
import math
import secrets
import struct

from conftest import anon

from app.db import get_conn
from app.services import semantic_search as sem

# ── helpers ────────────────────────────────────────────────────────────────

def _login(client, admin: bool = False):
    from app.auth.session import SessionManager
    login = f"v69_{secrets.token_hex(4)}"
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO users (login, full_name, email, is_admin) VALUES (?, 'V69', ?, ?)",
            (login, f"{login}@test.com", 1 if admin else 0),
        )
        uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
        conn.commit()
    session = SessionManager.create_session({"id": uid, "login": login})
    return session, {"id": uid, "login": login}


def _cookies(session):
    return {"flowdeck_session": session}


def _mkpage(title="Doc", body="hello world", workspace_id=None, fmt="blocks"):
    content = (json.dumps([{"type": "paragraph", "content": body}])
               if fmt == "blocks" else body)
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO pages (workspace, workspace_id, title, content, content_format)"
            " VALUES (?, ?, ?, ?, ?)",
            ("test", workspace_id, title, content, fmt),
        )
        pid = cur.lastrowid
        conn.commit()
    return pid


def _mkcollection(name="DB", description="desc"):
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO collections (name, description) VALUES (?, ?)", (name, description))
        cid = cur.lastrowid
        conn.commit()
    return cid


def _mkworkspace(owner_id, name="Team"):
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO workspaces (name, owner_id) VALUES (?, ?)", (name, owner_id))
        wid = cur.lastrowid
        conn.commit()
    return wid


# ── migration ──────────────────────────────────────────────────────────────

def test_migration_25_tables(client):
    with get_conn() as conn:
        tables = {r[0] for r in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
    assert "semantic_embeddings" in tables
    assert "semantic_index_state" in tables
    with get_conn() as conn:
        cols = {r[1] for r in conn.execute("PRAGMA table_info(pages)").fetchall()}
    assert "search_excluded" in cols
    with get_conn() as conn:
        v = conn.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]
    assert v >= 25


# ── units: chunk / embed / cosine ──────────────────────────────────────────

def test_chunk_text_short_and_empty(client):
    assert sem.chunk_text("") == []
    assert sem.chunk_text("hello") == ["hello"]


def test_chunk_text_splits_long(client):
    text = " ".join(f"word{i}" for i in range(800))
    chunks = sem.chunk_text(text)
    assert len(chunks) > 1
    assert all(len(c) <= sem.CHUNK_SIZE for c in chunks)
    # overlap: a middle word appears in two consecutive chunks
    assert any(w in chunks[0] and w in chunks[1] for w in chunks[1].split()[:20])


def test_embed_deterministic_and_normalized(client):
    a = sem.embed_text("hello world")
    b = sem.embed_text("hello world")
    assert a == b
    assert len(a) == sem.DIM * 4
    vals = struct.unpack(f"<{sem.DIM}f", a)
    assert math.isclose(sum(v * v for v in vals), 1.0, rel_tol=1e-5)


def test_cosine_identical_and_disjoint(client):
    a = sem.embed_text("alpha beta")
    assert math.isclose(sem.cosine(a, a), 1.0, rel_tol=1e-5)
    # disjoint single tokens collide with p≈1-(255/256)^2 ≈ tiny; use longer texts
    c = sem.embed_text("alpha beta gamma delta")
    d = sem.embed_text("epsilon zeta eta theta")
    assert 0.0 <= sem.cosine(c, d) < sem.cosine(c, c)


# ── indexing ───────────────────────────────────────────────────────────────

def test_index_page_and_idempotent(client):
    pid = _mkpage("Guide", "kubernetes deployment scaling tips")
    n = sem.index_resource("page", pid)
    assert n >= 1
    with get_conn() as conn:
        count = conn.execute(
            "SELECT COUNT(*) FROM semantic_embeddings WHERE resource_type='page'"
            " AND resource_id=?", (pid,)).fetchone()[0]
    assert count == n
    n2 = sem.index_resource("page", pid)
    assert n2 == n
    with get_conn() as conn:
        count2 = conn.execute(
            "SELECT COUNT(*) FROM semantic_embeddings WHERE resource_type='page'"
            " AND resource_id=?", (pid,)).fetchone()[0]
    assert count2 == n  # no duplicates


def test_index_skips_excluded(client):
    pid = _mkpage("Secret", "hidden content here")
    with get_conn() as conn:
        conn.execute("UPDATE pages SET search_excluded=1 WHERE id=?", (pid,))
        conn.commit()
    assert sem.index_resource("page", pid) == 0


def test_purge_removes_deleted(client):
    pid = _mkpage("Gone", "bye bye content")
    assert sem.index_resource("page", pid) >= 1
    with get_conn() as conn:
        conn.execute("UPDATE pages SET deleted_at='2026-01-01 00:00:00' WHERE id=?", (pid,))
        conn.commit()
    assert sem.purge_orphans() >= 1
    assert sem.index_resource("page", pid) == 0


def test_index_pending_picks_stale(client):
    pid = _mkpage("Fresh", "brand new content words")
    out = sem.index_pending(limit=50)
    assert out["indexed"] >= 1
    with get_conn() as conn:
        row = conn.execute(
            "SELECT indexed_at FROM semantic_index_state WHERE resource_type='page'"
            " AND resource_id=?", (pid,)).fetchone()
    assert row is not None


# ── vector + hybrid ────────────────────────────────────────────────────────

def test_vector_partial_overlap_recall(client):
    pid = _mkpage("Ops", "alpha beta gamma delta")
    sem.index_resource("page", pid)
    # FTS AND would need all terms; vector matches on shared "alpha".
    hits = sem.vector_search("alpha zeta omicron", limit=10)
    assert any(h["resource_id"] == pid for h in hits)


def test_hybrid_finds_by_keyword(client):
    session, user = _login(client)
    pid = _mkpage("Kubernetes Guide", "deploy pods and services")
    sem.index_resource("page", pid)
    r = client.get("/api/v2/search/hybrid?q=kubernetes", cookies=_cookies(session))
    assert r.status_code == 200, r.text
    ids = [x["id"] for x in r.json()["results"] if x["type"] == "page"]
    assert pid in ids
    assert "X-Total-Count" in r.headers


def test_hybrid_requires_auth(client):
    anon(client)
    r = client.get("/api/v2/search/hybrid?q=test")
    assert r.status_code == 401


def test_hybrid_empty_query_400(client):
    session, _ = _login(client)
    r = client.get("/api/v2/search/hybrid?q=", cookies=_cookies(session))
    assert r.status_code == 400


def test_hybrid_pagination(client):
    session, _ = _login(client)
    for i in range(3):
        sem.index_resource("page", _mkpage(f"Pagetopic {i}", f"pagetopic body {i}"))
    r = client.get("/api/v2/search/hybrid?q=pagetopic&limit=2&offset=0",
                   cookies=_cookies(session))
    assert r.status_code == 200
    body = r.json()
    assert len(body["results"]) <= 2
    assert body["limit"] == 2 and body["offset"] == 0


def test_hybrid_workspace_isolation(client):
    # Workspaces are open-by-default (viewer fallback); restriction is opt-in
    # via permission_type='restricted' + explicit grants.
    s1, u1 = _login(client)
    _s2, _u2 = _login(client)
    wid = _mkworkspace(u1["id"])
    pid = _mkpage("Team Secrets", "sekretwords vault", workspace_id=wid)
    with get_conn() as conn:
        conn.execute("UPDATE pages SET permission_type='restricted' WHERE id=?", (pid,))
        conn.commit()
    sem.index_resource("page", pid)
    r = client.get("/api/v2/search/hybrid?q=sekretwords", cookies=_cookies(s1))
    assert pid in [x["id"] for x in r.json()["results"] if x["type"] == "page"]
    r = client.get("/api/v2/search/hybrid?q=sekretwords", cookies=_cookies(_s2))
    assert pid not in [x["id"] for x in r.json()["results"] if x["type"] == "page"]


def test_hybrid_excluded_page_absent(client):
    session, _ = _login(client)
    pid = _mkpage("Hidden", "cloakwords invisible")
    with get_conn() as conn:
        conn.execute("UPDATE pages SET search_excluded=1 WHERE id=?", (pid,))
        conn.commit()
    sem.purge_orphans()
    r = client.get("/api/v2/search/hybrid?q=cloakwords", cookies=_cookies(session))
    assert pid not in [x["id"] for x in r.json()["results"] if x["type"] == "page"]


def test_hybrid_finds_collections(client):
    session, _ = _login(client)
    cid = _mkcollection("Customer CRM", "tracks zalonowords leads")
    sem.index_resource("collection", cid)
    r = client.get("/api/v2/search/hybrid?q=zalonowords", cookies=_cookies(session))
    assert r.status_code == 200
    assert cid in [x["id"] for x in r.json()["results"] if x["type"] == "collection"]


# ── ask ────────────────────────────────────────────────────────────────────

def test_ask_offline_with_citations(client):
    sem.reset_state()
    session, _ = _login(client)
    pid = _mkpage("Deploy Guide",
                  "To deploy the app, run the deploy script. Then verify the pods are ready.")
    sem.index_resource("page", pid)
    r = client.post("/api/v2/search/ask", json={"question": "how to deploy the app"},
                    cookies=_cookies(session))
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["offline"] is True
    assert any(c["id"] == pid for c in body["citations"])
    assert f"[[fdpage:{pid}]]" in body["answer_markdown"]


def test_ask_requires_auth(client):
    anon(client)
    r = client.post("/api/v2/search/ask", json={"question": "hi"})
    assert r.status_code == 401


def test_ask_empty_400(client):
    sem.reset_state()
    session, _ = _login(client)
    r = client.post("/api/v2/search/ask", json={"question": " "},
                    cookies=_cookies(session))
    assert r.status_code == 400


def test_ask_cached(client):
    sem.reset_state()
    session, _ = _login(client)
    pid = _mkpage("Caching", "the cache stores answers for reuse and speed")
    sem.index_resource("page", pid)
    payload = {"question": "what does the cache store"}
    r1 = client.post("/api/v2/search/ask", json=payload, cookies=_cookies(session))
    assert r1.json()["cached"] is False
    r2 = client.post("/api/v2/search/ask", json=payload, cookies=_cookies(session))
    assert r2.json()["cached"] is True
    assert r2.json()["answer_markdown"] == r1.json()["answer_markdown"]


def test_ask_acl_other_user(client):
    sem.reset_state()
    s1, u1 = _login(client)
    s2, _u2 = _login(client)
    wid = _mkworkspace(u1["id"])
    pid = _mkpage("Private Ops", "quagmirewords runbook steps", workspace_id=wid)
    with get_conn() as conn:
        conn.execute("UPDATE pages SET permission_type='restricted' WHERE id=?", (pid,))
        conn.commit()
    sem.index_resource("page", pid)
    r = client.post("/api/v2/search/ask", json={"question": "quagmirewords runbook"},
                    cookies=_cookies(s1))
    assert pid in [c["id"] for c in r.json()["citations"]]
    r = client.post("/api/v2/search/ask", json={"question": "quagmirewords runbook"},
                    cookies=_cookies(s2))
    assert pid not in [c["id"] for c in r.json()["citations"]]


def test_ask_rate_limit(client):
    sem.reset_state()
    session, _ = _login(client)
    last = None
    for _ in range(31):
        last = client.post("/api/v2/search/ask", json={"question": "ping"},
                           cookies=_cookies(session))
    assert last.status_code == 429


def test_index_status(client):
    session, _ = _login(client)
    pid = _mkpage("Status", "statuswords check")
    sem.index_resource("page", pid)
    r = client.get("/api/v2/search/index-status", cookies=_cookies(session))
    assert r.status_code == 200
    body = r.json()
    assert body["vectors"] >= 1
    assert body["model"] == "hash-256"