client.ts · Source

client.tspackages/search/src/client.tsraw ← src21,291 B← Back to the app
1/**
2 * @factory/search — thin OpenSearch client over an injectable fetch
3 * (mirrors the `@factory/storage` S3 client pattern: pure transport, no SDK).
4 *
5 * The client knows the OpenSearch HTTP API only; query shapes come from
6 * `queries.ts`, fusion from `rrf.ts`. Failures surface as `ApiError` with
7 * `SEARCH_UNAVAILABLE` (503) for transport/429/5xx errors.
8 *
9 * Hybrid search (search-architecture §4.5) is executed as TWO requests —
10 * BM25 (`_search` with the full-text body) and kNN (`_search` with a kNN-only
11 * body) — and the two rank lists are fused client-side with RRF (k = 60).
12 * Recorded decision: separate requests over the combined `query + knn` body,
13 * because the combined body's `_score` is BM25-dominated and the kNN ranking
14 * is not recoverable from it; two requests also make the two RRF inputs exact.
15 */
16 
17// NOTE: the API consumes this package under `node10` module resolution,
18// which only sees the package ROOT (no subpaths) — hence `@factory/shared`
19// (root re-export), not `@factory/shared/errors`. See the Node10 note in
20// packages/shared/src/index.ts.
21import { ApiError, ErrorCodes } from "@factory/shared";
22 
23import {
24  CONTENT_INDEX_ALIAS,
25  buildFullTextBody,
26  buildHybridKnnOnlyBody,
27  buildHybridBody,
28  buildPrefixBody,
29  contentIndex,
30  dateSortField,
31  decodeCursor,
32  decodeDateCursor,
33  encodeCursor,
34  encodeDateCursor,
35  withDateSort,
36  withSearchAfter,
37} from "./queries.js";
38import { fuseHybrid } from "./rrf.js";
39import { CHUNK_DOC_TYPE, displayType } from "./types.js";
40import type {
41  AnswerCitation,
42  SearchDoc,
43  SearchQuery,
44  SearchResult,
45  SearchResponse,
46  Suggestion,
47  ViewerContext,
48} from "./types.js";
49 
50export interface SearchClientOptions {
51  /** Base URL, e.g. `http://opensearch:9200`. */
52  baseUrl: string;
53  /** Injectable fetch (defaults to global fetch) — enables offline tests. */
54  fetchImpl?: typeof fetch;
55  /** Basic auth (dev); production uses IAM/signing at the network edge. */
56  auth?: { username: string; password: string };
57  /** Request timeout in ms (default 5000). */
58  timeoutMs?: number;
59  now?: () => number;
60  /** kNN list depth per hybrid pass (default 50, per search §4.5 `k: 50`). */
61  knnK?: number;
62}
63 
64interface OsHit {
65  _id: string;
66  _score?: number;
67  _source: SearchDoc & { highlight?: { title?: string[]; body?: string[] } };
68}
69 
70interface OsResponse {
71  hits?: {
72    total?: { value?: number };
73    hits?: OsHit[];
74  };
75}
76 
77export class SearchClient {
78  readonly baseUrl: string;
79  readonly fetchImpl: typeof fetch;
80  private readonly headers: Record<string, string>;
81  private readonly timeoutMs: number;
82  private readonly now: () => number;
83  private readonly knnK: number;
84 
85  constructor(options: SearchClientOptions) {
86    this.baseUrl = options.baseUrl.replace(/\/+$/, "");
87    this.fetchImpl = options.fetchImpl ?? fetch;
88    this.headers = { "content-type": "application/json" };
89    if (options.auth) {
90      this.headers.authorization =
91        "Basic " + Buffer.from(`${options.auth.username}:${options.auth.password}`).toString("base64");
92    }
93    this.timeoutMs = options.timeoutMs ?? 5000;
94    this.now = options.now ?? Date.now;
95    this.knnK = options.knnK ?? 50;
96  }
97 
98  /** Index one (or many chunked) documents via `_bulk` (single content index). */
99  async indexDocs(docs: SearchDoc[]): Promise<{ indexed: number }> {
100    if (docs.length === 0) return { indexed: 0 };
101    const lines: string[] = [];
102    for (const doc of docs) {
103      lines.push(JSON.stringify({ index: { _index: contentIndex(), _id: doc.docId } }));
104      lines.push(JSON.stringify(doc));
105    }
106    const res = await this.request("/_bulk", { method: "POST", body: lines.join("\n") + "\n" });
107    const parsed = (await res.json()) as { errors?: boolean; items?: { index?: { status?: number } }[] };
108    let indexed = 0;
109    for (const item of parsed.items ?? []) {
110      if (item.index && (item.index.status ?? 200) < 300) indexed++;
111    }
112    return { indexed };
113  }
114 
115  /** Delete one document by doc id (single content index). */
116  async deleteDoc(docId: string): Promise<void> {
117    await this.request(`/${contentIndex()}/_doc/${encodeURIComponent(docId)}`, { method: "DELETE" });
118  }
119 
120  /**
121   * Delete ALL chunk docs belonging to the given parent doc id(s) (TASK-0099 §3).
122   * Chunk docs have no `search_documents` mirror rows, so the worker cannot
123   * enumerate them by id (a shrunk body changes the count) — it deletes them in
124   * a parent-driven pass instead, matching on `parentDocId` AND
125   * `type: "chunk"` (both keyword fields in the mapping). One request handles
126   * any number of parents. A missing parent (404) is a no-op, like
127   * {@link deleteDoc}; 0 deletes is fine (idempotent).
128   */
129  async deleteChunksOf(parentDocIds: string | string[]): Promise<number> {
130    const ids = Array.isArray(parentDocIds) ? parentDocIds : [parentDocIds];
131    if (ids.length === 0) return 0;
132    const parentClause =
133      ids.length === 1
134        ? { term: { parentDocId: ids[0]! } }
135        : { terms: { parentDocId: ids } };
136    const body = JSON.stringify({
137      query: {
138        bool: {
139          filter: [parentClause, { term: { type: CHUNK_DOC_TYPE } }],
140        },
141      },
142    });
143    const res = await this.request(`/${contentIndex()}/_delete_by_query?conflicts=proceed`, {
144      method: "POST",
145      body,
146    });
147    const parsed = (await res.json()) as { deleted?: number };
148    return parsed.deleted ?? 0;
149  }
150 
151  /**
152   * Partial merge-update of one document by doc id (e.g. the AI `vector`
153   * field on a chunk doc — a full PUT would clobber title/body/grants).
154   */
155  async updateDoc(docId: string, doc: Record<string, unknown>): Promise<void> {
156    await this.request(`/${contentIndex()}/_update/${encodeURIComponent(docId)}`, {
157      method: "POST",
158      body: JSON.stringify({ doc }),
159    });
160  }
161 
162  /** True when the content index/alias exists. */
163  async indexExists(indexOrAlias: string = contentIndex()): Promise<boolean> {
164    try {
165      await this.request(`/${indexOrAlias}`, {});
166      return true;
167    } catch {
168      return false;
169    }
170  }
171 
172  /**
173   * T-177 (search health) — the document count of the content index,
174   * optionally scoped to a set of tenant workspaces. Counts ONLY whole docs
175   * (`type != "chunk"` — chunk docs are the RAG expansion of a parent doc and
176   * would double-count content). `size: 0` (no hits returned). A missing
177   * index resolves to 0 (the alias is created lazily on first write — 0 is
178   * the honest "nothing indexed yet", distinct from `indexExists` which is a
179   * separate probe). A 5xx/network failure throws `SEARCH_UNAVAILABLE` (via
180   * {@link request}) — the caller maps that to `healthy: false`, never to a
181   * guessed count.
182   */
183  async countDocs(workspaceIds?: string[]): Promise<number> {
184    const filter: Record<string, unknown>[] = [{
185      bool: { must_not: [{ term: { type: CHUNK_DOC_TYPE } }] },
186    }];
187    if (workspaceIds && workspaceIds.length > 0) {
188      filter.push({ terms: { workspaceId: workspaceIds } });
189    }
190    const out = await this.osRequest<OsResponse>(
191      `/${contentIndex()}/_search`,
192      { query: { bool: { filter } }, size: 0 }
193    );
194    return out.hits?.total?.value ?? 0;
195  }
196 
197  /**
198   * Create the content index (behind the alias) if missing (idempotent startup
199   * path). `mappings` is the full `{settings, mappings}` body from
200   * `contentMapping()`.
201   */
202  async createIndexIfMissing(
203    indexName: string,
204    body: { settings?: Record<string, unknown>; mappings?: Record<string, unknown> },
205  ): Promise<boolean> {
206    try {
207      await this.request(`/${indexName}`, { method: "PUT", body: JSON.stringify(body) });
208      return true;
209    } catch (err) {
210      // resource_already_exists_engine_exception → already there (idempotent).
211      if (err instanceof ApiError && /resource_already_exists/.test(err.message)) return false;
212      throw err;
213    }
214  }
215 
216  /**
217   * Create the content index AND point the alias at it (used by the
218   * `scripts/opensearch-reindex.ts` rollover). `actions` swap is atomic.
219   */
220  async createAlias(indexName: string, alias: string = CONTENT_INDEX_ALIAS): Promise<void> {
221    await this.request("/_aliases", {
222      method: "POST",
223      body: JSON.stringify({
224        actions: [
225          { remove: { index: indexName, alias } },
226          { add: { index: indexName, alias } },
227        ],
228      }),
229    });
230  }
231 
232  /**
233   * Whether `name` is an ALIAS (as opposed to a real index). 200 → alias,
234   * 404 → not an alias (either a real index, or nothing). Used by the
235   * self-heal path to detect the corrupted "alias name became a real index"
236   * state (OpenSearch was down at boot; the first `_bulk` auto-created a bare
237   * real index behind the alias name).
238   */
239  async isAlias(name: string = contentIndex()): Promise<boolean> {
240    try {
241      await this.request(`/_alias/${name}`, {});
242      return true;
243    } catch (err) {
244      if (err instanceof ApiError && err.status === 404) return false;
245      throw err; // 5xx / transport → treat as a real failure.
246    }
247  }
248 
249  /** Delete an index by name (the self-heal path rebuilds it with a mapping). */
250  async deleteIndex(name: string): Promise<void> {
251    await this.request(`/${name}`, { method: "DELETE" });
252  }
253 
254  /** Force a refresh (make recent writes queryable — tests / immediate lag). */
255  async refresh(indexOrAlias: string = contentIndex()): Promise<void> {
256    await this.request(`/${indexOrAlias}/_refresh`, {});
257  }
258 
259  /** The live index's `dense_vector` dimension (rollover guard, search §7.3). */
260  async getVectorDimension(indexOrAlias: string = contentIndex()): Promise<number | null> {
261    const res = await this.request(`/${indexOrAlias}/_mapping`, {});
262    const parsed = (await res.json()) as Record<string, { mappings?: { properties?: { vector?: { dimension?: number } } } }>;
263    for (const key of Object.keys(parsed)) {
264      const dim = parsed[key]?.mappings?.properties?.vector?.dimension;
265      if (typeof dim === "number") return dim;
266    }
267    return null;
268  }
269 
270  /**
271   * Which of `docIds` exist in the index (batched `_mget`, one request per 500).
272   * Returns a Set of the PRESENT ids. Used by the reconcile sweep to detect
273   * docs deleted/lost in the index (the contentHash sweep alone only catches
274   * SOURCE drift, not index loss).
275   */
276  async docExists(docIds: string[]): Promise<Set<string>> {
277    const present = new Set<string>();
278    if (docIds.length === 0) return present;
279    for (let i = 0; i < docIds.length; i += 500) {
280      const chunk = docIds.slice(i, i + 500);
281      const res = await this.request(`/${contentIndex()}/_mget`, {
282        method: "POST",
283        body: JSON.stringify({ ids: chunk }),
284      });
285      const parsed = (await res.json()) as { docs?: Array<{ found?: boolean; _id?: string }> };
286      for (const d of parsed.docs ?? []) {
287        if (d.found && d._id) present.add(d._id);
288      }
289    }
290    return present;
291  }
292 
293  /**
294   * TASK-0125 — fetch the `_source` of explicit doc ids (mget, 500-batched).
295   * The RAG query path uses it for the FULL chunk text (the search API's
296   * 160-char snippet is too thin for prompt context). Missing docs (deleted
297   * between retrieval and context assembly) are silently omitted — the
298   * caller drops their context block and keeps the answer consistent.
299   */
300  async getDocs(docIds: string[]): Promise<Array<{ docId: string; source: Record<string, unknown> }>> {
301    const out: Array<{ docId: string; source: Record<string, unknown> }> = [];
302    if (docIds.length === 0) return out;
303    for (let i = 0; i < docIds.length; i += 500) {
304      const chunk = docIds.slice(i, i + 500);
305      const res = await this.request(`/${contentIndex()}/_mget`, {
306        method: "POST",
307        body: JSON.stringify({ ids: chunk }),
308      });
309      const parsed = (await res.json()) as {
310        docs?: Array<{ found?: boolean; _id?: string; _source?: Record<string, unknown> }>;
311      };
312      for (const d of parsed.docs ?? []) {
313        if (d.found && d._id && d._source) out.push({ docId: d._id, source: d._source });
314      }
315    }
316    return out;
317  }
318 
319  /**
320   * Execute a search query. `fulltext` mode (default) is one request;
321   * `hybrid` runs BM25 + kNN and fuses with RRF. The query is tenant-scoped by
322   * the viewer's `workspaceIds` (single-index model — no index param).
323   *
324   * @param queryVector required for hybrid mode — the embedded query.
325   */
326  async search(
327    query: SearchQuery,
328    viewer: ViewerContext,
329    queryVector?: number[],
330  ): Promise<SearchResponse> {
331    const started = this.now();
332    const mode = query.mode ?? "fulltext";
333    if (mode === "hybrid" && !queryVector) {
334      throw new ApiError(ErrorCodes.AI_ERROR, { message: "hybrid search requires queryVector" });
335    }
336    if (mode === "hybrid") {
337      return this.hybridSearch(query, viewer, queryVector!, started);
338    }
339    return this.fullTextSearch(query, viewer, started);
340  }
341 
342  /**
343   * Prefix suggestions ("search as you type", search-architecture §4.2):
344   * prefix match on `title.keyword` (or the doc id) over the SAME
345   * mandatory tenant + permission filters as full-text search, no highlight,
346   * small size (default 10, cap 20). Returns minimal `{ id, type, title }`
347   * items (the source row id, not the `docId` — the UI deep-links by row id).
348   *
349   * The API's `GET /search/suggest` endpoint wraps this (TASK-0097 S3).
350   */
351  async suggest(prefix: string, viewer: ViewerContext, size = 10): Promise<Suggestion[]> {
352    const capped = Math.max(1, Math.min(size, 20));
353    const body = buildPrefixBody(prefix, viewer, capped);
354    const os = await this.osRequest<OsResponse>(`/${contentIndex()}/_search`, body);
355    const hits = os.hits?.hits ?? [];
356    const seen = new Set<string>();
357    const out: Suggestion[] = [];
358    for (const h of hits) {
359      const s = h._source;
360      // De-dupe by (type, id): chunk docs share the parent's title, so the
361      // display type (chunk → parent kind) makes a chunk hit collapse into
362      // its parent's suggestion.
363      const type = displayType(s);
364      if (seen.has(`${type}:${s.id}`)) continue;
365      seen.add(`${type}:${s.id}`);
366      out.push({ id: s.id, type, title: s.title });
367      if (out.length >= capped) break;
368    }
369    return out;
370  }
371 
372  private async fullTextSearch(
373    query: SearchQuery,
374    viewer: ViewerContext,
375    started: number,
376  ): Promise<SearchResponse> {
377    // T-155 date sorts are an ADDITIVE branch — `undefined`/`"relevance"` falls
378    // through to the original `_score` path below, byte-for-byte unchanged.
379    if (query.sortBy === "createdAt" || query.sortBy === "updatedAt") {
380      return this.dateSortedSearch(query, viewer, started);
381    }
382    let body: Record<string, unknown> = buildFullTextBody(query, viewer);
383    if (query.page) body = withSearchAfter(body, decodeCursor(query.page));
384    const os = await this.osRequest<OsResponse>(`/${contentIndex()}/_search`, body);
385    const hits = os.hits?.hits ?? [];
386    const pageSize = query.pageSize ?? 25;
387    const results = hits.map((h) => this.mapHit(h, "bm25"));
388    const last = hits[hits.length - 1];
389    const cursor =
390      hits.length >= pageSize && last ? encodeCursor([last._score ?? 0, last._source.updatedAt, last._id]) : undefined;
391    return {
392      results,
393      total: os.hits?.total?.value ?? hits.length,
394      ...(cursor !== undefined ? { cursor } : {}),
395      latencyMs: this.now() - started,
396    };
397  }
398 
399  /**
400   * T-155 — date-ordered search (`sortBy: "createdAt" | "updatedAt"`):
401   * newest-first on the stored date field (`_id` ascending tiebreak), with its
402   * own `(dateIso, id)` cursor. The relevance path is deliberately not touched
403   * (see `fullTextSearch`). The same mandatory tenant + permission + type +
404   * chip filters apply — `buildFullTextBody` builds them.
405   */
406  private async dateSortedSearch(
407    query: SearchQuery,
408    viewer: ViewerContext,
409    started: number,
410  ): Promise<SearchResponse> {
411    const field = dateSortField(query.sortBy!);
412    let body: Record<string, unknown> = buildFullTextBody(query, viewer);
413    if (query.page) body = withDateSort(body, query.sortBy!, decodeDateCursor(query.page));
414    else body = withDateSort(body, query.sortBy!);
415    const os = await this.osRequest<OsResponse>(`/${contentIndex()}/_search`, body);
416    const hits = os.hits?.hits ?? [];
417    const pageSize = query.pageSize ?? 25;
418    const results = hits.map((h) => this.mapHit(h, "bm25"));
419    const last = hits[hits.length - 1];
420    const lastDate = last?._source[field];
421    const cursor =
422      hits.length >= pageSize && last && lastDate
423        ? encodeDateCursor([lastDate, last._id])
424        : undefined;
425    return {
426      results,
427      total: os.hits?.total?.value ?? hits.length,
428      ...(cursor !== undefined ? { cursor } : {}),
429      latencyMs: this.now() - started,
430    };
431  }
432 
433  private async hybridSearch(
434    query: SearchQuery,
435    viewer: ViewerContext,
436    queryVector: number[],
437    started: number,
438  ): Promise<SearchResponse> {
439    const pageSize = query.pageSize ?? 25;
440    // Pass 1: BM25 (full-text body, sized to the fusion depth). TASK-0125:
441    // pass 1 is BM25-only — OpenSearch 2.15's `query.knn` cannot share a
442    // request body with a `bool`/full-text query, so the vector leg is its
443    // own request (pass 2) and the two are fused client-side with RRF.
444    const bm25Body: Record<string, unknown> = {
445      ...buildHybridBody(query, viewer, queryVector),
446      size: this.knnK,
447    };
448    // Pass 2: kNN-only (`query.knn`, pre-filtered), same mandatory filters.
449    const knnBody = buildHybridKnnOnlyBody(queryVector, viewer, this.knnK);
450    const [bm25Os, knnOs] = await Promise.all([
451      this.osRequest<OsResponse>(`/${contentIndex()}/_search`, bm25Body),
452      this.osRequest<OsResponse>(`/${contentIndex()}/_search`, knnBody),
453    ]);
454    const bm25Hits = bm25Os.hits?.hits ?? [];
455    const knnHits = knnOs.hits?.hits ?? [];
456    const bm25List = bm25Hits.map((h) => h._id);
457    const vecList = knnHits.map((h) => h._id);
458    const fused = fuseHybrid(bm25List, vecList);
459    const byId = new Map<string, OsHit>();
460    for (const h of bm25Hits) byId.set(h._id, h);
461    for (const h of knnHits) if (!byId.has(h._id)) byId.set(h._id, h);
462    const results: SearchResult[] = [];
463    for (const f of fused) {
464      const hit = byId.get(f.docId);
465      if (!hit) continue;
466      const source: SearchResult["source"] =
467        f.sources.includes("bm25") && f.sources.includes("vec") ? "bm25+vec" : f.sources.includes("vec") ? "vec" : "bm25";
468      results.push({ ...this.mapHit(hit, source), score: f.score });
469      if (results.length >= pageSize) break;
470    }
471    return {
472      results,
473      total: bm25Os.hits?.total?.value ?? results.length,
474      latencyMs: this.now() - started,
475    };
476  }
477 
478  private mapHit(hit: OsHit, source: SearchResult["source"]): SearchResult {
479    const s = hit._source;
480    const hl = s.highlight;
481    const snippet = hl?.body?.[0] ?? s.body.slice(0, 160);
482    const result: SearchResult = {
483      docId: hit._id,
484      // Chunk docs surface as their parent's public kind (displayType);
485      // whole docs map to themselves.
486      type: displayType(s),
487      id: s.id,
488      title: s.title,
489      snippet,
490      url: s.url,
491      updatedAt: s.updatedAt,
492      workspaceId: s.workspaceId,
493      score: hit._score ?? 0,
494      source,
495    };
496    if (s.icon !== undefined) result.icon = s.icon;
497    if (s.createdByName !== undefined) result.createdByName = s.createdByName;
498    if (s.pageId !== undefined) result.pageId = s.pageId;
499    if (s.blockId !== undefined) result.blockId = s.blockId;
500    if (s.databaseId !== undefined) result.databaseId = s.databaseId;
501    if (s.recordId !== undefined) result.recordId = s.recordId;
502    return result;
503  }
504 
505  private async request(path: string, init: { method?: string; body?: string }): Promise<Response> {
506    const controller = new AbortController();
507    const timer = setTimeout(() => controller.abort(), this.timeoutMs);
508    try {
509      const res = await this.fetchImpl(this.baseUrl + path, {
510        method: init.method ?? "GET",
511        headers: this.headers,
512        body: init.body ?? null,
513        signal: controller.signal,
514      });
515      if (!res.ok) {
516        throw new ApiError(ErrorCodes.SEARCH_UNAVAILABLE, {
517          message: `OpenSearch ${res.status} on ${path}`,
518          status: res.status,
519        });
520      }
521      return res;
522    } catch (err) {
523      if (err instanceof ApiError) throw err;
524      throw new ApiError(ErrorCodes.SEARCH_UNAVAILABLE, {
525        message: err instanceof Error ? err.message : "search request failed",
526      });
527    } finally {
528      clearTimeout(timer);
529    }
530  }
531 
532  private async osRequest<T>(path: string, body: Record<string, unknown>): Promise<T> {
533    const res = await this.request(path, { method: "POST", body: JSON.stringify(body) });
534    return (await res.json()) as T;
535  }
536}
537 
538/** Build RAG citations from the top search results (ai-architecture §6). */
539export function citationsFromResults(results: SearchResult[], limit = 5): AnswerCitation[] {
540  return results.slice(0, limit).map((r) => ({
541    docId: r.docId,
542    title: r.title,
543    url: r.url,
544    quote: r.snippet ?? "",
545    relevance: Math.min(1, r.score),
546  }));
547}
548