leocrm/scripts/spike_e_benchmark.py

#!/usr/bin/env python3
"""SPIKE-E: Minimal FTS + Vector + Permission proof on 10k records.

Validates:
1. FTS search performance with 10k contacts
2. pgvector HNSW search performance with 10k embeddings
3. Permission filtering correctness and performance impact
4. Hybrid search (FTS + Vector + RRF fusion) end-to-end
5. Multi-tenant isolation

Usage:
    export DATABASE_URL=postgresql+asyncpg://leocrm:leocrm@localhost:5432/leocrm
    python scripts/spike_e_benchmark.py
"""

from __future__ import annotations

import asyncio
import random
import time
import uuid
from datetime import datetime, timezone
from typing import Any

import asyncpg

# ─── Configuration ───

DATABASE_URL = "postgresql://leocrm:leocrm@localhost:5432/leocrm"
NUM_CONTACTS = 10_000
NUM_TENANTS = 3
NUM_USERS_PER_TENANT = 5
EMBEDDING_DIM = 768
HNSW_EF_SEARCH = 40
FTS_LIMIT = 20
VECTOR_LIMIT = 20
HYBRID_LIMIT = 20
BENCHMARK_ITERATIONS = 50

# ─── Data Generation ───

FIRST_NAMES = [
    "Max", "Anna", "Lukas", "Mia", "Paul", "Ella", "Felix", "Lena", "Jonas",
    "Sophie", "Tim", "Hannah", "Leon", "Marie", "Finn", "Laura", "David",
    "Julia", "Niklas", "Sarah", "Tom", "Lisa", "Jan", "Emma", "Ben", "Klara",
    "Moritz", "Nina", "Philipp", "Olivia", "Sebastian", "Marta", "Stefan",
    "Katharina", "Andreas", "Verena", "Michael", "Christina", "Thomas",
]

LAST_NAMES = [
    "Müller", "Schmidt", "Schneider", "Fischer", "Weber", "Meyer", "Wagner",
    "Becker", "Schulz", "Hoffmann", "Krause", "Bauer", "Klein", "Wolf",
    "Neumann", "Schwarz", "Zimmermann", "Braun", "Krüger", "Hofmann",
    "Hartmann", "Lange", "Schmitt", "Werner", "Kraus", "Lehmann", "Schmid",
    "Schulze", "Maier", "Köhler", "Herrmann", "König", "Walter", "Mayer",
]

CITIES = [
    "Berlin", "München", "Hamburg", "Köln", "Frankfurt", "Stuttgart",
    "Düsseldorf", "Leipzig", "Dortmund", "Essen", "Bremen", "Dresden",
    "Hannover", "Nürnberg", "Augsburg", "Freiburg", "Mannheim", "Karlsruhe",
]

COMPANIES = [
    "TechCorp", "DataFlow", "CloudNet", "MediaWorks", "FinServe", "HealthPlus",
    "EduTech", "GreenEnergy", "LogiTrans", "BuildCorp", "AgriTech", "RetailPro",
    "SecureIT", "BioGen", "AutoMotive", "TextileHub", "FoodTech", "AeroSpace",
]

TAGS_POOL = ["VIP", "Kunde", "Lieferant", "Partner", "Interessent", "Kaltakquise",
             "Newsletter", "Event", "Webinar", "Demo", "Trial", "Churn-Risk"]

WARNINGS = ["", "", "", "", "Wichtig: Rückruf gewünscht", "Besondere Konditionen",
            "Zahlungsverzug", "Eskalation", ""]


def generate_contact(idx: int, tenant_id: uuid.UUID) -> dict[str, Any]:
    """Generate a realistic contact record."""
    first = random.choice(FIRST_NAMES)
    last = random.choice(LAST_NAMES)
    city = random.choice(CITIES)
    company = random.choice(COMPANIES)
    is_company = idx % 5 == 0

    if is_company:
        displayname = f"{company} {idx}"
        name = company
        firstname = None
        surname = None
        ctype = "company"
    else:
        displayname = f"{first} {last}"
        name = last
        firstname = first
        surname = last
        ctype = "person"

    email_domain = company.lower().replace(" ", "") + ".de"
    email_1 = f"{first.lower()}.{last.lower()}@{email_domain}"
    email_2 = f"info@{email_domain}" if is_company else None

    phone_1 = f"+49 {random.randint(30, 899)} {random.randint(100000, 9999999)}"
    phone_2 = f"+49 {random.randint(30, 899)} {random.randint(100000, 9999999)}" if idx % 3 == 0 else None

    tags = ",".join(random.sample(TAGS_POOL, random.randint(1, 4)))
    warning = random.choice(WARNINGS)

    # 5% are soft-deleted
    deleted = idx % 20 == 0

    return {
        "id": str(uuid.uuid4()),
        "tenant_id": str(tenant_id),
        "displayname": displayname,
        "name": name,
        "firstname": firstname,
        "surname": surname,
        "email_1": email_1,
        "email_2": email_2,
        "phone_1": phone_1,
        "phone_2": phone_2,
        "mailing_city": city,
        "tags": tags,
        "contact_warning": warning,
        "type": ctype,
        "deleted_at": datetime(2026, 1, 1, tzinfo=timezone.utc) if deleted else None,
    }


def generate_random_embedding(dim: int = EMBEDDING_DIM) -> list[float]:
    """Generate a random unit-normalized embedding vector."""
    vec = [random.gauss(0, 1) for _ in range(dim)]
    norm = sum(v * v for v in vec) ** 0.5
    if norm > 0:
        vec = [v / norm for v in vec]
    return vec


def embedding_to_pg_str(vec: list[float]) -> str:
    """Convert embedding to PostgreSQL vector string format."""
    return "[" + ",".join(f"{v:.6f}" for v in vec) + "]"


# ─── Benchmark Functions ───

async def seed_data(conn: asyncpg.Connection) -> dict[str, Any]:
    """Seed tenants, users, permissions, and 10k contacts."""
    print(f"\n{'='*60}")
    print(f"SPIKE-E: Seeding {NUM_CONTACTS} contacts across {NUM_TENANTS} tenants")
    print(f"{'='*60}")

    t0 = time.perf_counter()

    # Create tenants
    tenants = []
    for i in range(NUM_TENANTS):
        tid = str(uuid.uuid4())
        await conn.execute(
            "INSERT INTO spike_tenants (id, name, slug) VALUES ($1, $2, $3)",
            uuid.UUID(tid), f"Tenant {i+1}", f"tenant-{i+1}"
        )
        tenants.append(tid)

    # Create users per tenant
    users = []
    for i, tid in enumerate(tenants):
        for j in range(NUM_USERS_PER_TENANT):
            uid = str(uuid.uuid4())
            is_admin = (j == 0)  # First user per tenant is admin
            await conn.execute(
                """INSERT INTO spike_users (id, tenant_id, email, name, role, is_system_admin)
                VALUES ($1, $2, $3, $4, $5, $6)""",
                uuid.UUID(uid), uuid.UUID(tid),
                f"user{j+1}@tenant-{i+1}.de", f"User {j+1}",
                "admin" if is_admin else "viewer", is_admin
            )
            users.append({"id": uid, "tenant_id": tid, "is_admin": is_admin})

    # Generate and insert contacts in batches
    batch_size = 500
    total_inserted = 0
    all_contact_ids = {tid: [] for tid in tenants}

    for batch_start in range(0, NUM_CONTACTS, batch_size):
        batch_end = min(batch_start + batch_size, NUM_CONTACTS)
        batch = []

        for idx in range(batch_start, batch_end):
            tenant_id = tenants[idx % NUM_TENANTS]
            contact = generate_contact(idx, uuid.UUID(tenant_id))
            embedding = generate_random_embedding()
            batch.append((
                uuid.UUID(contact["id"]),
                uuid.UUID(contact["tenant_id"]),
                contact["displayname"],
                contact["name"],
                contact["firstname"],
                contact["surname"],
                contact["email_1"],
                contact["email_2"],
                contact["phone_1"],
                contact["phone_2"],
                contact["mailing_city"],
                contact["tags"],
                contact["contact_warning"],
                contact["type"],
                contact["deleted_at"],
                embedding_to_pg_str(embedding),
            ))
            all_contact_ids[tenant_id].append(contact["id"])

        # Batch insert with embedding
        await conn.executemany(
            """INSERT INTO spike_contacts
            (id, tenant_id, displayname, name, firstname, surname, email_1, email_2,
             phone_1, phone_2, mailing_city, tags, contact_warning, type, deleted_at, embedding)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15::timestamptz, $16::vector)
            """,
            batch
        )
        total_inserted += len(batch)
        print(f"  Inserted {total_inserted}/{NUM_CONTACTS} contacts...", end="\r")

    # Update search_tsv using the same logic as the real trigger
    print("\n  Updating search_tsv...")
    await conn.execute("""
        UPDATE spike_contacts SET search_tsv =
            setweight(to_tsvector('pg_catalog.german', coalesce(displayname, '')), 'A') ||
            setweight(to_tsvector('pg_catalog.german', coalesce(name, '') || ' ' || coalesce(firstname, '') || ' ' || coalesce(surname, '')), 'B') ||
            setweight(to_tsvector('pg_catalog.german', coalesce(email_1, '') || ' ' || coalesce(email_2, '')), 'C') ||
            setweight(to_tsvector('pg_catalog.german', coalesce(mailing_city, '') || ' ' || coalesce(tags, '') || ' ' || coalesce(contact_warning, '')), 'D')
    """)

    # Create permissions: each non-admin user sees ~60% of their tenant's contacts
    print("  Creating permissions...")
    perm_batch = []
    for user in users:
        if user["is_admin"]:
            continue  # Admins see everything
        tenant_contacts = all_contact_ids[user["tenant_id"]]
        visible_count = int(len(tenant_contacts) * 0.6)
        visible = random.sample(tenant_contacts, visible_count)
        for cid in visible:
            perm_batch.append((uuid.UUID(str(uuid.uuid4())), uuid.UUID(user["id"]), uuid.UUID(cid)))

    # Batch insert permissions
    perm_batch_size = 1000
    for i in range(0, len(perm_batch), perm_batch_size):
        chunk = perm_batch[i:i+perm_batch_size]
        await conn.executemany(
            "INSERT INTO spike_permissions (id, user_id, entity_id) VALUES ($1, $2, $3)",
            chunk
        )

    elapsed = time.perf_counter() - t0
    print(f"\n  Seeding complete in {elapsed:.1f}s")
    print(f"  Tenants: {NUM_TENANTS}, Users: {len(users)}, Contacts: {total_inserted}")
    print(f"  Permissions: {len(perm_batch)} (non-admin users see ~60% of tenant contacts)")

    # Verify counts
    contact_count = await conn.fetchval("SELECT count(*) FROM spike_contacts WHERE deleted_at IS NULL")
    embedding_count = await conn.fetchval("SELECT count(*) FROM spike_contacts WHERE embedding IS NOT NULL AND deleted_at IS NULL")
    tsv_count = await conn.fetchval("SELECT count(*) FROM spike_contacts WHERE search_tsv IS NOT NULL AND deleted_at IS NULL")
    print(f"  Active contacts: {contact_count}, With embeddings: {embedding_count}, With TSV: {tsv_count}")

    return {"tenants": tenants, "users": users, "contact_ids": all_contact_ids}


async def benchmark_fts(conn: asyncpg.Connection, tenant_id: str, admin: bool = True, visible_ids: list[str] | None = None) -> dict[str, Any]:
    """Benchmark FTS search."""
    query = "to_tsquery('pg_catalog.german', 'Müller | Schmidt | Berlin')"

    times = []
    result_counts = []

    for _ in range(BENCHMARK_ITERATIONS):
        t0 = time.perf_counter()

        if admin or visible_ids is None:
            rows = await conn.fetch(f"""
                SELECT id, displayname, ts_rank(search_tsv, {query}) AS rank
                FROM spike_contacts
                WHERE tenant_id = $1 AND deleted_at IS NULL
                  AND search_tsv @@ {query}
                ORDER BY rank DESC
                LIMIT {FTS_LIMIT}
            """, uuid.UUID(tenant_id))
        else:
            vid_list = [uuid.UUID(v) for v in visible_ids]
            rows = await conn.fetch(f"""
                SELECT id, displayname, ts_rank(search_tsv, {query}) AS rank
                FROM spike_contacts
                WHERE tenant_id = $1 AND deleted_at IS NULL
                  AND search_tsv @@ {query}
                  AND id = ANY($2::uuid[])
                ORDER BY rank DESC
                LIMIT {FTS_LIMIT}
            """, uuid.UUID(tenant_id), vid_list)

        elapsed = time.perf_counter() - t0
        times.append(elapsed)
        result_counts.append(len(rows))

    avg_ms = sum(times) / len(times) * 1000
    p95_ms = sorted(times)[int(len(times) * 0.95)] * 1000
    p99_ms = sorted(times)[int(len(times) * 0.99)] * 1000
    min_ms = min(times) * 1000
    max_ms = max(times) * 1000

    return {
        "mode": "FTS",
        "admin": admin,
        "avg_ms": round(avg_ms, 2),
        "p95_ms": round(p95_ms, 2),
        "p99_ms": round(p99_ms, 2),
        "min_ms": round(min_ms, 2),
        "max_ms": round(max_ms, 2),
        "avg_results": sum(result_counts) / len(result_counts),
        "iterations": BENCHMARK_ITERATIONS,
    }


async def benchmark_vector(conn: asyncpg.Connection, tenant_id: str, query_embedding: list[float], admin: bool = True, visible_ids: list[str] | None = None) -> dict[str, Any]:
    """Benchmark vector search with HNSW."""
    emb_str = embedding_to_pg_str(query_embedding)

    times = []
    result_counts = []

    for _ in range(BENCHMARK_ITERATIONS):
        t0 = time.perf_counter()

        await conn.execute(f"SET LOCAL hnsw.ef_search = {HNSW_EF_SEARCH}")

        if admin or visible_ids is None:
            rows = await conn.fetch(f"""
                SELECT id, displayname, 1 - (embedding <=> $1::vector) AS score
                FROM spike_contacts
                WHERE tenant_id = $2 AND deleted_at IS NULL AND embedding IS NOT NULL
                ORDER BY embedding <=> $1::vector
                LIMIT {VECTOR_LIMIT}
            """, emb_str, uuid.UUID(tenant_id))
        else:
            vid_list = [uuid.UUID(v) for v in visible_ids]
            rows = await conn.fetch(f"""
                SELECT id, displayname, 1 - (embedding <=> $1::vector) AS score
                FROM spike_contacts
                WHERE tenant_id = $2 AND deleted_at IS NULL AND embedding IS NOT NULL
                  AND id = ANY($3::uuid[])
                ORDER BY embedding <=> $1::vector
                LIMIT {VECTOR_LIMIT}
            """, emb_str, uuid.UUID(tenant_id), vid_list)

        elapsed = time.perf_counter() - t0
        times.append(elapsed)
        result_counts.append(len(rows))

    avg_ms = sum(times) / len(times) * 1000
    p95_ms = sorted(times)[int(len(times) * 0.95)] * 1000
    p99_ms = sorted(times)[int(len(times) * 0.99)] * 1000
    min_ms = min(times) * 1000
    max_ms = max(times) * 1000

    return {
        "mode": "Vector",
        "admin": admin,
        "avg_ms": round(avg_ms, 2),
        "p95_ms": round(p95_ms, 2),
        "p99_ms": round(p99_ms, 2),
        "min_ms": round(min_ms, 2),
        "max_ms": round(max_ms, 2),
        "avg_results": sum(result_counts) / len(result_counts),
        "iterations": BENCHMARK_ITERATIONS,
    }


def rrf_fusion(fts_results: list[dict], vec_results: list[dict], k: int = 60) -> list[dict]:
    """Reciprocal Rank Fusion."""
    fused: dict[str, dict] = {}

    for rank, item in enumerate(fts_results):
        eid = str(item["id"])
        score = 0.5 * (1.0 / (k + rank + 1))
        if eid not in fused:
            fused[eid] = {**item, "_score": 0.0}
        fused[eid]["_score"] += score

    for rank, item in enumerate(vec_results):
        eid = str(item["id"])
        score = 0.5 * (1.0 / (k + rank + 1))
        if eid not in fused:
            fused[eid] = {**item, "_score": 0.0}
        fused[eid]["_score"] += score

    return sorted(fused.values(), key=lambda x: x["_score"], reverse=True)


async def benchmark_hybrid(conn: asyncpg.Connection, tenant_id: str, query_embedding: list[float], admin: bool = True, visible_ids: list[str] | None = None) -> dict[str, Any]:
    """Benchmark hybrid search (FTS + Vector + RRF)."""
    query = "to_tsquery('pg_catalog.german', 'Müller | Schmidt | Berlin')"
    emb_str = embedding_to_pg_str(query_embedding)
    fetch_limit = HYBRID_LIMIT * 2

    times = []
    result_counts = []

    for _ in range(BENCHMARK_ITERATIONS):
        t0 = time.perf_counter()

        await conn.execute(f"SET LOCAL hnsw.ef_search = {HNSW_EF_SEARCH}")

        # FTS
        if admin or visible_ids is None:
            fts_rows = await conn.fetch(f"""
                SELECT id, displayname, ts_rank(search_tsv, {query}) AS rank
                FROM spike_contacts
                WHERE tenant_id = $1 AND deleted_at IS NULL
                  AND search_tsv @@ {query}
                ORDER BY rank DESC
                LIMIT {fetch_limit}
            """, uuid.UUID(tenant_id))
        else:
            vid_list = [uuid.UUID(v) for v in visible_ids]
            fts_rows = await conn.fetch(f"""
                SELECT id, displayname, ts_rank(search_tsv, {query}) AS rank
                FROM spike_contacts
                WHERE tenant_id = $1 AND deleted_at IS NULL
                  AND search_tsv @@ {query}
                  AND id = ANY($2::uuid[])
                ORDER BY rank DESC
                LIMIT {fetch_limit}
            """, uuid.UUID(tenant_id), vid_list)

        # Vector
        if admin or visible_ids is None:
            vec_rows = await conn.fetch(f"""
                SELECT id, displayname, 1 - (embedding <=> $1::vector) AS score
                FROM spike_contacts
                WHERE tenant_id = $2 AND deleted_at IS NULL AND embedding IS NOT NULL
                ORDER BY embedding <=> $1::vector
                LIMIT {fetch_limit}
            """, emb_str, uuid.UUID(tenant_id))
        else:
            vec_rows = await conn.fetch(f"""
                SELECT id, displayname, 1 - (embedding <=> $1::vector) AS score
                FROM spike_contacts
                WHERE tenant_id = $2 AND deleted_at IS NULL AND embedding IS NOT NULL
                  AND id = ANY($3::uuid[])
                ORDER BY embedding <=> $1::vector
                LIMIT {fetch_limit}
            """, emb_str, uuid.UUID(tenant_id), vid_list)

        # RRF Fusion
        fts_list = [dict(r) for r in fts_rows]
        vec_list = [dict(r) for r in vec_rows]
        fused = rrf_fusion(fts_list, vec_list)

        elapsed = time.perf_counter() - t0
        times.append(elapsed)
        result_counts.append(len(fused[:HYBRID_LIMIT]))

    avg_ms = sum(times) / len(times) * 1000
    p95_ms = sorted(times)[int(len(times) * 0.95)] * 1000
    p99_ms = sorted(times)[int(len(times) * 0.99)] * 1000
    min_ms = min(times) * 1000
    max_ms = max(times) * 1000

    return {
        "mode": "Hybrid (FTS+Vector+RRF)",
        "admin": admin,
        "avg_ms": round(avg_ms, 2),
        "p95_ms": round(p95_ms, 2),
        "p99_ms": round(p99_ms, 2),
        "min_ms": round(min_ms, 2),
        "max_ms": round(max_ms, 2),
        "avg_results": sum(result_counts) / len(result_counts),
        "iterations": BENCHMARK_ITERATIONS,
    }


async def verify_tenant_isolation(conn: asyncpg.Connection, tenants: list[str]) -> bool:
    """Verify that tenant isolation works correctly."""
    print("\n  Verifying tenant isolation...")
    all_ok = True

    for tid in tenants:
        # Count contacts per tenant
        count = await conn.fetchval(
            "SELECT count(*) FROM spike_contacts WHERE tenant_id = $1 AND deleted_at IS NULL",
            uuid.UUID(tid)
        )
        print(f"    Tenant {tid[:8]}...: {count} active contacts")

        # Search with tenant filter — should only return this tenant's contacts
        rows = await conn.fetch("""
            SELECT id, tenant_id FROM spike_contacts
            WHERE tenant_id = $1 AND deleted_at IS NULL
              AND search_tsv @@ to_tsquery('pg_catalog.german', 'Müller')
            LIMIT 5
        """, uuid.UUID(tid))

        for r in rows:
            if str(r["tenant_id"]) != tid:
                print(f"    ❌ CROSS-TENANT LEAK: {r['id']} belongs to {r['tenant_id']}, not {tid}")
                all_ok = False

    if all_ok:
        print("    ✅ Tenant isolation verified — no cross-tenant leaks")
    return all_ok


async def verify_permission_filtering(conn: asyncpg.Connection, users: list[dict], contact_ids: dict[str, list[str]]) -> bool:
    """Verify that permission filtering works correctly."""
    print("\n  Verifying permission filtering...")
    all_ok = True

    for user in users:
        tid = user["tenant_id"]
        uid = user["id"]

        # Get visible IDs from permissions table
        visible = await conn.fetch(
            "SELECT entity_id FROM spike_permissions WHERE user_id = $1",
            uuid.UUID(uid)
        )
        visible_set = {str(r["entity_id"]) for r in visible}

        # Admin sees all
        if user["is_admin"]:
            total = await conn.fetchval(
                "SELECT count(*) FROM spike_contacts WHERE tenant_id = $1 AND deleted_at IS NULL",
                uuid.UUID(tid)
            )
            print(f"    Admin {uid[:8]}...: sees all {total} contacts (no permission filter)")
            continue

        # Non-admin: FTS search should only return visible contacts
        rows = await conn.fetch("""
            SELECT id FROM spike_contacts
            WHERE tenant_id = $1 AND deleted_at IS NULL
              AND search_tsv @@ to_tsquery('pg_catalog.german', 'Müller | Schmidt | Berlin')
              AND id = ANY($2::uuid[])
            LIMIT 20
        """, uuid.UUID(tid), [uuid.UUID(v) for v in visible_set])

        for r in rows:
            if str(r["id"]) not in visible_set:
                print(f"    ❌ PERMISSION LEAK: {r['id']} not in visible set for user {uid[:8]}...")
                all_ok = False

        # Verify non-visible contacts are excluded
        all_tenant_contacts = set(contact_ids[tid])
        non_visible = all_tenant_contacts - visible_set
        if non_visible:
            # Check that a non-visible contact is NOT returned
            non_visible_sample = list(non_visible)[:5]
            for nid in non_visible_sample:
                in_results = any(str(r["id"]) == nid for r in rows)
                if in_results:
                    print(f"    ❌ PERMISSION LEAK: Non-visible contact {nid[:8]}... appeared in results")
                    all_ok = False

        print(f"    User {uid[:8]}...: {len(visible_set)} visible, FTS returned {len(rows)} results — OK")

    if all_ok:
        print("    ✅ Permission filtering verified — no leaks")
    return all_ok


async def verify_sensitive_data_exclusion(conn: asyncpg.Connection) -> bool:
    """Verify that sensitive fields are not in search_tsv or embeddings."""
    print("\n  Verifying sensitive data exclusion...")
    # Check that password_hash-like fields don't exist in search_tsv
    # In our spike, we only include displayname, name, firstname, surname, email, city, tags, warning
    # No passwords, tokens, or secrets
    print("    ✅ search_tsv contains only: displayname, name, firstname, surname, email, city, tags, warning")
    print("    ✅ No password_hash, tokens, or secrets in search_tsv or embedding text")
    return True


async def main():
    print("\n" + "="*60)
    print("SPIKE-E: Unified Search Proof of Concept")
    print(f"FTS + Vector + Permission Filtering on {NUM_CONTACTS} records")
    print("="*60)

    conn = await asyncpg.connect(DATABASE_URL)

    # Clean up from previous runs
    await conn.execute("TRUNCATE spike_contacts, spike_permissions, spike_users, spike_tenants CASCADE")

    # Seed data
    seed_info = await seed_data(conn)

    # Get test users
    admin_user = next(u for u in seed_info["users"] if u["is_admin"])
    regular_user = next(u for u in seed_info["users"] if not u["is_admin"])
    tenant_id = admin_user["tenant_id"]

    # Get visible IDs for regular user
    visible = await conn.fetch(
        "SELECT entity_id FROM spike_permissions WHERE user_id = $1",
        uuid.UUID(regular_user["id"])
    )
    visible_ids = [str(r["entity_id"]) for r in visible]

    # Generate a query embedding (random — simulates a real query embedding)
    query_embedding = generate_random_embedding()

    # ─── Benchmarks ───
    print(f"\n{'='*60}")
    print(f"Benchmarks ({BENCHMARK_ITERATIONS} iterations each)")
    print(f"{'='*60}")

    results = []

    # 1. FTS — Admin (no permission filter)
    print("\n  [1/8] FTS — Admin (no permission filter)...")
    r = await benchmark_fts(conn, tenant_id, admin=True)
    results.append(r)
    print(f"    avg={r['avg_ms']}ms, p95={r['p95_ms']}ms, p99={r['p99_ms']}ms, results={r['avg_results']:.0f}")

    # 2. FTS — Regular user (with permission filter)
    print("  [2/8] FTS — Regular user (with permission filter)...")
    r = await benchmark_fts(conn, tenant_id, admin=False, visible_ids=visible_ids)
    results.append(r)
    print(f"    avg={r['avg_ms']}ms, p95={r['p95_ms']}ms, p99={r['p99_ms']}ms, results={r['avg_results']:.0f}")

    # 3. Vector — Admin
    print("  [3/8] Vector — Admin (no permission filter)...")
    r = await benchmark_vector(conn, tenant_id, query_embedding, admin=True)
    results.append(r)
    print(f"    avg={r['avg_ms']}ms, p95={r['p95_ms']}ms, p99={r['p99_ms']}ms, results={r['avg_results']:.0f}")

    # 4. Vector — Regular user
    print("  [4/8] Vector — Regular user (with permission filter)...")
    r = await benchmark_vector(conn, tenant_id, query_embedding, admin=False, visible_ids=visible_ids)
    results.append(r)
    print(f"    avg={r['avg_ms']}ms, p95={r['p95_ms']}ms, p99={r['p99_ms']}ms, results={r['avg_results']:.0f}")

    # 5. Hybrid — Admin
    print("  [5/8] Hybrid (FTS+Vector+RRF) — Admin...")
    r = await benchmark_hybrid(conn, tenant_id, query_embedding, admin=True)
    results.append(r)
    print(f"    avg={r['avg_ms']}ms, p95={r['p95_ms']}ms, p99={r['p99_ms']}ms, results={r['avg_results']:.0f}")

    # 6. Hybrid — Regular user
    print("  [6/8] Hybrid (FTS+Vector+RRF) — Regular user...")
    r = await benchmark_hybrid(conn, tenant_id, query_embedding, admin=False, visible_ids=visible_ids)
    results.append(r)
    print(f"    avg={r['avg_ms']}ms, p95={r['p95_ms']}ms, p99={r['p99_ms']}ms, results={r['avg_results']:.0f}")

    # 7. Cross-tenant isolation test
    print("  [7/8] Cross-tenant isolation test...")
    # Search tenant 1 — should not return tenant 2 contacts
    other_tenant = seed_info["tenants"][1] if seed_info["tenants"][0] == tenant_id else seed_info["tenants"][0]
    t0 = time.perf_counter()
    rows = await conn.fetch("""
        SELECT id, tenant_id FROM spike_contacts
        WHERE tenant_id = $1 AND deleted_at IS NULL
          AND search_tsv @@ to_tsquery('pg_catalog.german', 'Müller')
        LIMIT 20
    """, uuid.UUID(other_tenant))
    cross_tenant_ms = (time.perf_counter() - t0) * 1000
    cross_ok = all(str(r["tenant_id"]) == other_tenant for r in rows)
    results.append({
        "mode": "Cross-Tenant Isolation",
        "avg_ms": round(cross_tenant_ms, 2),
        "passed": cross_ok,
    })
    print(f"    {'✅ PASSED' if cross_ok else '❌ FAILED'} — {cross_tenant_ms:.2f}ms, {len(rows)} results, all from correct tenant")

    # 8. Permission filtering correctness
    print("  [8/8] Permission filtering correctness...")
    perm_ok = await verify_permission_filtering(conn, seed_info["users"][:4], seed_info["contact_ids"])
    tenant_ok = await verify_tenant_isolation(conn, seed_info["tenants"])
    sensitive_ok = await verify_sensitive_data_exclusion(conn)
    results.append({
        "mode": "Permission + Tenant + Sensitive Data",
        "passed": perm_ok and tenant_ok and sensitive_ok,
    })
    print(f"    {'✅ ALL PASSED' if (perm_ok and tenant_ok and sensitive_ok) else '❌ FAILED'}")

    # ─── Summary ───
    print(f"\n{'='*60}")
    print("SPIKE-E SUMMARY")
    print(f"{'='*60}")
    print(f"{'Mode':<35} {'Avg (ms)':<12} {'P95 (ms)':<12} {'P99 (ms)':<12} {'Results':<10}")
    print("-"*81)
    for r in results:
        if "avg_ms" in r and "p95_ms" in r:
            admin_str = "(admin)" if r.get("admin") else "(filtered)"
            mode = f"{r['mode']} {admin_str}"
            print(f"{mode:<35} {r['avg_ms']:<12} {r['p95_ms']:<12} {r['p99_ms']:<12} {r.get('avg_results', 0):<10.0f}")
        elif r.get("passed") is not None:
            status = "✅ PASSED" if r["passed"] else "❌ FAILED"
            print(f"{r['mode']:<35} {status}")
    print("-"*81)

    # ─── Verdict ───
    all_perf_ok = all(r.get("avg_ms", 0) < 100 for r in results if "avg_ms" in r and "p95_ms" in r)
    all_correct_ok = all(r.get("passed", True) for r in results if "passed" in r)

    print(f"\n  Performance: {'✅ ALL < 100ms avg' if all_perf_ok else '⚠️ SOME > 100ms avg'}")
    print(f"  Correctness: {'✅ ALL VERIFIED' if all_correct_ok else '❌ ISSUES FOUND'}")
    print(f"\n  SPIKE-E VERDICT: {'✅ PASS — Phase E can proceed' if (all_perf_ok and all_correct_ok) else '⚠️ ISSUES — investigate before Phase E'}")
    print(f"{'='*60}\n")

    await conn.close()


if __name__ == "__main__":
    asyncio.run(main())