import { Agent } from "undici"; import { didTag, handleTag, lookups, type LookupCache } from "#lib/api/lookupCache.ts"; // identity and account events always arrive. these two add a repo's own records export const LOOKUP_COLLECTIONS = ["org.tangled.repo.manifest", "org.tangled.git.refs"]; // bobbin applies the same event from its own stream. a lookup refetched before it has // would keep the old answer const REEVICT_MS = 10_000; const MIN_RETRY_MS = 1_000; const MAX_RETRY_MS = 60_000; const REPORT_MS = 60_000; interface JetstreamEvent { did?: unknown; kind?: unknown; identity?: { handle?: unknown }; commit?: { collection?: unknown }; } export const tagsForEvent = (event: JetstreamEvent): string[] => { if (typeof event.did !== "string") return []; const tags = [didTag(event.did)]; switch (event.kind) { case "identity": if (typeof event.identity?.handle === "string") tags.push(handleTag(event.identity.handle)); return tags; case "account": return tags; case "commit": return LOOKUP_COLLECTIONS.includes(String(event.commit?.collection)) ? tags : []; default: return []; } }; export const subscribeUrl = (url: string): string => { const parsed = new URL(url); for (const collection of LOOKUP_COLLECTIONS) parsed.searchParams.append("wantedCollections", collection); return parsed.href; }; interface Socket { onopen: (() => void) | null; onmessage: ((message: { data: unknown }) => void) | null; onclose: (() => void) | null; close(): void; } export interface LookupEventsOptions { url: string; cache?: LookupCache; connect?: (url: string) => Socket; } export interface LookupEvents { readonly connected: boolean; close(): void; } // the cache is cleared on every open because events missed while down are gone export const subscribeLookupEvents = ({ url, cache = lookups, // node's websocket would use the global dispatcher, whose http/2 can't upgrade connect = (target) => new WebSocket(target, { dispatcher: new Agent() } as never) as unknown as Socket }: LookupEventsOptions): LookupEvents => { const target = subscribeUrl(url); let socket: Socket | undefined; let connected = false; let closed = false; let unreachable = false; let retryMs = MIN_RETRY_MS; const evict = (tags: string[]) => { cache.evict(tags); setTimeout(() => cache.evict(tags), REEVICT_MS).unref?.(); }; const open = () => { const current = connect(target); socket = current; current.onopen = () => { cache.clear(); connected = true; unreachable = false; retryMs = MIN_RETRY_MS; console.info(`[lookups] following ${url}`); }; current.onmessage = ({ data }) => { if (typeof data !== "string") return; let event: JetstreamEvent; try { event = JSON.parse(data) as JetstreamEvent; } catch { return; } const tags = tagsForEvent(event); if (tags.length > 0) evict(tags); }; current.onclose = () => { if (socket !== current) return; if (connected) { console.warn(`[lookups] lost ${url}, cached lookups live out their ttl`); } else if (!unreachable) { console.warn(`[lookups] can't reach ${url}, retrying`); unreachable = true; } connected = false; if (closed) return; const wait = retryMs / 2 + Math.random() * (retryMs / 2); retryMs = Math.min(retryMs * 2, MAX_RETRY_MS); setTimeout(open, wait).unref?.(); }; }; open(); return { get connected() { return connected; }, close() { closed = true; socket?.close(); } }; }; export const startLookups = (config: { lookupCacheTtlMs: number; hydrantSubscribeUrl: string }) => { lookups.configure(config.lookupCacheTtlMs); if (config.lookupCacheTtlMs === 0) return; const events = subscribeLookupEvents({ url: config.hydrantSubscribeUrl }); setInterval(() => { const { hits, misses, evictions, clears } = lookups.stats; if (hits + misses + evictions + clears === 0) return; console.info( `[lookups] hits=${hits} misses=${misses} evictions=${evictions} clears=${clears} entries=${lookups.size} connected=${events.connected}` ); lookups.stats = { hits: 0, misses: 0, evictions: 0, clears: 0 }; }, REPORT_MS).unref?.(); };