fix(indexer): remove ensureSchema() and add startup schema-version guard

services/indexer/docker-compose.yml

@@ -8,7 +8,6 @@ services:
       POSTGRES_DB: ${POSTGRES_DB:-linkora}
     volumes:
       - postgres_data:/var/lib/postgresql/data
-      - ./migrations:/docker-entrypoint-initdb.d:ro
     ports:
       - "5432:5432"
     healthcheck:
@@ -17,6 +16,33 @@ services:
       timeout: 5s
       retries: 10
 
+  # One-shot container: applies all numbered migrations in filename order and
+  # exits.  The indexer will not start until this service completes successfully,
+  # ensuring the DB is never behind when the application boots.
+  migrate:
+    image: postgres:16-alpine
+    restart: "no"
+    depends_on:
+      postgres:
+        condition: service_healthy
+    volumes:
+      - ./migrations:/migrations:ro
+    environment:
+      PGPASSWORD: ${POSTGRES_PASSWORD:-linkora}
+    command: >
+      sh -c "
+        for f in \$(ls /migrations/*.sql | sort); do
+          echo \"[migrate] applying \$$f\";
+          psql -h postgres
+               -U ${POSTGRES_USER:-linkora}
+               -d ${POSTGRES_DB:-linkora}
+               -v ON_ERROR_STOP=1
+               -f \$$f
+          || exit 1;
+        done;
+        echo '[migrate] all migrations applied'
+      "
+
   indexer:
     build: .
     restart: unless-stopped
@@ -29,6 +55,8 @@ services:
     depends_on:
       postgres:
         condition: service_healthy
+      migrate:
+        condition: service_completed_successfully
 
 volumes:
   postgres_data:

services/indexer/migrate.sh

@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+# Apply all numbered migrations in filename order against $DATABASE_URL.
+#
+# Usage:
+#   DATABASE_URL=postgresql://user:pass@host/db bash services/indexer/migrate.sh
+#
+# The script is idempotent: every migration uses IF NOT EXISTS so re-running
+# against an already-migrated database is safe.
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+MIG_DIR="$SCRIPT_DIR/migrations"
+
+if [[ -z "${DATABASE_URL:-}" ]]; then
+  echo "error: DATABASE_URL is not set" >&2
+  exit 1
+fi
+
+shopt -s nullglob
+MIGRATIONS=("$MIG_DIR"/*.sql)
+shopt -u nullglob
+
+if [[ ${#MIGRATIONS[@]} -eq 0 ]]; then
+  echo "error: no migration files found in $MIG_DIR" >&2
+  exit 1
+fi
+
+echo "[migrate] applying ${#MIGRATIONS[@]} migration(s) from $MIG_DIR"
+for f in "${MIGRATIONS[@]}"; do
+  echo "[migrate] $(basename "$f")"
+  psql "$DATABASE_URL" -v ON_ERROR_STOP=1 -q -f "$f"
+done
+echo "[migrate] done"

services/indexer/migrations/README.md

@@ -43,15 +43,28 @@ If a future migration ever needs a destructive change (`DROP`, narrowing
 2. ship with an accompanying idempotency test and, where a rollback is
    meaningful, a matching `*_down.sql`.
 
-## Relationship to `ensureSchema()`
-
-The indexer also creates a subset of these tables at boot via `ensureSchema()`
-in `src/index.ts`, so dev/test environments can run without a separate migration
-step. The two must stay consistent. Notably, `indexer_state` is the
-**state-root** table (`ledger_sequence, state_root, computed_at`); the
-per-stream ingestion cursor lives in `indexer_cursor`. (An earlier revision of
-`006_raw_events.sql` also defined `indexer_state` as a cursor table, colliding
-with the state-root definition — that stale block has been removed.)
+## Starting the indexer
+
+**Migrations must be applied before the indexer starts.** At boot the indexer
+calls `assertSchemaVersion()` (`src/schema-version.ts`) which checks that all
+sentinel tables and columns exist and exits with a clear error if any are
+missing.
+
+Apply migrations with one of:
+
+```bash
+# Docker Compose (recommended) — migrate service runs automatically before indexer
+docker compose up
+
+# Shell script (CI / bare-metal)
+DATABASE_URL=postgresql://linkora:linkora@localhost/linkora bash migrate.sh
+```
+
+Notably, `indexer_state` is the **state-root** table
+(`ledger_sequence, state_root, computed_at`); the per-stream ingestion cursor
+lives in `indexer_cursor`. (An earlier revision of `006_raw_events.sql` also
+defined `indexer_state` as a cursor table, colliding with the state-root
+definition — that stale block has been removed.)
 
 ## Running the tests
 

services/indexer/src/__tests__/instrumented-pool.test.ts

@@ -0,0 +1,147 @@
+import { Pool, QueryResult } from "pg";
+import { InstrumentedPool } from "../instrumented-pool";
+
+// ── Logger mock ───────────────────────────────────────────────────────────────
+
+const mockWarn = jest.fn();
+const mockError = jest.fn();
+
+jest.mock("../logger", () => ({
+  logger: {
+    warn: (...args: unknown[]) => mockWarn(...args),
+    error: (...args: unknown[]) => mockError(...args),
+    info: jest.fn(),
+  },
+}));
+
+// ── Helpers ───────────────────────────────────────────────────────────────────
+
+const FAKE_RESULT: QueryResult = {
+  rows: [{ one: 1 }],
+  rowCount: 1,
+  command: "SELECT",
+  oid: 0,
+  fields: [],
+};
+
+function makePool(thresholdMs: number): InstrumentedPool {
+  return new InstrumentedPool(thresholdMs);
+}
+
+// ── Tests ─────────────────────────────────────────────────────────────────────
+
+describe("InstrumentedPool", () => {
+  let superQuerySpy: jest.SpyInstance;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    superQuerySpy = jest
+      .spyOn(Pool.prototype, "query")
+      .mockResolvedValue(FAKE_RESULT as never);
+  });
+
+  afterEach(() => {
+    superQuerySpy.mockRestore();
+  });
+
+  it("returns the query result unchanged on success", async () => {
+    const pool = makePool(5000);
+    const result = await pool.query("SELECT 1");
+    expect(result).toBe(FAKE_RESULT);
+  });
+
+  it("fires logger.warn when query duration exceeds the threshold", async () => {
+    const pool = makePool(50);
+
+    // Fake Date.now so that elapsed = 100 ms > 50 ms threshold.
+    const nowSpy = jest
+      .spyOn(Date, "now")
+      .mockReturnValueOnce(0)   // start
+      .mockReturnValueOnce(100); // end
+
+    await pool.query("SELECT slow_thing FROM big_table");
+
+    expect(mockWarn).toHaveBeenCalledTimes(1);
+    expect(mockWarn).toHaveBeenCalledWith(
+      expect.objectContaining({ dur: 100, sql: "SELECT slow_thing FROM big_table" }),
+      "slow-query"
+    );
+    nowSpy.mockRestore();
+  });
+
+  it("does not fire logger.warn when query duration is below the threshold", async () => {
+    const pool = makePool(5000);
+
+    const nowSpy = jest
+      .spyOn(Date, "now")
+      .mockReturnValueOnce(0)
+      .mockReturnValueOnce(10); // 10 ms < 5000 ms threshold
+
+    await pool.query("SELECT fast FROM table");
+
+    expect(mockWarn).not.toHaveBeenCalled();
+    nowSpy.mockRestore();
+  });
+
+  it("truncates the sql snippet in the log to 120 characters", async () => {
+    const pool = makePool(50);
+    const longSql = "SELECT " + "x, ".repeat(100) + "1";
+
+    const nowSpy = jest
+      .spyOn(Date, "now")
+      .mockReturnValueOnce(0)
+      .mockReturnValueOnce(200);
+
+    await pool.query(longSql);
+
+    const logged: { sql: string } = mockWarn.mock.calls[0][0];
+    expect(logged.sql.length).toBeLessThanOrEqual(120);
+    nowSpy.mockRestore();
+  });
+
+  it("logs '(prepared)' for QueryConfig objects", async () => {
+    const pool = makePool(50);
+
+    const nowSpy = jest
+      .spyOn(Date, "now")
+      .mockReturnValueOnce(0)
+      .mockReturnValueOnce(200);
+
+    await pool.query({ text: "SELECT $1", values: [1] });
+
+    expect(mockWarn).toHaveBeenCalledWith(
+      expect.objectContaining({ sql: "(prepared)" }),
+      "slow-query"
+    );
+    nowSpy.mockRestore();
+  });
+
+  it("fires logger.error and re-throws when the underlying query rejects", async () => {
+    superQuerySpy.mockRejectedValue(new Error("connection reset") as never);
+    const pool = makePool(5000);
+
+    const nowSpy = jest
+      .spyOn(Date, "now")
+      .mockReturnValueOnce(0)
+      .mockReturnValueOnce(25);
+
+    await expect(pool.query("SELECT 1")).rejects.toThrow("connection reset");
+
+    expect(mockError).toHaveBeenCalledTimes(1);
+    expect(mockError).toHaveBeenCalledWith(
+      expect.objectContaining({ dur: 25, sql: "SELECT 1" }),
+      "query-error"
+    );
+    // warn must NOT fire on the error path
+    expect(mockWarn).not.toHaveBeenCalled();
+    nowSpy.mockRestore();
+  });
+
+  it("does not swallow the error — caller receives the original rejection", async () => {
+    const boom = new Error("timeout");
+    superQuerySpy.mockRejectedValue(boom as never);
+    const pool = makePool(5000);
+
+    await expect(pool.query("SELECT 1")).rejects.toBe(boom);
+  });
+});

services/indexer/src/__tests__/schema-version.test.ts

@@ -0,0 +1,108 @@
+import { Pool } from "pg";
+import { assertSchemaVersion } from "../schema-version";
+
+// ── Logger mock ───────────────────────────────────────────────────────────────
+
+const mockInfo = jest.fn();
+const mockError = jest.fn();
+
+jest.mock("../logger", () => ({
+  logger: {
+    info: (...args: unknown[]) => mockInfo(...args),
+    error: (...args: unknown[]) => mockError(...args),
+    warn: jest.fn(),
+  },
+}));
+
+// ── process.exit mock ─────────────────────────────────────────────────────────
+
+const mockExit = jest
+  .spyOn(process, "exit")
+  .mockImplementation((_code?: string | number | null | undefined) => {
+    throw new Error(`process.exit(${_code})`);
+  });
+
+// ── Helpers ───────────────────────────────────────────────────────────────────
+
+const ALL_TABLES = [
+  "raw_events",
+  "indexer_cursor",
+  "indexer_state",
+  "device_tokens",
+  "sent_notifications",
+  "blocks",
+  "dm_keys",
+  "notification_preferences",
+];
+
+function makePool(tables: string[], hasContentTsv = true): Pool {
+  const pool = { query: jest.fn() } as unknown as Pool;
+  (pool.query as jest.Mock).mockImplementation((sql: string, params?: unknown[]) => {
+    if (typeof sql === "string" && sql.includes("pg_tables")) {
+      return Promise.resolve({ rows: tables.map((t) => ({ tablename: t })) });
+    }
+    if (typeof sql === "string" && sql.includes("information_schema.columns")) {
+      const count = hasContentTsv ? 1 : 0;
+      return Promise.resolve({ rows: [{ count }] });
+    }
+    return Promise.resolve({ rows: [] });
+  });
+  return pool;
+}
+
+// ── Tests ─────────────────────────────────────────────────────────────────────
+
+describe("assertSchemaVersion", () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  afterAll(() => {
+    mockExit.mockRestore();
+  });
+
+  it("resolves without error when all sentinel tables and columns are present", async () => {
+    const pool = makePool(ALL_TABLES, true);
+    await expect(assertSchemaVersion(pool)).resolves.toBeUndefined();
+    expect(mockExit).not.toHaveBeenCalled();
+    expect(mockInfo).toHaveBeenCalledWith(
+      expect.objectContaining({ requiredTables: ALL_TABLES.length }),
+      "Schema version check passed"
+    );
+  });
+
+  it("calls process.exit(1) when a required table is missing", async () => {
+    const missingTable = "notification_preferences";
+    const pool = makePool(ALL_TABLES.filter((t) => t !== missingTable), true);
+    await expect(assertSchemaVersion(pool)).rejects.toThrow("process.exit(1)");
+    expect(mockExit).toHaveBeenCalledWith(1);
+    expect(mockError).toHaveBeenCalledWith(
+      expect.objectContaining({ missingTables: [missingTable] }),
+      expect.stringContaining("out of date")
+    );
+  });
+
+  it("calls process.exit(1) when the sentinel column posts.content_tsv is missing", async () => {
+    const pool = makePool(ALL_TABLES, false);
+    await expect(assertSchemaVersion(pool)).rejects.toThrow("process.exit(1)");
+    expect(mockExit).toHaveBeenCalledWith(1);
+    expect(mockError).toHaveBeenCalledWith(
+      expect.objectContaining({ table: "posts", column: "content_tsv" }),
+      expect.stringContaining("out of date")
+    );
+  });
+
+  it("calls process.exit(1) with a list of all missing tables when multiple are absent", async () => {
+    const pool = makePool(["raw_events", "indexer_cursor"], true);
+    await expect(assertSchemaVersion(pool)).rejects.toThrow("process.exit(1)");
+    expect(mockExit).toHaveBeenCalledWith(1);
+    const [logObj] = mockError.mock.calls[0] as [{ missingTables: string[] }];
+    expect(logObj.missingTables.length).toBeGreaterThan(1);
+  });
+
+  it("does not call process.exit(1) when the table list is a superset of required tables", async () => {
+    const pool = makePool([...ALL_TABLES, "some_extra_table"], true);
+    await expect(assertSchemaVersion(pool)).resolves.toBeUndefined();
+    expect(mockExit).not.toHaveBeenCalled();
+  });
+});