3d9b76cea4
Check Cross-Plugin Imports / check (push) Has been cancelled
- SPIKE-E: FTS+Vector+Permission benchmark on 10k records (all <30ms) - E-PROV: supports_fts/vector/rag/graph capability flags on all providers - E-FTS/VEC: All 11 providers refactored to BaseSearchProvider with permission filtering - E-PERM: Over-fetch strategy for vector+permission (15x faster than ANY() filter) - E-FUSE: rrf_fusion_multi() for N-way RRF over FTS+Vector+RAG+Graph - E-LLM: Query understanding cleaned up to use central llm_complete() - E-CHUNK: Document chunking module + document_chunks table with HNSW index - E-EMB: Chunk embedding ARQ jobs (index_file_chunks, reindex_chunks) - E-RAG: RAG retrieval via FileSearchProvider.search_rag() - E-GRAPH: GraphRAG BFS traversal via GraphRAGSearchProvider.search_graph() - E-IX-EVT: Auto-indexing via outbox events + delete/cleanup handlers - E-IX-RE: Batch reindex with progress tracking + reindex_all job - E-DATA-LIFE: Lifecycle module (remove/rebuild/restore/correct) + API endpoints - E-K-MEM: AgentMemorySearchProvider - E-P-AI: AIChatSearchProvider - E-P-WF: WorkflowSearchProvider - E-P-COMM: ConversationSearchProvider verified (already on BaseSearchProvider) - E-API: Filter params (date_from/to, tags, sort) + /facets endpoint - E-TOOL: unified_search AI tool registered in ToolRegistry - E-MCP: Search tool in MCP server with normal RBAC/tenant checks - E-UI-CMD: CommandPalette (Cmd+K) with debounced search + recent searches - E-UI-FAC: SearchFacets, SearchResultCard, SavedSearches components - E-TEST: 40 new tests in test_unified_search_phase_e.py (105 total green) - E-DOC: api-documentation.md, plugin-development-guide.md, test-strategy.md updated 105 tests passing, TypeScript clean.
714 lines
28 KiB
Python
714 lines
28 KiB
Python
#!/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())
|