From a08e0016b539ae01b8982a30ea92004472a99703 Mon Sep 17 00:00:00 2001 From: Tom Scanlan Date: Fri, 19 Jun 2026 14:44:34 -0400 Subject: [PATCH 1/4] feat(search): feed Meili from backfill/refresh and add D1 reindex tool Wire the Meili search sink into the contrail config that the CLI loads, so an operator-run backfill/refresh (not just live ingest) populates the search index. Gated on SEARCH_SINK_URL so `pnpm generate` and the Worker (which attaches its own sink) stay unaffected. Add reindexEventsToSink plus `pnpm meili:reindex[:remote]`, which replays every records_event row through the sink (SELECT only, zero D1 writes, no PDS calls). This closes coverage for events whose authoring DID finished backfill before the sink existed, and gives a re-synced install full search coverage on demand. --- apps/web/package.json | 2 + apps/web/src/lib/contrail.config.test.ts | 74 ++++++++++++- apps/web/src/lib/contrail.config.ts | 16 +++ .../server/meili-sink.integration.test.ts | 43 ++++++++ apps/web/src/lib/search/server/reindex-cli.ts | 39 +++++++ .../web/src/lib/search/server/reindex.test.ts | 104 ++++++++++++++++++ apps/web/src/lib/search/server/reindex.ts | 77 +++++++++++++ 7 files changed, 354 insertions(+), 1 deletion(-) create mode 100644 apps/web/src/lib/search/server/reindex-cli.ts create mode 100644 apps/web/src/lib/search/server/reindex.test.ts create mode 100644 apps/web/src/lib/search/server/reindex.ts diff --git a/apps/web/package.json b/apps/web/package.json index 1ced8ee..ec05b5f 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -12,6 +12,8 @@ "prepare": "svelte-kit sync || echo ''", "backfill": "contrail backfill", "backfill:remote": "contrail backfill --remote", + "meili:reindex": "npx tsx src/lib/search/server/reindex-cli.ts", + "meili:reindex:remote": "npx tsx src/lib/search/server/reindex-cli.ts --remote", "check": "svelte-kit sync && svelte-check --tsconfig ./tsconfig.json", "check:watch": "svelte-kit sync && svelte-check --tsconfig ./tsconfig.json --watch", "test": "vitest run", diff --git a/apps/web/src/lib/contrail.config.test.ts b/apps/web/src/lib/contrail.config.test.ts index afc732f..326a4ba 100644 --- a/apps/web/src/lib/contrail.config.test.ts +++ b/apps/web/src/lib/contrail.config.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect } from 'vitest'; +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; import { config } from './contrail.config'; // The pipelineQuery handlers are pure (db unused for these), so they can be @@ -44,3 +44,75 @@ describe('listDiscoverableByUris pipelineQuery', () => { expect(source.params).toHaveLength(60); }); }); + +// The `contrail` CLI (`pnpm backfill` / `contrail refresh`) loads THIS config and +// builds `new Contrail(config)`, then fires `config.sinks` from applyEvents with +// phase:'backfill' (contrail-appview). The runtime Worker attaches its own sink in +// $lib/contrail/index.ts, so search stays current for live ingest — but the CLI +// only populates Meili if the config it loads carries the sink. Wire it here, +// gated on SEARCH_SINK_URL so it's active for an operator-run backfill and absent +// for `pnpm generate` / the Worker (which overrides sinks anyway). +describe('Meili search sink wiring (CLI backfill/refresh)', () => { + const ENV_KEYS = ['SEARCH_SINK_URL', 'SEARCH_SINK_API_KEY', 'SEARCH_INDEX'] as const; + const saved: Record = {}; + + beforeEach(() => { + for (const k of ENV_KEYS) { + saved[k] = process.env[k]; + delete process.env[k]; + } + vi.resetModules(); + }); + + afterEach(() => { + for (const k of ENV_KEYS) { + if (saved[k] === undefined) delete process.env[k]; + else process.env[k] = saved[k]; + } + vi.unstubAllGlobals(); + vi.resetModules(); + }); + + it('carries a Meili sink that upserts a backfilled event when SEARCH_SINK_URL is configured', async () => { + process.env.SEARCH_SINK_URL = 'http://meili.local'; + process.env.SEARCH_SINK_API_KEY = 'admin-key'; + process.env.SEARCH_INDEX = 'events-test'; + + const calls: { url: string; method: string }[] = []; + vi.stubGlobal( + 'fetch', + vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + calls.push({ url: String(input), method: init?.method ?? 'GET' }); + return new Response(null, { status: 202 }); + }) + ); + + const { config: loaded } = await import('./contrail.config'); + expect(loaded.sinks && loaded.sinks.length).toBeTruthy(); + + await loaded.sinks![0].onRecords( + [ + { + kind: 'created', + uri: 'at://did:plc:alice/community.lexicon.calendar.event/backfilled', + did: 'did:plc:alice', + collection: 'community.lexicon.calendar.event', + rkey: 'backfilled', + cid: 'bafycid', + record: { name: 'Backfilled Jazz Night', startsAt: '2099-01-01T10:00:00Z' }, + time_us: 1 + } + ], + { phase: 'backfill' } + ); + + const put = calls.find((c) => c.method === 'PUT'); + expect(put).toBeDefined(); + expect(put!.url).toBe('http://meili.local/indexes/events-test/documents?primaryKey=id'); + }); + + it('omits sinks when SEARCH_SINK_URL is unset (generate / Worker stay sink-free here)', async () => { + const { config: loaded } = await import('./contrail.config'); + expect(loaded.sinks ?? []).toHaveLength(0); + }); +}); diff --git a/apps/web/src/lib/contrail.config.ts b/apps/web/src/lib/contrail.config.ts index d9da450..3c7cd2b 100644 --- a/apps/web/src/lib/contrail.config.ts +++ b/apps/web/src/lib/contrail.config.ts @@ -1,6 +1,21 @@ import type { ContrailConfig } from '@atmo-dev/contrail'; import { SPACE_TYPE } from './spaces/config'; import { MAX_HYDRATION_URIS } from './search/constants'; +import { createMeiliSink, meiliSinkBackendFromEnv } from './search/server/meili-sink'; + +// The `contrail` CLI (`pnpm backfill` / `contrail refresh`) loads this config and +// fires `config.sinks` from applyEvents on the backfill/refresh paths, so a +// freshly-set-up or re-synced installation gets full search coverage, not just +// whatever flowed through live ingest. The backend is read from process.env +// (SEARCH_SINK_URL / SEARCH_SINK_API_KEY / SEARCH_INDEX), injected at runtime by +// the operator (e.g. `op run`), so no secret is committed. Gated on +// SEARCH_SINK_URL: absent it, the sink stays off, which keeps `pnpm generate` +// sink-free. The runtime Worker (`$lib/contrail/index.ts`) does not use this: it +// overrides `sinks` with its own env-armed sink, so there is never a double feed. +const searchSinks: NonNullable = + typeof process !== 'undefined' && process.env?.SEARCH_SINK_URL + ? [createMeiliSink(() => meiliSinkBackendFromEnv(process.env))] + : []; // Events hidden from discovery (preferences.showInDiscovery === false) are // excluded; a missing field defaults to true so pre-existing records without @@ -33,6 +48,7 @@ export const config: ContrailConfig = { // blob uploads are declared as standalone `scope.repo(...)` / // `scope.blob(...)` entries in `atproto/settings.ts`, not here. }, + ...(searchSinks.length ? { sinks: searchSinks } : {}), collections: { event: { collection: 'community.lexicon.calendar.event', diff --git a/apps/web/src/lib/search/server/meili-sink.integration.test.ts b/apps/web/src/lib/search/server/meili-sink.integration.test.ts index d7d9a59..86e4ada 100644 --- a/apps/web/src/lib/search/server/meili-sink.integration.test.ts +++ b/apps/web/src/lib/search/server/meili-sink.integration.test.ts @@ -114,3 +114,46 @@ run('MeiliSink ↔ read client, live against real Meilisearch', () => { expect(text.hits.map((h) => h.uri)).not.toContain(uri); }); }); + +// The `contrail` CLI loads contrail.config.ts and fires `config.sinks` on the +// backfill path (phase:'backfill'). This proves the config the CLI actually loads +// carries a sink that lands a backfilled event in real Meili, reachable via the +// read path — the end-to-end half the mocked config unit test can't cover. +run('contrail.config sink populates Meili on backfill (CLI path)', () => { + const readBackend: SearchBackend = { url: URL!, apiKey: KEY, indexUid: INDEX }; + const uri = 'at://did:plc:alice/community.lexicon.calendar.event/backfilled-via-config'; + const savedEnv: Record = {}; + + beforeAll(async () => { + // Point the config-resolved sink at the test Meili, exactly as an operator + // would export these before running `pnpm backfill:remote`. + for (const k of ['SEARCH_SINK_URL', 'SEARCH_SINK_API_KEY', 'SEARCH_INDEX'] as const) { + savedEnv[k] = process.env[k]; + } + process.env.SEARCH_SINK_URL = URL; + process.env.SEARCH_SINK_API_KEY = KEY; + process.env.SEARCH_INDEX = INDEX; + await applyMeiliSettings({ url: URL!, apiKey: KEY, indexUid: INDEX }); + }); + + it('indexes a backfilled event through the config sink so the read path finds it', async () => { + const { config } = await import('../../contrail.config'); + expect(config.sinks && config.sinks.length).toBeTruthy(); + + await config.sinks![0].onRecords( + [created(uri, { name: 'Backfilled Jazz Night', description: 'via cli backfill', startsAt: FUTURE })], + { phase: 'backfill' } + ); + + const text = await eventually( + () => searchEvents(readBackend, { q: 'jazz', limit: 10, offset: 0 }), + (r) => r.hits.some((h) => h.uri === uri) + ); + expect(text.hits.map((h) => h.uri)).toContain(uri); + + for (const k of ['SEARCH_SINK_URL', 'SEARCH_SINK_API_KEY', 'SEARCH_INDEX'] as const) { + if (savedEnv[k] === undefined) delete process.env[k]; + else process.env[k] = savedEnv[k]; + } + }); +}); diff --git a/apps/web/src/lib/search/server/reindex-cli.ts b/apps/web/src/lib/search/server/reindex-cli.ts new file mode 100644 index 0000000..6aaaf4d --- /dev/null +++ b/apps/web/src/lib/search/server/reindex-cli.ts @@ -0,0 +1,39 @@ +// CLI entry for the D1 -> Meili event reindex. Mirrors `contrail backfill`: +// pnpm meili:reindex # local D1 binding +// pnpm meili:reindex:remote # --remote -> getPlatformProxy production binding +// +// The Meili backend is resolved from the same env the sink uses, injected by the +// operator (so no secret is committed): +// SEARCH_SINK_URL / SEARCH_SINK_API_KEY / SEARCH_INDEX +// Reads records_event (SELECT only) and feeds the config sink. Zero D1 writes. +import { getPlatformProxy } from 'wrangler'; +import { config } from '../../contrail.config'; +import { reindexEventsToSink, type ReindexDb } from './reindex'; + +const remote = process.argv.includes('--remote'); +const binding = 'DB'; + +const sink = config.sinks?.[0]; +if (!sink) { + console.error( + 'No search sink configured. Export SEARCH_SINK_URL and SEARCH_SINK_API_KEY (and optionally SEARCH_INDEX) before running.' + ); + process.exit(1); +} + +const { env, dispose } = await getPlatformProxy({ + environment: remote ? 'production' : undefined +}); +try { + const db = (env as Record)[binding] as ReindexDb | undefined; + if (!db) throw new Error(`No "${binding}" binding in wrangler env (${remote ? 'production' : 'default'}).`); + + const total = await reindexEventsToSink({ + db, + sink, + onProgress: (n) => console.log(`fed ${n} event rows -> Meili sink`) + }); + console.log(`done: ${total} event rows reindexed to Meili (zero D1 writes)`); +} finally { + await dispose(); +} diff --git a/apps/web/src/lib/search/server/reindex.test.ts b/apps/web/src/lib/search/server/reindex.test.ts new file mode 100644 index 0000000..1ca3ed1 --- /dev/null +++ b/apps/web/src/lib/search/server/reindex.test.ts @@ -0,0 +1,104 @@ +import { describe, it, expect } from 'vitest'; +import { reindexEventsToSink, type ReindexDb } from './reindex'; +import { EVENT_COLLECTION } from './meili-sink'; + +// A fake D1 that serves `rows` via LIMIT/OFFSET and records the bind args, so we +// can assert pagination without a real database. +function fakeDb(rows: Array>) { + const binds: Array<{ limit: number; offset: number }> = []; + const db: ReindexDb = { + prepare() { + return { + bind(limit: unknown, offset: unknown) { + const l = Number(limit); + const o = Number(offset); + binds.push({ limit: l, offset: o }); + return { + async all() { + return { results: rows.slice(o, o + l) }; + } + }; + } + }; + } + }; + return { db, binds }; +} + +function recordingSink() { + const batches: Array<{ records: Array>; ctx: { phase: string } }> = []; + const sink = { + async onRecords(records: Array>, ctx: { phase: string }) { + batches.push({ records, ctx }); + } + } as unknown as Parameters[0]['sink']; + return { sink, batches }; +} + +function row(uri: string, record: Record, cid: string | null = 'bafycid') { + return { + uri, + did: uri.split('/')[2], + rkey: uri.split('/').pop(), + cid, + record: JSON.stringify(record), + time_us: 1 + }; +} + +describe('reindexEventsToSink', () => { + it('feeds every event row to the sink as a parsed created RecordEvent (phase backfill)', async () => { + const { db } = fakeDb([ + row('at://did:plc:a/community.lexicon.calendar.event/1', { name: 'One' }), + row('at://did:plc:b/community.lexicon.calendar.event/2', { name: 'Two' }) + ]); + const { sink, batches } = recordingSink(); + + const total = await reindexEventsToSink({ db, sink }); + + expect(total).toBe(2); + expect(batches).toHaveLength(1); + expect(batches[0].ctx).toEqual({ phase: 'backfill' }); + const recs = batches[0].records; + expect(recs[0]).toMatchObject({ + kind: 'created', + uri: 'at://did:plc:a/community.lexicon.calendar.event/1', + did: 'did:plc:a', + collection: EVENT_COLLECTION, + rkey: '1', + cid: 'bafycid', + // record arrives parsed, not as the stored JSON string + record: { name: 'One' }, + time_us: 1 + }); + }); + + it('pages through D1 with LIMIT/OFFSET until a short page', async () => { + const { db, binds } = fakeDb([ + row('at://did:plc:a/community.lexicon.calendar.event/1', {}), + row('at://did:plc:a/community.lexicon.calendar.event/2', {}), + row('at://did:plc:a/community.lexicon.calendar.event/3', {}) + ]); + const { sink, batches } = recordingSink(); + + const total = await reindexEventsToSink({ db, sink, batchSize: 2 }); + + expect(total).toBe(3); + expect(binds).toEqual([ + { limit: 2, offset: 0 }, + { limit: 2, offset: 2 } + ]); + expect(batches.map((b) => b.records.length)).toEqual([2, 1]); + }); + + it('maps a null cid to an empty string', async () => { + const { db } = fakeDb([ + row('at://did:plc:a/community.lexicon.calendar.event/1', { name: 'X' }, null) + ]); + const { sink, batches } = recordingSink(); + + await reindexEventsToSink({ db, sink }); + + expect(batches[0].records[0].cid).toBe(''); + }); +}); diff --git a/apps/web/src/lib/search/server/reindex.ts b/apps/web/src/lib/search/server/reindex.ts new file mode 100644 index 0000000..d32b439 --- /dev/null +++ b/apps/web/src/lib/search/server/reindex.ts @@ -0,0 +1,77 @@ +// D1 -> Meili reindex: replays every stored event record through a contrail +// Sink WITHOUT touching any PDS or writing any D1 row. The live sink only sees +// records as they are ingested, so events whose authoring DID finished backfill +// before the sink existed (or before a Meili outage recovered) never reach the +// search index. This reads them straight from `records_event` (SELECT only) and +// feeds them to the sink, which applies the same discoverable filter as live +// ingest (discoverable -> upsert, hidden -> delete). Idempotent: re-running just +// re-upserts the same docs. +import { EVENT_COLLECTION } from './meili-sink'; +import type { ContrailConfig } from '@atmo-dev/contrail'; + +type Sink = NonNullable[number]; +type RecordEvent = Parameters[0][number]; + +/** The slice of the D1 client this reindex needs: a prepared, bound, paged + * SELECT. Kept minimal so tests can supply a fake without a real database. */ +export interface ReindexDb { + prepare(query: string): { + bind(...args: unknown[]): { + all(): Promise<{ results?: Array> }>; + }; + }; +} + +export interface ReindexOptions { + db: ReindexDb; + sink: Sink; + /** Rows per page / per sink batch. Default 500. */ + batchSize?: number; + /** Called after each batch with the running total. */ + onProgress?: (total: number) => void; +} + +/** Pages `records_event` and feeds each batch to `sink.onRecords(..., {phase: + * 'backfill'})`. Returns the number of event rows fed. */ +export async function reindexEventsToSink(opts: ReindexOptions): Promise { + const { db, sink } = opts; + const batchSize = opts.batchSize ?? 500; + let offset = 0; + let total = 0; + + for (;;) { + const page = await db + .prepare( + 'SELECT uri, did, rkey, cid, record, time_us FROM records_event ORDER BY uri LIMIT ? OFFSET ?' + ) + .bind(batchSize, offset) + .all(); + const rows = page.results ?? []; + if (rows.length === 0) break; + + const records = rows.map( + (r): RecordEvent => ({ + kind: 'created', + uri: String(r.uri), + did: String(r.did), + collection: EVENT_COLLECTION, + rkey: String(r.rkey), + // records_event has no `collection` column (one table per collection) + // and may store a null cid; the sink wants a string. + cid: r.cid == null ? '' : String(r.cid), + // D1 stores `record` as a JSON string; the sink expects a parsed + // object (applyEvents passes safeParseJson(record) on the live path). + record: JSON.parse(String(r.record)) as Record, + time_us: Number(r.time_us) + }) + ); + + await sink.onRecords(records, { phase: 'backfill' }); + total += rows.length; + offset += batchSize; + opts.onProgress?.(total); + if (rows.length < batchSize) break; + } + + return total; +} -- 2.51.2 From 79d44a9776f0dde053ee8df36ff82a23e9364f36 Mon Sep 17 00:00:00 2001 From: Tom Scanlan Date: Fri, 19 Jun 2026 15:04:13 -0400 Subject: [PATCH 2/4] docs(search): document meili:reindex in README and tighten config comment --- README.md | 4 +++- apps/web/src/lib/contrail.config.ts | 17 ++++++++--------- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 6be2c39..26805b6 100644 --- a/README.md +++ b/README.md @@ -58,7 +58,9 @@ then set the search vars in `.env` (see `.env.example`): the read and write keys are kept separate on purpose so the browser-facing read path never holds the admin key. the index is populated by the same cron ingest that fills d1, so once configured a `pnpm backfill` (or normal ingest) will fill it. -**rollout order on an existing deployment.** the sink only indexes records applied _after_ it's enabled, so don't turn on the read path first or existing upcoming events vanish from search until they're next touched. instead: (1) set the write vars and let the sink arm, (2) run `pnpm backfill` and confirm the meili `events` index count looks right, then (3) set the read vars (`SEARCH_URL` / `SEARCH_API_KEY`). until step 3 the app keeps using the d1 fallback, so search stays working throughout. +**rollout order on an existing deployment.** the sink only indexes records applied _after_ it's enabled, so don't turn on the read path first or existing upcoming events vanish from search until they're next touched. instead: (1) set the write vars and let the sink arm, (2) populate the index (see below) and confirm the meili `events` index count looks right, then (3) set the read vars (`SEARCH_URL` / `SEARCH_API_KEY`). until step 3 the app keeps using the d1 fallback, so search stays working throughout. + +**populating the index.** backfill and refresh now feed the sink, so `pnpm backfill` fills meili as it walks each user's pds. on an existing deployment the event records are usually already in d1, so `pnpm meili:reindex` is faster: it replays the stored `community.lexicon.calendar.event` rows straight from d1 into the index with no network walk and no d1 writes (add `:remote` to target the deployed d1). both paths apply the same discoverable filter as live ingest, so re-running either is idempotent. ## contributing diff --git a/apps/web/src/lib/contrail.config.ts b/apps/web/src/lib/contrail.config.ts index 3c7cd2b..2afcfa5 100644 --- a/apps/web/src/lib/contrail.config.ts +++ b/apps/web/src/lib/contrail.config.ts @@ -3,15 +3,14 @@ import { SPACE_TYPE } from './spaces/config'; import { MAX_HYDRATION_URIS } from './search/constants'; import { createMeiliSink, meiliSinkBackendFromEnv } from './search/server/meili-sink'; -// The `contrail` CLI (`pnpm backfill` / `contrail refresh`) loads this config and -// fires `config.sinks` from applyEvents on the backfill/refresh paths, so a -// freshly-set-up or re-synced installation gets full search coverage, not just -// whatever flowed through live ingest. The backend is read from process.env -// (SEARCH_SINK_URL / SEARCH_SINK_API_KEY / SEARCH_INDEX), injected at runtime by -// the operator (e.g. `op run`), so no secret is committed. Gated on -// SEARCH_SINK_URL: absent it, the sink stays off, which keeps `pnpm generate` -// sink-free. The runtime Worker (`$lib/contrail/index.ts`) does not use this: it -// overrides `sinks` with its own env-armed sink, so there is never a double feed. +// The `contrail` CLI (`pnpm backfill` / `contrail refresh`) fires `config.sinks` +// on the backfill/refresh paths, so a fresh or re-synced install gets full search +// coverage, not just what flowed through live ingest. The backend is read from +// process.env (SEARCH_SINK_URL / SEARCH_SINK_API_KEY / SEARCH_INDEX), injected by +// the operator at runtime so no secret is committed; an unset SEARCH_SINK_URL +// omits the sink, keeping `pnpm generate` clean. The runtime Worker +// (`$lib/contrail/index.ts`) replaces `sinks` with its own per-invocation-env +// sink, so this one runs only under the CLI. const searchSinks: NonNullable = typeof process !== 'undefined' && process.env?.SEARCH_SINK_URL ? [createMeiliSink(() => meiliSinkBackendFromEnv(process.env))] -- 2.51.2 From 0517349bbad4364d1f871633630c8b6667aa3aeb Mon Sep 17 00:00:00 2001 From: Tom Scanlan Date: Fri, 19 Jun 2026 15:56:39 -0400 Subject: [PATCH 3/4] fix(search): harden meili:reindex (fail-loud config, apply index settings, skip poison rows) Adversarial-review follow-ups: - reindex-cli resolves the Meili backend explicitly and exits if SEARCH_SINK_API_KEY is missing, instead of borrowing config.sinks[0] (gated only on the URL) and silently no-opping while reporting success. - reindex-cli applies index settings up front: idempotent, ensures _geo/startsAt filters exist on a never-armed index, and doubles as an auth/connectivity check. - reindexEventsToSink parses each record defensively and skips a missing/corrupt row instead of throwing and aborting the whole run (the live path uses safeParseJson). - integration test restores SEARCH_SINK_* in afterAll so a failing assertion can't leak env into the rest of the process. --- .../server/meili-sink.integration.test.ts | 24 ++++++++--- apps/web/src/lib/search/server/reindex-cli.ts | 24 ++++++++--- .../web/src/lib/search/server/reindex.test.ts | 34 +++++++++++++++ apps/web/src/lib/search/server/reindex.ts | 43 ++++++++++++++----- 4 files changed, 103 insertions(+), 22 deletions(-) diff --git a/apps/web/src/lib/search/server/meili-sink.integration.test.ts b/apps/web/src/lib/search/server/meili-sink.integration.test.ts index 86e4ada..791ce20 100644 --- a/apps/web/src/lib/search/server/meili-sink.integration.test.ts +++ b/apps/web/src/lib/search/server/meili-sink.integration.test.ts @@ -7,7 +7,7 @@ // getmeili/meilisearch:v1.46 // MEILI_TEST_URL=http://localhost:7700 MEILI_TEST_KEY=masterKey \ // pnpm vitest run src/lib/search/server/meili-sink.integration.test.ts -import { describe, it, expect, beforeAll } from 'vitest'; +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; import { createMeiliSink, applyMeiliSettings, type MeiliSinkBackend } from './meili-sink'; import { searchEvents, nearMeEvents, type SearchBackend } from './meili'; @@ -136,12 +136,27 @@ run('contrail.config sink populates Meili on backfill (CLI path)', () => { await applyMeiliSettings({ url: URL!, apiKey: KEY, indexUid: INDEX }); }); + // Restore in afterAll, not inline: a failing assertion in the `it` must not + // leak SEARCH_SINK_* into the rest of the process. + afterAll(() => { + for (const k of ['SEARCH_SINK_URL', 'SEARCH_SINK_API_KEY', 'SEARCH_INDEX'] as const) { + if (savedEnv[k] === undefined) delete process.env[k]; + else process.env[k] = savedEnv[k]; + } + }); + it('indexes a backfilled event through the config sink so the read path finds it', async () => { const { config } = await import('../../contrail.config'); expect(config.sinks && config.sinks.length).toBeTruthy(); await config.sinks![0].onRecords( - [created(uri, { name: 'Backfilled Jazz Night', description: 'via cli backfill', startsAt: FUTURE })], + [ + created(uri, { + name: 'Backfilled Jazz Night', + description: 'via cli backfill', + startsAt: FUTURE + }) + ], { phase: 'backfill' } ); @@ -150,10 +165,5 @@ run('contrail.config sink populates Meili on backfill (CLI path)', () => { (r) => r.hits.some((h) => h.uri === uri) ); expect(text.hits.map((h) => h.uri)).toContain(uri); - - for (const k of ['SEARCH_SINK_URL', 'SEARCH_SINK_API_KEY', 'SEARCH_INDEX'] as const) { - if (savedEnv[k] === undefined) delete process.env[k]; - else process.env[k] = savedEnv[k]; - } }); }); diff --git a/apps/web/src/lib/search/server/reindex-cli.ts b/apps/web/src/lib/search/server/reindex-cli.ts index 6aaaf4d..fcec4df 100644 --- a/apps/web/src/lib/search/server/reindex-cli.ts +++ b/apps/web/src/lib/search/server/reindex-cli.ts @@ -5,28 +5,42 @@ // The Meili backend is resolved from the same env the sink uses, injected by the // operator (so no secret is committed): // SEARCH_SINK_URL / SEARCH_SINK_API_KEY / SEARCH_INDEX -// Reads records_event (SELECT only) and feeds the config sink. Zero D1 writes. +// Reads records_event (SELECT only) and feeds the sink. Zero D1 writes. import { getPlatformProxy } from 'wrangler'; -import { config } from '../../contrail.config'; +import { applyMeiliSettings, createMeiliSink, meiliSinkBackendFromEnv } from './meili-sink'; import { reindexEventsToSink, type ReindexDb } from './reindex'; const remote = process.argv.includes('--remote'); const binding = 'DB'; -const sink = config.sinks?.[0]; -if (!sink) { +// Resolve the backend ourselves rather than borrowing config.sinks[0]: that +// path is gated only on SEARCH_SINK_URL, so a missing/invalid SEARCH_SINK_API_KEY +// leaves the sink a silent no-op while we'd still report "N rows reindexed". +// Here a half-configured env fails loudly before we touch D1. +const backend = meiliSinkBackendFromEnv(process.env); +if (!backend) { console.error( 'No search sink configured. Export SEARCH_SINK_URL and SEARCH_SINK_API_KEY (and optionally SEARCH_INDEX) before running.' ); process.exit(1); } +// Apply index settings up front. It's idempotent, ensures the read path's +// filters (_geo / startsAt / endsAt) resolve even on a never-armed index, and +// doubles as an auth/connectivity check: a bad admin key or unreachable Meili +// throws here instead of letting every per-batch upsert silently fail. +await applyMeiliSettings(backend); +const sink = createMeiliSink(() => backend); + const { env, dispose } = await getPlatformProxy({ environment: remote ? 'production' : undefined }); try { const db = (env as Record)[binding] as ReindexDb | undefined; - if (!db) throw new Error(`No "${binding}" binding in wrangler env (${remote ? 'production' : 'default'}).`); + if (!db) + throw new Error( + `No "${binding}" binding in wrangler env (${remote ? 'production' : 'default'}).` + ); const total = await reindexEventsToSink({ db, diff --git a/apps/web/src/lib/search/server/reindex.test.ts b/apps/web/src/lib/search/server/reindex.test.ts index 1ca3ed1..213a260 100644 --- a/apps/web/src/lib/search/server/reindex.test.ts +++ b/apps/web/src/lib/search/server/reindex.test.ts @@ -101,4 +101,38 @@ describe('reindexEventsToSink', () => { expect(batches[0].records[0].cid).toBe(''); }); + + it('skips a row whose record is missing or not valid JSON without aborting the run', async () => { + const { db } = fakeDb([ + // malformed JSON — the live path uses safeParseJson, so reindex must not + // throw and abort coverage on one poison row. + { + uri: 'at://did:plc:a/community.lexicon.calendar.event/bad', + did: 'did:plc:a', + rkey: 'bad', + cid: 'c', + record: '{not json', + time_us: 1 + }, + // NULL record column. + { + uri: 'at://did:plc:a/community.lexicon.calendar.event/null', + did: 'did:plc:a', + rkey: 'null', + cid: 'c', + record: null, + time_us: 1 + }, + row('at://did:plc:a/community.lexicon.calendar.event/ok', { name: 'OK' }) + ]); + const { sink, batches } = recordingSink(); + + const total = await reindexEventsToSink({ db, sink }); + + // only the valid row is fed; the two bad rows are skipped, not thrown. + expect(total).toBe(1); + expect(batches.flatMap((b) => b.records.map((r) => r.uri))).toEqual([ + 'at://did:plc:a/community.lexicon.calendar.event/ok' + ]); + }); }); diff --git a/apps/web/src/lib/search/server/reindex.ts b/apps/web/src/lib/search/server/reindex.ts index d32b439..79d28b8 100644 --- a/apps/web/src/lib/search/server/reindex.ts +++ b/apps/web/src/lib/search/server/reindex.ts @@ -31,6 +31,21 @@ export interface ReindexOptions { onProgress?: (total: number) => void; } +/** Parse a stored `record` cell into a plain object, or null if it's missing / + * not valid JSON / not an object. The live ingest path uses safeParseJson, so a + * single poison row must be skipped, not allowed to abort the whole reindex. */ +function parseRecordObject(raw: unknown): Record | null { + if (raw == null) return null; + try { + const parsed = JSON.parse(String(raw)); + return parsed && typeof parsed === 'object' && !Array.isArray(parsed) + ? (parsed as Record) + : null; + } catch { + return null; + } +} + /** Pages `records_event` and feeds each batch to `sink.onRecords(..., {phase: * 'backfill'})`. Returns the number of event rows fed. */ export async function reindexEventsToSink(opts: ReindexOptions): Promise { @@ -49,8 +64,17 @@ export async function reindexEventsToSink(opts: ReindexOptions): Promise const rows = page.results ?? []; if (rows.length === 0) break; - const records = rows.map( - (r): RecordEvent => ({ + const records: RecordEvent[] = []; + for (const r of rows) { + // D1 stores `record` as a JSON string; the sink expects a parsed object. + // Skip (don't throw on) a missing/corrupt cell so one poison row can't + // abort coverage for the rest of the table. + const record = parseRecordObject(r.record); + if (!record) { + console.warn(`[reindex] skipping ${String(r.uri)}: record is missing or not valid JSON`); + continue; + } + records.push({ kind: 'created', uri: String(r.uri), did: String(r.did), @@ -59,16 +83,15 @@ export async function reindexEventsToSink(opts: ReindexOptions): Promise // records_event has no `collection` column (one table per collection) // and may store a null cid; the sink wants a string. cid: r.cid == null ? '' : String(r.cid), - // D1 stores `record` as a JSON string; the sink expects a parsed - // object (applyEvents passes safeParseJson(record) on the live path). - record: JSON.parse(String(r.record)) as Record, + record, time_us: Number(r.time_us) - }) - ); + }); + } - await sink.onRecords(records, { phase: 'backfill' }); - total += rows.length; - offset += batchSize; + if (records.length > 0) await sink.onRecords(records, { phase: 'backfill' }); + total += records.length; + // Advance by rows read (not records fed) so skipped rows don't stall paging. + offset += rows.length; opts.onProgress?.(total); if (rows.length < batchSize) break; } -- 2.51.2 From 5ed12502796bf6edca5ed75efe21a7ff81ce6a28 Mon Sep 17 00:00:00 2001 From: Tom Scanlan Date: Fri, 19 Jun 2026 21:42:15 -0400 Subject: [PATCH 4/4] fix(search): ensure index settings on first sink write; format; document remote parity Second review round: - createMeiliSink applies the index settings once before its first write, so a fresh-rollout 'pnpm backfill' (or reindex) can't auto-create a bare index whose _geo/startsAt filtered searches 400. Flag flips only on success so a transient Meili outage retries instead of disabling the sink. (TDD'd.) - meili:reindex:remote: clarify it resolves the D1 binding exactly like the pre-existing backfill:remote (same wrangler env.production), and log which D1 (local vs deployed) is targeted before connecting. - prettier: format README.md, contrail.config.ts, contrail.config.test.ts (were committed unformatted). --- README.md | 8 +++---- apps/web/src/lib/contrail.config.test.ts | 12 ++++++---- apps/web/src/lib/contrail.config.ts | 5 +--- .../src/lib/search/server/meili-sink.test.ts | 24 +++++++++++++++++++ apps/web/src/lib/search/server/meili-sink.ts | 20 +++++++++++++++- apps/web/src/lib/search/server/reindex-cli.ts | 13 +++++++--- 6 files changed, 66 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index 26805b6..dc94418 100644 --- a/README.md +++ b/README.md @@ -7,10 +7,11 @@ https://atmo.rsvp uses `community.lexicon.calendar.event` and `community.lexicon.calendar.rsvp`. features: + - event creation - rsvp to events -- add your events to any ical compatible calendar -(go to calendar/ when signed in and click "Add to your calendar") +- add your events to any ical compatible calendar + (go to calendar/ when signed in and click "Add to your calendar") - post your events/rsvps to bluesky or anywhere else with nice open-graph images - display comments - show what events your bsky follows are going to @@ -60,9 +61,8 @@ the read and write keys are kept separate on purpose so the browser-facing read **rollout order on an existing deployment.** the sink only indexes records applied _after_ it's enabled, so don't turn on the read path first or existing upcoming events vanish from search until they're next touched. instead: (1) set the write vars and let the sink arm, (2) populate the index (see below) and confirm the meili `events` index count looks right, then (3) set the read vars (`SEARCH_URL` / `SEARCH_API_KEY`). until step 3 the app keeps using the d1 fallback, so search stays working throughout. -**populating the index.** backfill and refresh now feed the sink, so `pnpm backfill` fills meili as it walks each user's pds. on an existing deployment the event records are usually already in d1, so `pnpm meili:reindex` is faster: it replays the stored `community.lexicon.calendar.event` rows straight from d1 into the index with no network walk and no d1 writes (add `:remote` to target the deployed d1). both paths apply the same discoverable filter as live ingest, so re-running either is idempotent. +**populating the index.** backfill and refresh now feed the sink, so `pnpm backfill` fills meili as it walks each user's pds. on an existing deployment the event records are usually already in d1, so `pnpm meili:reindex` is faster: it replays the stored `community.lexicon.calendar.event` rows straight from d1 into the index with no network walk and no d1 writes. both paths apply the same discoverable filter as live ingest, and the sink applies the index settings on its first write, so a fresh index gets the right filterable fields and re-running either is idempotent. `pnpm meili:reindex:remote` targets the deployed d1 and needs the same wrangler `env.production` that `pnpm backfill:remote` uses. ## contributing open for contributions by all :) - diff --git a/apps/web/src/lib/contrail.config.test.ts b/apps/web/src/lib/contrail.config.test.ts index 326a4ba..4cecaa6 100644 --- a/apps/web/src/lib/contrail.config.test.ts +++ b/apps/web/src/lib/contrail.config.test.ts @@ -6,7 +6,11 @@ import { config } from './contrail.config'; const listDiscoverableByUris = config.collections!.event.pipelineQueries!.listDiscoverableByUris; const run = async (search: string) => { - const source = await listDiscoverableByUris(undefined as never, new URLSearchParams(search), config); + const source = await listDiscoverableByUris( + undefined as never, + new URLSearchParams(search), + config + ); return { conditions: source.conditions ?? [], params: source.params }; }; @@ -22,9 +26,9 @@ describe('listDiscoverableByUris pipelineQuery', () => { const placeholders = source.conditions.join(' ').match(/\?/g) ?? []; expect(placeholders).toHaveLength(2); // The search surface must not leak events hidden from discovery. - expect( - source.conditions.some((c: string) => c.includes('preferences.showInDiscovery')) - ).toBe(true); + expect(source.conditions.some((c: string) => c.includes('preferences.showInDiscovery'))).toBe( + true + ); }); it('matches nothing when no uris are given', async () => { diff --git a/apps/web/src/lib/contrail.config.ts b/apps/web/src/lib/contrail.config.ts index 2afcfa5..c54943d 100644 --- a/apps/web/src/lib/contrail.config.ts +++ b/apps/web/src/lib/contrail.config.ts @@ -95,10 +95,7 @@ export const config: ContrailConfig = { .slice(0, MAX_HYDRATION_URIS); if (uris.length === 0) return { conditions: ['0 = 1'] }; return { - conditions: [ - `r.uri IN (${uris.map(() => '?').join(', ')})`, - DISCOVERABLE_CONDITION - ], + conditions: [`r.uri IN (${uris.map(() => '?').join(', ')})`, DISCOVERABLE_CONDITION], params: uris }; }, diff --git a/apps/web/src/lib/search/server/meili-sink.test.ts b/apps/web/src/lib/search/server/meili-sink.test.ts index 6a49ed3..aa19357 100644 --- a/apps/web/src/lib/search/server/meili-sink.test.ts +++ b/apps/web/src/lib/search/server/meili-sink.test.ts @@ -186,6 +186,30 @@ describe('createMeiliSink onRecords', () => { expect(calls).toHaveLength(0); }); + + it('applies index settings once, before the first write (fresh-index safety)', async () => { + const { fn, calls } = fakeFetch(); + const sink = createMeiliSink(() => BACKEND, fn); + + // Two batches on the same sink: a fresh-rollout `pnpm backfill` must not + // let PUT /documents auto-create a bare index whose _geo/startsAt searches + // 400. Settings are applied exactly once, and before the first write. + await sink.onRecords( + [created('at://did:plc:alice/community.lexicon.calendar.event/a', { name: 'A' })], + { phase: 'backfill' } + ); + await sink.onRecords( + [created('at://did:plc:alice/community.lexicon.calendar.event/b', { name: 'B' })], + { phase: 'backfill' } + ); + + const settings = calls.filter((c) => c.method === 'PATCH' && c.url.endsWith('/settings')); + expect(settings).toHaveLength(1); + const firstSettingsIdx = calls.findIndex((c) => c.url.endsWith('/settings')); + const firstPutIdx = calls.findIndex((c) => c.method === 'PUT'); + expect(firstSettingsIdx).toBeGreaterThanOrEqual(0); + expect(firstSettingsIdx).toBeLessThan(firstPutIdx); + }); }); describe('fetch is invoked detached (workerd Illegal invocation guard)', () => { diff --git a/apps/web/src/lib/search/server/meili-sink.ts b/apps/web/src/lib/search/server/meili-sink.ts index bf4df7c..18c1f4a 100644 --- a/apps/web/src/lib/search/server/meili-sink.ts +++ b/apps/web/src/lib/search/server/meili-sink.ts @@ -92,7 +92,15 @@ export class MeiliEventIndex { async applySettings(): Promise { await this.request('PATCH', `/indexes/${this.indexUid}/settings`, { searchableAttributes: ['name', 'description'], - filterableAttributes: ['_geo', 'startsAt', 'endsAt', 'status', 'mode', 'did', 'locationTypes'], + filterableAttributes: [ + '_geo', + 'startsAt', + 'endsAt', + 'status', + 'mode', + 'did', + 'locationTypes' + ], sortableAttributes: ['_geo', 'startsAt', 'endsAt'] }); } @@ -148,6 +156,12 @@ export function createMeiliSink( getBackend: () => MeiliSinkBackend | null, fetchFn?: typeof fetch ): Sink { + // Apply the read-path's filterable/sortable settings once, before the first + // write. Otherwise a fresh-rollout `pnpm backfill` (or reindex) lets PUT + // /documents auto-create a bare index whose _geo/startsAt filtered searches + // 400. Only flip the flag on success, so a transient Meili outage retries + // next batch instead of permanently disabling the sink in the Worker. + let settingsApplied = false; return { async onRecords(events: RecordEvent[]): Promise { const backend = getBackend(); @@ -180,6 +194,10 @@ export function createMeiliSink( if (docs.length === 0 && deletes.length === 0) return; const index = new MeiliEventIndex(backend, fetchFn); + if (!settingsApplied) { + await index.applySettings(); + settingsApplied = true; + } // Order doesn't matter across upsert/remove within a batch: a given uri // is deduplicated to a single RecordEvent, so it's either a create or a // delete, never both. diff --git a/apps/web/src/lib/search/server/reindex-cli.ts b/apps/web/src/lib/search/server/reindex-cli.ts index fcec4df..fb3b46a 100644 --- a/apps/web/src/lib/search/server/reindex-cli.ts +++ b/apps/web/src/lib/search/server/reindex-cli.ts @@ -1,6 +1,8 @@ -// CLI entry for the D1 -> Meili event reindex. Mirrors `contrail backfill`: -// pnpm meili:reindex # local D1 binding -// pnpm meili:reindex:remote # --remote -> getPlatformProxy production binding +// CLI entry for the D1 -> Meili event reindex. Resolves the D1 binding exactly +// like `contrail backfill`, so `:remote` needs the same wrangler `env.production` +// that `backfill:remote` already relies on: +// pnpm meili:reindex # default env -> local D1 binding +// pnpm meili:reindex:remote # --remote -> getPlatformProxy production env // // The Meili backend is resolved from the same env the sink uses, injected by the // operator (so no secret is committed): @@ -32,6 +34,11 @@ if (!backend) { await applyMeiliSettings(backend); const sink = createMeiliSink(() => backend); +// Say which D1 we're about to touch, so an operator can't mistake a default-env +// (local) run for a deployed one, or vice-versa. +console.log( + `reindex target: ${remote ? 'production env (deployed D1)' : 'default env (local D1)'}` +); const { env, dispose } = await getPlatformProxy({ environment: remote ? 'production' : undefined }); -- 2.51.2