opensearch-reindex.ts

opensearch-reindex.tsscripts/opensearch-reindex.tsraw ← scripts18,180 B← Back to the app
1/**
2 * AI Software Factory — OpenSearch content-index reindex/rollover (TASK-0099 §5).
3 *
4 * Rolls the `factory-content` alias to a NEW index (`factory-content-<to>`,
5 * default: the next version after the current alias target) with a fresh
6 * mapping (knn_vector `vector` — faiss/l2 on OpenSearch 2.15, see below —
7 * dimension/m/efConstruction tunable),
8 * backfills it from the `search_documents` Postgres mirror in 500-doc
9 * `_bulk` batches, asserts the count equality, then ATOMICALLY swaps the
10 * alias (`POST /_aliases`) and optionally deletes the old index.
11 *
12 * Why: `dense_vector` dimensions are IMMUTABLE per index — a model change
13 * (1536 → 3072, …) cannot re-map in place; it must roll a new index forward
14 * through the alias. This script is the runbook for that (and for the v1 →
15 * v2 hybrid-foundation mapping change, TASK-0099).
16 *
17 * Self-contained by design (recorded decision): the root has no
18 * `@factory/*` hoist and the repo allows no new dependencies, so the
19 * mapping is re-declared below (mirror of packages/search mapping.ts —
20 * keep the two in sync) and OpenSearch is driven with raw `fetch`.
21 * Postgres uses the repo's existing `@prisma/client` (no new dep).
22 *
23 *   pnpm search:reindex --i-know-this-swaps-the-alias [--to v2]
24 *       [--dimension 1536] [--dry-run] [--strict-equal] [--cleanup-v1]
25 *
26 * Flags:
27 *   --i-know-this-swaps-the-alias  required (unless --dry-run): this command
28 *                                  re-points the LIVE search alias.
29 *   --to <ver>        target version suffix (default: current+1, or v1 fresh).
30 *   --dimension <n>   dense_vector dimension (default 1536).
31 *   --dry-run         read-only: print the plan, create/write nothing.
32 *   --strict-equal    fail (exit 1) if target count != mirror count
33 *                     (default: warn only).
34 *   --cleanup-v1      delete the OLD index after a successful swap.
35 *
36 * Env: SEARCH_URL (http://localhost:9200), SEARCH_USERNAME (admin),
37 *      SEARCH_PASSWORD (changeme), DATABASE_URL
38 *      (postgresql://factory:factory@localhost:5432/factory),
39 *      SEARCH_KNN_M (16), SEARCH_KNN_EF_CONSTRUCTION (100).
40 *
41 * Backfill source: the `search_documents` MIRROR (whole docs only — chunk
42 * docs have no mirror rows). Phase-12 loose end: backfill from the SOURCE
43 * tables (url/grants/createdByName) instead of the mirror; until then a
44 * fresh v2 index contains whole docs only, and chunk docs reappear as the
45 * worker re-indexes pages/records (or on the next content event).
46 */
47import { PrismaClient } from "@prisma/client";
48 
49// ---------------------------------------------------------------------------
50// Flags + env
51// ---------------------------------------------------------------------------
52 
53const ALIAS = "factory-content";
54const BULK_BATCH = 500; // _bulk line batch (mirrors the worker's BATCH)
55 
56interface Flags {
57  guard: boolean;
58  to: string | null;
59  dimension: number;
60  dryRun: boolean;
61  strictEqual: boolean;
62  cleanupV1: boolean;
63}
64 
65function fail(msg: string): never {
66  console.error(`✖ ${msg}`);
67  process.exit(1);
68}
69 
70function parseFlags(argv: string[]): Flags {
71  const f: Flags = {
72    guard: false,
73    to: null,
74    dimension: 1536,
75    dryRun: false,
76    strictEqual: false,
77    cleanupV1: false,
78  };
79  for (let i = 0; i < argv.length; i++) {
80    const a = argv[i]!;
81    switch (a) {
82      case "--i-know-this-swaps-the-alias":
83        f.guard = true;
84        break;
85      case "--to": {
86        const v = argv[++i];
87        if (!v || !/^v[0-9]+$/.test(v))
88          fail(`--to expects a version like "v2", got: ${v ?? "<none>"}`);
89        f.to = v;
90        break;
91      }
92      case "--dimension": {
93        const v = argv[++i];
94        const n = v ? Number.parseInt(v, 10) : NaN;
95        if (!v || !Number.isInteger(n) || n <= 0)
96          fail(`--dimension expects a positive integer, got: ${v ?? "<none>"}`);
97        f.dimension = n;
98        break;
99      }
100      case "--dry-run":
101        f.dryRun = true;
102        break;
103      case "--strict-equal":
104        f.strictEqual = true;
105        break;
106      case "--cleanup-v1":
107        f.cleanupV1 = true;
108        break;
109      default:
110        fail(`unknown flag: ${a} (see header of scripts/opensearch-reindex.ts)`);
111    }
112  }
113  return f;
114}
115 
116// ---------------------------------------------------------------------------
117// OpenSearch (raw fetch — no SDK dependency)
118// ---------------------------------------------------------------------------
119 
120interface Os {
121  base: string;
122  user: string;
123  pass: string;
124}
125 
126async function os(osCfg: Os, path: string, init?: RequestInit): Promise<unknown> {
127  const headers: Record<string, string> = {
128    "content-type": "application/json",
129    authorization: "Basic " + Buffer.from(`${osCfg.user}:${osCfg.pass}`).toString("base64"),
130  };
131  const res = await fetch(osCfg.base.replace(/\/+$/, "") + path, { ...init, headers });
132  const text = await res.text();
133  if (!res.ok) {
134    throw new Error(
135      `OpenSearch ${init?.method ?? "GET"} ${path} → HTTP ${res.status}: ${text.slice(0, 500)}`,
136    );
137  }
138  return text ? (JSON.parse(text) as unknown) : null;
139}
140 
141async function osCurrentAliasTarget(osCfg: Os): Promise<string | null> {
142  const res = await os(osCfg, `/_alias/${ALIAS}`);
143  const idxs = Object.keys((res as Record<string, unknown>) ?? {});
144  return idxs.length > 0 ? idxs[0]! : null;
145}
146 
147interface IndexInfo {
148  dimension: number | null;
149  count: number | null;
150}
151 
152async function osIndexInfo(osCfg: Os, index: string): Promise<IndexInfo> {
153  const mapping = (await os(osCfg, `/${index}/_mapping`)) as Record<
154    string,
155    { mappings?: Record<string, unknown> }
156  >;
157  const m = mapping[index]?.mappings as
158    { properties?: { vector?: { dimension?: number } } } | undefined;
159  const dimension = m?.properties?.vector?.dimension ?? null;
160  const countRes = (await os(osCfg, `/${index}/_count`)) as { count: number };
161  return { dimension, count: countRes.count ?? 0 };
162}
163 
164// ---------------------------------------------------------------------------
165// Mapping (MIRROR of packages/search/src/mapping.ts contentMapping — keep in
166// sync; re-declared here because the script must stay dependency-free).
167// ---------------------------------------------------------------------------
168 
169function contentMapping(
170  dimension: number,
171  m: number,
172  efConstruction: number,
173): Record<string, unknown> {
174  return {
175    settings: {
176      index: {
177        refresh_interval: "1s", // lag SLA: keep 1 s (search §1.4)
178        number_of_replicas: 1,
179        knn: true,
180        "knn.algo_param.ef_search": 100,
181      },
182    },
183    mappings: {
184      properties: {
185        docId: { type: "keyword" },
186        // MIRRORS packages/search contentMapping: plain keyword — the
187        // OpenSearch in docker/CI (2.15) rejects a mapping-level `enum` on
188        // keyword mappers; the allowed set is enforced at the application
189        // layer (the client only indexes known SearchDocType values).
190        type: { type: "keyword" },
191        id: { type: "keyword" },
192        workspaceId: { type: "keyword" },
193        teamId: { type: "keyword" },
194        title: {
195          type: "text",
196          fields: { keyword: { type: "keyword", ignore_above: 512 } },
197        },
198        body: { type: "text" },
199        chunkId: { type: "keyword" },
200        chunkIndex: { type: "integer" },
201        parentDocId: { type: "keyword" },
202        vectorModel: { type: "keyword" },
203        vector: {
204          // MIRRORS the package (packages/search/src/mapping.ts — keep in
205          // sync): OpenSearch 2.15 has the 1.x `knn_vector` + `method` HNSW
206          // form — NO `dense_vector` handler. TASK-0125 (verified live):
207          // engine = faiss (nmslib hard-rejects the kNN pre-filter that the
208          // hybrid leg needs) and space_type = l2 (FAISS 2.15 rejects
209          // cosinesimil/inner_product; l2 ranks identically to cosine for
210          // L2-normalized real embeddings).
211          type: "knn_vector",
212          dimension,
213          method: {
214            name: "hnsw",
215            space_type: "l2",
216            engine: "faiss",
217            parameters: { m, ef_construction: efConstruction },
218          },
219        },
220        grants: {
221          properties: {
222            public: { type: "boolean" },
223            workspace: { type: "boolean" },
224            user: { type: "boolean" },
225            teams: { type: "keyword" },
226            users: { type: "keyword" },
227          },
228        },
229        archived: { type: "boolean" },
230        pageId: { type: "keyword" },
231        blockId: { type: "keyword" },
232        databaseId: { type: "keyword" },
233        recordId: { type: "keyword" },
234        createdBy: { type: "keyword" },
235        createdByName: { type: "keyword" },
236        icon: { type: "keyword", index: false },
237        url: { type: "keyword", index: false },
238        size: { type: "long" },
239        mimeType: { type: "keyword" },
240        createdAt: { type: "date" },
241        updatedAt: { type: "date" },
242      },
243    },
244  };
245}
246 
247// ---------------------------------------------------------------------------
248// Mirror → index doc (whole docs only; chunk docs have no mirror rows)
249// ---------------------------------------------------------------------------
250 
251interface MirrorRow {
252  id: string;
253  workspaceId: string;
254  type: string;
255  refId: string;
256  title: string;
257  body: string | null;
258  permissions: unknown;
259  metadata: unknown;
260  indexedAt: Date;
261}
262 
263/** The index `_id` / `docId`: `<type>:<mirror refId>` (user refId is already
264 *  workspace-scoped: `user:<ws>:<uid>`). */
265function docIdFor(row: MirrorRow): string {
266  return `${row.type}:${row.refId}`;
267}
268 
269/** Rebuild the wire-form index doc from a mirror row (metadata = the
270 *  deep-link facet snapshot the worker stored in upsertMirror). */
271function mirrorToDoc(row: MirrorRow): Record<string, unknown> {
272  const meta = (row.metadata ?? {}) as Record<string, unknown>;
273  const str = (k: string): string | undefined =>
274    typeof meta[k] === "string" ? (meta[k] as string) : undefined;
275  const doc: Record<string, unknown> = {
276    docId: docIdFor(row),
277    type: row.type,
278    // the doc's `id` is the plain row id (user docs: the bare uid)
279    id: row.type === "user" ? (row.refId.split(":").pop() ?? row.refId) : row.refId,
280    workspaceId: row.workspaceId,
281    title: row.title,
282    body: row.body ?? "",
283    grants: (row.permissions ?? {}) as Record<string, unknown>,
284    archived: typeof meta.archived === "boolean" ? meta.archived : false,
285    url: str("url") ?? "",
286    createdAt: str("createdAt") ?? row.indexedAt.toISOString(),
287    updatedAt: str("updatedAt") ?? row.indexedAt.toISOString(),
288  };
289  // facet fields — only present when the mirror recorded them
290  for (const k of [
291    "teamId",
292    "pageId",
293    "blockId",
294    "databaseId",
295    "recordId",
296    "createdBy",
297    "icon",
298    "size",
299    "mimeType",
300  ] as const) {
301    const v = meta[k];
302    if (v !== null && v !== undefined) doc[k] = v;
303  }
304  return doc;
305}
306 
307// ---------------------------------------------------------------------------
308// Main
309// ---------------------------------------------------------------------------
310 
311async function main(): Promise<void> {
312  const flags = parseFlags(process.argv.slice(2));
313  if (!flags.guard && !flags.dryRun) {
314    fail(
315      "refusing to reindex without the guard flag — this command re-points the LIVE " +
316        `"${ALIAS}" alias.\n  Re-run with --i-know-this-swaps-the-alias (or --dry-run to inspect only).`,
317    );
318  }
319 
320  const osCfg: Os = {
321    base: process.env.SEARCH_URL ?? "http://localhost:9200",
322    user: process.env.SEARCH_USERNAME ?? "admin",
323    pass: process.env.SEARCH_PASSWORD ?? "changeme",
324  };
325  const dimension = flags.dimension;
326  const knnM = Number.parseInt(process.env.SEARCH_KNN_M ?? "16", 10);
327  const efConstruction = Number.parseInt(process.env.SEARCH_KNN_EF_CONSTRUCTION ?? "100", 10);
328 
329  console.log(`▸ reindex plan: alias="${ALIAS}" target-dim=${dimension} dry-run=${flags.dryRun}`);
330 
331  // 1. Resolve the current alias target + live state.
332  const current = await osCurrentAliasTarget(osCfg).catch((e: Error) => {
333    if (/404/.test(e.message)) return null;
334    throw e;
335  });
336  console.log(`  current alias target: ${current ?? "(none — fresh cluster)"}`);
337 
338  // 2. Target version: explicit --to, else next after current (v1 → v2),
339  //    else v1 for a fresh cluster.
340  let target: string;
341  if (flags.to) {
342    target = `${ALIAS}-${flags.to}`;
343  } else if (current) {
344    const m = /-v([0-9]+)$/.exec(current);
345    if (!m) fail(`cannot derive the next version from "${current}" — pass --to explicitly`);
346    target = `${ALIAS}-v${Number.parseInt(m[1]!, 10) + 1}`;
347  } else {
348    target = `${ALIAS}-v1`;
349  }
350  console.log(`  target index: ${target}`);
351 
352  // 3. Read the mirror count (the backfill source of truth).
353  const prisma = new PrismaClient({
354    datasources: {
355      db: {
356        url: process.env.DATABASE_URL ?? "postgresql://factory:factory@localhost:5432/factory",
357      },
358    },
359  });
360  let mirrorCount = 0;
361  try {
362    mirrorCount = await prisma.searchDocument.count();
363  } finally {
364    if (flags.dryRun) await prisma.$disconnect();
365  }
366  console.log(`  mirror rows (search_documents, whole docs): ${mirrorCount}`);
367 
368  // 4. Existing target index? (idempotent re-run: same --to re-bakes + re-swaps)
369  let targetInfo: IndexInfo | null = null;
370  try {
371    targetInfo = await osIndexInfo(osCfg, target);
372  } catch (e) {
373    if (!/404/.test((e as Error).message)) throw e;
374  }
375  if (targetInfo) {
376    console.log(`  target already exists: dim=${targetInfo.dimension} docs=${targetInfo.count}`);
377    if (targetInfo.dimension !== null && targetInfo.dimension !== dimension) {
378      fail(
379        `target "${target}" exists with dimension ${targetInfo.dimension} ≠ requested ${dimension} — ` +
380          `dense_vector dimensions are immutable. Pick another --to (or delete the index).`,
381      );
382    }
383  }
384 
385  if (flags.dryRun) {
386    console.log(
387      `\n[dry-run] plan only — would:\n` +
388        `  1. ${targetInfo ? "reuse" : "create"} ${target} (dim=${dimension}, m=${knnM}, ef_construction=${efConstruction})\n` +
389        `  2. bulk-index ${mirrorCount} mirror docs in ${BULK_BATCH}-doc batches\n` +
390        `  3. refresh + count check (strict=${flags.strictEqual})\n` +
391        `  4. atomically move "${ALIAS}": ${current ?? "(none)"} → ${target}\n` +
392        `  ${flags.cleanupV1 && current ? `5. delete old index ${current}\n` : ""}` +
393        `No writes were made.`,
394    );
395    return;
396  }
397 
398  // 5. Create the target index (idempotent — only when missing).
399  if (!targetInfo) {
400    await os(osCfg, `/${target}`, {
401      method: "PUT",
402      body: JSON.stringify(contentMapping(dimension, knnM, efConstruction)),
403    });
404    console.log(
405      `  created ${target} (dim=${dimension}, m=${knnM}, ef_construction=${efConstruction})`,
406    );
407  }
408 
409  // 6. Backfill from the mirror (cursor-paged reads, 500-doc _bulk writes).
410  let written = 0;
411  let cursor: string | null = null;
412  const PAGE = 1000;
413  for (;;) {
414    // exactOptionalPropertyTypes: `where?: X` does not accept `X | undefined`,
415    // so the cursor-paged read is a plain conditional (no `| undefined`).
416    const rows: MirrorRow[] = cursor
417      ? ((await prisma.searchDocument.findMany({
418          where: { id: { gt: cursor } },
419          orderBy: { id: "asc" },
420          take: PAGE,
421        })) as MirrorRow[])
422      : ((await prisma.searchDocument.findMany({
423          orderBy: { id: "asc" },
424          take: PAGE,
425        })) as MirrorRow[]);
426    if (rows.length === 0) break;
427    const line: string[] = [];
428    for (const row of rows) {
429      line.push(`{"index":{"_index":"${target}","_id":"${docIdFor(row)}"}}`);
430      line.push(JSON.stringify(mirrorToDoc(row)));
431    }
432    const docCount = line.length / 2;
433    for (let i = 0; i < line.length; i += BULK_BATCH * 2) {
434      const batchDocs = Math.min(BULK_BATCH, docCount - i / 2);
435      const body = line.slice(i, i + BULK_BATCH * 2).join("\n") + "\n";
436      const res = (await os(osCfg, `/${target}/_bulk`, { method: "POST", body })) as {
437        errors?: boolean;
438        items?: { index?: { error?: { type?: string; reason?: string }; status?: number } }[];
439      };
440      if (res.errors) {
441        const bad = res.items?.find((it) => it.index?.error) ?? null;
442        throw new Error(`_bulk errors on ${target}: ${JSON.stringify(bad?.index?.error ?? {})}`);
443      }
444      written += batchDocs;
445    }
446    cursor = rows[rows.length - 1]!.id;
447    if (rows.length < PAGE) break;
448  }
449  await prisma.$disconnect();
450  console.log(`  bulk-indexed ${written} docs (mirror rows read: ${written})`);
451 
452  // 7. Refresh + count check (idempotent docs: the target holds exactly the
453  //    mirror's whole docs — chunk docs have no mirror rows, see header).
454  await os(osCfg, `/${target}/_refresh`, { method: "POST" });
455  const after = await osIndexInfo(osCfg, target);
456  console.log(`  target count after backfill: ${after.count} (mirror: ${mirrorCount})`);
457  const equal = after.count === mirrorCount;
458  if (!equal) {
459    const msg = `count mismatch: target ${after.count} ≠ mirror ${mirrorCount}`;
460    if (flags.strictEqual) fail(msg);
461    console.warn(`  ⚠ ${msg} (continue — pass --strict-equal to hard-fail)`);
462  } else {
463    console.log(`  ✓ count equality holds (${mirrorCount})`);
464  }
465 
466  // 8. Atomic alias swap (one /_aliases call — never remove-then-add).
467  const actions: Record<string, unknown>[] = [{ add: { index: target, alias: ALIAS } }];
468  if (current) actions.unshift({ remove: { index: current, alias: ALIAS } });
469  await os(osCfg, "/_aliases", { method: "POST", body: JSON.stringify({ actions }) });
470  console.log(`  ✓ alias "${ALIAS}" now → ${target}${current ? ` (was ${current})` : ""}`);
471 
472  // 9. Optional cleanup of the old index.
473  if (flags.cleanupV1 && current) {
474    await os(osCfg, `/${current}`, { method: "DELETE" });
475    console.log(`  ✓ deleted old index ${current}`);
476  } else if (current) {
477    console.log(`  (kept ${current} — pass --cleanup-v1 to delete it after verifying)`);
478  }
479 
480  console.log("✔ reindex complete");
481}
482 
483main().catch((e: Error) => fail(e.message));
484