diff --git a/.changeset/record-filter-and-identity-events.md b/.changeset/record-filter-and-identity-events.md new file mode 100644 index 0000000..69925f7 --- /dev/null +++ b/.changeset/record-filter-and-identity-events.md @@ -0,0 +1,8 @@ +--- +"@atmo-dev/contrail": minor +--- + +Add per-collection `recordFilter` and apply Jetstream `#identity` handle changes during ingest. + +- `CollectionConfig.recordFilter?: (record) => boolean` runs against each create/update during ingest; returning false drops the record before it reaches the DB. Useful for narrowing high-volume collections to just the records you care about (e.g. only `app.bsky.feed.post` records mentioning a particular URL). Deletes are not filtered, so they still tear down any record the filter previously let through. Throws are caught, logged, and treated as drops. +- Jetstream `#identity` events (handle changes) now flow through to the `identities` table via a new `applyIdentityEvent` helper. UPDATE-only — unknown DIDs are no-ops so we don't materialize partial rows lacking PDS. diff --git a/packages/contrail/src/core/identity.ts b/packages/contrail/src/core/identity.ts index 05040fd..fe207b3 100644 --- a/packages/contrail/src/core/identity.ts +++ b/packages/contrail/src/core/identity.ts @@ -119,6 +119,24 @@ export async function resolveActor( return resolved.did; } +/** + * Apply a handle change from a Jetstream `#identity` event. + * + * UPDATE-only — does not create a row for unknown DIDs (we'd lack PDS, and + * partial rows confuse the rest of the pipeline). PDS column is left + * untouched; it gets refreshed lazily via `getPDS` / next slingshot resolve. + */ +export async function applyIdentityEvent( + db: Database, + did: string, + handle: string +): Promise { + await db + .prepare("UPDATE identities SET handle = ?, resolved_at = ? WHERE did = ?") + .bind(handle, Date.now(), did) + .run(); +} + export async function refreshStaleIdentities( db: Database, dids: string[] diff --git a/packages/contrail/src/core/jetstream.ts b/packages/contrail/src/core/jetstream.ts index 566c406..abc0287 100644 --- a/packages/contrail/src/core/jetstream.ts +++ b/packages/contrail/src/core/jetstream.ts @@ -7,7 +7,7 @@ import { buildFeedTargetCaps, } from "./types"; import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; -import { refreshStaleIdentities } from "./identity"; +import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; import { backfillFollowersFromConstellation } from "./constellation"; const BATCH_SIZE = 50; @@ -37,6 +37,7 @@ export async function ingestEvents( events: IngestEvent[]; lastCursor: number | null; newlyKnownDids: string[]; + identityUpdates: Map; }> { const log = getLogger(config); const startTimeUs = Date.now() * 1000; @@ -56,6 +57,7 @@ export async function ingestEvents( const seenUris = new Map(); // uri -> time_us of first occurrence const duplicateUris: string[] = []; const newlyKnownDids = new Set(); + const identityUpdates = new Map(); const subscription = new JetstreamSubscription({ url: urls, @@ -86,6 +88,9 @@ export async function ingestEvents( const uri = `at://${event.did}/${commit.collection}/${commit.rkey}`; + const short = shortNameForNsid(config, commit.collection); + const collectionCfg = short ? config.collections[short] : undefined; + if (dependentCollections.has(commit.collection) && knownDids) { if (!knownDids.has(event.did)) { filteredUnknownDid++; @@ -96,10 +101,7 @@ export async function ingestEvents( // pointing at a `subject` DID), drop records whose subject isn't a // DID we care about. Trims network-wide social graph to the // subjects our discoverable users overlap with. - const short = shortNameForNsid(config, commit.collection); - const subjectField = short - ? config.collections[short]?.subjectField - : undefined; + const subjectField = collectionCfg?.subjectField; if (subjectField && commit.operation !== "delete") { const subj = (commit.record as Record | undefined)?.[ subjectField @@ -110,6 +112,17 @@ export async function ingestEvents( } } + if (collectionCfg?.recordFilter && commit.operation !== "delete") { + const rec = commit.record as Record | undefined; + let keep = false; + try { + keep = !!(rec && collectionCfg.recordFilter(rec)); + } catch (err) { + log.warn(`[ingest] recordFilter threw for ${uri}: ${err}`); + } + if (!keep) continue; + } + const prev = seenUris.get(uri); if (prev !== undefined) { duplicateUris.push(uri); @@ -147,6 +160,8 @@ export async function ingestEvents( newlyKnownDids.add(event.did); } } + } else if (event.kind === "identity") { + identityUpdates.set(event.did, event.identity.handle); } if (event.time_us >= startTimeUs) { @@ -202,7 +217,7 @@ export async function ingestEvents( ); } - return { events: collected, lastCursor, newlyKnownDids: [...newlyKnownDids] }; + return { events: collected, lastCursor, newlyKnownDids: [...newlyKnownDids], identityUpdates }; } // Run a full ingest cycle: init schema, load cursor, ingest, apply, save cursor @@ -250,7 +265,7 @@ export async function runIngestCycle( } } - const { events, lastCursor, newlyKnownDids } = await ingestEvents( + const { events, lastCursor, newlyKnownDids, identityUpdates } = await ingestEvents( config, cursor, timeoutMs, @@ -275,6 +290,19 @@ export async function runIngestCycle( await applyEvents(db, batch, config, { pubsub }); } + // Apply handle changes from #identity events. UPDATE-only, so unknown + // DIDs are no-ops — we don't want to create partial rows lacking PDS. + if (identityUpdates.size > 0) { + for (const [did, handle] of identityUpdates) { + try { + await applyIdentityEvent(db, did, handle); + } catch (err) { + log.warn(`[ingest] identity update failed for ${did}: ${err}`); + } + } + log.log(`[ingest] applied ${identityUpdates.size} identity event(s)`); + } + // Refresh stale/missing identities for DIDs in this batch const uniqueDids = [...new Set(events.map((e) => e.did))]; if (uniqueDids.length > 0) { diff --git a/packages/contrail/src/core/persistent.ts b/packages/contrail/src/core/persistent.ts index e53740c..882c9b8 100644 --- a/packages/contrail/src/core/persistent.ts +++ b/packages/contrail/src/core/persistent.ts @@ -8,7 +8,7 @@ import { shortNameForNsid, } from "./types"; import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db"; -import { refreshStaleIdentities } from "./identity"; +import { refreshStaleIdentities, applyIdentityEvent } from "./identity"; import { backfillFollowersFromConstellation } from "./constellation"; import { createIngestState } from "./jetstream"; import type { IngestState } from "./jetstream"; @@ -230,13 +230,13 @@ async function streamAndFlush( if (event.kind === "commit") { const { commit } = event; + const short = shortNameForNsid(config, commit.collection); + const collectionCfg = short ? config.collections[short] : undefined; + if (dependentCollections.has(commit.collection) && knownDids) { if (!knownDids.has(event.did)) continue; // Subject filter: skip records whose subject DID isn't known. - const short = shortNameForNsid(config, commit.collection); - const subjectField = short - ? config.collections[short]?.subjectField - : undefined; + const subjectField = collectionCfg?.subjectField; if (subjectField && commit.operation !== "delete") { const subj = (commit.record as Record | undefined)?.[ subjectField @@ -245,6 +245,17 @@ async function streamAndFlush( } } + if (collectionCfg?.recordFilter && commit.operation !== "delete") { + const rec = commit.record as Record | undefined; + let keep = false; + try { + keep = !!(rec && collectionCfg.recordFilter(rec)); + } catch (err) { + log.warn(`recordFilter threw for ${commit.collection}/${commit.rkey}: ${err}`); + } + if (!keep) continue; + } + const now = Date.now(); const uri = `at://${event.did}/${commit.collection}/${commit.rkey}`; @@ -266,6 +277,12 @@ async function streamAndFlush( opts.newlyKnownDids?.add(event.did); } } + } else if (event.kind === "identity") { + try { + await applyIdentityEvent(db, event.did, event.identity.handle); + } catch (err) { + log.warn(`Identity update failed for ${event.did}: ${err}`); + } } if (buffer.length >= batchSize) { diff --git a/packages/contrail/src/core/types.ts b/packages/contrail/src/core/types.ts index 48cea2a..c59f116 100644 --- a/packages/contrail/src/core/types.ts +++ b/packages/contrail/src/core/types.ts @@ -153,6 +153,14 @@ export interface CollectionConfig { * knownDids — useful for trimming network-wide social graphs to the * subjects we care about. */ subjectField?: string; + /** Per-record predicate run during ingest. Returning false drops the + * record before it hits the buffer / DB. Runs only for create/update; + * deletes always pass through (the delete may target a record that *did* + * pass an earlier version of the filter). Thrown errors are caught, + * logged, and treated as "drop". Note: Jetstream filters only by + * `wantedCollections`, so non-matching records still travel over the wire + * — this trims what gets persisted, not bandwidth. */ + recordFilter?: (record: Record) => boolean; } export interface ProfileConfig { diff --git a/packages/contrail/tests/persistent.test.ts b/packages/contrail/tests/persistent.test.ts index f248cf0..c48c7db 100644 --- a/packages/contrail/tests/persistent.test.ts +++ b/packages/contrail/tests/persistent.test.ts @@ -1,13 +1,16 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; import type { ContrailConfig, Database } from "../src/core/types"; +import { resolveConfig } 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 +const applyIdentityEventMock = vi.fn().mockResolvedValue(undefined); vi.mock("../src/core/identity", () => ({ refreshStaleIdentities: vi.fn().mockResolvedValue(undefined), + applyIdentityEvent: (...args: unknown[]) => applyIdentityEventMock(...args), })); let db: Database; @@ -292,9 +295,79 @@ describe("runPersistent", () => { expect(row!.count_rsvp_going).toBe(1); }); + it("drops records that fail a collection's recordFilter", async () => { + // Filter accepts only events whose `name` contains "keep". The other + // events have well-formed records but should never reach the DB. + const filterConfig = resolveConfig({ + namespace: "com.example", + collections: { + event: { + collection: "community.lexicon.calendar.event", + recordFilter: (r) => + typeof r.name === "string" && r.name.includes("keep"), + }, + }, + }); + + const freshDb = createTestDb(); + await initSchema(freshDb, filterConfig); + + const events = [ + { + kind: "commit" as const, + did: "did:plc:a", + time_us: 7000, + commit: { + collection: "community.lexicon.calendar.event", + operation: "create", + rkey: "drop1", + cid: "c1", + record: { name: "drop me", startsAt: "2026-04-01T10:00:00Z", mode: "online" }, + }, + }, + { + kind: "commit" as const, + did: "did:plc:b", + time_us: 7001, + commit: { + collection: "community.lexicon.calendar.event", + operation: "create", + rkey: "keep1", + cid: "c2", + record: { name: "keep this", startsAt: "2026-04-01T10:00:00Z", mode: "online" }, + }, + }, + ]; + + const controller = new AbortController(); + const promise = runPersistent(freshDb, filterConfig, { + batchSize: 100, + flushIntervalMs: 50, + signal: controller.signal, + createSubscription: () => mockSubscription(events) as any, + }); + + await new Promise((r) => setTimeout(r, 200)); + controller.abort(); + await promise; + + const result = await queryRecords(freshDb, filterConfig, { + collection: "community.lexicon.calendar.event", + limit: 100, + }); + expect(result.records.length).toBe(1); + expect(result.records[0]!.uri).toContain("/keep1"); + }); + it("skips non-commit events", async () => { + applyIdentityEventMock.mockClear(); const events = [ - { kind: "identity" as const, did: "did:plc:someone", time_us: 4000 }, + { + kind: "identity" as const, + did: "did:plc:someone", + time_us: 4000, + identity: { did: "did:plc:someone", handle: "newhandle.test", seq: 1, time: "2026-04-01T10:00:00Z" }, + }, { kind: "commit" as const, did: "did:plc:real", @@ -327,5 +400,12 @@ describe("runPersistent", () => { limit: 100, }); expect(result.records.length).toBe(1); + // Identity events update the identities table even though they don't + // produce records. + expect(applyIdentityEventMock).toHaveBeenCalledWith( + expect.anything(), + "did:plc:someone", + "newhandle.test" + ); }); });