From dd01fe76c95a66c293f5ef5f771245f89b3ac11e Mon Sep 17 00:00:00 2001 From: Simon Axelsson Date: Mon, 7 Sep 2026 18:08:09 +0200 Subject: [PATCH] feat: add durable versioned search definition repository --- .../search-definition-repository.ts | 140 ++++++++++++++++++ 1 file changed, 140 insertions(+) create mode 100644 src/repositories/search-definition-repository.ts diff --git a/src/repositories/search-definition-repository.ts b/src/repositories/search-definition-repository.ts new file mode 100644 index 0000000..f0984a5 --- /dev/null +++ b/src/repositories/search-definition-repository.ts @@ -0,0 +1,140 @@ +import { resolveStoreMode } from "../platform/dependency-mode"; +import { SiaxError } from "../platform/errors"; +import type { SearchIndexDefinition, SearchFieldType } from "../search/search-types"; +import { getRepositorySql, repositoryDatabaseUrl, type RepositorySql } from "./db"; + +export interface SearchDefinitionRepository { + save(definition: SearchIndexDefinition, actorId?: string): Promise; + latest(indexName: string): Promise; + listLatest(): Promise; +} + +interface DefinitionRow { + index_name: string; + version: number; + fields: Record; + default_search_fields: string[]; +} + +function toDefinition(row: DefinitionRow): SearchIndexDefinition { + return { + index_name: row.index_name, + version: Number(row.version), + fields: { ...row.fields }, + default_search_fields: [...row.default_search_fields], + }; +} + +function fingerprint(definition: SearchIndexDefinition): string { + return JSON.stringify({ + index_name: definition.index_name, + version: definition.version, + fields: Object.fromEntries(Object.entries(definition.fields).sort(([a], [b]) => a.localeCompare(b))), + default_search_fields: [...definition.default_search_fields], + }); +} + +export class InMemorySearchDefinitionRepository implements SearchDefinitionRepository { + private readonly versions = new Map(); + + async save(definition: SearchIndexDefinition): Promise { + const list = this.versions.get(definition.index_name) ?? []; + const existing = list.find((item) => item.version === definition.version); + if (existing) { + if (fingerprint(existing) !== fingerprint(definition)) { + throw new SiaxError("CONFLICT", "0pt.search", "search definition version reused with different schema"); + } + return { ...existing, fields: { ...existing.fields }, default_search_fields: [...existing.default_search_fields] }; + } + const expected = (list[list.length - 1]?.version ?? 0) + 1; + if (definition.version !== expected) { + throw new SiaxError("CONFLICT", "0pt.search", `next search definition version must be ${expected}`); + } + const copy = { ...definition, fields: { ...definition.fields }, default_search_fields: [...definition.default_search_fields] }; + list.push(copy); + this.versions.set(definition.index_name, list); + return copy; + } + + async latest(indexName: string): Promise { + const list = this.versions.get(indexName) ?? []; + const item = list[list.length - 1]; + return item ? { ...item, fields: { ...item.fields }, default_search_fields: [...item.default_search_fields] } : undefined; + } + + async listLatest(): Promise { + const result: SearchIndexDefinition[] = []; + for (const key of this.versions.keys()) { + const latest = await this.latest(key); + if (latest) result.push(latest); + } + return result.sort((a, b) => a.index_name.localeCompare(b.index_name)); + } +} + +export class PostgresSearchDefinitionRepository implements SearchDefinitionRepository { + constructor(private readonly sql: RepositorySql) {} + + async save(definition: SearchIndexDefinition, actorId?: string): Promise { + return this.sql.begin(async (tx) => { + await tx`SELECT pg_advisory_xact_lock(hashtext(${`0pt-search:${definition.index_name}`}))`; + const same = await tx` + SELECT index_name, version, fields, default_search_fields + FROM opt_search_index_definitions + WHERE index_name = ${definition.index_name} AND version = ${definition.version} + LIMIT 1`; + if (same[0]) { + const existing = toDefinition(same[0]); + if (fingerprint(existing) !== fingerprint(definition)) { + throw new SiaxError("CONFLICT", "0pt.search", "search definition version reused with different schema"); + } + return existing; + } + + const latest = await tx<{ version: number }[]>` + SELECT COALESCE(max(version), 0)::int AS version + FROM opt_search_index_definitions + WHERE index_name = ${definition.index_name}`; + const expected = (latest[0]?.version ?? 0) + 1; + if (definition.version !== expected) { + throw new SiaxError("CONFLICT", "0pt.search", `next search definition version must be ${expected}`); + } + + const rows = await tx` + INSERT INTO opt_search_index_definitions + (index_name, version, fields, default_search_fields, created_by) + VALUES + (${definition.index_name}, ${definition.version}, + ${tx.json(definition.fields as Parameters[0])}, + ${tx.array(definition.default_search_fields)}, ${actorId ?? null}) + RETURNING index_name, version, fields, default_search_fields`; + return toDefinition(rows[0]); + }); + } + + async latest(indexName: string): Promise { + const rows = await this.sql` + SELECT index_name, version, fields, default_search_fields + FROM opt_search_index_definitions + WHERE index_name = ${indexName} + ORDER BY version DESC + LIMIT 1`; + return rows[0] ? toDefinition(rows[0]) : undefined; + } + + async listLatest(): Promise { + const rows = await this.sql` + SELECT DISTINCT ON (index_name) + index_name, version, fields, default_search_fields + FROM opt_search_index_definitions + ORDER BY index_name, version DESC`; + return rows.map(toDefinition); + } +} + +export function createSearchDefinitionRepository(databaseUrl?: string): SearchDefinitionRepository { + const mode = resolveStoreMode("0pt.search_definitions", databaseUrl ?? repositoryDatabaseUrl()); + return mode.mode === "postgres" + ? new PostgresSearchDefinitionRepository(getRepositorySql(mode.url)) + : new InMemorySearchDefinitionRepository(); +}