openvibers/openvibe.search Commits

Search Fabric hybrid ranking: pgvector embeddings of public documents through OpenVibe.AI, fused with the keyword rank by RRF (k 60); keyword ranking whenever vectors are not there, and explain says why (#126)

OP OpenVibers committed 16 days ago · 0402cf1446 · 15 files, +1048 −52

Changes

ddiff --git a/.env.example b/.env.example
iindex db808b0..4bd209e 100644
---- a/.env.example
++++ b/.env.example
@@@ -64,6 +64,21 @@ VALKEY_PREFIX=
# Owners whose documents the ladder's own open index reads first.
#SEARCH_OPEN_INDEX_OWNERS=sources
++# Semantic ranking (SF-2, README "Semantic ranking"): every public document gets one local
++# `bge-small-en-v1.5` embedding (384 dimensions) through OpenVibe.AI's POST /api/v1/embed, and a
++# query's keyword and vector ranks are fused with reciprocal rank fusion (k = 60). AI must serve
++# the embedding through its local provider (AI_LOCAL_EMBED_MODEL=bge-small-en-v1.5) and Network must
++# grant this principal `ai.run.create` with namespace `ai.embed` for audience openvibe.ai.
++# The job is on unless SEARCH_EMBED=0. At most SEARCH_EMBED_PER_MIN documents are embedded per
++# rolling minute (batches of 32); SEARCH_EMBED_INTERVAL_MS is how often a pass starts.
++#SEARCH_EMBED=1
++#SEARCH_EMBED_MODEL=bge-small-en-v1.5
++#SEARCH_EMBED_DIM=384
++#SEARCH_EMBED_PER_MIN=120
++#SEARCH_EMBED_INTERVAL_MS=60000
++# Restore-drill mode: background work stays off (the drill also blanks OV_OAUTH_CLIENT_SECRET).
++#SEARCH_DRILL=1
++
# Saved searches per signed-in person.
#SEARCH_SAVED_MAX_PER_SUBJECT=50
# Notifications of new matches through OpenVibe.Network (README "Saved searches"): off unless 1, and
ddiff --git a/migrations/0006_embeddings.sql b/migrations/0006_embeddings.sql
nnew file mode 100644
iindex 0000000..7a422c4
---- /dev/null
++++ b/migrations/0006_embeddings.sql
@@@ -0,0 +1,28 @@
++-- phase: expand
++-- Semantic ranking (SF-2, plan 2026-10-28 §4.5): one embedding per public document for the Search Fabric's network
++-- scope. PostgreSQL 18 with pgvector 0.8.1 has the extension already (created by the superuser in ov_search); PGlite
++-- has no `vector` type, so this migration checks pg_extension first and creates nothing there. The DDL runs through
++-- EXECUTE because the vector type must not even be parsed on a database without the extension.
++DO $embeddings$
++BEGIN
++ IF EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'vector') THEN
++ -- owner, type and key are the document's identity exactly as the documents table keys it (owner, type, id);
++ -- one row per document, so switching models replaces the old vector instead of mixing two vector spaces.
++ EXECUTE $ddl$
++ CREATE TABLE IF NOT EXISTS search_embeddings (
++ owner text COLLATE "C" NOT NULL,
++ type text COLLATE "C" NOT NULL,
++ key text COLLATE "C" NOT NULL,
++ model text COLLATE "C" NOT NULL,
++ dim integer NOT NULL,
++ embedding vector(384) NOT NULL,
++ content_hash text COLLATE "C" NOT NULL,
++ updated_at timestamptz NOT NULL DEFAULT now(),
++ PRIMARY KEY (owner, type, key)
++ )
++ $ddl$;
++ EXECUTE 'CREATE INDEX IF NOT EXISTS search_embeddings_embedding_hnsw
++ ON search_embeddings USING hnsw (embedding vector_cosine_ops)';
++ END IF;
++END
++$embeddings$;
ddiff --git a/scripts/embed-backfill.js b/scripts/embed-backfill.js
nnew file mode 100644
iindex 0000000..97cfcbb
---- /dev/null
++++ b/scripts/embed-backfill.js
@@@ -0,0 +1,67 @@
++#!/usr/bin/env node
++'use strict';
++/**
++ * Embed every public document once, for a first deploy of semantic ranking (SF-2).
++ *
++ * node scripts/embed-backfill.js # run the job over everything, respecting SEARCH_EMBED_PER_MIN
++ * node scripts/embed-backfill.js --dry # report what would be embedded; no AI call, no write
++ *
++ * It uses the same configuration, vector store and embedder as the service (server/fabric/embedder.js), so
++ * the model, the 2 000-character text, the public-only rule and the batches are exactly the running job's.
++ * On a database without pgvector (PGlite) or with SEARCH_EMBED=0 it says so and exits 0. A failed AI call
++ * backs off and is retried; the script exits non-zero only on an unexpected error. No secret is printed.
++ */
++require('dotenv').config();
++const { load } = require('../server/config');
++const { openDb } = require('../server/db');
++const { createPgVectorStore } = require('../server/fabric/vector-store');
++const { createEmbeddingClient } = require('../server/fabric/embed');
++const { createEmbedder } = require('../server/fabric/embedder');
++
++const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, Math.max(1000, ms)));
++
++async function main() {
++ const dry = process.argv.slice(2).includes('--dry');
++ const config = load();
++ const log = {
++ log: (m) => console.log(m),
++ warn: (m) => console.warn(m),
++ error: (m) => console.error(m),
++ };
++ const db = await openDb(config, { log });
++ try {
++ const vectors = createPgVectorStore({ db, model: config.embed.model, dim: config.embed.dim });
++ const embeddings = createEmbeddingClient({ config, log });
++ const embedder = createEmbedder({ db, vectors, embeddings, config, log });
++
++ const before = await embedder.plan();
++ if (before.skipped) {
++ console.log(`[embed-backfill] ${before.skipped}`);
++ return;
++ }
++ console.log(`[embed-backfill] ${before.public} public document(s): ${before.embedded} embedded, ${before.pending} pending`);
++ if (dry) {
++ console.log(`[embed-backfill] --dry: would embed ${before.pending} document(s), batches of 32, at most ${config.embed.perMinute}/minute`);
++ return;
++ }
++
++ let done = 0;
++ for (;;) {
++ const r = await embedder.tick();
++ if (r.skipped) { console.log(`[embed-backfill] stopped: ${r.skipped}`); break; }
++ done += r.embedded || 0;
++ console.log(`[embed-backfill] ${done} embedded, ${r.pending || 0} pending${r.failed ? ` (${r.error})` : ''}`);
++ if (!r.pending) break;
++ if (r.failed) await sleep((r.retry_at || Date.now() + 2000) - Date.now());
++ else if (!r.embedded) await sleep(60000); // the rolling minute's budget is spent
++ }
++ console.log(`[embed-backfill] done: ${done} document(s) embedded`);
++ } finally {
++ await db.close();
++ }
++}
++
++main().catch((err) => {
++ console.error(`[embed-backfill] failed: ${err && err.message ? err.message : err}`);
++ process.exitCode = 1;
++});
ddiff --git a/server/api/query.js b/server/api/query.js
iindex 4e9ba7f..e7485c5 100644
---- a/server/api/query.js
++++ b/server/api/query.js
@@@ -27,9 +27,12 @@ const { toMatch, snippetHtml } = require('../engine/pg');
const { EXPOSURE } = require('../document');
const { AuthError } = require('../auth');
const { OWNER_RE, TYPE_RE, ID_RE } = require('./documents');
++const { fuseRrf } = require('../fabric/rrf');
++const { createQueryEmbeddingCache } = require('../fabric/embed');
const FACET_KEY_RE = /^[a-z][a-z0-9_]{0,39}$/;
const LANG_RE = /^[a-zA-Z]{2,3}(-[A-Za-z0-9]{2,8})*$/;
++const HYBRID_CANDIDATES = 100;
class QueryError extends Error {
constructor(code, detail) { super(detail); this.code = code; }
@@@ -74,13 +77,16 @@ function decodeCursor(s, hash) {
try { c = JSON.parse(Buffer.from(String(s), 'base64url').toString('utf8')); } catch { c = null; }
if (!c || c.h !== hash || !Number.isInteger(c.i)) throw new QueryError('search.bad_cursor', 'cursor does not belong to this query');
// n: the first page's reference time for freshness, so later pages rank on the same clock.
++ // f: a hybrid page is an offset into the fused list (the keyword and vector ranks are recomputed for
++ // each page, so the fused order is stable; `i` carries the last rid for the shared cursor shape).
++ if (c.k === 'f' && Number.isInteger(c.o) && c.o >= 0) return { offset: c.o, rid: c.i, now: Number.isFinite(c.n) ? c.n : null };
if (c.k === 'r' && typeof c.r === 'number') return { rank: c.r, rid: c.i, now: Number.isFinite(c.n) ? c.n : null };
if (c.k === 't' && typeof c.t === 'string') return { sort_at: c.t, rid: c.i };
throw new QueryError('search.bad_cursor', 'cursor is malformed');
}
/** What a viewer gets for one document: never the ACL, the hash or the owner's internals. */
--function projection(doc, row, snip) {
++function projection(doc, row, snip, ranks) {
const out = {
owner: doc.owner, type: doc.type, id: doc.id, revision: doc.revision,
visibility: doc.visibility,
@@@ -96,6 +102,7 @@ function projection(doc, row, snip) {
indexable: row.index_decision === 'index',
};
if (snip !== undefined) out.snippet_html = snippetHtml(snip);
++ if (ranks !== undefined) out.ranks = ranks;
return out;
}
@@@ -118,12 +125,57 @@ function canView(viewer, row, doc) {
/**
* One search as a viewer: the query API, saved-search runs and the HTML page all go through here.
-- * run({ text, query, viewer, limit, cursor, facetKeys, now, since }) → { results, next_cursor, facets? }
++ * run({ text, query, viewer, limit, cursor, facetKeys, now, since }) → { results, next_cursor, explain, facets? }
* `query` is the query-string-shaped filter source (owner, type, lang, facet.<key>). `since` (epoch ms,
* internal callers only: the saved-search notifier) keeps documents indexed after it; the public
* API never passes it. Throws QueryError for a malformed request.
++ *
++ * With a vector store and embeddings configured, a query of at least two characters also ranks the
++ * public listed documents by cosine distance (top 100) and fuses that list with the keyword top 100 by
++ * reciprocal rank fusion (k = 60). The keyword list still supplies snippets, filters, facet counts and
++ * the ACL: the vector list is public-only, and every fused hit is re-checked by canView. When vectors
++ * are not available, AI fails or the query is too short, the answer is exactly today's keyword page and
++ * `explain` says which reason applied. Keyword paging keeps its rank/rid cursor; a hybrid page carries
++ * an offset into the fused list. `ranks` on each hit are 1-based positions in the two lists (null when
++ * the hit is absent from one).
*/
--function createSearcher({ config, store, engine, now: clock = () => Date.now() }) {
++function createSearcher({ config, store, engine, vectorStore = null, embeddings = null, now: clock = () => Date.now() }) {
++ const queryEmbeddings = embeddings ? createQueryEmbeddingCache({ client: embeddings, now: clock }) : null;
++
++ /** 'no_vectors' when the extension/table or a current-model embedding is missing, else null. */
++ async function vectorAvailability() {
++ if (!vectorStore || typeof vectorStore.available !== 'function') return 'no_vectors';
++ try {
++ if (!await vectorStore.available()) return 'no_vectors';
++ if (typeof vectorStore.hasEmbeddings === 'function' && !await vectorStore.hasEmbeddings()) return 'no_vectors';
++ } catch {
++ return 'no_vectors';
++ }
++ return null;
++ }
++
++ /** The ranking this query would get: { ranking: 'hybrid', vector } or { ranking: 'keyword', reason }. */
++ async function semanticState(text) {
++ const unavailable = await vectorAvailability();
++ if (unavailable) return { ranking: 'keyword', reason: unavailable };
++ if (String(text).trim().length < 2) return { ranking: 'keyword', reason: 'short_query' };
++ if (!queryEmbeddings) return { ranking: 'keyword', reason: 'ai_unavailable' };
++ const embedded = await queryEmbeddings.get(text);
++ if (!embedded.ok) return { ranking: 'keyword', reason: 'ai_unavailable' };
++ if (vectorStore.dim && embedded.vector.length !== vectorStore.dim) return { ranking: 'keyword', reason: 'ai_unavailable' };
++ return { ranking: 'hybrid', vector: embedded.vector };
++ }
++
++ /** The two top-100 lists and their RRF order; snippets come from the keyword side. */
++ async function hybridCandidates({ match, filters, viewer, now, vector }) {
++ const [keyword, vectorHits] = await Promise.all([
++ engine.search({ match, filters, viewer, limit: HYBRID_CANDIDATES, now }),
++ vectorStore.nearest({ vector, filters, viewer, limit: HYBRID_CANDIDATES }),
++ ]);
++ const snips = new Map(keyword.hits.map((h) => [h.rid, h.snip]));
++ return fuseRrf(keyword.hits, vectorHits).map((h) => ({ ...h, snip: snips.get(h.rid) }));
++ }
++
async function run({ text = '', query = {}, filters = null, viewer, limit, cursor, facetKeys = [], now = clock(), since = null }) {
if (text.length > 500) throw new QueryError('search.bad_query', 'q is longer than 500 characters');
filters = filters || parseFilters(query, config);
@@@ -131,33 +183,71 @@ function createSearcher({ config, store, engine, now: clock = () => Date.now() }
limit = Math.min(Math.max(parseInt(limit, 10) || config.query.defaultLimit, 1), config.query.maxLimit);
if (facetKeys.length > 5 || facetKeys.some(k => !FACET_KEY_RE.test(k))) throw new QueryError('search.bad_query', 'facets: at most 5 well-formed keys');
-- const match = text.trim() ? toMatch(text, { maxTerms: config.query.maxTerms }) : null;
++ const trimmed = text.trim();
++ const match = trimmed ? toMatch(text, { maxTerms: config.query.maxTerms }) : null;
const hash = queryHash([match, filters]);
const after = decodeCursor(cursor, hash);
-- if (text.trim() && !match) {
-- return { results: [], next_cursor: null, ...(facetKeys.length ? { facets: Object.fromEntries(facetKeys.map(k => [k, []])) } : {}) };
++ if (trimmed && !match) {
++ const unavailable = await vectorAvailability();
++ return {
++ results: [], next_cursor: null,
++ explain: { ranking: 'keyword', reason: unavailable || 'short_query' },
++ ...(facetKeys.length ? { facets: Object.fromEntries(facetKeys.map(k => [k, []])) } : {}),
++ };
}
-- let page;
++
++ const results = [];
++ let next = null;
++ let explain;
let refNow = now;
if (match) {
-- if (after && after.rank === undefined) throw new QueryError('search.bad_cursor', 'cursor does not belong to this query');
-- if (after && after.now != null) refNow = after.now;
-- page = await engine.search({ match, filters, viewer, limit, after, now: refNow });
++ const state = await semanticState(trimmed);
++ if (state.ranking === 'hybrid') {
++ if (after && after.offset === undefined) throw new QueryError('search.bad_cursor', 'cursor does not belong to this query');
++ if (after && after.now != null) refNow = after.now;
++ const fused = await hybridCandidates({ match, filters, viewer, now: refNow, vector: state.vector });
++ let consumed = after ? after.offset : 0;
++ let lastRid = after ? after.rid : null;
++ for (let i = consumed; i < fused.length && results.length < limit; i++) {
++ const f = fused[i];
++ consumed = i + 1;
++ lastRid = f.rid;
++ const found = await store.byRid(f.rid);
++ // Defence in depth: the vector store is public-only, the keyword SQL is ACL-aware.
++ if (!found || !canView(viewer, found.row, found.doc)) continue;
++ results.push(projection(found.doc, found.row, f.snip, f.ranks));
++ }
++ if (consumed < fused.length) next = encodeCursor({ k: 'f', o: consumed, i: lastRid, n: refNow, h: hash });
++ explain = { ranking: 'hybrid' };
++ } else {
++ if (after && after.rank === undefined) throw new QueryError('search.bad_cursor', 'cursor does not belong to this query');
++ if (after && after.now != null) refNow = after.now;
++ const page = await engine.search({ match, filters, viewer, limit, after, now: refNow });
++ let position = 0;
++ for (const h of page.hits) {
++ position++;
++ const found = await store.byRid(h.rid);
++ // Defence in depth: the engine already filtered; re-check against the stored row.
++ if (!found || !canView(viewer, found.row, found.doc)) continue;
++ results.push(projection(found.doc, found.row, h.snip, { keyword: position, vector: null }));
++ }
++ next = page.next ? encodeCursor({ k: 'r', r: page.next.rank, i: page.next.rid, n: refNow, h: hash }) : null;
++ explain = { ranking: 'keyword', reason: state.reason };
++ }
} else {
if (after && after.sort_at === undefined) throw new QueryError('search.bad_cursor', 'cursor does not belong to this query');
-- page = await engine.browse({ filters, viewer, limit, after });
-- }
-- const results = [];
-- for (const h of page.hits) {
-- const found = await store.byRid(h.rid);
-- // Defence in depth: the engine already filtered; re-check against the stored row.
-- if (!found || !canView(viewer, found.row, found.doc)) continue;
-- results.push(projection(found.doc, found.row, match ? h.snip : undefined));
++ const page = await engine.browse({ filters, viewer, limit, after });
++ for (const h of page.hits) {
++ const found = await store.byRid(h.rid);
++ if (!found || !canView(viewer, found.row, found.doc)) continue;
++ results.push(projection(found.doc, found.row));
++ }
++ next = page.next ? encodeCursor({ k: 't', t: page.next.sort_at, i: page.next.rid, h: hash }) : null;
++ const unavailable = await vectorAvailability();
++ explain = { ranking: 'keyword', reason: unavailable || 'short_query' };
}
-- const next = page.next
-- ? encodeCursor(match ? { k: 'r', r: page.next.rank, i: page.next.rid, n: refNow, h: hash } : { k: 't', t: page.next.sort_at, i: page.next.rid, h: hash })
-- : null;
-- const body = { results, next_cursor: next };
++
++ const body = { results, next_cursor: next, explain };
if (facetKeys.length) body.facets = await engine.facetCounts({ match, filters, viewer, keys: facetKeys });
return body;
}
@@@ -166,6 +256,7 @@ function createSearcher({ config, store, engine, now: clock = () => Date.now() }
run,
parseFilters: (query) => parseFilters(query, config),
hasWords: (text) => toMatch(text, { maxTerms: config.query.maxTerms }) !== null,
++ vectorAvailability,
};
}
@@@ -200,7 +291,7 @@ function withViewer(auth, handler, limit = null) {
};
}
--function queryRouter({ config, store, engine, auth, limits, searcher = createSearcher({ config, store, engine }) }) {
++function queryRouter({ config, store, engine, auth, limits, vectorStore = null, embeddings = null, searcher = createSearcher({ config, store, engine, vectorStore, embeddings }) }) {
const router = express.Router();
const guarded = (limit, h) => withViewer(auth, h, limit);
@@@ -254,4 +345,4 @@ function queryRouter({ config, store, engine, auth, limits, searcher = createSea
return router;
}
--module.exports = { queryRouter, createSearcher, withViewer, parseFilters, QueryError, canView, projection, toMatch, one };
++module.exports = { queryRouter, createSearcher, withViewer, parseFilters, QueryError, canView, projection, toMatch, one, fuseRrf };
ddiff --git a/server/app.js b/server/app.js
iindex 176e647..30417cc 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -21,7 +21,7 @@ const { createLimits } = require('./actor-limits');
const { createIndexNow } = require('openvibe-shared/indexnow');
const pkg = require('../package.json');
--function createApp({ config, db, store, engine, auth, outbox, purges, purger, saved, searcher: givenSearcher = null, notifier = null, fetchImpl = globalThis.fetch, log = console, now }) {
++function createApp({ config, db, store, engine, auth, outbox, purges, purger, saved, searcher: givenSearcher = null, notifier = null, vectorStore = null, embeddings = null, fetchImpl = globalThis.fetch, log = console, now }) {
const app = express();
app.disable('x-powered-by');
app.set('trust proxy', 'loopback');
@@@ -71,7 +71,7 @@ function createApp({ config, db, store, engine, auth, outbox, purges, purger, sa
// GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).
release.mount(app, { registry: metrics.registry });
-- const searcher = givenSearcher || createSearcher({ config, store, engine, now });
++ const searcher = givenSearcher || createSearcher({ config, store, engine, vectorStore, embeddings, now });
// One per-actor limiter for the API routes below (server/actor-limits.js); the Events webhook,
// health, ready, release.json and metrics above are never limited.
const limits = createLimits({ config, now: now || (() => Date.now()), registry: metrics.registry, log });
ddiff --git a/server/config.js b/server/config.js
iindex 6dfac9c..0c946c4 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -98,6 +98,21 @@ function load(env = process.env) {
},
maxBodyBytes: int(env.SEARCH_MAX_BODY_BYTES, 256 * 1024),
++ // Restore-drill mode (SEARCH_DRILL=1, `ovhost drill search`): background jobs that write or call out
++ // stay off. The drill's empty OV_OAUTH_CLIENT_SECRET is a second switch for anything that needs AI.
++ drill: env.SEARCH_DRILL === '1',
++
++ // Semantic ranking (SF-2): one local `bge-small-en-v1.5` embedding (384 dimensions) per public document,
++ // through OpenVibe.AI's /api/v1/embed. SEARCH_EMBED=0 turns the job off; SEARCH_EMBED_PER_MIN bounds how
++ // many documents one rolling minute may embed; SEARCH_EMBED_INTERVAL_MS is how often a pass starts.
++ embed: {
++ enabled: env.SEARCH_EMBED !== '0',
++ model: String(env.SEARCH_EMBED_MODEL || 'bge-small-en-v1.5').trim() || 'bge-small-en-v1.5',
++ dim: Math.max(1, int(env.SEARCH_EMBED_DIM, 384)),
++ perMinute: Math.max(1, int(env.SEARCH_EMBED_PER_MIN, 120)),
++ intervalMs: Math.max(1000, int(env.SEARCH_EMBED_INTERVAL_MS, 60000)),
++ },
++
// The Search Fabric's web ladder (server/fabric): OpenVibe.AI's research endpoint for the web
// providers, its own open index first. SEARCH_WEB=off leaves the ladder's providers off (the
// open index and cached answers still work); SEARCH_WEB_DAILY bounds paid DeepSeek calls per day.
ddiff --git a/server/engine/pg.js b/server/engine/pg.js
iindex 6f81744..27bc6f0 100644
---- a/server/engine/pg.js
++++ b/server/engine/pg.js
@@@ -29,6 +29,25 @@ const HEADLINE = `StartSel=${MARK_OPEN}, StopSel=${MARK_CLOSE}, MaxWords=16, Min
const DAY_MS = 24 * 3600 * 1000;
++/**
++ * The filter predicate shared by the full-text engine and the vector store (alias d = documents).
++ * Filters never change visibility; the caller adds that. `since` is internal (saved searches).
++ */
++function filterClause(filters, params) {
++ const parts = [];
++ if (filters.owner) { parts.push('d.owner = @f_owner'); params.f_owner = filters.owner; }
++ if (filters.type) { parts.push('d.type = @f_type'); params.f_type = filters.type; }
++ if (filters.language) { parts.push('d.language = @f_lang'); params.f_lang = filters.language; }
++ // Internal only (saved-search notifications): documents indexed (a new revision) after `since`, epoch ms.
++ if (filters.since != null) { parts.push('d.indexed_at > @f_since::bigint'); params.f_since = Number(filters.since); }
++ (filters.facets || []).forEach(([key, values], i) => {
++ params[`f_fk${i}`] = key;
++ params[`f_fv${i}`] = values.map(String);
++ parts.push(`EXISTS (SELECT 1 FROM doc_facets x WHERE x.rid = d.rid AND x.key = @f_fk${i} AND x.value = ANY(@f_fv${i}::text[]))`);
++ });
++ return parts.length ? ` AND ${parts.join(' AND ')}` : '';
++}
++
/**
* Freshness multiplier for a document dated `iso` at reference time `nowMs`:
*
@@@ -110,21 +129,6 @@ function createEngine(db, { freshness = { weight: 1, halfLifeDays: 30 } } = {})
)`;
}
-- function filterClause(filters, params) {
-- const parts = [];
-- if (filters.owner) { parts.push('d.owner = @f_owner'); params.f_owner = filters.owner; }
-- if (filters.type) { parts.push('d.type = @f_type'); params.f_type = filters.type; }
-- if (filters.language) { parts.push('d.language = @f_lang'); params.f_lang = filters.language; }
-- // Internal only (saved-search notifications): documents indexed (a new revision) after `since`, epoch ms.
-- if (filters.since != null) { parts.push('d.indexed_at > @f_since::bigint'); params.f_since = Number(filters.since); }
-- (filters.facets || []).forEach(([key, values], i) => {
-- params[`f_fk${i}`] = key;
-- params[`f_fv${i}`] = values.map(String);
-- parts.push(`EXISTS (SELECT 1 FROM doc_facets x WHERE x.rid = d.rid AND x.key = @f_fk${i} AND x.value = ANY(@f_fv${i}::text[]))`);
-- });
-- return parts.length ? ` AND ${parts.join(' AND ')}` : '';
-- }
--
/**
* The visible, matching set as (rid, rank) rows. match = toMatch()'s query. rank is minus ts_rank times the
* freshness boost at reference time `now` (ms), so lower ranks first and pages of one query
@@@ -293,4 +297,4 @@ function snippetHtml(snip) {
return esc.split(MARK_OPEN).join('<mark>').split(MARK_CLOSE).join('</mark>');
}
--module.exports = { createEngine, toMatch, tsquery, snippetHtml, freshnessBoost, fold, words };
++module.exports = { createEngine, toMatch, tsquery, snippetHtml, freshnessBoost, fold, words, filterClause };
ddiff --git a/server/fabric/embed.js b/server/fabric/embed.js
nnew file mode 100644
iindex 0000000..45a18f2
---- /dev/null
++++ b/server/fabric/embed.js
@@@ -0,0 +1,136 @@
++'use strict';
++/**
++ * Embeddings through OpenVibe.AI, and the short-lived cache the query path keeps in front of them.
++ *
++ * POST <config.fabric.aiUrl>/api/v1/embed?wait=30000 { input: [text, ...] }
++ * -> { run: { status, output: { vectors, dimensions }, provenance: { model } } }
++ *
++ * Search calls AI as itself: a Network client-credentials token (client `search`, audience
++ * openvibe.ai) carrying ai.run.create and the ai.embed namespace. The token is fetched lazily, so a
++ * Search without the Network OAuth client boots and simply degrades to keyword ranking. Nothing here
++ * throws: a network failure, a non-2xx answer, a failed run or malformed vectors becomes
++ * { ok: false, error } so the query can answer with reason 'ai_unavailable' and the embedder can back off.
++ *
++ * Query vectors are cached for 10 minutes by normalized text (LRU, 500 entries): the same search
++ * typed twice costs one embedding call, and a popular query never grows the cache past its bound.
++ */
++const { createServiceTokenClient } = require('openvibe-sdk/auth');
++const { normalizeQuery } = require('./cache');
++
++const AI_TIMEOUT_MS = 30000;
++const MAX_INPUTS = 64;
++
++function shortError(err) {
++ return String((err && (err.message || err.name)) || 'request failed').replace(/\s+/g, ' ').trim().slice(0, 160) || 'request failed';
++}
++
++function createEmbeddingClient({ config, fetchImpl = globalThis.fetch, tokenClient = null, now = () => Date.now(), log = console }) {
++ const fabric = config.fabric || {};
++ const oauth = config.oauth || {};
++ const embedConfig = config.embed || {};
++ const aiUrl = String(fabric.aiUrl || '').replace(/\/+$/, '');
++ const model = String(embedConfig.model || 'bge-small-en-v1.5');
++ let tokens = null;
++
++ function available() {
++ const missing = [];
++ if (!aiUrl) missing.push('OV_AI_INTERNAL_URL');
++ if (!oauth.clientSecret) missing.push('OV_OAUTH_CLIENT_SECRET');
++ return { configured: missing.length === 0, reason: missing.length ? `${missing.join(' and ')} unset` : null };
++ }
++
++ function client() {
++ if (!tokens) {
++ tokens = tokenClient || createServiceTokenClient({
++ network: config.networkInternalUrl,
++ clientId: oauth.clientId,
++ clientSecret: oauth.clientSecret,
++ audience: 'openvibe.ai',
++ fetch: fetchImpl,
++ now,
++ });
++ }
++ return tokens;
++ }
++
++ /** Embed 1..64 texts. → { ok, vectors, dimensions, model, ms } | { ok: false, error, ms }. */
++ async function embed(texts) {
++ const started = now();
++ const ms = () => Math.max(0, Math.round(now() - started));
++ const fail = (error) => ({ ok: false, error, ms: ms() });
++ const inputs = (Array.isArray(texts) ? texts : [texts]).map((t) => String(t == null ? '' : t));
++ if (!inputs.length) return fail('no text to embed');
++ if (inputs.length > MAX_INPUTS) return fail(`at most ${MAX_INPUTS} texts per embedding call`);
++
++ const state = available();
++ if (!state.configured) return fail(state.reason);
++
++ try {
++ const token = await client().getToken({ audience: 'openvibe.ai' });
++ const timeout = AbortSignal.timeout(AI_TIMEOUT_MS);
++ const res = await fetchImpl(`${aiUrl}/api/v1/embed?wait=30000`, {
++ method: 'POST',
++ headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', Accept: 'application/json' },
++ body: JSON.stringify({ input: inputs }),
++ signal: timeout,
++ });
++ if (!res.ok) return fail(`AI answered ${res.status}`);
++ let json;
++ try { json = await res.json(); } catch { return fail('AI answered no JSON'); }
++ const run = json && json.run;
++ if (!run) return fail('AI answered no run');
++ if (run.status !== 'succeeded' && run.status !== 'cached') return fail(`AI run ${run.status || 'did not finish'}`);
++ const output = run.output || {};
++ const vectors = output.vectors;
++ if (!Array.isArray(vectors) || vectors.length !== inputs.length
++ || vectors.some((v) => !Array.isArray(v) || !v.length || v.some((n) => typeof n !== 'number' || !Number.isFinite(n)))) {
++ return fail('AI answered invalid vectors');
++ }
++ return {
++ ok: true,
++ vectors,
++ dimensions: Number(output.dimensions) || (vectors[0] ? vectors[0].length : 0),
++ model: (run.provenance && run.provenance.model) || model,
++ ms: ms(),
++ };
++ } catch (err) {
++ log.warn(`[embed] AI: ${shortError(err)}`);
++ return fail(shortError(err));
++ }
++ }
++
++ return { embed, available, model };
++}
++
++/**
++ * Query-embedding cache: normalized text -> vector, 10 minutes, LRU 500. `get()` returns
++ * { ok, vector, cached } or { ok: false, error }; a failure is never cached, so the next search tries again.
++ */
++function createQueryEmbeddingCache({ client, now = () => Date.now(), ttlMs = 10 * 60 * 1000, max = 500 }) {
++ const entries = new Map(); // insertion order is the LRU order
++
++ async function get(text) {
++ const key = normalizeQuery(text);
++ if (!key) return { ok: false, error: 'empty query' };
++ const hit = entries.get(key);
++ if (hit) {
++ if (now() - hit.at <= ttlMs) {
++ entries.delete(key);
++ entries.set(key, hit);
++ return { ok: true, vector: hit.vector, cached: true, model: hit.model };
++ }
++ entries.delete(key);
++ }
++ const r = await client.embed([key]);
++ if (!r.ok) return { ok: false, error: r.error };
++ const vector = r.vectors && r.vectors[0];
++ if (!Array.isArray(vector) || !vector.length) return { ok: false, error: 'AI answered no vector' };
++ entries.set(key, { vector, at: now(), model: r.model || null });
++ while (entries.size > max) entries.delete(entries.keys().next().value);
++ return { ok: true, vector, cached: false, model: r.model || null };
++ }
++
++ return { get, size: () => entries.size, clear: () => entries.clear() };
++}
++
++module.exports = { createEmbeddingClient, createQueryEmbeddingCache, shortError, AI_TIMEOUT_MS, MAX_INPUTS };
ddiff --git a/server/fabric/embedder.js b/server/fabric/embedder.js
nnew file mode 100644
iindex 0000000..fa49e3f
---- /dev/null
++++ b/server/fabric/embedder.js
@@@ -0,0 +1,166 @@
++'use strict';
++/**
++ * The embedding job (SF-2): keeps search_embeddings in step with the public documents table.
++ *
++ * It is a job loop in the codebase's style (createPurger/createSavedNotifier): start() puts an unref'd
++ * timer on the process, tick() is one pass (overlapping calls share it), stop() clears it and waits.
++ * Every pass takes the public, published documents and compares their documents.hash with the stored
++ * search_embeddings.content_hash; a document with no row, a changed hash, another model or another
++ * dimension is pending. It embeds up to 32 at a time — title + summary + body, cut to 2 000 characters
++ * — through OpenVibe.AI, and stores each vector under the documents table's identity (owner, type, id).
++ *
++ * It is off when SEARCH_EMBED=0, under the restore-drill flag (SEARCH_DRILL=1), or when AI is not
++ * configured. Never embeds a document whose visibility is not 'public' (drafts, unpublished,
++ * unlisted, members and private documents have no vector at all). At most SEARCH_EMBED_PER_MIN
++ * documents are embedded per rolling minute. A failed AI call stores nothing and backs off
++ * exponentially (2 s .. 1 h); the documents stay pending and are retried after the delay.
++ *
++ * createEmbedder({ db, vectors, embeddings, config, now?, log? }) -> { start, stop, tick, state, running }
++ */
++const { identity } = require('./vector-store');
++
++const BATCH_SIZE = 32;
++const MAX_CHARS = 2000;
++const BACKOFF_MS = [2000, 10000, 60000, 300000, 900000, 1800000, 3600000];
++
++/** Why the job is off, or null when it may run. `embeddings.available()` is synchronous by contract. */
++function disabledReason(config, embeddings) {
++ const embed = config.embed || {};
++ if (!embed.enabled) return 'off (SEARCH_EMBED=0)';
++ if (config.drill) return 'off (SEARCH_DRILL=1)';
++ let state;
++ try { state = embeddings && embeddings.available ? embeddings.available() : { configured: false, reason: 'no embedding client' }; } catch { state = { configured: false, reason: 'embedding client unavailable' }; }
++ if (!state.configured) return `off (${state.reason || 'AI is not configured'})`;
++ return null;
++}
++
++/** The text one document contributes: title + summary + body, cut to 2 000 characters. */
++function embeddingText(doc) {
++ const d = typeof doc === 'string' ? JSON.parse(doc) : (doc || {});
++ return [d.title, d.summary, d.body].filter((x) => typeof x === 'string' && x.length).join('\n').slice(0, MAX_CHARS);
++}
++
++function createEmbedder({ db, vectors, embeddings, config, now = () => Date.now(), log = console }) {
++ const embed = config.embed || {};
++ const perMinute = Math.max(1, Number(embed.perMinute) || 120);
++ const intervalMs = Math.max(1000, Number(embed.intervalMs) || 60000);
++ const off = disabledReason(config, embeddings);
++
++ let timer = null;
++ let ticking = null;
++ let failures = 0;
++ let nextTryAt = 0;
++ let windowStart = now();
++ let usedThisWindow = 0;
++
++ function budgetLeft() {
++ const at = now();
++ if (at - windowStart >= 60000) { windowStart = at; usedThisWindow = 0; }
++ return Math.max(0, perMinute - usedThisWindow);
++ }
++
++ function backoff(error) {
++ failures++;
++ nextTryAt = now() + BACKOFF_MS[Math.min(failures - 1, BACKOFF_MS.length - 1)];
++ log.warn(`[embed] ${error}; retrying in ${Math.round((nextTryAt - now()) / 1000)}s`);
++ return nextTryAt;
++ }
++
++ async function publicDocuments() {
++ return await db.prepare(`SELECT rid, owner, type, id, hash, doc FROM documents
++ WHERE deleted = 0 AND visibility = 'public' AND publication_state = 'published'
++ ORDER BY rid`).all();
++ }
++
++ async function doTick() {
++ if (off) return { skipped: off };
++ if (now() < nextTryAt) return { backoff: true, retry_at: nextTryAt, failures };
++ if (!vectors || typeof vectors.available !== 'function') return { skipped: 'no_vectors' };
++ if (!await vectors.available()) return { skipped: 'no_vectors' };
++
++ const docs = await publicDocuments();
++ const existing = await vectors.existing();
++ const pending = docs.filter((row) => {
++ const prev = existing.get(identity(row.owner, row.type, row.id));
++ return !prev || prev.model !== vectors.model || prev.dim !== vectors.dim || prev.content_hash !== row.hash;
++ });
++ if (!pending.length) return { embedded: 0, batches: 0, pending: 0 };
++
++ let embedded = 0;
++ let batches = 0;
++ let at = 0;
++ while (at < pending.length) {
++ const left = budgetLeft();
++ if (left <= 0) break;
++ const rows = pending.slice(at, at + Math.min(BATCH_SIZE, left, pending.length - at));
++ const texts = rows.map((row) => embeddingText(row.doc));
++ const r = await embeddings.embed(texts);
++ usedThisWindow += rows.length;
++ if (!r || !r.ok) {
++ const retryAt = backoff((r && r.error) || 'AI did not answer');
++ return { embedded, batches, pending: pending.length - embedded, failed: 1, error: (r && r.error) || 'AI did not answer', retry_at: retryAt };
++ }
++ const vectorsOut = r.vectors || [];
++ if (vectorsOut.length !== rows.length || vectorsOut.some((v) => !Array.isArray(v) || v.length !== vectors.dim)) {
++ const retryAt = backoff(`AI answered ${vectorsOut.length} vector(s) of the wrong dimension`);
++ return { embedded, batches, pending: pending.length - embedded, failed: 1, error: 'invalid vectors', retry_at: retryAt };
++ }
++ const updatedAt = new Date(now()).toISOString();
++ for (let i = 0; i < rows.length; i++) {
++ const row = rows[i];
++ await vectors.put({
++ owner: row.owner, type: row.type, key: row.id,
++ embedding: vectorsOut[i], contentHash: row.hash, updatedAt,
++ });
++ }
++ failures = 0;
++ nextTryAt = 0;
++ embedded += rows.length;
++ batches++;
++ at += rows.length;
++ }
++ return { embedded, batches, pending: Math.max(0, pending.length - embedded) };
++ }
++
++ /** What one pass would do, without calling AI or writing anything (scripts/embed-backfill.js --dry). */
++ async function plan() {
++ if (off) return { skipped: off };
++ if (!vectors || typeof vectors.available !== 'function') return { skipped: 'no_vectors' };
++ if (!await vectors.available()) return { skipped: 'no_vectors' };
++ const docs = await publicDocuments();
++ const existing = await vectors.existing();
++ const pending = docs.filter((row) => {
++ const prev = existing.get(identity(row.owner, row.type, row.id));
++ return !prev || prev.model !== vectors.model || prev.dim !== vectors.dim || prev.content_hash !== row.hash;
++ });
++ return { available: true, public: docs.length, embedded: docs.length - pending.length, pending: pending.length };
++ }
++
++ /** One pass (overlapping calls share it). */
++ function tick() {
++ if (!ticking) ticking = doTick().finally(() => { ticking = null; });
++ return ticking;
++ }
++
++ function start() {
++ if (timer || off) return;
++ timer = setInterval(() => { tick().catch((err) => log.warn(`[embed] tick: ${err.message}`)); }, intervalMs);
++ timer.unref?.();
++ }
++
++ function stop() {
++ if (timer) clearInterval(timer);
++ timer = null;
++ return ticking ? ticking.catch(() => {}) : Promise.resolve();
++ }
++
++ return {
++ start, stop, tick, plan,
++ running: () => Boolean(timer),
++ state: () => off || (timer ? 'running' : 'stopped'),
++ failures: () => failures,
++ nextTryAt: () => nextTryAt,
++ };
++}
++
++module.exports = { createEmbedder, disabledReason, embeddingText, BATCH_SIZE, MAX_CHARS, BACKOFF_MS };
ddiff --git a/server/fabric/index.js b/server/fabric/index.js
iindex 53132cd..64aa180 100644
---- a/server/fabric/index.js
++++ b/server/fabric/index.js
@@@ -25,6 +25,37 @@ const MAX_EXPLAIN = 20;
const MAX_CITATIONS = 20;
const FRESHNESS_MS = { day: 86400000, week: 7 * 86400000, month: 30 * 86400000, year: 365 * 86400000 };
++/**
++ * Does the pinned search.fabric-result@1 already allow the semantic-ranking additions? The fields are
++ * optional and additive (a contract diff is in the task's answer), but until Contracts ships them the
++ * answer must stay valid: with the current schema they are left out, not sent and rejected.
++ */
++const FABRIC_RANK_FIELDS = (() => {
++ try {
++ const schema = contracts.schema('search.fabric-result@1');
++ const explain = schema && schema.properties && schema.properties.route
++ && schema.properties.route.properties && schema.properties.route.properties.explain;
++ const item = explain && explain.items && explain.items.properties;
++ const result = schema && schema.properties && schema.properties.results
++ && schema.properties.results.items && schema.properties.results.items.properties;
++ return {
++ explainRanking: Boolean(item && item.ranking),
++ explainReason: Boolean(item && item.reason),
++ resultRanks: Boolean(result && result.ranks),
++ };
++ } catch {
++ return { explainRanking: false, explainReason: false, resultRanks: false };
++ }
++})();
++
++function withRanking(row, page) {
++ const explain = page && page.explain;
++ if (!explain) return row;
++ if (FABRIC_RANK_FIELDS.explainRanking && explain.ranking) row.ranking = explain.ranking;
++ if (FABRIC_RANK_FIELDS.explainReason && explain.reason) row.reason = explain.reason;
++ return row;
++}
++
/**
* The vertical scopes and the document kinds Search actually indexes for them. A scope with no entry
* (images and people today) has nothing to narrow yet and says so instead of guessing. Several
@@@ -79,7 +110,7 @@ function documentResult(doc, scope) {
const snippet = doc.snippet_html ? plainText(doc.snippet_html, 600)
: doc.summary ? plainText(doc.summary, 600) : null;
const site = hostOf(url);
-- return {
++ const out = {
scope,
kind: 'document',
title: clip(doc.title || url, 300),
@@@ -95,6 +126,8 @@ function documentResult(doc, scope) {
id: doc.id ? clip(doc.id, 200) : null,
score: null,
};
++ if (FABRIC_RANK_FIELDS.resultRanks && doc.ranks) out.ranks = doc.ranks;
++ return out;
}
/** An open-index document as a web page: it is in Search's index, but it cites like any web source. */
@@@ -174,7 +207,7 @@ function createFabric({ config, searcher, cache, web, db, now = () => Date.now()
return {
results,
next_cursor: page.next_cursor || null,
-- explain: [{ scope: 'network', step: 'index', ok: true, count: results.length, ms: msSince(started, now) }],
++ explain: [withRanking({ scope: 'network', step: 'index', ok: true, count: results.length, ms: msSince(started, now) }, page)],
};
})();
}
@@@ -200,7 +233,7 @@ function createFabric({ config, searcher, cache, web, db, now = () => Date.now()
return {
results,
next_cursor: page.next_cursor || null,
-- explain: [{ scope: 'mine', step: 'index', ok: true, count: results.length, ms: msSince(started, now) }],
++ explain: [withRanking({ scope: 'mine', step: 'index', ok: true, count: results.length, ms: msSince(started, now) }, page)],
};
})();
}
@@@ -233,7 +266,7 @@ function createFabric({ config, searcher, cache, web, db, now = () => Date.now()
}
return {
results, next_cursor: null,
-- explain: [{ scope, step: 'index', ok: true, count: results.length, ms: msSince(started, now) }],
++ explain: [withRanking({ scope, step: 'index', ok: true, count: results.length, ms: msSince(started, now) }, pages[0])],
};
})();
}
ddiff --git a/server/fabric/rrf.js b/server/fabric/rrf.js
nnew file mode 100644
iindex 0000000..ac84649
---- /dev/null
++++ b/server/fabric/rrf.js
@@@ -0,0 +1,65 @@
++'use strict';
++/**
++ * Reciprocal rank fusion (RRF, k = 60) for the Search Fabric's network scope. Two ranked lists — the
++ * keyword top 100 and the vector top 100 — are merged by
++ *
++ * score(d) = 1 / (k + keywordRank(d)) + 1 / (k + vectorRank(d))
++ *
++ * where a rank is 1-based and a document absent from a list contributes nothing. k = 60 is the
++ * standard RRF constant: it flattens the head enough that a document near the top of both lists beats
++ * one that is first in only one of them, without letting a single list's first place dominate.
++ *
++ * Ties are broken by the keyword rank, then the vector rank, then rid, so the order is stable for the
++ * same two lists. `fuseRrf` is pure; the query path calls it with the engine's hits and the vector
++ * store's nearest hits, both already ordered best first.
++ */
++const RRF_K = 60;
++
++function rankOf(hit) {
++ if (hit && typeof hit === 'object') return hit.rid;
++ return hit;
++}
++
++function fuseRrf(keywordHits = [], vectorHits = [], { k = RRF_K, limit = null } = {}) {
++ const byRid = new Map();
++
++ function entry(rid) {
++ let e = byRid.get(rid);
++ if (!e) {
++ e = { rid, score: 0, ranks: { keyword: null, vector: null } };
++ byRid.set(rid, e);
++ }
++ return e;
++ }
++
++ keywordHits.forEach((hit, i) => {
++ const rid = rankOf(hit);
++ if (rid == null) return;
++ const e = entry(rid);
++ e.ranks.keyword = i + 1;
++ e.score += 1 / (k + i + 1);
++ });
++ vectorHits.forEach((hit, i) => {
++ const rid = rankOf(hit);
++ if (rid == null) return;
++ const e = entry(rid);
++ e.ranks.vector = i + 1;
++ e.score += 1 / (k + i + 1);
++ });
++
++ const order = [...byRid.values()].sort((a, b) => {
++ if (b.score !== a.score) return b.score - a.score;
++ const ka = a.ranks.keyword == null ? Infinity : a.ranks.keyword;
++ const kb = b.ranks.keyword == null ? Infinity : b.ranks.keyword;
++ if (ka !== kb) return ka - kb;
++ const va = a.ranks.vector == null ? Infinity : a.ranks.vector;
++ const vb = b.ranks.vector == null ? Infinity : b.ranks.vector;
++ if (va !== vb) return va - vb;
++ const na = Number(a.rid), nb = Number(b.rid);
++ if (Number.isFinite(na) && Number.isFinite(nb)) return na - nb;
++ return String(a.rid).localeCompare(String(b.rid));
++ });
++ return limit == null ? order : order.slice(0, limit);
++}
++
++module.exports = { fuseRrf, RRF_K };
ddiff --git a/server/fabric/vector-store.js b/server/fabric/vector-store.js
nnew file mode 100644
iindex 0000000..994fa88
---- /dev/null
++++ b/server/fabric/vector-store.js
@@@ -0,0 +1,101 @@
++'use strict';
++/**
++ * The vector side of the Search Fabric's network scope (SF-2): a small interface the PostgreSQL/pgvector
++ * implementation satisfies and the tests replace with a fake, because PGlite has no `vector` type.
++ *
++ * available() the extension and search_embeddings table are both there
++ * hasEmbeddings() at least one vector for the configured model exists
++ * existing() Map(identity -> { model, dim, content_hash }) for the embedder
++ * put(row) store/replace one document's vector (owner, type, key is the identity)
++ * nearest({ vector, filters, limit }) the top hits by cosine distance, public listed only
++ *
++ * Rows are keyed by the documents table's identity (owner, type, key=id), so one document has at most one
++ * vector: switching models replaces it instead of mixing two vector spaces. `nearest` always joins
++ * documents and keeps only public listed, non-deleted rows, so a document that stops being public can
++ * never come back through a stale vector. Filters are the engine's own SQL predicate (owner, type, lang,
++ * facets, since) — visibility is decided here, not by the caller.
++ */
++const { EXPOSURE } = require('../document');
++const { filterClause } = require('../engine/pg');
++
++const IDENTITY_SEP = '\u0000';
++
++function identity(owner, type, id) {
++ return `${owner}${IDENTITY_SEP}${type}${IDENTITY_SEP}${id}`;
++}
++
++function createPgVectorStore({ db, model = 'bge-small-en-v1.5', dim = 384 }) {
++ let availability = null;
++
++ async function available() {
++ if (availability !== null) return availability;
++ try {
++ const ext = await db.prepare("SELECT 1 AS ok FROM pg_extension WHERE extname = 'vector'").get();
++ if (!ext) return (availability = false);
++ const table = await db.prepare("SELECT to_regclass('search_embeddings') AS name").get();
++ return (availability = Boolean(table && table.name));
++ } catch {
++ // A database without pg_extension (or a proxy that refuses the query) has no vectors.
++ return (availability = false);
++ }
++ }
++
++ async function hasEmbeddings() {
++ if (!await available()) return false;
++ const row = await db.prepare('SELECT 1 AS ok FROM search_embeddings WHERE model = @model LIMIT 1').get({ model });
++ return Boolean(row);
++ }
++
++ /** One round trip for the embedder: every stored vector's identity and content hash. */
++ async function existing() {
++ const out = new Map();
++ if (!await available()) return out;
++ const rows = await db.prepare('SELECT owner, type, key, model, dim, content_hash FROM search_embeddings').all();
++ for (const row of rows) {
++ out.set(identity(row.owner, row.type, row.key), {
++ model: row.model,
++ dim: Number(row.dim),
++ content_hash: row.content_hash,
++ });
++ }
++ return out;
++ }
++
++ async function put({ owner, type, key, embedding, contentHash, updatedAt = new Date().toISOString() }) {
++ if (!Array.isArray(embedding) || embedding.length !== dim) throw new Error(`embedding must have ${dim} dimensions`);
++ await db.prepare(`INSERT INTO search_embeddings (owner, type, key, model, dim, embedding, content_hash, updated_at)
++ VALUES (@owner, @type, @key, @model, @dim, @embedding::vector, @content_hash, @updated_at)
++ ON CONFLICT (owner, type, key) DO UPDATE SET
++ model = EXCLUDED.model, dim = EXCLUDED.dim, embedding = EXCLUDED.embedding,
++ content_hash = EXCLUDED.content_hash, updated_at = EXCLUDED.updated_at`).run({
++ owner, type, key,
++ model,
++ dim: embedding.length,
++ embedding: JSON.stringify(embedding),
++ content_hash: contentHash,
++ updated_at: updatedAt,
++ });
++ }
++
++ /** Top hits by cosine distance among the public listed documents the filters allow. */
++ async function nearest({ vector, filters = {}, limit = 100 }) {
++ if (!Array.isArray(vector) || vector.length !== dim) return [];
++ const params = { qvec: JSON.stringify(vector), r_model: model, r_dim: dim, limit };
++ const where = filterClause(filters, params);
++ const rows = await db.prepare(`SELECT d.rid AS rid, (e.embedding <=> @qvec::vector) AS distance
++ FROM search_embeddings e
++ JOIN documents d ON d.owner = e.owner AND d.type = e.type AND d.id = e.key
++ WHERE e.model = @r_model AND e.dim = @r_dim
++ AND d.deleted = 0 AND d.exposure = ${EXPOSURE.public_listed}${where}
++ ORDER BY e.embedding <=> @qvec::vector, d.rid
++ LIMIT @limit`).all(params);
++ return rows.map((row) => ({ rid: Number(row.rid), distance: Number(row.distance) }));
++ }
++
++ return {
++ name: 'pgvector', model, dim,
++ available, hasEmbeddings, existing, put, nearest,
++ };
++}
++
++module.exports = { createPgVectorStore, identity, IDENTITY_SEP };
ddiff --git a/server/index.js b/server/index.js
iindex a3dc589..4c489af 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -23,8 +23,11 @@ const { createSearcher } = require('./api/query');
const { createNetworkPush } = require('./network-push');
const { createSavedNotifier } = require('./saved-notifier');
const { createApp } = require('./app');
++const { createPgVectorStore } = require('./fabric/vector-store');
++const { createEmbeddingClient } = require('./fabric/embed');
++const { createEmbedder } = require('./fabric/embedder');
--async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, networkPush = null, log = console, listen = true } = {}) {
++async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, networkPush = null, log = console, listen = true, vectorStore: givenVectorStore = null, embeddingClient: givenEmbeddings = null, embedder: givenEmbedder = null } = {}) {
config = config || load();
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
const db = givenDb || await openDb(config, { log });
@@@ -48,13 +51,18 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const jwksUrl = `${config.networkInternalUrl}/api/.well-known/jwks`;
const jwks = jwksClient(jwksUrl, { log, fetch: fetchImpl });
const auth = createAuth({ config, jwks, log });
-- const searcher = createSearcher({ config, store, engine, now });
++ // Semantic ranking (SF-2): the pgvector store (inert on PGlite: available() is false), the AI embedding
++ // client, and the job that keeps one embedding per public document. Tests inject a fake vector store.
++ const vectorStore = givenVectorStore || createPgVectorStore({ db, model: config.embed.model, dim: config.embed.dim });
++ const embeddingClient = givenEmbeddings || createEmbeddingClient({ config, fetchImpl, now, log });
++ const embedder = givenEmbedder || createEmbedder({ db, vectors: vectorStore, embeddings: embeddingClient, config, now, log });
++ const searcher = createSearcher({ config, store, engine, vectorStore, embeddings: embeddingClient, now });
// Saved-search notifications through Network: off unless SEARCH_SAVED_NOTIFY=1 with the `search` client's secret.
const notifier = createSavedNotifier({
config, saved, searcher, now, log,
network: networkPush || createNetworkPush({ config, fetchImpl }),
});
-- const app = createApp({ config, db, store, engine, auth, outbox, purges, purger, saved, searcher, notifier, fetchImpl, log, now });
++ const app = createApp({ config, db, store, engine, auth, outbox, purges, purger, saved, searcher, notifier, fetchImpl, log, now, vectorStore, embeddings: embeddingClient });
// The background refresher is not started under the test runtime (the tests stub the fetch and load
// the keys once below); a real boot refreshes on an unref'd timer. keyLoaded settles the first load.
@@@ -63,6 +71,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
outbox.start();
purger.start();
notifier.start();
++ embedder.start();
const pruneTimer = setInterval(async () => { try { await outbox.outbox.prune(); } catch (err) { log.error(`[outbox] prune: ${err.message}`); } }, 6 * 3600 * 1000);
pruneTimer.unref?.();
// Expired Search Fabric answers are cheap to keep but pointless to serve; drop them hourly.
@@@ -86,6 +95,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
await outbox.stop();
await purger.stop();
await notifier.stop();
++ await embedder.stop();
if (server) {
server.closeAllConnections?.();
await new Promise(resolve => server.close(() => resolve()));
@@@ -95,7 +105,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
return {
config, db, engine, store, outbox, purges, purger, saved, searcher, notifier, jwks, keyLoaded, auth, app, server, close,
-- queryCache, fabric: app.locals.fabric,
++ queryCache, vectorStore, embeddingClient, embedder, fabric: app.locals.fabric,
};
}
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 1a10c8a..c6159f4 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -43,7 +43,7 @@ function userToken({ subjectId = ids.newId('user'), role = 'user', aud = ['openv
}, key);
}
--async function boot({ env = {}, fetchImpl, jwksFetch, networkPush, now } = {}) {
++async function boot({ env = {}, fetchImpl, jwksFetch, networkPush, now, vectorStore, embeddingClient, embedder } = {}) {
const config = load({
NODE_ENV: 'test',
PORT: '0',
@@@ -55,7 +55,7 @@ async function boot({ env = {}, fetchImpl, jwksFetch, networkPush, now } = {}) {
const inner = fetchImpl || globalThis.fetch;
const jwks = jwksFetch || stubJwks;
const fetchImplBoth = async (url, opts) => String(url).includes('/api/.well-known/jwks') ? jwks(url, opts) : inner(url, opts);
-- const h = await start({ config, db: testdb.db, log: silent, fetchImpl: fetchImplBoth, networkPush, now });
++ const h = await start({ config, db: testdb.db, log: silent, fetchImpl: fetchImplBoth, networkPush, now, vectorStore, embeddingClient, embedder });
const base = `http://127.0.0.1:${h.server.address().port}`;
return { ...h, base, async stop() { await h.close(); await testdb.close(); } };
}
ddiff --git a/test/hybrid.test.js b/test/hybrid.test.js
nnew file mode 100644
iindex 0000000..3cc4261
---- /dev/null
++++ b/test/hybrid.test.js
@@@ -0,0 +1,265 @@
++'use strict';
++/**
++ * Semantic ranking (SF-2): the pgvector migration's no-op on PGlite, reciprocal rank fusion (k = 60), the
++ * embedding job's batching/skip/backoff rules, and the query's hybrid/fallback answers. PGlite has no
++ * `vector` type, so the vector store is a fake with the same interface as createPgVectorStore.
++ */
++const assert = require('assert');
++const { boot, request, doc, suite } = require('./helpers');
++const { testDb } = require('./db');
++const { fuseRrf } = require('../server/fabric/rrf');
++const { embeddingText } = require('../server/fabric/embedder');
++const { createQueryEmbeddingCache } = require('../server/fabric/embed');
++const { identity } = require('../server/fabric/vector-store');
++
++const t = suite('hybrid');
++
++class FakeVectors {
++ constructor({ available = true, has = true, model = 'bge-small-en-v1.5', dim = 384 } = {}) {
++ this._available = available;
++ this._has = has;
++ this.model = model;
++ this.dim = dim;
++ this.embedded = new Map();
++ this.puts = [];
++ this.hits = [];
++ }
++ async available() { return this._available; }
++ async hasEmbeddings() { return this._has; }
++ async existing() { return new Map(this.embedded); }
++ async put(row) {
++ this.puts.push(row);
++ this.embedded.set(identity(row.owner, row.type, row.key), { model: this.model, dim: this.dim, content_hash: row.contentHash });
++ }
++ async nearest({ limit = 100 }) { return this.hits.slice(0, limit); }
++}
++
++function fakeEmbeddings({ fail = false, dim = 384 } = {}) {
++ const client = {
++ calls: [],
++ fail,
++ model: 'bge-small-en-v1.5',
++ available: () => ({ configured: true, reason: null }),
++ async embed(texts) {
++ client.calls.push(texts);
++ if (client.fail) return { ok: false, error: 'AI down', ms: 1 };
++ return {
++ ok: true,
++ vectors: texts.map((_text, i) => Array.from({ length: dim }, (_, j) => ((i + j) % 17) / 17 + 0.01)),
++ dimensions: dim,
++ model: client.model,
++ ms: 1,
++ };
++ },
++ };
++ return client;
++}
++
++async function ridMap(db) {
++ const rows = await db.prepare('SELECT rid, id FROM documents').all();
++ return Object.fromEntries(rows.map((r) => [r.id, Number(r.rid)]));
++}
++
++t('the embeddings migration succeeds on PGlite and creates nothing', async () => {
++ const tdb = await testDb();
++ try {
++ const ext = await tdb.db.prepare("SELECT 1 AS ok FROM pg_extension WHERE extname = 'vector'").get();
++ assert.ok(!ext, 'PGlite has no vector extension');
++ const table = await tdb.db.prepare("SELECT to_regclass('search_embeddings') AS name").get();
++ assert.strictEqual(table && table.name, null, 'no table without the extension');
++ } finally {
++ await tdb.close();
++ }
++});
++
++t('RRF k=60: a document near the top of both lists wins, ties break by keyword rank', () => {
++ const fused = fuseRrf(
++ [{ rid: 1 }, { rid: 2 }, { rid: 3 }, { rid: 4 }],
++ [{ rid: 3 }, { rid: 4 }, { rid: 1 }, { rid: 5 }],
++ );
++ assert.deepStrictEqual(fused.map((h) => h.rid), [1, 3, 4, 2, 5]);
++ assert.deepStrictEqual(fused.find((h) => h.rid === 1).ranks, { keyword: 1, vector: 3 });
++ assert.deepStrictEqual(fused.find((h) => h.rid === 2).ranks, { keyword: 2, vector: null });
++ assert.deepStrictEqual(fused.find((h) => h.rid === 5).ranks, { keyword: null, vector: 4 });
++ assert.ok(fused[0].score > fused[2].score, 'the shared head beats a document in only one list');
++});
++
++t('the query embedding cache normalizes text, keeps 10 minutes and evicts the LRU past 500', async () => {
++ const clock = { t: 1000 };
++ let calls = 0;
++ const client = {
++ async embed(texts) {
++ calls++;
++ return { ok: true, vectors: texts.map(() => Array.from({ length: 384 }, () => calls)), dimensions: 384, model: 'bge-small-en-v1.5' };
++ },
++ };
++ const cache = createQueryEmbeddingCache({ client, now: () => clock.t, ttlMs: 100, max: 2 });
++ const first = await cache.get('Hello World');
++ const second = await cache.get('hello world');
++ assert.strictEqual(calls, 1, 'the normalized text is one cache entry');
++ assert.strictEqual(second.cached, true);
++ assert.deepStrictEqual(first.vector, second.vector);
++ clock.t += 101;
++ await cache.get('hello world');
++ assert.strictEqual(calls, 2, 'an expired entry is embedded again');
++ await cache.get('third text');
++ assert.strictEqual(cache.size(), 2, 'the LRU bound holds');
++});
++
++t('the embedder batches 32, skips unchanged and non-public documents, and re-embeds changed ones', async () => {
++ const vectors = new FakeVectors();
++ const embeddings = fakeEmbeddings();
++ const svc = await boot({ env: { SEARCH_EMBED: '1', SEARCH_EMBED_PER_MIN: '120' }, vectorStore: vectors, embeddingClient: embeddings });
++ try {
++ for (let i = 0; i < 40; i++) await svc.store.apply(doc({ id: `pub-${i}`, title: `Public ${i}`, summary: `Summary ${i}`, body: `Body ${i}` }));
++ await svc.store.apply(doc({ id: 'private', visibility: 'private', acl: { subjects: ['usr_test'] }, title: 'Private', body: 'secret' }));
++ await svc.store.apply(doc({ id: 'draft', visibility: 'draft', title: 'Draft', body: 'draft' }));
++ await svc.store.apply(doc({ id: 'unpublished', publication_state: 'draft', title: 'Unpublished', body: 'unpublished' }));
++
++ const first = await svc.embedder.tick();
++ assert.strictEqual(first.embedded, 40);
++ assert.deepStrictEqual(embeddings.calls.map((b) => b.length), [32, 8], 'batches of at most 32');
++ assert.strictEqual(vectors.puts.length, 40);
++ assert.ok(vectors.puts.every((p) => !['private', 'draft', 'unpublished'].includes(p.key)), 'never a non-public document');
++ assert.strictEqual(embeddingText(doc({ title: 't', summary: 's', body: 'x'.repeat(3000) })).length, 2000);
++
++ embeddings.calls = [];
++ const second = await svc.embedder.tick();
++ assert.strictEqual(second.embedded, 0, 'unchanged documents are skipped');
++ assert.strictEqual(embeddings.calls.length, 0, 'and cost no AI call');
++
++ await svc.store.apply(doc({ id: 'pub-0', revision: 2, title: 'Changed 0', summary: 'Summary 0', body: 'Body 0' }));
++ const third = await svc.embedder.tick();
++ assert.strictEqual(third.embedded, 1, 'the changed content hash is re-embedded');
++ assert.strictEqual(embeddings.calls.length, 1);
++ assert.strictEqual(embeddings.calls[0].length, 1);
++ } finally {
++ await svc.stop();
++ }
++});
++
++t('the embedder respects SEARCH_EMBED_PER_MIN per rolling minute', async () => {
++ const clock = { t: 1800000000000 };
++ const vectors = new FakeVectors();
++ const embeddings = fakeEmbeddings();
++ const svc = await boot({ env: { SEARCH_EMBED_PER_MIN: '33' }, now: () => clock.t, vectorStore: vectors, embeddingClient: embeddings });
++ try {
++ for (let i = 0; i < 40; i++) await svc.store.apply(doc({ id: `rate-${i}`, title: `Rate ${i}` }));
++ const first = await svc.embedder.tick();
++ assert.strictEqual(first.embedded, 33, 'at most 33 documents in the first minute');
++ assert.strictEqual(first.pending, 7);
++ assert.deepStrictEqual(embeddings.calls.map((b) => b.length), [32, 1]);
++
++ const second = await svc.embedder.tick();
++ assert.strictEqual(second.embedded, 0, 'the minute budget is spent');
++ assert.strictEqual(second.pending, 7);
++ assert.strictEqual(embeddings.calls.length, 2);
++
++ clock.t += 60000;
++ const third = await svc.embedder.tick();
++ assert.strictEqual(third.embedded, 7, 'the next rolling minute continues');
++ assert.strictEqual(third.pending, 0);
++ } finally {
++ await svc.stop();
++ }
++});
++
++t('the embedder backs off after a failed AI call and retries after the delay', async () => {
++ const clock = { t: 1790000000000 };
++ const vectors = new FakeVectors();
++ const embeddings = fakeEmbeddings({ fail: true });
++ const svc = await boot({ env: { SEARCH_EMBED_PER_MIN: '10' }, now: () => clock.t, vectorStore: vectors, embeddingClient: embeddings });
++ try {
++ await svc.store.apply(doc({ id: 'retry-me', title: 'Retry me' }));
++ const first = await svc.embedder.tick();
++ assert.strictEqual(first.failed, 1);
++ assert.strictEqual(vectors.puts.length, 0, 'a failed call stores nothing');
++ assert.strictEqual(embeddings.calls.length, 1);
++
++ const during = await svc.embedder.tick();
++ assert.strictEqual(during.backoff, true, 'the second tick is inside the backoff window');
++ assert.strictEqual(embeddings.calls.length, 1, 'no AI call during backoff');
++
++ clock.t += 2500;
++ embeddings.fail = false;
++ const after = await svc.embedder.tick();
++ assert.strictEqual(after.embedded, 1, 'the document is embedded after the backoff');
++ assert.strictEqual(vectors.puts.length, 1);
++ } finally {
++ await svc.stop();
++ }
++});
++
++t('query falls back to keyword with no_vectors, short_query and ai_unavailable', async () => {
++ const noVectors = await boot({ embeddingClient: fakeEmbeddings() });
++ try {
++ await noVectors.store.apply(doc({ id: 'plain', title: 'Hello plain' }));
++ const r = await request(noVectors.base, 'GET', '/api/v1/search?q=hello');
++ assert.strictEqual(r.status, 200, r.text);
++ assert.deepStrictEqual(r.body.explain, { ranking: 'keyword', reason: 'no_vectors' });
++ assert.deepStrictEqual(r.body.results.map((x) => x.id), ['plain']);
++ } finally {
++ await noVectors.stop();
++ }
++
++ const shortVectors = new FakeVectors();
++ const shortEmbeddings = fakeEmbeddings();
++ const short = await boot({ vectorStore: shortVectors, embeddingClient: shortEmbeddings });
++ try {
++ await short.store.apply(doc({ id: 'a-doc', title: 'A' }));
++ const r = await request(short.base, 'GET', '/api/v1/search?q=a');
++ assert.deepStrictEqual(r.body.explain, { ranking: 'keyword', reason: 'short_query' });
++ assert.strictEqual(shortEmbeddings.calls.length, 0, 'a one-character query is never embedded');
++ } finally {
++ await short.stop();
++ }
++
++ const aiDownVectors = new FakeVectors();
++ const aiDown = await boot({ vectorStore: aiDownVectors, embeddingClient: fakeEmbeddings({ fail: true }) });
++ try {
++ await aiDown.store.apply(doc({ id: 'hello-doc', title: 'Hello there' }));
++ const r = await request(aiDown.base, 'GET', '/api/v1/search?q=hello');
++ assert.deepStrictEqual(r.body.explain, { ranking: 'keyword', reason: 'ai_unavailable' });
++ assert.deepStrictEqual(r.body.results.map((x) => x.id), ['hello-doc']);
++ assert.deepStrictEqual(r.body.results[0].ranks, { keyword: 1, vector: null });
++ } finally {
++ await aiDown.stop();
++ }
++});
++
++t('hybrid: keyword and vector top-100 are fused, ranks are per hit, and a hidden vector hit is dropped', async () => {
++ const vectors = new FakeVectors();
++ const embeddings = fakeEmbeddings();
++ const svc = await boot({ vectorStore: vectors, embeddingClient: embeddings });
++ try {
++ await svc.store.apply(doc({ id: 'both', title: 'Alpha both', body: 'alpha shared' }));
++ await svc.store.apply(doc({ id: 'keyword-only', title: 'Alpha keyword', body: 'alpha only' }));
++ await svc.store.apply(doc({ id: 'vector-only', title: 'Gamma vector', body: 'gamma only' }));
++ await svc.store.apply(doc({ id: 'private-hit', visibility: 'private', acl: { subjects: ['usr_test'] }, title: 'Alpha secret', body: 'alpha secret' }));
++ const rid = await ridMap(svc.db);
++ vectors.hits = [
++ { rid: rid.both, distance: 0.1 },
++ { rid: rid['private-hit'], distance: 0.2 }, // even a stale/incorrect vector hit must not leak
++ { rid: rid['vector-only'], distance: 0.3 },
++ ];
++
++ const r = await request(svc.base, 'GET', '/api/v1/search?q=alpha&limit=10');
++ assert.strictEqual(r.status, 200, r.text);
++ assert.deepStrictEqual(r.body.explain, { ranking: 'hybrid' });
++ assert.deepStrictEqual(r.body.results.map((x) => x.id), ['both', 'keyword-only', 'vector-only']);
++ assert.deepStrictEqual(r.body.results.find((x) => x.id === 'both').ranks, { keyword: 1, vector: 1 });
++ assert.deepStrictEqual(r.body.results.find((x) => x.id === 'keyword-only').ranks, { keyword: 2, vector: null });
++ assert.deepStrictEqual(r.body.results.find((x) => x.id === 'vector-only').ranks, { keyword: null, vector: 3 });
++ assert.ok(!JSON.stringify(r.body).includes('private-hit'), 'visibility still decides, not the vector list');
++
++ // Paging walks the fused list without duplicates.
++ const p1 = await request(svc.base, 'GET', '/api/v1/search?q=alpha&limit=2');
++ const p2 = await request(svc.base, 'GET', `/api/v1/search?q=alpha&limit=2&cursor=${encodeURIComponent(p1.body.next_cursor)}`);
++ assert.deepStrictEqual([...p1.body.results, ...p2.body.results].map((x) => x.id), ['both', 'keyword-only', 'vector-only']);
++ assert.strictEqual(p2.body.next_cursor, null);
++ } finally {
++ await svc.stop();
++ }
++});
++
++t.run();