diff --git a/tests/integration/search-durable-definition.integration.test.ts b/tests/integration/search-durable-definition.integration.test.ts new file mode 100644 index 0000000..e0f9277 --- /dev/null +++ b/tests/integration/search-durable-definition.integration.test.ts @@ -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 | 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); + }); +});