diff --git a/apps/contrail-e2e/tests/cursor-resume.test.ts b/apps/contrail-e2e/tests/cursor-resume.test.ts index 1b01066..9161f5a 100644 --- a/apps/contrail-e2e/tests/cursor-resume.test.ts +++ b/apps/contrail-e2e/tests/cursor-resume.test.ts @@ -24,7 +24,7 @@ import { describe, it, expect, beforeAll, afterAll } from "vitest"; import pg from "pg"; import { CredentialManager, Client } from "@atcute/client"; import "@atcute/atproto"; -import { Contrail } from "@atmo-dev/contrail"; +import { runPersistent } from "@atmo-dev/contrail"; import { createPostgresDatabase } from "@atmo-dev/contrail/postgres"; import { config as baseConfig } from "../config"; import { @@ -73,15 +73,11 @@ describe("cursor resume (ingester stops, events pile up, ingester restarts)", () it("resumes from saved cursor and picks up commits issued while down", async () => { const db = createPostgresDatabase(pool); - // Go through the Contrail wrapper so the config is resolved; raw - // runPersistent expects a resolved config. - const contrail = new Contrail({ ...baseConfig, db }); - await contrail.init(); const runOpts = { batchSize: 50, flushIntervalMs: 500 }; // ---- Phase 1: start ingester, publish A, verify indexed ---- const c1 = new AbortController(); - const ingest1 = contrail.runPersistent({ ...runOpts, signal: c1.signal }); + const ingest1 = runPersistent(db, baseConfig, { ...runOpts, signal: c1.signal }); const nameA = `A-${Date.now()}`; const aRes = await client.post("com.atproto.repo.createRecord", { @@ -164,7 +160,7 @@ describe("cursor resume (ingester stops, events pile up, ingester restarts)", () // ---- Phase 4: restart ingester, assert replay ---- const c2 = new AbortController(); - const ingest2 = contrail.runPersistent({ ...runOpts, signal: c2.signal }); + const ingest2 = runPersistent(db, baseConfig, { ...runOpts, signal: c2.signal }); try { const indexedB = await waitFor( diff --git a/apps/contrail-e2e/tests/ingest-roundtrip.test.ts b/apps/contrail-e2e/tests/ingest-roundtrip.test.ts index b597245..e9b67ce 100644 --- a/apps/contrail-e2e/tests/ingest-roundtrip.test.ts +++ b/apps/contrail-e2e/tests/ingest-roundtrip.test.ts @@ -19,7 +19,7 @@ import { describe, it, expect, beforeAll, afterAll } from "vitest"; import pg from "pg"; import { CredentialManager, Client } from "@atcute/client"; import "@atcute/atproto"; -import { Contrail } from "@atmo-dev/contrail"; +import { Contrail, runPersistent } from "@atmo-dev/contrail"; import { createHandler } from "@atmo-dev/contrail/server"; import { createPostgresDatabase } from "@atmo-dev/contrail/postgres"; import { config as baseConfig } from "../config"; @@ -64,11 +64,8 @@ describe("ingest roundtrip (devnet PDS → Jetstream → Contrail)", () => { await contrail.init(); handle = createHandler(contrail); - // In-process ingester via Contrail so the config is resolved (raw - // runPersistent expects a resolved config — grouped counts silently - // don't update otherwise). ingestController = new AbortController(); - ingestPromise = contrail.runPersistent({ + ingestPromise = runPersistent(db, baseConfig, { batchSize: 50, flushIntervalMs: 500, signal: ingestController.signal, diff --git a/packages/contrail/src/core/persistent.ts b/packages/contrail/src/core/persistent.ts index 4dade65..c1e1c18 100644 --- a/packages/contrail/src/core/persistent.ts +++ b/packages/contrail/src/core/persistent.ts @@ -1,6 +1,6 @@ import type { JetstreamSubscription } from "@atcute/jetstream"; -import type { ContrailConfig, IngestEvent, Database, Logger } from "./types"; -import { getCollectionNsids, getDependentNsids, DEFAULT_FEED_MAX_ITEMS } from "./types"; +import type { ContrailConfig, IngestEvent, Database, Logger, ResolvedContrailConfig } from "./types"; +import { getCollectionNsids, getDependentNsids, DEFAULT_FEED_MAX_ITEMS, resolveConfig } from "./types"; import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; import { refreshStaleIdentities } from "./identity"; import { createIngestState } from "./jetstream"; @@ -29,6 +29,13 @@ export async function runPersistent( config: ContrailConfig, options?: PersistentIngestOptions, ): Promise { + // Internals (applyEvents, count updates, query planning) read `_resolved` + // and silently skip features when it's missing. The Contrail class resolves + // in its constructor; callers using this raw export must also get a resolved + // config, so do it defensively here. resolveConfig is idempotent. + if (!(config as ResolvedContrailConfig)._resolved) { + config = resolveConfig(config); + } const log = getLogger(config, options); const batchSize = options?.batchSize ?? 50; const flushIntervalMs = options?.flushIntervalMs ?? 5_000; diff --git a/packages/contrail/tests/persistent.test.ts b/packages/contrail/tests/persistent.test.ts index 5fba923..f248cf0 100644 --- a/packages/contrail/tests/persistent.test.ts +++ b/packages/contrail/tests/persistent.test.ts @@ -1,8 +1,9 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; -import type { Database } from "../src/core/types"; -import { createTestDbWithSchema, TEST_CONFIG } from "./helpers"; +import type { ContrailConfig, Database } from "../src/core/types"; +import { createTestDb, createTestDbWithSchema, TEST_CONFIG } from "./helpers"; import { runPersistent } from "../src/core/persistent"; import { getLastCursor, queryRecords } from "../src/core/db/records"; +import { initSchema } from "../src/core/db/schema"; // Mock identity resolution to avoid network calls in tests vi.mock("../src/core/identity", () => ({ @@ -201,6 +202,96 @@ describe("runPersistent", () => { await promise; }); + it("resolves config internally when given an unresolved ContrailConfig", async () => { + // Regression test for the silent grouped-count bug: passing a raw + // ContrailConfig (no `_resolved`) used to let total counts update while + // grouped count columns stayed at 0. runPersistent must defensively + // resolve so consumers of the raw export get the same behavior as those + // going through the Contrail class wrapper. + const rawConfig: ContrailConfig = { + namespace: "com.example", + collections: { + event: { + collection: "community.lexicon.calendar.event", + relations: { + rsvps: { + collection: "rsvp", + groupBy: "status", + groups: { + going: "community.lexicon.calendar.rsvp#going", + notgoing: "community.lexicon.calendar.rsvp#notgoing", + }, + }, + }, + }, + rsvp: { + collection: "community.lexicon.calendar.rsvp", + references: { + event: { collection: "event", field: "subject.uri" }, + }, + }, + }, + }; + // Sanity: the raw object must NOT have _resolved — that's the whole point. + expect((rawConfig as any)._resolved).toBeUndefined(); + + const freshDb = createTestDb(); + await initSchema(freshDb, rawConfig); + + const eventUri = "at://did:plc:alice/community.lexicon.calendar.event/evt1"; + const events = [ + { + kind: "commit" as const, + did: "did:plc:alice", + time_us: 9000, + commit: { + collection: "community.lexicon.calendar.event", + operation: "create", + rkey: "evt1", + cid: "cidE", + record: { name: "E", startsAt: "2026-04-01T10:00:00Z", mode: "online" }, + }, + }, + { + kind: "commit" as const, + did: "did:plc:alice", + time_us: 9001, + commit: { + collection: "community.lexicon.calendar.rsvp", + operation: "create", + rkey: "r1", + cid: "cidR", + record: { + subject: { uri: eventUri, cid: "cidE" }, + status: "community.lexicon.calendar.rsvp#going", + }, + }, + }, + ]; + + const controller = new AbortController(); + const promise = runPersistent(freshDb, rawConfig, { + batchSize: 100, + flushIntervalMs: 50, + signal: controller.signal, + createSubscription: () => mockSubscription(events) as any, + }); + await new Promise((r) => setTimeout(r, 300)); + controller.abort(); + await promise; + + // Query the raw count columns directly — using queryRecords would hide + // the bug since it hydrates via the resolved config anyway. + const row = await freshDb + .prepare(`SELECT count_rsvp, count_rsvp_going FROM records_event WHERE uri = ?`) + .bind(eventUri) + .first<{ count_rsvp: number; count_rsvp_going: number }>(); + expect(row).not.toBeNull(); + expect(row!.count_rsvp).toBe(1); + // This is the assertion that would fail before the fix. + expect(row!.count_rsvp_going).toBe(1); + }); + it("skips non-commit events", async () => { const events = [ { kind: "identity" as const, did: "did:plc:someone", time_us: 4000 },