feat: add durable versioned search definition repository

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<SearchIndexDefinition>;
+  latest(indexName: string): Promise<SearchIndexDefinition | undefined>;
+  listLatest(): Promise<SearchIndexDefinition[]>;
+}
+
+interface DefinitionRow {
+  index_name: string;
+  version: number;
+  fields: Record<string, SearchFieldType>;
+  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<string, SearchIndexDefinition[]>();
+
+  async save(definition: SearchIndexDefinition): Promise<SearchIndexDefinition> {
+    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<SearchIndexDefinition | undefined> {
+    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<SearchIndexDefinition[]> {
+    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<SearchIndexDefinition> {
+    return this.sql.begin(async (tx) => {
+      await tx`SELECT pg_advisory_xact_lock(hashtext(${`0pt-search:${definition.index_name}`}))`;
+      const same = await tx<DefinitionRow[]>`
+        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<DefinitionRow[]>`
+        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<RepositorySql["json"]>[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<SearchIndexDefinition | undefined> {
+    const rows = await this.sql<DefinitionRow[]>`
+      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<SearchIndexDefinition[]> {
+    const rows = await this.sql<DefinitionRow[]>`
+      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();
+}