From 88b8bfe6732d32635d5535edaf765cf4d9ebed78 Mon Sep 17 00:00:00 2001 From: dawn Date: Fri, 2 Oct 2026 00:00:50 +0300 Subject: [PATCH] sitemap: read hydrant's jetstream from index.tangled.network, rewinding on reconnect Signed-off-by: dawn --- sitemap/package.json | 4 + sitemap/pnpm-lock.yaml | 168 ++++++++++++++++++++++++ sitemap/src/hydrant.ts | 18 +-- sitemap/src/index.ts | 151 ++++++--------------- sitemap/src/jetstream.ts | 164 +++++++++++++++++++++++ sitemap/src/types.ts | 31 +---- sitemap/test/events.ts | 62 +++++++++ sitemap/test/jetstream.test.ts | 62 +++++++++ sitemap/test/sitemap.test.ts | 120 +++++------------ sitemap/test/sync.test.ts | 231 ++++++++++----------------------- sitemap/wrangler.dev.jsonc | 8 +- sitemap/wrangler.jsonc | 8 +- 12 files changed, 616 insertions(+), 411 deletions(-) create mode 100644 sitemap/src/jetstream.ts create mode 100644 sitemap/test/events.ts create mode 100644 sitemap/test/jetstream.test.ts diff --git a/sitemap/package.json b/sitemap/package.json index 87bf48c56..a4de81c57 100644 --- a/sitemap/package.json +++ b/sitemap/package.json @@ -15,5 +15,9 @@ "typescript": "^5.9.3", "vitest": "~3.0.7", "wrangler": "^4.75.0" + }, + "dependencies": { + "@atcute/jetstream": "^2.2.0", + "valibot": "^1.5.0" } } diff --git a/sitemap/pnpm-lock.yaml b/sitemap/pnpm-lock.yaml index d87fa5dc0..48c2b203e 100644 --- a/sitemap/pnpm-lock.yaml +++ b/sitemap/pnpm-lock.yaml @@ -7,6 +7,13 @@ settings: importers: .: + dependencies: + '@atcute/jetstream': + specifier: ^2.2.0 + version: 2.2.0(@atcute/cid@2.5.0)(@atcute/lexicons@2.1.1)(typescript@5.9.3) + valibot: + specifier: ^1.5.0 + version: 1.5.0(typescript@5.9.3) devDependencies: '@cloudflare/workers-types': specifier: ^4.20260317.1 @@ -23,6 +30,41 @@ importers: packages: + '@atcute/atproto@4.0.4': + resolution: {integrity: sha512-aEp5wQVPVbZcFJUwoi54tyT+ehRI/xtPe9T0zIvcryHAVXF/zxHtNboM+htbZLOAGsDFsqzgCyxph0Bafw7L8w==} + peerDependencies: + '@atcute/lexicons': ^2.0.0 + + '@atcute/cbor@2.3.8': + resolution: {integrity: sha512-gazXBcNTr2U3cRaCjfRbq2GKHJnWapUUA3mgehL+0tbVXEty2wf/LIwF9z46gjqjX2DYGGwkybF8lTI3dzHaNQ==} + peerDependencies: + '@atcute/cid': ^2.5.0 + + '@atcute/cid@2.5.0': + resolution: {integrity: sha512-CAZyzUdaQhuC+S7tGjoC7vgyDI2X4JU+CPO/7WjEU7HlNsAGXjSkD++5TlfnSWGDqUeqpjMZjUfFBZYHKzPHew==} + + '@atcute/firehose@1.2.0': + resolution: {integrity: sha512-u6Oh137haGOzWNFbJzakUjjDOHvVNORY5r5H4v9OfHqVzFOYwSDptMYsD2TlRf4BbdtmHfjCb7pL1/D0eMfh6Q==} + peerDependencies: + '@atcute/lexicons': ^2.1.0 + + '@atcute/jetstream@2.2.0': + resolution: {integrity: sha512-2Aux4AxUj+mRLNyI1L7kycZM/q9UuAF4pGcyuWVfGLrK95OgAe4BSyCn33HbuKidBYZPwQjDn+j6B8fz5cOUYg==} + peerDependencies: + '@atcute/lexicons': ^2.0.0 + + '@atcute/lexicons@2.1.1': + resolution: {integrity: sha512-kHyqUW8g/Fq9LTctGh0TVBx0AmtecV6VSrRprrSbqKqHe9q8C6y+YPL3EMNq2l93XA1iDHkkvYwLdTcKL/lNbA==} + + '@atcute/multibase@1.2.5': + resolution: {integrity: sha512-cReTONgYpQo/VHD3ZmzPNoyBKJgSk1J4h//cvvdVVJBMar+SjlQ/sUXeTjQfuyfmDKv+TLKhmutLcvbMcQ9Rvw==} + + '@atcute/uint8array@1.2.0': + resolution: {integrity: sha512-KoBGTbV4lS8zXNu91osM3FecrH3NUnxNPsBAkaNtn6uf/p240245qoLn7keFpRM0npX522X8R+XBBb2rvtE7Rg==} + + '@atcute/util-text@1.3.4': + resolution: {integrity: sha512-u2UAM7iSM09sQaSG9jtxWFSPgB8boVj50/BoyMvYnhVgGBu+nXIuAcdDUQCsZA44YZgTPPhN/b86JX+jH7SPzQ==} + '@cloudflare/kv-asset-handler@0.5.0': resolution: {integrity: sha512-jxQYkj8dSIzc0cD6cMMNdOc1UVjqSqu8BZdor5s8cGjW2I8BjODt/kWPVdY+u9zj3ms75Q5qaZgnxUad83+eAg==} engines: {node: '>=22.0.0'} @@ -560,6 +602,12 @@ packages: '@jridgewell/trace-mapping@0.3.9': resolution: {integrity: sha512-3Belt6tdc8bPgAtbcmdtNJlirVoTmEb5e2gC94PnkwEW9jI6CAHUeoG85tjWP5WquqfavoMtMwiG4P926ZKKuQ==} + '@mary-ext/event-iterator@1.1.0': + resolution: {integrity: sha512-uL6XFBsfAq+h6wnmioLTDMKzgbuKKQTo/2k512W1w1M5SgVroS6+D4MOD+j1K0loEYUJD9rIKV0Fo5ETZlgafw==} + + '@mary-ext/simple-event-emitter@1.0.2': + resolution: {integrity: sha512-A644JdOf+KfTGyX/EYf8AI1Y18mM7369KrhtOVkdEwRa2QIC0Botv7XWnK6/XU7HuQfvtFpCufmHJC3hi4YMuw==} + '@napi-rs/lzma-linux-x64-gnu@1.5.1': resolution: {integrity: sha512-oTXEIha4SsuXdTA4Iyskj0kpdx2yVXdhd75c2v3xGrHFfVMsbhTPZU/nMPL4sWKo4pBHm3aucLaqGlF696dTyQ==} engines: {node: ^22.20 || ^24.12 || >=25} @@ -567,6 +615,9 @@ packages: os: [linux] libc: [glibc] + '@oomfware/eval@0.1.0': + resolution: {integrity: sha512-EIkukTd3zDQEHUjcfFRDq79Jgm4OEyUrEUvp/SwgjfoH2POaE5cvaswIonFwdt8fSzJF5xtAdgcgIoKD8sPpjg==} + '@poppinss/colors@4.1.6': resolution: {integrity: sha512-H9xkIdFswbS8n1d6vmRd8+c10t2Qe+rZITbbDHHkQixH5+2x1FDGmi/0K+WgWiqQFKPSlIYB7jlH6Kpfn6Fleg==} @@ -721,6 +772,9 @@ packages: '@speed-highlight/core@1.2.24': resolution: {integrity: sha512-qeW2e1l78afw8VhRPfPQ1Gjj+KU5XFQ/OFV5ti6eTa9bruO7mJyZtA4vw0ofqmA3tKCkROE9xLk3VZoeRc98nw==} + '@standard-schema/spec@1.1.0': + resolution: {integrity: sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==} + '@types/estree@1.0.9': resolution: {integrity: sha512-GhdPgy1el4/ImP05X05Uw4cw2/M93BCUmnEvWZNStlCzEKME4Fkk+YpoA5OiHNQmoS7Cafb8Xa3Pya8m1Qrzeg==} @@ -812,9 +866,15 @@ packages: engines: {node: '>=18'} hasBin: true + esm-env@1.2.2: + resolution: {integrity: sha512-Epxrv+Nr/CaL4ZcFGPJIYLWFom+YeV1DqMLHJoEd9SYRxNbaFruBwfEX/kkHUJf55j2+TUbmDcmuilbP1TmXHA==} + estree-walker@3.0.3: resolution: {integrity: sha512-7RUKfXgSMMkzt6ZuXmqapOurLGPPfgj6l9uRZ7lRGolvk0y2yocc35LdcxKC5PQZdn2DMqioAQ2NoWcrTKmm6g==} + event-target-polyfill@0.0.4: + resolution: {integrity: sha512-Gs6RLjzlLRdT8X9ZipJdIZI/Y6/HhRLyq9RdDlCsnpxr/+Nn6bU2EFGuC94GjxqhM+Nmij2Vcq98yoHrU8uNFQ==} + expect-type@1.4.0: resolution: {integrity: sha512-KfYbmpRm0VbLjEvVa9yGwCi9GI34xvi7A/HXYWQO65CSD2u3MczUJSuwXKFIxlGsgBQizV9q5J9NHj4VG0n+pA==} engines: {node: '>=12.0.0'} @@ -855,6 +915,14 @@ packages: engines: {node: ^10 || ^12 || ^13.7 || ^14 || >=15.0.1} hasBin: true + partysocket@1.3.0: + resolution: {integrity: sha512-1zToNyolZFK/7nuAw/K2bZrNzFqaZyRoCEkS+9vG6WSC5ikrN6qWRe96q6ImU51uptz2r+dAwSkwhJVdQi4LiA==} + peerDependencies: + react: '>=17' + peerDependenciesMeta: + react: + optional: true + path-to-regexp@6.3.0: resolution: {integrity: sha512-Yhpw4T9C6hPpgPeA28us07OJeqZ5EzQTkbfwuhsUg0c237RomFoETJgmp2sa3F/41gfLE6G5cqcYwznmeEeOlQ==} @@ -944,6 +1012,17 @@ packages: unenv@2.0.0-rc.24: resolution: {integrity: sha512-i7qRCmY42zmCwnYlh9H2SvLEypEFGye5iRmEMKjcGi7zk9UquigRjFtTLz0TYqr0ZGLZhaMHl/foy1bZR+Cwlw==} + unicode-segmenter@0.17.3: + resolution: {integrity: sha512-hKZwqBjJDmqNrq1+LjDxck1qLzJFcLLJM2Xq92ORRHOuau6GSHeELmn+uyk6zNH1CK9mogrUJGP9WJCJUQXMAw==} + + valibot@1.5.0: + resolution: {integrity: sha512-nil6AkP2TChWL43Z5uJ6GTxX01CUA+g8LWUM+N/rB9NBbkUMaUsi9PUNzlUPgoKASgmx9f7eGOYpJ04/fSa6FQ==} + peerDependencies: + typescript: '>=5' + peerDependenciesMeta: + typescript: + optional: true + vite-node@3.0.9: resolution: {integrity: sha512-w3Gdx7jDcuT9cNn9jExXgOyKmf5UOTb6WMHz8LGAm54eS1Elf5OuBhCxl6zJxGhEeIkgsE1WbHuoL0mj/UXqXg==} engines: {node: ^18.0.0 || ^20.0.0 || >=22.0.0} @@ -1049,6 +1128,10 @@ packages: utf-8-validate: optional: true + yocto-queue@1.2.2: + resolution: {integrity: sha512-4LCcse/U2MHZ63HAJVE+v71o7yOdIe4cZ70Wpf8D/IyjDKYQLV5GD46B+hSTjJsvV5PztjvHoU580EftxjDZFQ==} + engines: {node: '>=12.20'} + youch-core@0.3.3: resolution: {integrity: sha512-ho7XuGjLaJ2hWHoK8yFnsUGy2Y5uDpqSTq1FkHLK4/oqKtyUU1AFbOOxY4IpC9f0fTLjwYbslUz0Po5BpD1wrA==} @@ -1057,6 +1140,65 @@ packages: snapshots: + '@atcute/atproto@4.0.4(@atcute/lexicons@2.1.1)': + dependencies: + '@atcute/lexicons': 2.1.1 + + '@atcute/cbor@2.3.8(@atcute/cid@2.5.0)': + dependencies: + '@atcute/cid': 2.5.0 + '@atcute/multibase': 1.2.5 + '@atcute/uint8array': 1.2.0 + + '@atcute/cid@2.5.0': + dependencies: + '@atcute/multibase': 1.2.5 + '@atcute/uint8array': 1.2.0 + + '@atcute/firehose@1.2.0(@atcute/cid@2.5.0)(@atcute/lexicons@2.1.1)': + dependencies: + '@atcute/cbor': 2.3.8(@atcute/cid@2.5.0) + '@atcute/lexicons': 2.1.1 + '@atcute/uint8array': 1.2.0 + '@mary-ext/event-iterator': 1.1.0 + '@mary-ext/simple-event-emitter': 1.0.2 + partysocket: 1.3.0 + transitivePeerDependencies: + - '@atcute/cid' + - react + + '@atcute/jetstream@2.2.0(@atcute/cid@2.5.0)(@atcute/lexicons@2.1.1)(typescript@5.9.3)': + dependencies: + '@atcute/atproto': 4.0.4(@atcute/lexicons@2.1.1) + '@atcute/firehose': 1.2.0(@atcute/cid@2.5.0)(@atcute/lexicons@2.1.1) + '@atcute/lexicons': 2.1.1 + '@mary-ext/event-iterator': 1.1.0 + '@mary-ext/simple-event-emitter': 1.0.2 + partysocket: 1.3.0 + valibot: 1.5.0(typescript@5.9.3) + transitivePeerDependencies: + - '@atcute/cid' + - react + - typescript + + '@atcute/lexicons@2.1.1': + dependencies: + '@atcute/uint8array': 1.2.0 + '@atcute/util-text': 1.3.4 + '@oomfware/eval': 0.1.0 + '@standard-schema/spec': 1.1.0 + esm-env: 1.2.2 + + '@atcute/multibase@1.2.5': + dependencies: + '@atcute/uint8array': 1.2.0 + + '@atcute/uint8array@1.2.0': {} + + '@atcute/util-text@1.3.4': + dependencies: + unicode-segmenter: 0.17.3 + '@cloudflare/kv-asset-handler@0.5.0': {} '@cloudflare/unenv-preset@2.16.1(unenv@2.0.0-rc.24)(workerd@1.20260903.1)': @@ -1362,9 +1504,17 @@ snapshots: '@jridgewell/resolve-uri': 3.1.2 '@jridgewell/sourcemap-codec': 1.6.0 + '@mary-ext/event-iterator@1.1.0': + dependencies: + yocto-queue: 1.2.2 + + '@mary-ext/simple-event-emitter@1.0.2': {} + '@napi-rs/lzma-linux-x64-gnu@1.5.1': optional: true + '@oomfware/eval@0.1.0': {} + '@poppinss/colors@4.1.6': dependencies: kleur: 4.1.5 @@ -1456,6 +1606,8 @@ snapshots: '@speed-highlight/core@1.2.24': {} + '@standard-schema/spec@1.1.0': {} + '@types/estree@1.0.9': {} '@vitest/expect@3.0.9': @@ -1592,10 +1744,14 @@ snapshots: '@esbuild/win32-ia32': 0.28.1 '@esbuild/win32-x64': 0.28.1 + esm-env@1.2.2: {} + estree-walker@3.0.3: dependencies: '@types/estree': 1.0.9 + event-target-polyfill@0.0.4: {} + expect-type@1.4.0: {} fdir@6.5.0(picomatch@4.0.7): @@ -1629,6 +1785,10 @@ snapshots: nanoid@3.3.18: {} + partysocket@1.3.0: + dependencies: + event-target-polyfill: 0.0.4 + path-to-regexp@6.3.0: {} pathe@2.0.3: {} @@ -1747,6 +1907,12 @@ snapshots: dependencies: pathe: 2.0.3 + unicode-segmenter@0.17.3: {} + + valibot@1.5.0(typescript@5.9.3): + optionalDependencies: + typescript: 5.9.3 + vite-node@3.0.9(supports-color@10.2.2): dependencies: cac: 6.7.14 @@ -1847,6 +2013,8 @@ snapshots: ws@8.21.0: {} + yocto-queue@1.2.2: {} + youch-core@0.3.3: dependencies: '@poppinss/exception': 1.2.3 diff --git a/sitemap/src/hydrant.ts b/sitemap/src/hydrant.ts index 559b32fa5..155f71b64 100644 --- a/sitemap/src/hydrant.ts +++ b/sitemap/src/hydrant.ts @@ -1,4 +1,5 @@ -import type { HydrantEvent, HydrantRecordValue, UrlEntry } from './types'; +import type { JetstreamEvent } from '@atcute/jetstream'; +import type { RecordValue, UrlEntry } from './types'; import { sanitizeLastmod } from './xml'; // uri is the record's at-uri, deletes carry no record so it's what finds their url @@ -7,7 +8,7 @@ export type ExtractedAction = | { type: 'upsert'; uri: string; url: string; entry: UrlEntry }; interface CollectionSpec { - buildPath: (owner: string, rkey: string, record?: HydrantRecordValue) => string; + buildPath: (owner: string, rkey: string, record?: RecordValue) => string; changefreq: UrlEntry['changefreq']; priority: number; } @@ -33,26 +34,27 @@ export const COLLECTION_SPECS: Record = { // owners are dids rather than handles, so a url never moves when someone // changes handle, and the web redirects it to the handle anyway export function extractUrlAction( - event: HydrantEvent, + event: JetstreamEvent, baseUrl: string, now = new Date(), ): ExtractedAction | null { - if (event.type !== 'record' || !event.record) { + if (event.kind !== 'commit') { return null; } - const { collection, rkey, action, did, record } = event.record; + const { collection, rkey, operation } = event.commit; const spec = COLLECTION_SPECS[collection]; if (!spec) { return null; } - const uri = `at://${did}/${collection}/${rkey}`; + const record = event.commit.operation === 'delete' ? undefined : (event.commit.record as RecordValue); + const uri = `at://${event.did}/${collection}/${rkey}`; const normalizedBase = baseUrl.replace(/\/+$/, ''); - const path = spec.buildPath(did, rkey, record); + const path = spec.buildPath(event.did, rkey, record); const url = `${normalizedBase}/${path}`; - if (action === 'delete') { + if (operation === 'delete') { return { type: 'delete', uri, url }; } diff --git a/sitemap/src/index.ts b/sitemap/src/index.ts index f6ad1049b..d9054f756 100644 --- a/sitemap/src/index.ts +++ b/sitemap/src/index.ts @@ -1,5 +1,6 @@ import type { Env, ShardInfo, UrlEntry, WorkerState } from './types'; -import { extractUrlAction, type ExtractedAction } from './hydrant'; +import { COLLECTION_SPECS, extractUrlAction, type ExtractedAction } from './hydrant'; +import { readJetstream, subscribeJetstream, type Subscribe } from './jetstream'; import { flushShardToR2, formatShardFilename, @@ -17,6 +18,21 @@ import { } from './xml'; const KEY_WORKER_STATE = 'worker_state'; +const DEFAULT_HYDRANT_URL = 'https://index.tangled.network'; + +function hydrantUrl(env: Env): string { + return (env.HYDRANT_URL || DEFAULT_HYDRANT_URL).replace(/\/+$/, ''); +} + +export interface SyncDeps { + subscribe: Subscribe; + idleMs: number; +} + +const defaultDeps: SyncDeps = { + subscribe: subscribeJetstream, + idleMs: 3000, +}; export function parseConfigLimits(env: Env): { maxUrls: number; maxBytes: number } { const parsedUrls = parseInt(env.SHARD_MAX_URLS || '', 10); @@ -215,7 +231,7 @@ export async function loadWorkerState(kv: KVNamespace): Promise { } } -export async function syncFromHydrant(env: Env): Promise<{ +export async function syncFromHydrant(env: Env, deps: SyncDeps = defaultDeps): Promise<{ success: boolean; syncedEvents: number; mutatedUrls: number; @@ -228,6 +244,7 @@ export async function syncFromHydrant(env: Env): Promise<{ const now = new Date(); const state = await loadWorkerState(env.SITEMAP_KV); + const hydrant = hydrantUrl(env); const kvStore = new SitemapKvStore(env.SITEMAP_KV); // Initialize shard 1 if needed @@ -270,13 +287,16 @@ export async function syncFromHydrant(env: Env): Promise<{ mutated = true; } - let highestCursor = state.cursor; + let cursor = Number(state.cursor) || 0; let totalEventCount = 0; let totalMutationCount = 0; let page = 0; const MAX_PAGES = 10; const parsedBatch = parseInt(env.SYNC_BATCH_SIZE || '', 10); const BATCH_SIZE = !isNaN(parsedBatch) && parsedBatch > 0 ? parsedBatch : 10000; + const collections = Object.keys(COLLECTION_SPECS); + const startedUs = Date.now() * 1000; + const seen = new Set(); // Track historical shards modified by in-place updates or deletes: shardIndex -> { upserts, deletes } const historicalModifications = new Map< @@ -310,8 +330,20 @@ export async function syncFromHydrant(env: Env): Promise<{ while (page < MAX_PAGES) { page++; - const fetchResult = await fetchHydrantStreamPages(env, highestCursor, BATCH_SIZE); - if (!fetchResult.ok) { + const batch = await readJetstream( + { + url: hydrant, + cursor, + collections, + maxEvents: BATCH_SIZE, + idleMs: deps.idleMs, + untilUs: startedUs, + maxReconnects: 3, + seen, + }, + deps.subscribe, + ); + if (batch.error && batch.events.length === 0) { if (page === 1) { return { success: false, @@ -319,23 +351,14 @@ export async function syncFromHydrant(env: Env): Promise<{ mutatedUrls: 0, newCursor: state.cursor, totalUrls: state.totalUrls, - error: fetchResult.error, + error: batch.error, }; } break; } - const hydrantEvents = fetchResult.events; - if (hydrantEvents.length === 0) { - break; - } - - totalEventCount += hydrantEvents.length; - for (const evt of hydrantEvents) { - highestCursor = String(Math.max(Number(highestCursor), evt.id)); - } - - const actions = hydrantEvents.flatMap((evt) => extractUrlAction(evt, baseUrl, now) ?? []); + totalEventCount += batch.events.length; + const actions = batch.events.flatMap((evt) => extractUrlAction(evt, baseUrl, now) ?? []); await kvStore.prime(actions); for (const action of actions) { @@ -366,7 +389,8 @@ export async function syncFromHydrant(env: Env): Promise<{ totalMutationCount++; } - if (hydrantEvents.length < BATCH_SIZE) { + cursor = batch.cursor; + if (batch.error || !batch.full) { break; } } @@ -449,7 +473,7 @@ export async function syncFromHydrant(env: Env): Promise<{ // Compute total URL count state.totalUrls = state.shards.reduce((acc, s) => acc + s.count, 0); - state.cursor = highestCursor; + state.cursor = String(cursor); // Commit ordering: granular KV entries saved first, then worker state is canonical commit marker if (mutated || totalEventCount > 0 || kvStore.hasDirty()) { @@ -466,92 +490,3 @@ export async function syncFromHydrant(env: Env): Promise<{ totalUrls: state.totalUrls, }; } - -export async function fetchHydrantStreamPages( - env: Env, - startCursor: string, - maxEvents = 10000, - idleTimeoutMs = 3000, -): Promise<{ ok: boolean; events: any[]; error?: string }> { - const client = env.HYDRANT ?? { fetch }; - const endpoint = env.HYDRANT - ? `http://hydrant.internal/stream?cursor=${startCursor}` - : `${(env.HYDRANT_URL || 'https://api.tangled.org').replace(/\/+$/, '')}/stream?cursor=${startCursor}`; - - try { - const res = await client.fetch(endpoint, { - headers: { Upgrade: 'websocket' }, - }); - - const ws = (res as any).webSocket; - if (!ws) { - // Fallback for mocks / non-websocket responses in tests - if (!res.ok) { - const errText = await res.text(); - return { ok: false, events: [], error: `Hydrant returned HTTP ${res.status}: ${errText}` }; - } - const body = await res.json(); - const events: any[] = Array.isArray(body) ? body : (body as any)?.events ?? []; - return { ok: true, events }; - } - - ws.accept(); - - return new Promise((resolve) => { - const events: any[] = []; - let timer: ReturnType | null = null; - - function resetIdleTimer() { - if (timer) clearTimeout(timer); - timer = setTimeout(() => { - cleanup(); - resolve({ ok: true, events }); - }, idleTimeoutMs); - } - - function cleanup() { - if (timer) clearTimeout(timer); - try { - ws.close(1000, 'batch finished'); - } catch {} - } - - ws.addEventListener('message', (event: any) => { - try { - const raw = typeof event.data === 'string' ? event.data : new TextDecoder().decode(event.data); - const parsed = JSON.parse(raw); - if (parsed && typeof parsed === 'object') { - if (parsed.type === 'error') { - cleanup(); - resolve({ ok: false, events, error: parsed.error || 'Hydrant stream error' }); - return; - } - events.push(parsed); - if (events.length >= maxEvents) { - cleanup(); - resolve({ ok: true, events }); - return; - } - } - resetIdleTimer(); - } catch (err: any) { - console.error('[sitemap] Error parsing ws frame:', err); - } - }); - - ws.addEventListener('error', (err: any) => { - cleanup(); - resolve({ ok: events.length > 0, events, error: err?.message || 'WebSocket error' }); - }); - - ws.addEventListener('close', () => { - if (timer) clearTimeout(timer); - resolve({ ok: true, events }); - }); - - resetIdleTimer(); - }); - } catch (err: any) { - return { ok: false, events: [], error: err?.message || 'Network error' }; - } -} diff --git a/sitemap/src/jetstream.ts b/sitemap/src/jetstream.ts new file mode 100644 index 000000000..167bf5fb7 --- /dev/null +++ b/sitemap/src/jetstream.ts @@ -0,0 +1,164 @@ +import { jetstreamEventSchema, type JetstreamEvent } from '@atcute/jetstream'; +import * as v from 'valibot'; + +// jetstream only roughly orders events by time_us, so every connection starts +// this far behind the newest event we've seen and `seen` drops the repeats +export const REWIND_US = 10_000_000; + +export type Subscribe = ( + url: string, + cursor: number, + collections: string[], + onClose: () => void, +) => AsyncIterable; + +async function connect(url: URL): Promise { + const res = await fetch(url, { headers: { Upgrade: 'websocket' } }); + if (!res.webSocket) throw new Error(`jetstream upgrade failed: http ${res.status}`); + res.webSocket.accept(); + return res.webSocket; +} + +function parseEvent(data: string | ArrayBuffer): JetstreamEvent | undefined { + try { + const raw = JSON.parse(typeof data === 'string' ? data : new TextDecoder().decode(data)); + const result = v.safeParse(jetstreamEventSchema, raw); + return result.success ? result.output : undefined; + } catch { + return undefined; + } +} + +// atcute's JetstreamSubscription rides on partysocket, which throws on every +// message in workerd because it clones events with their own constructor. so +// this is a plain socket using atcute's schema, and readJetstream reconnects +export const subscribeJetstream: Subscribe = (url, cursor, collections, onClose) => ({ + [Symbol.asyncIterator]() { + const endpoint = new URL('/subscribe', url); + endpoint.searchParams.set('cursor', String(cursor)); + for (const collection of collections) endpoint.searchParams.append('wantedCollections', collection); + + const queue: JetstreamEvent[] = []; + let closed = false; + let wake: (() => void) | undefined; + let socket: WebSocket | undefined; + + const finish = () => { + if (closed) return; + closed = true; + onClose(); + wake?.(); + }; + + const opened = connect(endpoint).then( + (ws) => { + socket = ws; + ws.addEventListener('message', (msg) => { + const evt = parseEvent(msg.data); + if (!evt) return; + queue.push(evt); + wake?.(); + }); + ws.addEventListener('close', finish); + ws.addEventListener('error', finish); + }, + (err) => { + console.warn('[sitemap] jetstream connect failed:', err); + finish(); + }, + ); + + return { + async next(): Promise> { + await opened; + while (queue.length === 0 && !closed) { + await new Promise((resolve) => (wake = resolve)); + } + const evt = queue.shift(); + return evt ? { value: evt, done: false } : { value: undefined, done: true }; + }, + async return(): Promise> { + closed = true; + wake?.(); + socket?.close(1000, 'done'); + return { value: undefined, done: true }; + }, + }; + }, +}); + +export interface ReadOptions { + url: string; + cursor: number; + collections: string[]; + maxEvents: number; + // quiet for this long means the replay reached the live tip + idleMs: number; + // anything newer happened after the run started, so the backlog is done + untilUs: number; + maxReconnects: number; + seen: Set; +} + +export interface ReadResult { + events: JetstreamEvent[]; + cursor: number; + full: boolean; + error?: string; +} + +type Outcome = 'dropped' | 'idle' | 'caught-up' | 'full'; + +function eventKey(evt: JetstreamEvent): string { + const commit = evt.kind === 'commit' ? `${evt.commit.collection}/${evt.commit.rkey}@${evt.commit.rev}` : ''; + return `${evt.time_us}:${evt.did}:${evt.kind}:${commit}`; +} + +export async function readJetstream(opts: ReadOptions, subscribe = subscribeJetstream): Promise { + const events: JetstreamEvent[] = []; + let cursor = opts.cursor; + let drops = 0; + + while (true) { + let dropped!: () => void; + const drop = new Promise<'dropped'>((resolve) => (dropped = () => resolve('dropped'))); + const from = Math.max(0, cursor - REWIND_US); + const it = subscribe(opts.url, from, opts.collections, () => dropped())[Symbol.asyncIterator](); + + let outcome: Outcome; + try { + outcome = await drain(it, drop); + } finally { + // not awaited, a fake stream parked forever would never let return() finish + it.return?.()?.catch(() => {}); + } + + if (outcome !== 'dropped') { + return { events, cursor, full: outcome === 'full' }; + } + if (++drops > opts.maxReconnects) { + return { events, cursor, full: false, error: `jetstream disconnected ${drops} times` }; + } + } + + async function drain(it: AsyncIterator, drop: Promise<'dropped'>): Promise { + while (true) { + let timer: ReturnType | null = null; + const idle = new Promise<'idle'>((resolve) => (timer = setTimeout(() => resolve('idle'), opts.idleMs))); + const next = await Promise.race([it.next(), drop, idle]).finally(() => clearTimeout(timer)); + + if (next === 'dropped' || next === 'idle') return next; + if (next.done) return 'dropped'; + + const evt = next.value; + const key = eventKey(evt); + if (opts.seen.has(key)) continue; + opts.seen.add(key); + events.push(evt); + cursor = Math.max(cursor, evt.time_us); + + if (evt.time_us >= opts.untilUs) return 'caught-up'; + if (events.length >= opts.maxEvents) return 'full'; + } + } +} diff --git a/sitemap/src/types.ts b/sitemap/src/types.ts index 2eff7166f..19d74c853 100644 --- a/sitemap/src/types.ts +++ b/sitemap/src/types.ts @@ -1,7 +1,6 @@ export interface Env { SITEMAP_KV: KVNamespace; SITEMAP_BUCKET: R2Bucket; - HYDRANT?: Fetcher; BASE_URL?: string; HYDRANT_URL?: string; SHARD_MAX_URLS?: string; @@ -9,40 +8,12 @@ export interface Env { SYNC_BATCH_SIZE?: string; } -export type DbAction = 'create' | 'update' | 'delete'; - -export interface HydrantRecordValue { +export interface RecordValue { name?: string; createdAt?: string; - handle?: string; - description?: string; [key: string]: unknown; } -export interface HydrantRecordPayload { - live?: boolean; - did: string; - handle?: string; - rev?: string; - collection: string; - rkey: string; - action: DbAction; - record?: HydrantRecordValue; - cid?: string; -} - -export interface HydrantIdentityPayload { - did: string; - handle?: string; -} - -export interface HydrantEvent { - id: number; - type: 'record' | 'identity' | 'account'; - record?: HydrantRecordPayload; - identity?: HydrantIdentityPayload; -} - export interface UrlEntry { loc: string; lastmod?: string; diff --git a/sitemap/test/events.ts b/sitemap/test/events.ts new file mode 100644 index 000000000..6453dd9f7 --- /dev/null +++ b/sitemap/test/events.ts @@ -0,0 +1,62 @@ +import type { JetstreamEvent } from '@atcute/jetstream'; +import type { Subscribe } from '../src/jetstream'; + +const SECOND = 1_000_000; + +export function commit( + seconds: number, + did: string, + collection: string, + rkey: string, + operation: 'create' | 'update' | 'delete' = 'create', + record: Record = {}, +): JetstreamEvent { + const base = { rev: `rev-${seconds}`, collection, rkey, operation }; + return { + did, + time_us: seconds * SECOND, + kind: 'commit', + commit: operation === 'delete' ? base : { ...base, cid: 'bafy', record }, + } as JetstreamEvent; +} + +export function identity(seconds: number, did: string, handle: string): JetstreamEvent { + return { + did, + time_us: seconds * SECOND, + kind: 'identity', + identity: { did, handle, seq: seconds, time: new Date(0).toISOString() }, + } as JetstreamEvent; +} + +const parked = () => new Promise(() => {}); + +// replays whatever source() holds from the cursor on, then goes quiet like a +// caught up jetstream. dropAfter[n] closes the nth connection after that many events +export function fakeJetstream(source: () => JetstreamEvent[], dropAfter: (number | undefined)[] = []) { + const cursors: number[] = []; + const subscribe: Subscribe = (_url, cursor, _collections, onClose) => { + const connection = cursors.push(cursor) - 1; + const limit = dropAfter[connection]; + const pending = source().filter((e) => e.time_us >= cursor); + return { + async *[Symbol.asyncIterator]() { + for (const [i, evt] of pending.entries()) { + if (i === limit) break; + yield evt; + } + if (limit !== undefined) onClose(); + await parked(); + }, + }; + }; + return { subscribe, cursors }; +} + +// a jetstream that refuses every connection +export const deadJetstream: Subscribe = (_url, _cursor, _collections, onClose) => ({ + async *[Symbol.asyncIterator]() { + queueMicrotask(onClose); + await parked(); + }, +}); diff --git a/sitemap/test/jetstream.test.ts b/sitemap/test/jetstream.test.ts new file mode 100644 index 000000000..543bcf1b8 --- /dev/null +++ b/sitemap/test/jetstream.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, it } from 'vitest'; +import { readJetstream, REWIND_US, type ReadOptions } from '../src/jetstream'; +import { commit, deadJetstream, fakeJetstream } from './events'; + +const events = Array.from({ length: 30 }, (_, i) => commit(100 + i, 'did:plc:alice', 'sh.tangled.repo', `repo-${i}`)); + +function options(overrides: Partial = {}): ReadOptions { + return { + url: 'https://index.tangled.network', + cursor: 0, + collections: ['sh.tangled.repo'], + maxEvents: 1000, + idleMs: 5, + untilUs: Number.MAX_SAFE_INTEGER, + maxReconnects: 3, + seen: new Set(), + ...overrides, + }; +} + +describe('readJetstream', () => { + it('rewinds on reconnect and drops the replayed events', async () => { + const jetstream = fakeJetstream(() => events, [12]); + const res = await readJetstream(options(), jetstream.subscribe); + + expect(res.error).toBeUndefined(); + expect(res.events.map((e) => e.time_us)).toEqual(events.map((e) => e.time_us)); + expect(jetstream.cursors).toEqual([0, events[11].time_us - REWIND_US]); + expect(res.cursor).toBe(events[29].time_us); + }); + + it('starts every read behind the saved cursor', async () => { + const jetstream = fakeJetstream(() => events); + const res = await readJetstream(options({ cursor: events[20].time_us }), jetstream.subscribe); + + expect(jetstream.cursors).toEqual([events[20].time_us - REWIND_US]); + expect(res.events).toHaveLength(20); + }); + + it('stops once events reach the time the run started', async () => { + const jetstream = fakeJetstream(() => events); + const res = await readJetstream(options({ untilUs: events[4].time_us }), jetstream.subscribe); + + expect(res.events).toHaveLength(5); + expect(res.full).toBe(false); + }); + + it('says when it stopped because the batch filled up', async () => { + const jetstream = fakeJetstream(() => events); + const res = await readJetstream(options({ maxEvents: 10 }), jetstream.subscribe); + + expect(res.events).toHaveLength(10); + expect(res.full).toBe(true); + }); + + it('gives up after too many disconnects', async () => { + const res = await readJetstream(options({ maxReconnects: 2 }), deadJetstream); + + expect(res.events).toHaveLength(0); + expect(res.error).toBe('jetstream disconnected 3 times'); + }); +}); diff --git a/sitemap/test/sitemap.test.ts b/sitemap/test/sitemap.test.ts index 8715d2fa0..fa78cdffa 100644 --- a/sitemap/test/sitemap.test.ts +++ b/sitemap/test/sitemap.test.ts @@ -1,3 +1,4 @@ +import type { JetstreamEvent } from '@atcute/jetstream'; import { describe, expect, it } from 'vitest'; import { extractUrlAction } from '../src/hydrant'; import { @@ -5,7 +6,8 @@ import { getStaticUrls, parseUrlsetXml, } from '../src/sitemap'; -import type { HydrantEvent, ShardInfo, UrlEntry } from '../src/types'; +import type { ShardInfo, UrlEntry } from '../src/types'; +import { commit, identity } from './events'; import { buildSitemapIndexXml, buildUrlsetXml, @@ -82,43 +84,28 @@ describe('Hydrant Event Processing', () => { const now = new Date('2026-09-08T12:00:00Z'); it('builds repo urls from the owner did and the record name', () => { - const event: HydrantEvent = { - id: 101, - type: 'record', - record: { - did: 'did:plc:alice123', - collection: 'sh.tangled.repo', - rkey: 'cool-tool', - action: 'create', - record: { - name: 'cool-tool', - createdAt: '2026-07-15T00:00:00Z', - }, - }, - }; + const event = commit(101, 'did:plc:alice123', 'sh.tangled.repo', '3mhgx3cxdfq22', 'create', { + name: 'cool-tool', + createdAt: '2026-07-15T00:00:00Z', + }); const action = extractUrlAction(event, baseUrl, now); - expect(action).not.toBeNull(); expect(action?.type).toBe('upsert'); + expect(action?.uri).toBe('at://did:plc:alice123/sh.tangled.repo/3mhgx3cxdfq22'); expect(action?.url).toBe('https://tangled.org/did:plc:alice123/cool-tool'); expect(action?.type === 'upsert' && action.entry.lastmod).toBe('2026-07-15'); expect(action?.type === 'upsert' && action.entry.priority).toBe(0.8); }); + it('falls back to the rkey for repos named by it', () => { + const event = commit(101, 'did:plc:alice123', 'sh.tangled.repo', 'cool-tool'); + expect(extractUrlAction(event, baseUrl, now)?.url).toBe('https://tangled.org/did:plc:alice123/cool-tool'); + }); + it('extracts profile creation event', () => { - const event: HydrantEvent = { - id: 102, - type: 'record', - record: { - did: 'did:plc:bob456', - collection: 'sh.tangled.actor.profile', - rkey: 'self', - action: 'create', - record: { - createdAt: '2026-06-01T00:00:00Z', - }, - }, - }; + const event = commit(102, 'did:plc:bob456', 'sh.tangled.actor.profile', 'self', 'create', { + createdAt: '2026-06-01T00:00:00Z', + }); const action = extractUrlAction(event, baseUrl, now); expect(action?.type).toBe('upsert'); @@ -126,76 +113,35 @@ describe('Hydrant Event Processing', () => { }); it('extracts string creation event', () => { - const event: HydrantEvent = { - id: 103, - type: 'record', - record: { - did: 'did:plc:alice123', - collection: 'sh.tangled.string', - rkey: 'tid-3xyz', - action: 'create', - record: { - createdAt: '2026-08-01T00:00:00Z', - }, - }, - }; + const event = commit(103, 'did:plc:alice123', 'sh.tangled.string', 'tid-3xyz', 'create', { + createdAt: '2026-08-01T00:00:00Z', + }); const action = extractUrlAction(event, baseUrl, now); - expect(action?.type).toBe('upsert'); expect(action?.url).toBe('https://tangled.org/did:plc:alice123/strings/tid-3xyz'); expect(action?.type === 'upsert' && action.entry.changefreq).toBe('weekly'); }); it('emits delete action with the record uri', () => { - const event: HydrantEvent = { - id: 104, - type: 'record', - record: { - did: 'did:plc:alice123', - collection: 'sh.tangled.repo', - rkey: 'deleted-repo', - action: 'delete', - }, - }; + const event = commit(104, 'did:plc:alice123', 'sh.tangled.repo', '3mhgx3cxdfq22', 'delete'); const action = extractUrlAction(event, baseUrl, now); expect(action?.type).toBe('delete'); - expect(action?.uri).toBe('at://did:plc:alice123/sh.tangled.repo/deleted-repo'); + expect(action?.uri).toBe('at://did:plc:alice123/sh.tangled.repo/3mhgx3cxdfq22'); }); - it('ignores pull requests, comments, identity events and unknown collections', () => { - const pullEvent: HydrantEvent = { - id: 105, - type: 'record', - record: { - did: 'did:plc:alice123', - collection: 'sh.tangled.repo.pull', - rkey: 'pull-1', - action: 'create', - }, - }; - - expect(extractUrlAction(pullEvent, baseUrl, now)).toBeNull(); - - const commentEvent: HydrantEvent = { - id: 106, - type: 'record', - record: { - did: 'did:plc:alice123', - collection: 'sh.tangled.feed.comment', - rkey: 'comment-1', - action: 'create', - }, - }; - - expect(extractUrlAction(commentEvent, baseUrl, now)).toBeNull(); - - const identityEvent: HydrantEvent = { - id: 107, - type: 'identity', - identity: { did: 'did:plc:alice123', handle: 'alice.tangled.org' }, - }; - expect(extractUrlAction(identityEvent, baseUrl, now)).toBeNull(); + it('ignores pull requests, comments, identity and account events', () => { + const account = { + did: 'did:plc:alice123', + time_us: 107, + kind: 'account', + account: { did: 'did:plc:alice123', active: true, seq: 1, time: now.toISOString() }, + } as JetstreamEvent; + + expect(extractUrlAction(commit(105, 'did:plc:alice123', 'sh.tangled.repo.pull', 'pull-1'), baseUrl, now)).toBeNull(); + expect(extractUrlAction(commit(106, 'did:plc:alice123', 'sh.tangled.feed.comment', 'c-1'), baseUrl, now)).toBeNull(); + expect(extractUrlAction(identity(100, 'did:plc:alice123', 'alice.tangled.org'), baseUrl, now)).toBeNull(); + expect(extractUrlAction(account, baseUrl, now)).toBeNull(); }); }); diff --git a/sitemap/test/sync.test.ts b/sitemap/test/sync.test.ts index 67bd15747..f299d121e 100644 --- a/sitemap/test/sync.test.ts +++ b/sitemap/test/sync.test.ts @@ -1,11 +1,14 @@ import { describe, expect, it } from 'vitest'; -import { parseConfigLimits, syncFromHydrant } from '../src/index'; -import type { Env, HydrantEvent } from '../src/types'; +import type { JetstreamEvent } from '@atcute/jetstream'; +import { parseConfigLimits, syncFromHydrant, type SyncDeps } from '../src/index'; +import type { Env } from '../src/types'; +import { commit, deadJetstream, fakeJetstream, identity } from './events'; import { gzipDecompress } from '../src/xml'; class MockKV { store = new Map(); putCount = 0; + deleteCount = 0; async get(key: string): Promise { return this.store.get(key) ?? null; @@ -17,6 +20,7 @@ class MockKV { } async delete(key: string): Promise { + this.deleteCount++; this.store.delete(key); } } @@ -56,6 +60,10 @@ class MockR2Bucket { } } +function deps(subscribe: SyncDeps['subscribe']): SyncDeps { + return { subscribe, idleMs: 5 }; +} + describe('Full syncFromHydrant workflow with Sharding & Multi-Sync', () => { it('handles multi-sync, updates historical shards in-place, and enforces caps', async () => { const kv = new MockKV(); @@ -64,70 +72,36 @@ describe('Full syncFromHydrant workflow with Sharding & Multi-Sync', () => { // ========================================== // SYNC 1: Initial creation + multi-shard rollover // ========================================== - const sync1Events: HydrantEvent[] = [ - { - id: 1, - type: 'identity', - identity: { did: 'did:plc:dawn', handle: 'dawn' }, - }, - { - id: 2, - type: 'record', - record: { - did: 'did:plc:dawn', - collection: 'sh.tangled.actor.profile', - rkey: 'self', - action: 'create', - record: { createdAt: '2026-08-01T00:00:00Z' }, - }, - }, + const sync1Events: JetstreamEvent[] = [ + identity(1, 'did:plc:dawn', 'dawn.tngl.sh'), + commit(2, 'did:plc:dawn', 'sh.tangled.actor.profile', 'self', 'create', { createdAt: '2026-08-01T00:00:00Z' }), ]; // 58 repo events -> 61 total URLs (2 static + 1 profile + 58 repos) for (let i = 1; i <= 58; i++) { - sync1Events.push({ - id: 10 + i, - type: 'record', - record: { - did: 'did:plc:dawn', - collection: 'sh.tangled.repo', - rkey: `repo-${i}`, - action: 'create', - record: { - name: `repo-${i}`, - createdAt: '2026-08-15T00:00:00Z', - }, - }, - }); + sync1Events.push( + commit(10 + i, 'did:plc:dawn', 'sh.tangled.repo', `repo-${i}`, 'create', { + name: `repo-${i}`, + createdAt: '2026-08-15T00:00:00Z', + }), + ); } let currentMockEvents = sync1Events; - const mockHydrantFetcher = { - fetch: async (url: string | Request) => { - const reqUrl = typeof url === 'string' ? url : url.url; - const u = new URL(reqUrl); - const cursor = Number(u.searchParams.get('cursor') || '0'); - const filtered = currentMockEvents.filter((e) => e.id > cursor); - return new Response(JSON.stringify(filtered), { - status: 200, - headers: { 'Content-Type': 'application/json' }, - }); - }, - }; + const jetstream = fakeJetstream(() => currentMockEvents); const env: Env = { SITEMAP_KV: kv as any, SITEMAP_BUCKET: bucket as any, - HYDRANT: mockHydrantFetcher as any, BASE_URL: 'https://tangled.org', SHARD_MAX_URLS: '25', // 25 max per shard }; - const res1 = await syncFromHydrant(env); + const res1 = await syncFromHydrant(env, deps(jetstream.subscribe)); expect(res1.success).toBe(true); expect(res1.syncedEvents).toBe(60); expect(res1.totalUrls).toBe(61); - expect(res1.newCursor).toBe('68'); + expect(res1.newCursor).toBe('68000000'); // Shards: 25 + 25 + 11 = 61 const state1Raw = await kv.get('worker_state'); @@ -148,56 +122,25 @@ describe('Full syncFromHydrant workflow with Sharding & Multi-Sync', () => { // - Asserts in-place update in historical shard 1 (repo-2) // - Asserts delete against historical sealed shard 1 (repo-1) // ========================================== - const sync2Events: HydrantEvent[] = [ - // New repo from same user - { - id: 100, - type: 'record', - record: { - did: 'did:plc:dawn', - collection: 'sh.tangled.repo', - rkey: 'new-tool', - action: 'create', - record: { createdAt: '2026-09-01T00:00:00Z' }, - }, - }, - // Delete repo-1 which is inside sealed shard 1 - { - id: 101, - type: 'record', - record: { - did: 'did:plc:dawn', - collection: 'sh.tangled.repo', - rkey: 'repo-1', - action: 'delete', - }, - }, - // In-place update for repo-2 inside shard 1 - { - id: 102, - type: 'record', - record: { - did: 'did:plc:dawn', - collection: 'sh.tangled.repo', - rkey: 'repo-2', - action: 'update', - record: { createdAt: '2026-09-05T00:00:00Z' }, - }, - }, + currentMockEvents = [ + commit(100, 'did:plc:dawn', 'sh.tangled.repo', 'new-tool', 'create', { createdAt: '2026-09-01T00:00:00Z' }), + commit(101, 'did:plc:dawn', 'sh.tangled.repo', 'repo-1', 'delete'), + commit(102, 'did:plc:dawn', 'sh.tangled.repo', 'repo-2', 'update', { createdAt: '2026-09-05T00:00:00Z' }), ]; - - currentMockEvents = sync2Events; const kvPutsBeforeSync2 = kv.putCount; - const res2 = await syncFromHydrant(env); + const res2 = await syncFromHydrant(env, deps(jetstream.subscribe)); expect(res2.success).toBe(true); expect(res2.syncedEvents).toBe(3); expect(res2.mutatedUrls).toBe(3); - expect(res2.newCursor).toBe('102'); + expect(res2.newCursor).toBe('102000000'); expect(res2.totalUrls).toBe(61); // -1 deleted, +1 in-place updated, +1 new in active shard // new-tool's shard and record url and the worker state, not every url again expect(kv.putCount - kvPutsBeforeSync2).toBe(3); + // the second run picked up 10s before where the first one stopped + expect(jetstream.cursors).toEqual([0, 58000000]); + const state2Raw = await kv.get('worker_state'); const state2 = JSON.parse(state2Raw!); @@ -224,7 +167,7 @@ describe('Full syncFromHydrant workflow with Sharding & Multi-Sync', () => { const kvPutsBefore = kv.putCount; currentMockEvents = []; - const res3 = await syncFromHydrant(env); + const res3 = await syncFromHydrant(env, deps(jetstream.subscribe)); expect(res3.success).toBe(true); expect(res3.syncedEvents).toBe(0); expect(res3.mutatedUrls).toBe(0); @@ -234,42 +177,32 @@ describe('Full syncFromHydrant workflow with Sharding & Multi-Sync', () => { expect(kv.putCount).toBe(kvPutsBefore); // ========================================== - // SYNC 4: Fetch failure propagation + // SYNC 4: Connection failure propagation // ========================================== - mockHydrantFetcher.fetch = async () => new Response('Internal error', { status: 500 }); - const res4 = await syncFromHydrant(env); + const res4 = await syncFromHydrant(env, deps(deadJetstream)); expect(res4.success).toBe(false); - expect(res4.error).toContain('HTTP 500'); - expect(res4.newCursor).toBe('102'); // Cursor not falsely advanced + expect(res4.error).toContain('disconnected'); + expect(res4.newCursor).toBe('102000000'); // Cursor not falsely advanced }); it('finds deleted and renamed repos through their record uri', async () => { const kv = new MockKV(); const bucket = new MockR2Bucket(); - const repo = (id: number, rkey: string, action: 'create' | 'update' | 'delete', name?: string): HydrantEvent => ({ - id, - type: 'record', - record: { - did: 'did:plc:alice', - collection: 'sh.tangled.repo', - rkey, - action, - record: name ? { name } : undefined, - }, - }); - let events = [repo(1, '3mtid1', 'create', 'old-name'), repo(2, '3mtid2', 'create', 'doomed')]; - const env: Env = { - SITEMAP_KV: kv as any, - SITEMAP_BUCKET: bucket as any, - HYDRANT: { fetch: async () => new Response(JSON.stringify(events)) } as any, - BASE_URL: 'https://tangled.org', - }; + let events = [ + commit(1, 'did:plc:alice', 'sh.tangled.repo', '3mtid1', 'create', { name: 'old-name' }), + commit(2, 'did:plc:alice', 'sh.tangled.repo', '3mtid2', 'create', { name: 'doomed' }), + ]; + const jetstream = fakeJetstream(() => events); + const env: Env = { SITEMAP_KV: kv as any, SITEMAP_BUCKET: bucket as any, BASE_URL: 'https://tangled.org' }; - await syncFromHydrant(env); + await syncFromHydrant(env, deps(jetstream.subscribe)); expect(await kv.get('r:at://did:plc:alice/sh.tangled.repo/3mtid2')).toBe('https://tangled.org/did:plc:alice/doomed'); - events = [repo(20, '3mtid1', 'update', 'new-name'), repo(21, '3mtid2', 'delete')]; - await syncFromHydrant(env); + events = [ + commit(20, 'did:plc:alice', 'sh.tangled.repo', '3mtid1', 'update', { name: 'new-name' }), + commit(21, 'did:plc:alice', 'sh.tangled.repo', '3mtid2', 'delete'), + ]; + await syncFromHydrant(env, deps(jetstream.subscribe)); const shard = await gzipDecompress(await (await bucket.get('sitemaps/sitemap-0001.xml.gz'))!.arrayBuffer()); expect(shard).toContain('https://tangled.org/did:plc:alice/new-name'); @@ -284,48 +217,28 @@ describe('Full syncFromHydrant workflow with Sharding & Multi-Sync', () => { const kv = new MockKV(); const bucket = new MockR2Bucket(); - // 15 total events, with batch size = 5 (so 3 full pages + 1 final check) - const allEvents: HydrantEvent[] = Array.from({ length: 15 }, (_, i) => ({ - id: i + 1, - type: 'record', - record: { - did: 'did:plc:multi', - collection: 'sh.tangled.repo', - rkey: `repo-${i + 1}`, - action: 'create', - record: { createdAt: '2026-08-01T00:00:00Z' }, - }, - })); - - let fetchCount = 0; - const mockHydrant = { - fetch: async (url: string | Request) => { - fetchCount++; - const reqUrl = typeof url === 'string' ? url : url.url; - const u = new URL(reqUrl); - const cursor = Number(u.searchParams.get('cursor') || '0'); - // Return up to 5 events after cursor - const slice = allEvents.filter((e) => e.id > cursor).slice(0, 5); - return new Response(JSON.stringify(slice), { - status: 200, - headers: { 'Content-Type': 'application/json' }, - }); - }, - }; + const allEvents = [ + identity(0, 'did:plc:multi', 'multi-user'), + ...Array.from({ length: 15 }, (_, i) => + commit(i + 1, 'did:plc:multi', 'sh.tangled.repo', `repo-${i + 1}`, 'create', { + createdAt: '2026-08-01T00:00:00Z', + }), + ), + ]; + const jetstream = fakeJetstream(() => allEvents); const env: Env = { SITEMAP_KV: kv as any, SITEMAP_BUCKET: bucket as any, - HYDRANT: mockHydrant as any, BASE_URL: 'https://tangled.org', - SYNC_BATCH_SIZE: '5', + SYNC_BATCH_SIZE: '4', }; - const res = await syncFromHydrant(env); + const res = await syncFromHydrant(env, deps(jetstream.subscribe)); expect(res.success).toBe(true); - expect(res.syncedEvents).toBe(15); - expect(res.newCursor).toBe('15'); - expect(fetchCount).toBe(4); // 3 pages of 5, plus 1 empty terminal check + expect(res.syncedEvents).toBe(16); + expect(res.newCursor).toBe('15000000'); + expect(jetstream.cursors).toHaveLength(5); // 4 full pages of 4, then a quiet one // Verify all 15 repos were indexed for (let i = 1; i <= 15; i++) { @@ -336,30 +249,20 @@ describe('Full syncFromHydrant workflow with Sharding & Multi-Sync', () => { it('enforces byte-size cap rollover', async () => { const kv = new MockKV(); const bucket = new MockR2Bucket(); + const jetstream = fakeJetstream(() => [ + identity(1, 'did:plc:alice', 'alice'), + commit(2, 'did:plc:alice', 'sh.tangled.repo', 'long-repository-name-to-consume-bytes'), + ]); // Tiny 350-byte max cap -> will force rollover after ~2 URLs const env: Env = { SITEMAP_KV: kv as any, SITEMAP_BUCKET: bucket as any, - HYDRANT: { - fetch: async () => new Response(JSON.stringify([ - { - id: 1, - type: 'record', - record: { - did: 'did:plc:alice', - collection: 'sh.tangled.repo', - rkey: 'long-repository-name-to-consume-bytes', - action: 'create', - }, - }, - ])), - } as any, BASE_URL: 'https://tangled.org', SHARD_MAX_BYTES: '350', // very low byte limit }; - const res = await syncFromHydrant(env); + const res = await syncFromHydrant(env, deps(jetstream.subscribe)); expect(res.success).toBe(true); const stateRaw = await kv.get('worker_state'); diff --git a/sitemap/wrangler.dev.jsonc b/sitemap/wrangler.dev.jsonc index 7b55544db..8149cc735 100644 --- a/sitemap/wrangler.dev.jsonc +++ b/sitemap/wrangler.dev.jsonc @@ -27,15 +27,9 @@ "bucket_name": "tangled-sitemaps-dev", }, ], - "vpc_services": [ - { - "binding": "HYDRANT", - "service_id": "01a05715-1857-7561-abab-104563fe3970", - }, - ], "vars": { "BASE_URL": "https://next.tangled.org", - "HYDRANT_URL": "https://api.tangled.org", + "HYDRANT_URL": "https://index.tangled.network", "SHARD_MAX_URLS": "50000", }, } diff --git a/sitemap/wrangler.jsonc b/sitemap/wrangler.jsonc index cac7ed846..fb519abbb 100644 --- a/sitemap/wrangler.jsonc +++ b/sitemap/wrangler.jsonc @@ -25,15 +25,9 @@ "bucket_name": "tangled-sitemaps", }, ], - "vpc_services": [ - { - "binding": "HYDRANT", - "service_id": "01a05715-1857-7561-abab-104563fe3970", - }, - ], "vars": { "BASE_URL": "https://tangled.org", - "HYDRANT_URL": "https://api.tangled.org", + "HYDRANT_URL": "https://index.tangled.network", "SHARD_MAX_URLS": "50000", }, } -- 2.51.2