test: prove search definitions survive restart across Postgres and OpenSearch

2026-09-07 18:12:50 +02:00
parent 6d02a5cd34
commit ed1abc4cd4
1 changed files with 117 additions and 0 deletions
@@ -0,0 +1,117 @@
+import fs from "node:fs";
+import path from "node:path";
+import postgres from "postgres";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+import { closeRepositorySqlForTests, getRepositorySql } from "@/src/repositories/db";
+import { PostgresSearchDefinitionRepository } from "@/src/repositories/search-definition-repository";
+import { HttpOpenSearchAdapter } from "@/src/search/opensearch-adapter";
+import { SearchService, type DocumentSource } from "@/src/search/search-service";
+import type { SearchDocument } from "@/src/search/search-types";
+
+const OPENSEARCH_URL = process.env.OPENSEARCH_URL;
+const DATABASE_URL = process.env.SEARCH_DEFINITION_TEST_DATABASE_URL;
+const run = OPENSEARCH_URL && DATABASE_URL ? describe : describe.skip;
+let admin: ReturnType<typeof postgres> | undefined;
+let reachable = false;
+let reason = "";
+
+beforeAll(async () => {
+  if (!OPENSEARCH_URL || !DATABASE_URL) return;
+  try {
+    admin = postgres(DATABASE_URL, { max: 1, prepare: false, onnotice: () => undefined });
+    await admin`SELECT 1`;
+    await admin.unsafe(fs.readFileSync(path.join(process.cwd(), "sql", "013_search_index_definitions.sql"), "utf8"));
+    const health = await new HttpOpenSearchAdapter({ baseUrl: OPENSEARCH_URL }).health();
+    if (!(["green", "yellow"] as string[]).includes(health.status)) throw new Error(`OpenSearch ${health.status}`);
+    reachable = true;
+  } catch (error) {
+    reason = error instanceof Error ? error.message : String(error);
+  }
+}, 30_000);
+
+afterAll(async () => {
+  await admin?.end({ timeout: 5 });
+  await closeRepositorySqlForTests();
+});
+
+run("durable SearchDefinitionRepository + live OpenSearch", () => {
+  it("reloads a definition after a fresh SearchService and rebuilds/querys the projection", async (ctx) => {
+    if (!reachable) {
+      console.warn(`[realinfra] search durable definition dependencies unavailable (${reason}) — skipping honestly`);
+      ctx.skip();
+      return;
+    }
+    const stamp = `${Date.now()}_${Math.random().toString(16).slice(2)}`.replace(/[^a-z0-9_]/gi, "_");
+    const indexName = `companies_${stamp}`.slice(0, 60);
+    const tenant = `t_${stamp}`;
+    const repo = new PostgresSearchDefinitionRepository(getRepositorySql(DATABASE_URL!));
+    const adapter = new HttpOpenSearchAdapter({ baseUrl: OPENSEARCH_URL! });
+
+    const writer = new SearchService(adapter, { definitionRepository: repo });
+    await writer.registerIndexDefinitionDurable({
+      index_name: indexName,
+      version: 1,
+      fields: { name: "text", org_nr: "keyword" },
+      default_search_fields: ["name"],
+    }, "realinfra-test");
+
+    // New instance, empty in-memory cache — this is the restart condition.
+    const reader = new SearchService(adapter, { definitionRepository: repo });
+    const loaded = await reader.resolveIndexDefinition(indexName);
+    expect(loaded.version).toBe(1);
+    expect(loaded.fields.name).toBe("text");
+
+    const docs: SearchDocument[] = [{
+      doc_id: "company-1",
+      tenant_id: tenant,
+      purpose_tags: ["company_enrichment"],
+      rights_tags: ["internal"],
+      source_refs: ["0pt:dataset:companies@1"],
+      payload: { name: "Acme Aktiebolag", org_nr: "5560000000" },
+    }];
+    const source: DocumentSource = {
+      name: "0pt:dataset:companies@1",
+      async *iterate() { for (const doc of docs) yield doc; },
+    };
+
+    const rebuilt = await reader.rebuild(indexName, source, { tenant_id: tenant });
+    expect(rebuilt.docs_indexed).toBe(1);
+    expect(rebuilt.switched).toBe(true);
+
+    const result = await reader.query(indexName, {
+      tenant_id: tenant,
+      purpose: "company_enrichment",
+      rights_tag: "internal",
+      text: "Acme",
+    });
+    expect(result.total).toBe(1);
+    expect(result.hits[0].doc_id).toBe("company-1");
+
+    // A foreign tenant sees no hit even though it queries the same physical alias.
+    const foreign = await reader.query(indexName, {
+      tenant_id: `foreign_${tenant}`,
+      purpose: "company_enrichment",
+      rights_tag: "internal",
+      text: "Acme",
+    });
+    expect(foreign.total).toBe(0);
+  });
+
+  it("enforces sequential schema versions and rejects same-version drift", async (ctx) => {
+    if (!reachable) return ctx.skip();
+    const stamp = `${Date.now()}_${Math.random().toString(16).slice(2)}`.replace(/[^a-z0-9_]/gi, "_");
+    const indexName = `schema_${stamp}`.slice(0, 60);
+    const repo = new PostgresSearchDefinitionRepository(getRepositorySql(DATABASE_URL!));
+    const v1 = { index_name: indexName, version: 1, fields: { name: "text" as const }, default_search_fields: ["name"] };
+    const first = await repo.save(v1, "test");
+    const replay = await repo.save(v1, "test");
+    expect(first.created).toBe(true);
+    expect(replay.created).toBe(false);
+    expect(replay.created_at).toBe(first.created_at);
+
+    await expect(repo.save({ ...v1, fields: { name: "keyword" } }, "test")).rejects.toMatchObject({ code: "CONFLICT" });
+    await expect(repo.save({ ...v1, version: 3 }, "test")).rejects.toMatchObject({ code: "CONFLICT" });
+    expect((await repo.save({ ...v1, version: 2 }, "test")).definition.version).toBe(2);
+  });
+});