Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep, type WorkflowStepConfig } from 'cloudflare:workers';import { applyChanges, countUrls, getState, markStale, markSynced, putState, saveAccounts, saveDeclarations, saveHandles, staleOwners,} from './db';import { COLLECTION_SPECS, DECLARATION, extractDeclaration, extractUrlAction } from './hydrant';import { hydrantRepos, type RepoHandle, type Repos } from './identity';import { readJetstream, subscribeJetstream, type Subscribe } from './jetstream';import { getStaticUrls, renderShards, resetSitemap } from './sitemap';import type { Env } from './types';import { PROTOCOL_MAX_BYTES, PROTOCOL_MAX_URLS } from './xml';
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; repos: (env: Env) => Repos; idleMs: number; closeMs: number;}
const defaultDeps: SyncDeps = { subscribe: subscribeJetstream, // the vpc service only takes http and routes by itself, so the host there means nothing repos: (env) => (env.HYDRANT ? hydrantRepos(env.HYDRANT, 'http://hydrant') : hydrantRepos(globalThis, hydrantUrl(env))), idleMs: 3000, closeMs: 10_000,};
export function parseConfigLimits(env: Env): { maxUrls: number; maxBytes: number } { const parsedUrls = parseInt(env.SHARD_MAX_URLS || '', 10); const parsedBytes = parseInt(env.SHARD_MAX_BYTES || '', 10);
const maxUrls = !isNaN(parsedUrls) && parsedUrls > 0 ? Math.min(parsedUrls, PROTOCOL_MAX_URLS) : PROTOCOL_MAX_URLS;
const maxBytes = !isNaN(parsedBytes) && parsedBytes > 0 ? Math.min(parsedBytes, PROTOCOL_MAX_BYTES) : PROTOCOL_MAX_BYTES;
return { maxUrls, maxBytes };}
export default { async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> { const url = new URL(request.url); const pathname = url.pathname;
if (pathname === '/sitemap.xml') { const obj = await env.SITEMAP_BUCKET.get('sitemap.xml'); if (!obj) { return new Response('Sitemap index not found. Has sync run yet?', { status: 404 }); }
return new Response(obj.body, { headers: { 'Content-Type': 'application/xml; charset=utf-8', 'Cache-Control': 'public, max-age=86400, s-maxage=86400', }, }); }
if (pathname.startsWith('/sitemaps/')) { const filename = pathname.replace(/^\/sitemaps\//, ''); const obj = await env.SITEMAP_BUCKET.get(`sitemaps/${filename}`); if (!obj) { return new Response('Sitemap shard not found', { status: 404 }); }
return new Response(obj.body, { headers: { 'Content-Type': 'application/gzip', 'Cache-Control': 'public, max-age=3600, stale-while-revalidate=86400', }, }); }
return new Response('Not Found', { status: 404 }); },};
export interface Steps { do<T>(name: string, fn: () => Promise<T>): Promise<T>;}
// the workflow checkpoints every step, tests and local runs just call themconst inline: Steps = { do: (_name, fn) => fn() };
export interface SyncOptions { reset?: boolean; startedAt?: Date;}
export interface SyncResult { syncedEvents: number; mutatedUrls: number; walkedRepos: number; lookedUpHandles: number; newCursor: string; totalUrls: number;}
const MAX_PAGES = 10;
const REPO_PAGE = 1000;const WALK_PAGES = 10;const MAX_WALK_STEPS = 100;
// a page of reads that all time out still fits well inside a stepconst LOOKUP_PAGE = 200;const LOOKUP_CONCURRENCY = 8;const MAX_LOOKUP_STEPS = 10;
// the first run, and the first after a reset, pages through every handle hydrant has// instead of reading owners one by one. a walk that gets cut off carries on next runasync function walkRepos( db: D1Database, repos: Repos, now: Date,): Promise<{ repos: number; done: boolean; failed: boolean }> { if (await getState(db, 'handles_walked')) return { repos: 0, done: true, failed: false }; let after = await getState(db, 'handles_after'); let count = 0; try { for (let page = 0; page < WALK_PAGES; page++) { const found = await repos.list(after, REPO_PAGE); await saveHandles(db, found); count += found.length; if (found.length < REPO_PAGE) { await putState(db, 'handles_walked', now.toISOString()); return { repos: count, done: true, failed: false }; } after = found[found.length - 1].did; await putState(db, 'handles_after', after); } } catch (err) { // the sitemap still goes out, just with dids where handles are missing console.warn(`[sitemap] couldn't page through hydrant's repos: ${err}`); return { repos: count, done: false, failed: true }; } return { repos: count, done: false, failed: false };}
// owners new since the walk or touched by an identity event. a failed read stays// stale for the next runasync function lookUpHandles( db: D1Database, repos: Repos, after: string,): Promise<{ looked: number; failed: number; last: string }> { const dids = await staleOwners(db, after, LOOKUP_PAGE); const found: RepoHandle[] = []; let failed = 0; let next = 0; const worker = async () => { while (next < dids.length) { const did = dids[next++]; try { found.push({ did, handle: await repos.get(did) }); } catch { failed++; } } }; await Promise.all(Array.from({ length: LOOKUP_CONCURRENCY }, worker)); await saveHandles(db, found); if (failed > 0) console.warn(`[sitemap] ${failed} of ${dids.length} handle reads failed`); return { looked: dids.length, failed, last: dids.at(-1) ?? after };}
export async function syncFromHydrant( env: Env, deps: SyncDeps = defaultDeps, steps: Steps = inline, options: SyncOptions = {},): Promise<SyncResult> { const db = env.SITEMAP_DB; const baseUrl = env.BASE_URL || 'https://tangled.org'; const limits = parseConfigLimits(env); // a workflow replays this on resume, so the time has to come from the instance const now = options.startedAt ?? new Date(); const parsedBatch = parseInt(env.SYNC_BATCH_SIZE || '', 10); const batchSize = !isNaN(parsedBatch) && parsedBatch > 0 ? parsedBatch : 10000; const seen = new Set<string>(); let syncedEvents = 0; let mutatedUrls = 0;
if (options.reset) { await steps.do('reset', () => resetSitemap(db, env.SITEMAP_BUCKET)); }
mutatedUrls += await steps.do('seed', async () => { if ((await countUrls(db)) > 0) return 0; const seeds = getStaticUrls(baseUrl, now).map((entry) => ({ type: 'upsert' as const, url: entry.loc, entry })); return applyChanges(db, seeds, limits, undefined, now); });
for (let page = 1; page <= MAX_PAGES; page++) { const batch = await steps.do(`batch ${page}`, async () => { // read here rather than carried over, so a retried step starts from what landed const cursor = Number(await getState(db, 'cursor')) || 0; const read = await readJetstream( { url: hydrantUrl(env), cursor, collections: [...Object.keys(COLLECTION_SPECS), DECLARATION], maxEvents: batchSize, idleMs: deps.idleMs, closeMs: deps.closeMs, untilUs: now.getTime() * 1000, maxReconnects: 3, seen, }, deps.subscribe, ); if (read.events.length === 0) { if (read.error) throw new Error(read.error); return { events: 0, mutations: 0, more: false }; } const changes = read.events.flatMap((evt) => extractUrlAction(evt, baseUrl, now) ?? []); // before the cursor moves, so a retried batch can't skip it await markStale(db, read.events.flatMap((evt) => (evt.kind === 'identity' ? [evt.did] : []))); await saveAccounts(db, read.events.flatMap((evt) => (evt.kind === 'account' ? [evt.account] : []))); await saveDeclarations(db, read.events.flatMap((evt) => extractDeclaration(evt) ?? [])); const mutations = await applyChanges(db, changes, limits, String(read.cursor), now); return { events: read.events.length, mutations, more: read.full && !read.error }; }); syncedEvents += batch.events; mutatedUrls += batch.mutations; if (!batch.more) break; }
const repos = deps.repos(env); let walkedRepos = 0; let walked = false; for (let step = 1; step <= MAX_WALK_STEPS && !walked; step++) { const res = await steps.do(`walk ${step}`, () => walkRepos(db, repos, now)); walkedRepos += res.repos; walked = res.done; if (res.failed) break; }
// until the walk is through, most owners are still in it let lookedUpHandles = 0; let after = ''; for (let step = 1; walked && step <= MAX_LOOKUP_STEPS; step++) { const res = await steps.do(`lookups ${step}`, () => lookUpHandles(db, repos, after)); lookedUpHandles += res.looked; // nothing answering means hydrant is down, so render with what we have if (res.looked < LOOKUP_PAGE || res.failed === res.looked) break; after = res.last; }
await steps.do('render', () => renderShards(db, env.SITEMAP_BUCKET, baseUrl, now));
return steps.do('finish', async () => { if (syncedEvents > 0) await markSynced(db, now); return { syncedEvents, mutatedUrls, walkedRepos, lookedUpHandles, newCursor: (await getState(db, 'cursor')) ?? '0', totalUrls: await countUrls(db), }; });}
const STEP_CONFIG: WorkflowStepConfig = { retries: { limit: 3, delay: '30 seconds', backoff: 'exponential' }, timeout: '15 minutes',};
// started by its own schedule at midnight, or by hand with// wrangler workflows trigger <name> '{"reset":true}'export class SyncWorkflow extends WorkflowEntrypoint<Env, SyncOptions> { async run(event: Readonly<WorkflowEvent<SyncOptions>>, step: WorkflowStep): Promise<SyncResult> { const steps: Steps = { do: (name, fn) => step.do(name, STEP_CONFIG, fn as () => Promise<never>) }; const res = await syncFromHydrant(this.env, defaultDeps, steps, { reset: event.payload?.reset, startedAt: event.timestamp, }); console.log(`[sitemap] sync finished: ${JSON.stringify(res)}`); return res; }}