diff --git a/app/config.ts b/app/config.ts index 0879d4c..1f9e991 100644 --- a/app/config.ts +++ b/app/config.ts @@ -18,6 +18,7 @@ export const config: ContrailConfig = { collection: "community.lexicon.calendar.rsvp", groupBy: "status", count: true, + countDistinct: "did", groups: { interested: "community.lexicon.calendar.rsvp#interested", going: "community.lexicon.calendar.rsvp#going", diff --git a/examples/cloudflare-workers/README.md b/examples/cloudflare-workers/README.md index 64c8602..9a0e294 100644 --- a/examples/cloudflare-workers/README.md +++ b/examples/cloudflare-workers/README.md @@ -18,9 +18,11 @@ npx wrangler d1 create contrail Copy the `database_id` from the output into `wrangler.jsonc`. +> **Note:** The `contrail` dependency installs from GitHub. This may take a minute on first install since it also pulls in the AT Protocol client libraries. + ## Configure -Edit `config.ts` to define your collections, queryable fields, relations, and references. See the [Contrail README](../../README.md) for all options. +Edit `config.ts` to define your collections, queryable fields, relations, and references. See the [Contrail README](https://github.com/flo-bit/contrail) for all options. ## Develop diff --git a/src/contrail.ts b/src/contrail.ts index fd5a0b9..522e841 100644 --- a/src/contrail.ts +++ b/src/contrail.ts @@ -3,7 +3,8 @@ import { resolveConfig, validateConfig } from "./core/types"; import { initSchema } from "./core/db/schema"; import { queryRecords } from "./core/db/records"; import type { QueryOptions, SortOption } from "./core/db/records"; -import { runIngestCycle } from "./core/jetstream"; +import { runIngestCycle, createIngestState } from "./core/jetstream"; +import type { IngestState } from "./core/jetstream"; import { discoverDIDs, backfillAll } from "./core/backfill"; import type { BackfillAllOptions, BackfillProgress } from "./core/backfill"; import { processNotifyUris } from "./core/router/notify"; @@ -16,6 +17,7 @@ export interface ContrailOptions extends ContrailConfig { export class Contrail { readonly config: ResolvedContrailConfig; private _db?: Database; + private _ingestState: IngestState = createIngestState(); constructor(options: ContrailOptions) { const { db, ...configInput } = options; @@ -46,7 +48,7 @@ export class Contrail { /** Run one Jetstream ingestion cycle (catches up to present, then stops). */ async ingest(options?: { timeoutMs?: number }, db?: Database): Promise { - await runIngestCycle(this.getDb(db), this.config, options?.timeoutMs); + await runIngestCycle(this.getDb(db), this.config, options?.timeoutMs, this._ingestState); } /** Discover users from relays. Returns discovered DIDs. */ diff --git a/src/core/backfill.ts b/src/core/backfill.ts index 17cfb73..2514943 100644 --- a/src/core/backfill.ts +++ b/src/core/backfill.ts @@ -424,14 +424,21 @@ async function insertDiscoveredDIDs( ): Promise { if (dids.length === 0) return; - const stmt = db.prepare( - "INSERT INTO backfills (did, collection, completed) VALUES (?, ?, 0) ON CONFLICT DO NOTHING" - ); - - const batch = dids.map((did) => stmt.bind(did, collection)); - - for (let i = 0; i < batch.length; i += 50) { - await db.batch(batch.slice(i, i + 50)); + // Use multi-row INSERT to reduce the number of statements + const CHUNK_SIZE = 50; + for (let i = 0; i < dids.length; i += CHUNK_SIZE) { + const chunk = dids.slice(i, i + CHUNK_SIZE); + const placeholders = chunk.map(() => "(?, ?, 0)").join(", "); + const bindings: string[] = []; + for (const did of chunk) { + bindings.push(did, collection); + } + await db + .prepare( + `INSERT INTO backfills (did, collection, completed) VALUES ${placeholders} ON CONFLICT DO NOTHING` + ) + .bind(...bindings) + .run(); } } diff --git a/src/core/db/index.ts b/src/core/db/index.ts index 814f0be..7c07f0a 100644 --- a/src/core/db/index.ts +++ b/src/core/db/index.ts @@ -1,4 +1,4 @@ export { initSchema } from "./schema"; -export { getLastCursor, saveCursor, applyEvents, queryRecords, pruneFeedItems } from "./records"; -export type { QueryOptions, SortOption } from "./records"; +export { getLastCursor, saveCursor, applyEvents, lookupExistingRecords, queryRecords, pruneFeedItems } from "./records"; +export type { QueryOptions, SortOption, ExistingRecordInfo } from "./records"; export type { RecordSource } from "../types"; diff --git a/src/core/db/records.ts b/src/core/db/records.ts index e339abe..3b18834 100644 --- a/src/core/db/records.ts +++ b/src/core/db/records.ts @@ -35,82 +35,100 @@ function getInboundRelations( } /** - * Build statements that fully recount child records for affected parents. - * Instead of +1/-1, we SELECT COUNT(*) so the count is always accurate. - * This runs for create, update, and delete operations. + * Collect recount targets from a single event into a shared map. + * The map is keyed by `parentCollection:relationName:targetValue` to deduplicate + * across the entire batch — so 50 RSVPs to the same event produce one recount, not 50. */ -function buildCountStatements( - db: Database, +function collectCountTargets( event: IngestEvent, config: ContrailConfig, existingRecordJson: string | null, -): Statement[] { + targets: Map +): void { const inbound = getInboundRelations(config, event.collection); - if (inbound.length === 0) return []; + if (inbound.length === 0) return; const record = event.record ? JSON.parse(event.record) : null; const existingRecord = existingRecordJson ? JSON.parse(existingRecordJson) : null; - const statements: Statement[] = []; - const recountedTargets = new Set(); - for (const { parentCollection, relationName, rel } of inbound) { if (rel.count === false) continue; const field = getRelationField(rel); - const matchColumn = rel.match === "did" ? "did" : "uri"; - const childTable = recordsTableName(rel.collection); - const parentTable = recordsTableName(parentCollection); - // Collect target URIs/DIDs that need recounting (current + old if changed) - const targets: string[] = []; + const values: string[] = []; if (record) { const t = getNestedValue(record, field); - if (t) targets.push(t); + if (t) values.push(t); } if (existingRecord) { const t = getNestedValue(existingRecord, field); - if (t && !targets.includes(t)) targets.push(t); + if (t && !values.includes(t)) values.push(t); } - for (const targetValue of targets) { + for (const targetValue of values) { const key = `${parentCollection}:${relationName}:${targetValue}`; - if (recountedTargets.has(key)) continue; - recountedTargets.add(key); + if (!targets.has(key)) { + targets.set(key, { parentCollection, relationName, rel, targetValue }); + } + } + } +} - const setClauses: string[] = []; - const setBindings: (string | number)[] = []; +/** + * Build deduplicated count UPDATE statements from collected targets. + * One UPDATE per unique parent+relation+target, regardless of how many + * events in the batch affected that target. + */ +function buildBatchCountStatements( + db: Database, + config: ContrailConfig, + targets: Map +): Statement[] { + const statements: Statement[] = []; - // Total count - const totalCol = countColumnName(rel.collection); - setClauses.push( - `${totalCol} = (SELECT COUNT(*) FROM ${childTable} WHERE json_extract(record, '$.${field}') = ?)` - ); - setBindings.push(targetValue); - - // Grouped counts - if (rel.groupBy) { - const mapping = (config as ResolvedContrailConfig)._resolved?.relations[parentCollection]?.[relationName]; - if (mapping?.groups) { - for (const [, fullToken] of Object.entries(mapping.groups)) { - const groupCol = countColumnName(fullToken); - setClauses.push( - `${groupCol} = (SELECT COUNT(*) FROM ${childTable} WHERE json_extract(record, '$.${field}') = ? AND json_extract(record, '$.${rel.groupBy}') = ?)` - ); - setBindings.push(targetValue, fullToken); - } + for (const { parentCollection, relationName, rel, targetValue } of targets.values()) { + const field = getRelationField(rel); + const matchColumn = rel.match === "did" ? "did" : "uri"; + const childTable = recordsTableName(rel.collection); + const parentTable = recordsTableName(parentCollection); + + const setClauses: string[] = []; + const setBindings: (string | number)[] = []; + + const countExpr = rel.countDistinct + ? `COUNT(DISTINCT ${rel.countDistinct})` + : "COUNT(*)"; + + // Total count + const totalCol = countColumnName(rel.collection); + setClauses.push( + `${totalCol} = (SELECT ${countExpr} FROM ${childTable} WHERE json_extract(record, '$.${field}') = ?)` + ); + setBindings.push(targetValue); + + // Grouped counts + if (rel.groupBy) { + const mapping = (config as ResolvedContrailConfig)._resolved?.relations[parentCollection]?.[relationName]; + if (mapping?.groups) { + for (const [, fullToken] of Object.entries(mapping.groups)) { + const groupCol = countColumnName(fullToken); + setClauses.push( + `${groupCol} = (SELECT ${countExpr} FROM ${childTable} WHERE json_extract(record, '$.${field}') = ? AND json_extract(record, '$.${rel.groupBy}') = ?)` + ); + setBindings.push(targetValue, fullToken); } } + } - if (setClauses.length > 0) { - statements.push( - db - .prepare( - `UPDATE ${parentTable} SET ${setClauses.join(", ")} WHERE ${matchColumn} = ?` - ) - .bind(...setBindings, targetValue) - ); - } + if (setClauses.length > 0) { + statements.push( + db + .prepare( + `UPDATE ${parentTable} SET ${setClauses.join(", ")} WHERE ${matchColumn} = ?` + ) + .bind(...setBindings, targetValue) + ); } } @@ -122,7 +140,8 @@ function buildCountStatements( function buildFtsStatements( db: Database, event: IngestEvent, - config: ContrailConfig + config: ContrailConfig, + existingMap: Map ): Statement[] { const colConfig = config.collections[event.collection]; if (!colConfig) return []; @@ -142,8 +161,10 @@ function buildFtsStatements( const content = buildFtsContent(record, fields); if (!content) return []; - // Delete-then-insert handles both create and update - stmts.push(db.prepare(`DELETE FROM ${table} WHERE uri = ?`).bind(event.uri)); + // Only delete existing FTS row if this is an update (record already existed) + if (existingMap.has(event.uri)) { + stmts.push(db.prepare(`DELETE FROM ${table} WHERE uri = ?`).bind(event.uri)); + } stmts.push( db.prepare(`INSERT INTO ${table} (uri, content) VALUES (?, ?)`).bind(event.uri, content) ); @@ -277,56 +298,97 @@ export async function saveCursor( .run(); } +// --- Existing record lookup --- + +export interface ExistingRecordInfo { + cid: string | null; + record: string | null; +} + +/** + * Look up existing records for a set of events, grouped by collection. + * Returns a map of uri → { cid, record }. + * When includeRecord is false, record will always be null (saves reading large blobs). + */ +export async function lookupExistingRecords( + db: Database, + events: { uri: string; collection: string }[], + includeRecord: boolean = true +): Promise> { + const result = new Map(); + if (events.length === 0) return result; + + const byCollection = new Map(); + for (const e of events) { + const uris = byCollection.get(e.collection) ?? []; + uris.push(e.uri); + byCollection.set(e.collection, uris); + } + + const selectCols = includeRecord ? "uri, cid, record" : "uri, cid"; + for (const [collection, uris] of byCollection) { + const table = recordsTableName(collection); + for (let i = 0; i < uris.length; i += 50) { + const chunk = uris.slice(i, i + 50); + const placeholders = chunk.map(() => "?").join(","); + const rows = await db + .prepare(`SELECT ${selectCols} FROM ${table} WHERE uri IN (${placeholders})`) + .bind(...chunk) + .all<{ uri: string; cid: string | null; record?: string | null }>(); + for (const row of rows.results ?? []) { + result.set(row.uri, { + cid: row.cid, + record: includeRecord ? (row.record ?? null) : null, + }); + } + } + } + + return result; +} + // --- Events --- export async function applyEvents( db: Database, events: IngestEvent[], config?: ContrailConfig, - options?: { skipReplayDetection?: boolean; skipFeedFanout?: boolean } + options?: { + skipReplayDetection?: boolean; + skipFeedFanout?: boolean; + /** Pre-fetched existing records — skips the internal lookup when provided */ + existing?: Map; + } ): Promise { if (events.length === 0) return; - // Look up existing records for replay detection and count recounts. - const existingCids = new Map(); - const existingRecords = new Map(); const followCollections = config ? getFeedFollowCollections(config) : []; const hasCountingRelations = config ? Object.values(config.collections).some(c => Object.values(c.relations ?? {}).some(r => r.count !== false) ) : false; const needRecordContent = followCollections.length > 0 || hasCountingRelations; - if (config && !options?.skipReplayDetection) { - // Group events by collection to query the correct tables - const byCollection = new Map(); - for (const e of events) { - const uris = byCollection.get(e.collection) ?? []; - uris.push(e.uri); - byCollection.set(e.collection, uris); - } - - const selectCols = needRecordContent ? "uri, cid, record" : "uri, cid"; - for (const [collection, uris] of byCollection) { - const table = recordsTableName(collection); - for (let i = 0; i < uris.length; i += 50) { - const chunk = uris.slice(i, i + 50); - const placeholders = chunk.map(() => "?").join(","); - const rows = await db - .prepare(`SELECT ${selectCols} FROM ${table} WHERE uri IN (${placeholders})`) - .bind(...chunk) - .all<{ uri: string; cid: string | null; record?: string | null }>(); - for (const row of rows.results ?? []) { - existingCids.set(row.uri, row.cid); - if (needRecordContent && row.record) { - existingRecords.set(row.uri, row.record); - } - } - } - } + // Use pre-fetched data or look up existing records + let existingMap: Map; + if (options?.existing) { + existingMap = options.existing; + } else if (config && !options?.skipReplayDetection) { + existingMap = await lookupExistingRecords(db, events, needRecordContent); + } else { + existingMap = new Map(); } const batch: Statement[] = []; + // Build a record-content map for feed statements (needs string values) + const existingRecordStrings = new Map(); + for (const [uri, info] of existingMap) { + existingRecordStrings.set(uri, info.record); + } + + // Collect all count recount targets across the batch, deduplicated + const countTargets = new Map(); + for (const e of events) { const table = recordsTableName(e.collection); @@ -349,24 +411,29 @@ export async function applyEvents( } if (config) { - // Recount is idempotent — always run it for create/update/delete. - const existingRecordJson = existingRecords.get(e.uri) ?? null; - batch.push(...buildCountStatements(db, e, config, existingRecordJson)); + // Collect count targets (deduplicated across the whole batch) + const existingRecordJson = existingMap.get(e.uri)?.record ?? null; + collectCountTargets(e, config, existingRecordJson, countTargets); // Feed fanout still needs replay detection - const existing = existingCids.get(e.uri); + const existingInfo = existingMap.get(e.uri); const isReplay = e.operation === "delete" - ? existing === undefined - : existing === e.cid; + ? existingInfo === undefined + : existingInfo?.cid === e.cid; if (!isReplay && !options?.skipFeedFanout) { - batch.push(...buildFeedStatements(db, e, config, existingRecords)); + batch.push(...buildFeedStatements(db, e, config, existingRecordStrings)); } - batch.push(...buildFtsStatements(db, e, config)); + batch.push(...buildFtsStatements(db, e, config, existingMap)); } } + // Build deduplicated count statements — one UPDATE per unique target + if (config) { + batch.push(...buildBatchCountStatements(db, config, countTargets)); + } + await db.batch(batch); } @@ -447,15 +514,25 @@ export async function queryRecords( // Cursor = AT URI of last seen record. Look it up to get keyset values. if (cursor) { + // Only select the columns needed for cursor pagination + let cursorSelect: string; + if (sort?.recordField) { + cursorSelect = `time_us, json_extract(record, '$.${sort.recordField}') as sort_value`; + } else if (sort?.countType) { + const sortCol = countColumnName(sort.countType); + cursorSelect = `time_us, ${sortCol}`; + } else { + cursorSelect = "time_us"; + } + const cursorRow = await db - .prepare(`SELECT * FROM ${table} WHERE uri = ?`) + .prepare(`SELECT ${cursorSelect} FROM ${table} WHERE uri = ?`) .bind(cursor) .first(); if (cursorRow) { if (sort?.recordField) { - const cursorRecord = cursorRow.record ? JSON.parse(cursorRow.record) : null; - const sortValue = cursorRecord ? getNestedValue(cursorRecord, sort.recordField) : null; + const sortValue = cursorRow.sort_value; const field = `json_extract(r.record, '$.${sort.recordField}')`; const cmp = sort.direction === "desc" ? "<" : ">"; conditions.push(`(${field} ${cmp} ? OR (${field} = ? AND r.time_us < ?))`); diff --git a/src/core/jetstream.ts b/src/core/jetstream.ts index 7d6261a..9794c86 100644 --- a/src/core/jetstream.ts +++ b/src/core/jetstream.ts @@ -5,13 +5,19 @@ import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } fr import { refreshStaleIdentities } from "./identity"; const BATCH_SIZE = 50; - -// Cache state in memory across ingest cycles (survives within the same Worker isolate) -let cachedKnownDids: Set | undefined; -let schemaInitialized = false; -let lastFeedPruneMs = 0; const FEED_PRUNE_INTERVAL_MS = 60 * 60 * 1000; // 1 hour +/** Mutable state that persists across ingest cycles within the same process. */ +export interface IngestState { + cachedKnownDids?: Set; + schemaInitialized: boolean; + lastFeedPruneMs: number; +} + +export function createIngestState(): IngestState { + return { schemaInitialized: false, lastFeedPruneMs: 0 }; +} + function getLogger(config: ContrailConfig): Logger { return config.logger ?? console; } @@ -98,13 +104,15 @@ export async function ingestEvents( export async function runIngestCycle( db: Database, config: ContrailConfig, - timeoutMs: number = 25_000 + timeoutMs: number = 25_000, + state?: IngestState ): Promise { const log = getLogger(config); + const s = state ?? createIngestState(); - if (!schemaInitialized) { + if (!s.schemaInitialized) { await initSchema(db, config); - schemaInitialized = true; + s.schemaInitialized = true; } const cursor = await getLastCursor(db); @@ -119,15 +127,15 @@ export async function runIngestCycle( let knownDids: Set | undefined; if (dependentCollections.length > 0) { - if (cachedKnownDids) { - knownDids = cachedKnownDids; + if (s.cachedKnownDids) { + knownDids = s.cachedKnownDids; log.log(`Using cached known DIDs (${knownDids.size} users)`); } else { const result = await db .prepare("SELECT did FROM identities") .all<{ did: string }>(); knownDids = new Set((result.results ?? []).map((r) => r.did)); - cachedKnownDids = knownDids; + s.cachedKnownDids = knownDids; log.log(`Loaded ${knownDids.size} known DIDs from database`); } } @@ -162,13 +170,13 @@ export async function runIngestCycle( } // Prune feed items hourly - if (config.feeds && Date.now() - lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) { + if (config.feeds && Date.now() - s.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) { const maxItems = Math.max( ...Object.values(config.feeds).map((f) => f.maxItems ?? DEFAULT_FEED_MAX_ITEMS) ); const pruned = await pruneFeedItems(db, maxItems); if (pruned > 0) log.log(`Pruned ${pruned} old feed items`); - lastFeedPruneMs = Date.now(); + s.lastFeedPruneMs = Date.now(); } log.log(`Ingestion complete. Stored ${events.length} events.`); diff --git a/src/core/router/collection.ts b/src/core/router/collection.ts index 6766099..4dbd08f 100644 --- a/src/core/router/collection.ts +++ b/src/core/router/collection.ts @@ -1,5 +1,5 @@ import type { Hono } from "hono"; -import type { ContrailConfig, ResolvedContrailConfig, Database, RecordRow, QueryableField, RecordSource } from "../types"; +import type { ContrailConfig, ResolvedContrailConfig, Database, RecordRow, QueryableField, RecordSource, RelationConfig } from "../types"; import { getCollectionNames, countColumnName, recordsTableName } from "../types"; import { queryRecords } from "../db"; import type { SortOption } from "../db/records"; @@ -29,13 +29,16 @@ export async function runPipeline( const cursor = params.get("cursor") || undefined; const actor = params.get("actor") || params.get("did") || undefined; const wantProfiles = params.get("profiles") === "true"; + const wantBackfill = params.get("backfill") === "true"; let did: string | undefined; if (actor) { const resolved = await resolveActor(db, actor); if (!resolved) throw new Error("Could not resolve actor"); did = resolved; - await backfillUser(db, did, collection, Date.now() + 10_000, config); + if (wantBackfill) { + await backfillUser(db, did, collection, Date.now() + 10_000, config); + } } const filters: Record = {}; @@ -192,10 +195,13 @@ export function registerCollectionRoutes( const relations = colConfig.relations ?? {}; const references = colConfig.references ?? {}; + const relMap = (config as ResolvedContrailConfig)._resolved?.relations[collection] ?? {}; const table = recordsTableName(collection); + const countCols = getRelationCountColumns(relations, relMap); + const selectCols = `uri, did, rkey, cid, record, time_us, indexed_at${countCols.length > 0 ? ", " + countCols.map(c => c.column).join(", ") : ""}`; const row = await db - .prepare(`SELECT * FROM ${table} WHERE uri = ?`) + .prepare(`SELECT ${selectCols} FROM ${table} WHERE uri = ?`) .bind(uri) .first(); @@ -274,6 +280,24 @@ export function registerCollectionRoutes( } } +function getRelationCountColumns( + relations: Record, + relMap: Record +): { column: string }[] { + const cols: { column: string }[] = []; + for (const [relName, rel] of Object.entries(relations)) { + if (rel.count === false) continue; + cols.push({ column: countColumnName(rel.collection) }); + const mapping = relMap[relName]; + if (mapping?.groups) { + for (const [, fullToken] of Object.entries(mapping.groups as Record)) { + cols.push({ column: countColumnName(fullToken) }); + } + } + } + return cols; +} + function extractCounts( row: any, relations: Record diff --git a/src/core/router/notify.ts b/src/core/router/notify.ts index 92164f9..1199003 100644 --- a/src/core/router/notify.ts +++ b/src/core/router/notify.ts @@ -1,7 +1,6 @@ import type { Hono } from "hono"; import type { Database, ContrailConfig, IngestEvent } from "../types"; -import { recordsTableName } from "../types"; -import { applyEvents } from "../db/records"; +import { applyEvents, lookupExistingRecords } from "../db/records"; import { getPDS } from "../client"; import type { Did } from "@atcute/lexicons"; @@ -53,19 +52,29 @@ export async function processNotifyUris( const events: IngestEvent[] = []; const errors: string[] = []; + // Validate and parse all URIs first + const validUris: { uri: string; parsed: { did: string; collection: string; rkey: string } }[] = []; for (const uri of uris) { const parsed = parseAtUri(uri); if (!parsed) { errors.push(`invalid AT URI: ${uri}`); continue; } - - // Only accept collections we're tracking if (!config.collections[parsed.collection]) { errors.push(`collection not tracked: ${parsed.collection}`); continue; } + validUris.push({ uri, parsed }); + } + + // Single batch lookup for all existing records (cid + record in one query) + const existing = await lookupExistingRecords( + db, + validUris.map(({ uri, parsed }) => ({ uri, collection: parsed.collection })), + true + ); + for (const { uri, parsed } of validUris) { const pds = await getPDS(parsed.did as Did, db); if (!pds) { errors.push(`could not resolve PDS for ${parsed.did}`); @@ -80,16 +89,10 @@ export async function processNotifyUris( ); const now = Date.now() * 1000; // microseconds - - // Check if this record already exists locally - const table = recordsTableName(parsed.collection); - const existing = await db - .prepare(`SELECT cid FROM ${table} WHERE uri = ?`) - .bind(uri) - .first<{ cid: string | null }>(); + const existingInfo = existing.get(uri); if (result) { - if (existing?.cid === result.cid) { + if (existingInfo?.cid === result.cid) { // Same CID — nothing changed continue; } @@ -99,19 +102,14 @@ export async function processNotifyUris( did: parsed.did, collection: parsed.collection, rkey: parsed.rkey, - operation: existing ? "update" : "create", + operation: existingInfo ? "update" : "create", cid: result.cid, record: JSON.stringify(result.value), time_us: now, indexed_at: now, }); - } else if (existing) { + } else if (existingInfo) { // Record gone from PDS but exists locally — delete it. - const existingRecord = await db - .prepare(`SELECT record FROM ${table} WHERE uri = ?`) - .bind(uri) - .first<{ record: string | null }>(); - events.push({ uri, did: parsed.did, @@ -119,7 +117,7 @@ export async function processNotifyUris( rkey: parsed.rkey, operation: "delete", cid: null, - record: existingRecord?.record ?? null, + record: existingInfo.record, time_us: now, indexed_at: now, }); @@ -127,7 +125,8 @@ export async function processNotifyUris( } if (events.length > 0) { - await applyEvents(db, events, config); + // Pass pre-fetched existing records so applyEvents skips re-querying + await applyEvents(db, events, config, { existing }); } return { diff --git a/src/core/types.ts b/src/core/types.ts index 1512e13..5f8e56d 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -24,6 +24,8 @@ export interface RelationConfig { groupBy?: string; /** Enable materialized count columns on the parent. Defaults to true. */ count?: boolean; + /** Count distinct values of a field (e.g. "did" for unique users) instead of total records. */ + countDistinct?: string; /** Pre-resolved group mappings: shortName → full token (e.g. { going: "community.lexicon.calendar.rsvp#going" }). Auto-computed from groupBy if omitted. */ groups?: Record; }