From 708d1b028fa4e7d58432e5247d2291b42a682ebb Mon Sep 17 00:00:00 2001 From: Philippe Serhal Date: Sun, 8 Nov 2026 18:49:43 -0500 Subject: [PATCH 1/5] feat(social): index collections from Jetstream Nothing on the network can list every collection, so browse needs its own index. drainJetstream replays Jetstream from a saved cursor until it catches up or runs out of time, and keeps collections plus distinct-account like and follow counts in Redis. Every operation is idempotent, so replaying overlap is safe. Local dev and tests run the same store on an in-memory stand-in for the Redis commands it uses. --- .../utils/atproto/utils/collections-index.ts | 362 +++++++++++++++++ server/utils/atproto/utils/collections.ts | 4 +- server/utils/atproto/utils/memory-redis.ts | 97 +++++ shared/types/social.ts | 10 + .../server/utils/collections-index.spec.ts | 369 ++++++++++++++++++ 5 files changed, 840 insertions(+), 2 deletions(-) create mode 100644 server/utils/atproto/utils/collections-index.ts create mode 100644 server/utils/atproto/utils/memory-redis.ts create mode 100644 test/unit/server/utils/collections-index.spec.ts diff --git a/server/utils/atproto/utils/collections-index.ts b/server/utils/atproto/utils/collections-index.ts new file mode 100644 index 0000000000..05cda221b0 --- /dev/null +++ b/server/utils/atproto/utils/collections-index.ts @@ -0,0 +1,362 @@ +import { on } from 'node:events' +import { Redis } from '@upstash/redis' +import { createMemoryRedis, type IndexRedis } from './memory-redis' +import * as dev from '#shared/types/lexicons/dev' +import { parseSubjectRef } from '#shared/utils/collections' +import { COLLECTION_SUBJECT_REF } from '#shared/utils/constants' +import type { CollectionStats } from '#shared/types/social' +import { toCollectionFields, type IndexedCollection, type Reaction } from './collections' + +/** + * A global index of dev.npmx.collection records, fed from Jetstream. Nothing else on the network + * can enumerate every collection, so this is what powers browse. It is an accelerator, not a + * source of truth: the PDS is, and gaps are acceptable. + */ + +const JETSTREAM_URL = + 'wss://jetstream.us-east.bsky.network/xrpc/network.bsky.jetstream.subscribeEvents' +const COLLECTION_NSID = dev.npmx.collection.$nsid +const LIKE_NSID = dev.npmx.feed.like.$nsid +const FOLLOW_NSID = dev.npmx.graph.follow.$nsid + +const US_PER_MS = 1000 +// First run only: Jetstream maps a timestamp cursor to a seq. Retention is about 36h, and an +// older cursor starts at the retention floor anyway. +export const INITIAL_LOOKBACK_US = 35 * 60 * 60 * 1000 * US_PER_MS +// Within this lag of now means we've caught up. +const CAUGHT_UP_LAG_US = 10 * 1000 * US_PER_MS +const CURSOR_SAVE_EVERY = 100 +const MAX_EVENT_ATTEMPTS = 3 + +export type BrowseSort = 'recent' | 'liked' | 'followed' + +// Rankings count distinct accounts, so one account can't stack records. Removing one of an +// account's duplicate records undercounts by one; that only costs the account doing it. +const authorOf = (recordKey: string) => recordKey.split('/')[0]! + +const REACTION_SORT = { like: 'liked', follow: 'followed' } as const satisfies Record< + Reaction, + BrowseSort +> +const REACTIONS = Object.keys(REACTION_SORT) as Reaction[] + +// Exported for tests, which check what's stored +export const KEY = { + cursor: 'collections-index:cursor', + failedEvent: 'collections-index:failed-event', + sorted: (sort: BrowseSort) => `collections-index:${sort}`, + item: (uri: string) => `collections-index:item:${uri}`, + record: (reaction: Reaction, recordKey: string) => `collections-index:${reaction}:${recordKey}`, + authors: (reaction: Reaction, uri: string) => + `collections-index:${reaction === 'like' ? 'likers' : 'followers'}:${uri}`, +} + +type StoredEntry = IndexedCollection & Pick + +/** Browse's index in Redis: an entry per collection plus a sorted set per sort order. */ +export class CollectionsIndexStore { + constructor(private readonly redis: IndexRedis) {} + + /** A Jetstream seq. */ + async getCursor() { + return (await this.redis.get(KEY.cursor)) ?? null + } + + async setCursor(seq: number) { + await this.redis.set(KEY.cursor, seq) + } + + async clearCursor() { + await this.redis.del(KEY.cursor) + } + + /** How many runs in a row have failed to apply the event at `seq`, counting this one. */ + async recordFailedEvent(seq: number) { + const last = await this.redis.get<{ seq: number; attempts: number }>(KEY.failedEvent) + const attempts = last?.seq === seq ? last.attempts + 1 : 1 + await this.redis.set(KEY.failedEvent, { seq, attempts }) + return attempts + } + + /** `seenUs` ranks in recent only the first time a collection is seen, so edits don't bump it. */ + async upsertCollection(entry: IndexedCollection, seenUs: number) { + const stored: StoredEntry = { + ...entry, + updatedAt: new Date(Math.floor(seenUs / US_PER_MS)).toISOString(), + } + await Promise.all([ + this.redis.set(KEY.item(entry.uri), stored), + this.redis.zadd(KEY.sorted('recent'), { nx: true }, { score: seenUs, member: entry.uri }), + ]) + } + + async removeCollection(uri: string) { + await Promise.all([ + this.redis.del(KEY.item(uri)), + this.redis.zrem(KEY.sorted('recent'), uri), + ...REACTIONS.flatMap(reaction => [ + this.redis.zrem(KEY.sorted(REACTION_SORT[reaction]), uri), + this.redis.del(KEY.authors(reaction, uri)), + ]), + ]) + } + + /** + * `recordKey` is `/` of a like or follow. Ignored if the collection + * isn't indexed (made-up URIs). Every step is idempotent, so a replay or a retry after a partial + * failure converges. + */ + async addReaction(reaction: Reaction, recordKey: string, subjectUri: string) { + if (!(await this.redis.exists(KEY.item(subjectUri)))) return + await this.redis.sadd(KEY.authors(reaction, subjectUri), authorOf(recordKey)) + await this.syncScore(reaction, subjectUri) + // Last, so removeReaction never finds a record whose author wasn't counted + await this.redis.set(KEY.record(reaction, recordKey), subjectUri) + } + + async removeReaction(reaction: Reaction, recordKey: string) { + const subjectUri = await this.redis.getdel(KEY.record(reaction, recordKey)) + if (!subjectUri) return + await this.redis.srem(KEY.authors(reaction, subjectUri), authorOf(recordKey)) + await this.syncScore(reaction, subjectUri) + } + + // Set rather than incremented, so a lost or repeated update can't drift the score. Collections + // nobody has liked or followed stay out, rather than filling those sorts in arbitrary order. + private async syncScore(reaction: Reaction, uri: string) { + const key = KEY.sorted(REACTION_SORT[reaction]) + const authors = await this.redis.scard(KEY.authors(reaction, uri)) + await (authors > 0 + ? this.redis.zadd(key, { score: authors, member: uri }) + : this.redis.zrem(key, uri)) + } + + /** `scanned` counts index slots read, including stale ones skipped, for paging. */ + async list(sort: BrowseSort, offset: number, limit: number) { + const uris = await this.redis.zrange(KEY.sorted(sort), offset, offset + limit - 1, { + rev: true, + }) + if (uris.length === 0) return { entries: [], scanned: 0 } + // The reaction sorts' scores are the counts, and a missing member has none + const [stored, likes, follows] = await Promise.all([ + this.redis.mget<(StoredEntry | null)[]>(...uris.map(KEY.item)), + this.redis.zmscore(KEY.sorted(REACTION_SORT.like), uris), + this.redis.zmscore(KEY.sorted(REACTION_SORT.follow), uris), + ]) + const entries = stored.flatMap((entry, i) => { + if (!entry) return [] + const { updatedAt, ...fields } = entry + const stats: CollectionStats = { + likes: likes?.[i] ?? 0, + follows: follows?.[i] ?? 0, + updatedAt, + } + return [{ ...fields, stats }] + }) + return { entries, scanned: uris.length } + } +} + +let localStore: CollectionsIndexStore | undefined + +export function getCollectionsIndexStore(): CollectionsIndexStore { + const config = useRuntimeConfig() + if (!import.meta.dev && config.upstash?.redisRestUrl && config.upstash?.redisRestToken) { + return new CollectionsIndexStore( + new Redis({ url: config.upstash.redisRestUrl, token: config.upstash.redisRestToken }), + ) + } + return (localStore ??= new CollectionsIndexStore(createMemoryRedis())) +} + +/** A Jetstream record as an index entry. Null if the record isn't a valid collection. */ +function toIndexedCollection( + did: string, + rkey: string, + cid: string | undefined, + record: unknown, +): IndexedCollection | null { + const result = dev.npmx.collection.$safeValidate(record) + if (!result.success || !cid) return null + return toCollectionFields(did, rkey, cid, result.value) +} + +const EVENT_TYPE = 'network.bsky.jetstream.subscribeEvents' + +/** A Jetstream v2 frame's `payload`. Commits carry their fields at the top level. */ +export type JetstreamPayload = { + $type: string + seq?: number + time?: string + did?: string + operation?: 'create' | 'update' | 'delete' + collection?: string + rkey?: string + cid?: string + record?: unknown + name?: string + message?: string +} + +const payloadTimeUs = (payload: JetstreamPayload) => + payload.time ? Date.parse(payload.time) * US_PER_MS : Number.NaN + +/** Applies one Jetstream event to the index. Idempotent, so replaying overlap is safe. */ +export async function applyJetstreamEvent(store: CollectionsIndexStore, event: JetstreamPayload) { + const { did, rkey, operation } = event + if (event.$type !== `${EVENT_TYPE}#commit` || !did || !rkey || !operation) return + + if (event.collection === COLLECTION_NSID) { + const uri = COLLECTION_SUBJECT_REF(did, rkey) + const entry = + operation === 'delete' ? null : toIndexedCollection(did, rkey, event.cid, event.record) + const seenUs = payloadTimeUs(event) + // An edit that made the record invalid should take it out of browse too + if (entry) { + await store.upsertCollection(entry, Number.isFinite(seenUs) ? seenUs : Date.now() * US_PER_MS) + } else await store.removeCollection(uri) + return + } + + const reaction = + event.collection === LIKE_NSID ? 'like' : event.collection === FOLLOW_NSID ? 'follow' : null + if (!reaction) return + const recordKey = `${did}/${rkey}` + if (operation === 'delete') { + await store.removeReaction(reaction, recordKey) + return + } + if (operation !== 'create') return + const subject = + reaction === 'like' + ? dev.npmx.feed.like.$safeValidate(event.record) + : dev.npmx.graph.follow.$safeValidate(event.record) + if (!subject.success) return + const subjectUri = + 'subjectRef' in subject.value ? subject.value.subjectRef : subject.value.subject + // Package likes are counted via Constellation; only collections matter here. + if (parseSubjectRef(subjectUri)?.kind !== 'collection') return + await store.addReaction(reaction, recordKey, subjectUri) +} + +export type DrainResult = { + events: number + fromCursor: number + toCursor: number + caughtUp: boolean +} + +export type DrainDeps = { + store: CollectionsIndexStore + /** + * Opens Jetstream at `url` and yields raw messages, ending when `signal` aborts. Stopping + * iteration must close it. + */ + subscribe: (url: string, signal: AbortSignal) => AsyncIterable + nowUs?: () => number + maxDurationMs?: number +} + +/** + * `identity` events (~15/min network-wide) are the clock. Replay scans history server-side and + * sends nothing when nothing matches, so with commits alone a quiet stream could mean "caught up" + * or "still scanning". + */ +function jetstreamUrl(cursor: number) { + const params = new URLSearchParams([ + ['collections', COLLECTION_NSID], + ['collections', LIKE_NSID], + ['collections', FOLLOW_NSID], + ['kinds', 'commit'], + ['kinds', 'identity'], + ['cursor', String(cursor)], + ]) + return `${JETSTREAM_URL}?${params}` +} + +/** + * Replays Jetstream from the stored cursor until caught up or out of time, saving the cursor as + * it goes so a run that times out still makes progress. + */ +export async function drainJetstream({ + store, + subscribe, + nowUs = () => Date.now() * US_PER_MS, + // Replay runs at about an hour of history per second, so this covers Jetstream's whole + // retention. Stays under the 300s function timeout, so the final cursor write runs, and + // under the 5-minute cron interval, since nothing stops two runs from overlapping. + maxDurationMs = 180_000, +}: DrainDeps): Promise { + const startedAt = nowUs() + const deadline = startedAt + maxDurationMs * US_PER_MS + const storedCursor = await store.getCursor() + const fromCursor = storedCursor ?? startedAt - INITIAL_LOOKBACK_US + + let cursor = fromCursor + let received = 0 + let events = 0 + let caughtUp = false + + // Ends the stream even if Jetstream goes quiet or never connects + const signal = AbortSignal.timeout(maxDurationMs) + for await (const message of subscribe(jetstreamUrl(fromCursor), signal)) { + received++ + let payload: JetstreamPayload | undefined + try { + payload = JSON.parse(message)?.payload + } catch { + continue + } + if (!payload) continue + if (payload.$type === `${EVENT_TYPE}#info`) { + console.warn(`[collections-index] Jetstream ${payload.name}: ${payload.message}`) + continue + } + try { + await applyJetstreamEvent(store, payload) + } catch (error) { + const attempts = + payload.seq == null ? MAX_EVENT_ATTEMPTS : await store.recordFailedEvent(payload.seq) + console.error('[collections-index] Failed to apply event', payload.seq, attempts, error) + // Stop and let the next run retry it, unless it keeps failing: then skip it, or it would + // stall the index for good + if (attempts < MAX_EVENT_ATTEMPTS) break + } + events++ + if (payload.seq != null) cursor = payload.seq + if (events % CURSOR_SAVE_EVERY === 0) await store.setCursor(cursor) + + const now = nowUs() + if (payloadTimeUs(payload) >= now - CAUGHT_UP_LAG_US) { + caughtUp = true + break + } + if (now >= deadline) break + } + + // Not even the identity clock arrived, so Jetstream refused the connection. Most likely the + // cursor is past retention (HTTP 400 CursorTooOld), which would fail every run: start over. + if (received === 0 && storedCursor !== null) { + await store.clearCursor() + return { events, fromCursor, toCursor: fromCursor, caughtUp } + } + + await store.setCursor(cursor) + return { events, fromCursor, toCursor: cursor, caughtUp } +} + +/** Jetstream's text messages until the socket closes or errors, or `signal` aborts. */ +export async function* webSocketMessages(url: string, signal: AbortSignal): AsyncGenerator { + const socket = new WebSocket(url) + socket.addEventListener('error', () => console.warn('[collections-index] Jetstream socket error')) + try { + for await (const [event] of on(socket, 'message', { close: ['close', 'error'], signal })) { + yield String((event as MessageEvent).data) + } + } catch (error) { + // Aborting is how a run ends on time, not a failure + if (!signal.aborted) throw error + } finally { + socket.close() + } +} diff --git a/server/utils/atproto/utils/collections.ts b/server/utils/atproto/utils/collections.ts index ff05ab6fa2..36221aa2af 100644 --- a/server/utils/atproto/utils/collections.ts +++ b/server/utils/atproto/utils/collections.ts @@ -67,10 +67,10 @@ export type CollectionsDeps = { moderation: Pick } -type IndexedCollection = Omit +export type IndexedCollection = Omit /** A record's listing fields, as browse indexes them and summaries show them. */ -function toCollectionFields( +export function toCollectionFields( did: string, rkey: string, cid: string, diff --git a/server/utils/atproto/utils/memory-redis.ts b/server/utils/atproto/utils/memory-redis.ts new file mode 100644 index 0000000000..f2a1d0b267 --- /dev/null +++ b/server/utils/atproto/utils/memory-redis.ts @@ -0,0 +1,97 @@ +import type { Redis } from '@upstash/redis' + +/** The Redis commands the collections index uses. */ +export type IndexRedis = Pick< + Redis, + | 'get' + | 'set' + | 'del' + | 'exists' + | 'getdel' + | 'mget' + | 'sadd' + | 'srem' + | 'scard' + | 'smembers' + | 'zadd' + | 'zrem' + | 'zrange' + | 'zmscore' +> + +type ZMember = { score: number; member: string } + +// Upstash round-trips values through JSON, so callers never share references with the store +const clone = (value: T): T => structuredClone(value) + +/** + * An in-memory stand-in for the subset of Upstash Redis in `IndexRedis`, so local dev and tests + * run the same index code as production. Not shared across serverless instances. + */ +export function createMemoryRedis(): IndexRedis { + const strings = new Map() + const sets = new Map>() + const zsets = new Map>() + const del = (key: string) => [strings, sets, zsets].some(map => map.delete(key)) + + const redis = { + get: async (key: string) => (strings.has(key) ? clone(strings.get(key)) : null), + set: async (key: string, value: unknown) => { + strings.set(key, clone(value)) + return 'OK' + }, + del: async (...keys: string[]) => keys.filter(del).length, + exists: async (...keys: string[]) => + keys.filter(key => strings.has(key) || sets.has(key) || zsets.has(key)).length, + getdel: async (key: string) => { + const value = strings.has(key) ? clone(strings.get(key)) : null + strings.delete(key) + return value + }, + mget: async (...keys: string[]) => + keys.map(key => (strings.has(key) ? clone(strings.get(key)) : null)), + sadd: async (key: string, ...members: string[]) => { + const set = sets.get(key) ?? new Set() + sets.set(key, set) + const before = set.size + for (const member of members) set.add(member) + return set.size - before + }, + srem: async (key: string, ...members: string[]) => + members.filter(member => sets.get(key)?.delete(member)).length, + scard: async (key: string) => sets.get(key)?.size ?? 0, + smembers: async (key: string) => [...(sets.get(key) ?? [])], + // Like Upstash, the options argument is optional and told apart from a member by `score` + zadd: async (key: string, ...args: (ZMember | { nx?: boolean; xx?: boolean })[]) => { + const opts = (args[0] && !('score' in args[0]) ? args.shift() : {}) as { + nx?: boolean + xx?: boolean + } + const members = args as ZMember[] + const zset = zsets.get(key) ?? new Map() + zsets.set(key, zset) + let added = 0 + for (const { score, member } of members) { + const exists = zset.has(member) + if ((opts.nx && exists) || (opts.xx && !exists)) continue + if (!exists) added++ + zset.set(member, score) + } + return added + }, + zrem: async (key: string, ...members: string[]) => + members.filter(member => zsets.get(key)?.delete(member)).length, + // Score order, ties by member, as Redis does; `stop` is inclusive + zrange: async (key: string, start: number, stop: number, opts?: { rev?: boolean }) => { + const sorted = [...(zsets.get(key) ?? [])] + .sort(([a, sa], [b, sb]) => sa - sb || (a < b ? -1 : a > b ? 1 : 0)) + .map(([member]) => member) + if (opts?.rev) sorted.reverse() + return sorted.slice(start, stop + 1) + }, + // Null for a missing member, as Redis does + zmscore: async (key: string, members: string[]) => + members.map(member => zsets.get(key)?.get(member) ?? null), + } + return redis as unknown as IndexRedis +} diff --git a/shared/types/social.ts b/shared/types/social.ts index b37ccd1ccd..b856f5df38 100644 --- a/shared/types/social.ts +++ b/shared/types/social.ts @@ -39,6 +39,16 @@ export type CollectionSummary = { createdAt: string } +/** Browse's counts and edit time for a collection, from the Jetstream index. */ +export type CollectionStats = { + likes: number + follows: number + /** When the index last saw the record created or edited */ + updatedAt: string +} + +export type BrowseCollection = CollectionSummary & { stats: CollectionStats } + /** Why an included collection could not be shown */ export type CollectionUnavailableReason = 'unresolvable' | 'hidden' diff --git a/test/unit/server/utils/collections-index.spec.ts b/test/unit/server/utils/collections-index.spec.ts new file mode 100644 index 0000000000..2fd2feb503 --- /dev/null +++ b/test/unit/server/utils/collections-index.spec.ts @@ -0,0 +1,369 @@ +import { describe, expect, it, vi } from 'vite-plus/test' +import { COLLECTION_SUBJECT_REF, PACKAGE_SUBJECT_REF } from '#shared/utils/constants' + +const { CollectionsIndexStore, KEY, applyJetstreamEvent, drainJetstream, INITIAL_LOOKBACK_US } = + await import('#server/utils/atproto/utils/collections-index') +const { createMemoryRedis } = await import('#server/utils/atproto/utils/memory-redis') + +// The store under test, on the in-memory Redis that local dev also uses +let redis = createMemoryRedis() +async function newStore(cursor?: number) { + redis = createMemoryRedis() + const store = new CollectionsIndexStore(redis) + if (cursor !== undefined) await store.setCursor(cursor) + return store +} +const likers = async (uri: string) => (await redis.smembers(KEY.authors('like', uri))).sort() +const followers = async (uri: string) => (await redis.smembers(KEY.authors('follow', uri))).sort() + +const ALICE = 'did:plc:alice' +const BOB = 'did:plc:bob' +const RKEY_A = '3m5k5baifas2a' +const RKEY_B = '3m5k5baifas2b' +const A = COLLECTION_SUBJECT_REF(ALICE, RKEY_A) +const B = COLLECTION_SUBJECT_REF(ALICE, RKEY_B) + +const collectionRecord = (name: string, items = 1) => ({ + $type: 'dev.npmx.collection', + name, + items: Array.from({ length: items }, () => ({ subjectRef: PACKAGE_SUBJECT_REF('zod') })), + createdAt: '2026-11-01T00:00:00.000Z', +}) + +const TYPE = 'network.bsky.jetstream.subscribeEvents' +// `seq` doubles as the event time in µs, which keeps fixtures readable. Jetstream times have µs +// precision but Date only ms, so fixtures that care about order step by at least 1000. +const timeOf = (seq: number) => new Date(seq / 1000).toISOString() + +const commit = ( + did: string, + seq: number, + collection: string, + operation: 'create' | 'update' | 'delete', + rkey: string, + record?: unknown, +) => ({ + $type: `${TYPE}#commit`, + did, + seq, + time: timeOf(seq), + operation, + collection, + rkey, + ...(operation === 'delete' ? {} : { cid: `cid-${seq}`, record }), +}) + +const identity = (did: string, seq: number) => ({ + $type: `${TYPE}#identity`, + did, + seq, + time: timeOf(seq), + identity: { did }, +}) + +const likeOf = (subjectRef: string) => ({ + $type: 'dev.npmx.feed.like', + subjectRef, + createdAt: '2026-11-01T00:00:00.000Z', +}) + +const followOf = (subject: string) => ({ + $type: 'dev.npmx.graph.follow', + subject, + createdAt: '2026-11-01T00:00:00.000Z', +}) + +// Shaped like real Jetstream v2 payloads, including the identity events used as a clock +const fixture = [ + identity(BOB, 1_000), + commit(ALICE, 2_000, 'dev.npmx.collection', 'create', RKEY_A, collectionRecord('Parsers', 2)), + commit(ALICE, 3_000, 'dev.npmx.collection', 'create', RKEY_B, collectionRecord('Bundlers')), + commit(BOB, 4_000, 'dev.npmx.feed.like', 'create', 'like1', likeOf(A)), + commit(BOB, 5_000, 'dev.npmx.feed.like', 'create', 'like2', likeOf(PACKAGE_SUBJECT_REF('zod'))), + commit(ALICE, 6_000, 'dev.npmx.collection', 'update', RKEY_A, collectionRecord('Arg parsers', 3)), + identity(BOB, 7_000), + commit(ALICE, 8_000, 'dev.npmx.feed.like', 'create', 'like3', likeOf(A)), + commit(BOB, 9_000, 'dev.npmx.feed.like', 'delete', 'like1'), + commit(ALICE, 10_000, 'dev.npmx.collection', 'create', 'bad', { name: 'missing items' }), + commit(BOB, 11_000, 'dev.npmx.feed.like', 'create', 'like4', likeOf(B)), + commit(ALICE, 12_000, 'dev.npmx.collection', 'delete', RKEY_B), +] + +async function names( + store: InstanceType, + sort: 'recent' | 'liked' | 'followed', + offset = 0, +) { + return (await store.list(sort, offset, 10)).entries.map(c => c.name) +} + +async function applyAll(events: unknown[]) { + const store = await newStore() + for (const event of events) await applyJetstreamEvent(store, event as never) + return store +} + +describe('applyJetstreamEvent', () => { + it('builds the index from a stream of events', async () => { + const store = await applyAll(fixture) + + expect((await store.list('recent', 0, 10)).entries).toEqual([ + { + uri: A, + cid: 'cid-6000', + did: ALICE, + rkey: RKEY_A, + name: 'Arg parsers', + description: undefined, + itemCount: 3, + createdAt: '2026-11-01T00:00:00.000Z', + // The edit's time; counts are distinct accounts + stats: { likes: 1, follows: 0, updatedAt: timeOf(6_000) }, + }, + ]) + // like1 removed, like3 kept; package likes and likes of deleted collections don't count + expect(await likers(A)).toEqual([ALICE]) + expect(await redis.exists(KEY.record('like', `${BOB}/like2`))).toBe(0) + }) + + it('keeps first-seen order on edit', async () => { + const store = await applyAll(fixture.slice(0, 6)) + expect(await names(store, 'recent')).toEqual(['Bundlers', 'Arg parsers']) + }) + + it('is idempotent under replay', async () => { + const store = await applyAll([...fixture, ...fixture]) + expect(await likers(A)).toEqual([ALICE]) + expect(await names(store, 'recent')).toEqual(['Arg parsers']) + }) + + it('drops a collection whose edit makes it invalid', async () => { + const store = await applyAll([ + fixture[1], + commit(ALICE, 20_000, 'dev.npmx.collection', 'update', RKEY_A, { name: 'broken' }), + ]) + expect(await names(store, 'recent')).toEqual([]) + }) + + it('sorts by likes', async () => { + const store = await applyAll([ + fixture[1], + fixture[2], + commit(BOB, 30_000, 'dev.npmx.feed.like', 'create', 'l1', likeOf(B)), + commit(ALICE, 31_000, 'dev.npmx.feed.like', 'create', 'l2', likeOf(B)), + commit(BOB, 32_000, 'dev.npmx.feed.like', 'create', 'l3', likeOf(A)), + ]) + expect(await names(store, 'liked')).toEqual(['Bundlers', 'Parsers']) + expect(await names(store, 'liked', 1)).toEqual(['Parsers']) + }) + + it('counts each account once, however many like records it makes', async () => { + const store = await applyAll([ + fixture[1], + fixture[2], + ...Array.from({ length: 5 }, (_, i) => + commit(BOB, 30_000 + i, 'dev.npmx.feed.like', 'create', `spam${i}`, likeOf(A)), + ), + commit(BOB, 40_000, 'dev.npmx.feed.like', 'create', 'l1', likeOf(B)), + commit(ALICE, 41_000, 'dev.npmx.feed.like', 'create', 'l2', likeOf(B)), + ]) + expect(await names(store, 'liked')).toEqual(['Bundlers', 'Parsers']) + }) + + it('ranks by distinct followers, separately from likes', async () => { + const store = await applyAll([ + fixture[1], + fixture[2], + commit(BOB, 30_000, 'dev.npmx.graph.follow', 'create', 'f1', followOf(A)), + commit(BOB, 31_000, 'dev.npmx.graph.follow', 'create', 'f2', followOf(A)), + commit(ALICE, 32_000, 'dev.npmx.graph.follow', 'create', 'f3', followOf(A)), + commit(BOB, 33_000, 'dev.npmx.graph.follow', 'create', 'f4', followOf(B)), + commit(ALICE, 34_000, 'dev.npmx.graph.follow', 'create', 'f5', followOf(B)), + commit(ALICE, 35_000, 'dev.npmx.graph.follow', 'delete', 'f5'), + commit(ALICE, 36_000, 'dev.npmx.graph.follow', 'create', 'bad', { subject: 'nope' }), + commit(ALICE, 37_000, 'dev.npmx.feed.like', 'create', 'l1', likeOf(B)), + ]) + expect(await followers(A)).toEqual([ALICE, BOB]) + expect(await followers(B)).toEqual([BOB]) + expect(await names(store, 'followed')).toEqual(['Parsers', 'Bundlers']) + // Unliked collections stay out of the liked sort + expect(await names(store, 'liked')).toEqual(['Bundlers']) + }) + + it('ignores likes of collections it has not indexed', async () => { + const madeUp = COLLECTION_SUBJECT_REF(BOB, '3m5k5baifasz') + const store = await applyAll([ + commit(BOB, 30_000, 'dev.npmx.feed.like', 'create', 'l1', likeOf(madeUp)), + ]) + expect(await redis.exists(KEY.record('like', `${BOB}/l1`))).toBe(0) + expect(await names(store, 'liked')).toEqual([]) + }) +}) + +async function* messages(events: unknown[], onDone?: () => void) { + try { + for (const payload of events) yield JSON.stringify({ $type: 'message', payload }) + } finally { + onDone?.() + } +} + +describe('drainJetstream', () => { + const NOW = 2_000_000_000_000_000 + + it('starts a first run from a timestamp cursor within the lookback', async () => { + const store = await newStore() + const subscribe = vi.fn((_url: string) => messages([])) + + const result = await drainJetstream({ store, subscribe, nowUs: () => NOW }) + + const url = new URL(subscribe.mock.calls[0]![0]) + expect(result.fromCursor).toBe(NOW - INITIAL_LOOKBACK_US) + expect(url.searchParams.get('cursor')).toBe(String(NOW - INITIAL_LOOKBACK_US)) + expect(url.searchParams.getAll('collections')).toEqual([ + 'dev.npmx.collection', + 'dev.npmx.feed.like', + 'dev.npmx.graph.follow', + ]) + expect(url.searchParams.getAll('kinds')).toEqual(['commit', 'identity']) + expect(await store.getCursor()).toBe(NOW - INITIAL_LOOKBACK_US) + }) + + it('skips info frames', async () => { + const store = await newStore(5) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + + const result = await drainJetstream({ + store, + subscribe: () => + messages([ + { $type: `${TYPE}#info`, name: 'OutdatedCursor', message: 'starting at seq 9' }, + identity(BOB, NOW), + ]), + nowUs: () => NOW, + }) + + expect(warn).toHaveBeenCalledOnce() + expect(result).toMatchObject({ events: 1, toCursor: NOW, caughtUp: true }) + }) + + it('resumes from the stored cursor and stops once caught up', async () => { + const store = await newStore(NOW - 60_000_000) + let closed = false + const events = [ + commit( + ALICE, + NOW - 50_000_000, + 'dev.npmx.collection', + 'create', + RKEY_A, + collectionRecord('A'), + ), + identity(BOB, NOW - 1_000_000), + commit(ALICE, NOW, 'dev.npmx.collection', 'create', RKEY_B, collectionRecord('never read')), + ] + + const result = await drainJetstream({ + store, + subscribe: () => messages(events, () => (closed = true)), + nowUs: () => NOW, + }) + + expect(result).toEqual({ + events: 2, + fromCursor: NOW - 60_000_000, + toCursor: NOW - 1_000_000, + caughtUp: true, + }) + expect(closed).toBe(true) + expect(await store.getCursor()).toBe(NOW - 1_000_000) + expect(await names(store, 'recent')).toEqual(['A']) + }) + + it('stops at the deadline with progress saved', async () => { + const store = await newStore(NOW - 3_600_000_000) + let now = NOW + const events = Array.from({ length: 5 }, (_, i) => identity(BOB, NOW - 3_000_000_000 + i)) + + const result = await drainJetstream({ + store, + subscribe: () => messages(events), + nowUs: () => (now += 20_000_000), + maxDurationMs: 45_000, + }) + + expect(result.caughtUp).toBe(false) + // Clock reads: start (+20s, deadline +65s), then +40s, +60s, +80s → stops after the third + expect(result.events).toBe(3) + expect(await store.getCursor()).toBe(NOW - 3_000_000_000 + 2) + }) + + it('does not advance past an event it failed to apply', async () => { + const store = await newStore(NOW - 60_000_000) + vi.spyOn(store, 'upsertCollection').mockRejectedValueOnce(new Error('redis down')) + vi.spyOn(console, 'error').mockImplementation(() => {}) + + const result = await drainJetstream({ + store, + subscribe: () => + messages([ + identity(BOB, NOW - 50_000_000), + commit( + ALICE, + NOW - 40_000_000, + 'dev.npmx.collection', + 'create', + RKEY_A, + collectionRecord('A'), + ), + ]), + nowUs: () => NOW, + }) + + expect(result.toCursor).toBe(NOW - 50_000_000) + expect(await store.getCursor()).toBe(NOW - 50_000_000) + }) + + it("retries a failing event on later runs, then skips it so it can't stall the index", async () => { + const store = await newStore(NOW - 60_000_000) + vi.spyOn(store, 'upsertCollection').mockRejectedValue(new Error('bad event')) + vi.spyOn(console, 'error').mockImplementation(() => {}) + const run = () => + drainJetstream({ + store, + subscribe: () => + messages([ + commit( + ALICE, + NOW - 50_000_000, + 'dev.npmx.collection', + 'create', + RKEY_A, + collectionRecord('A'), + ), + identity(BOB, NOW - 40_000_000), + ]), + nowUs: () => NOW, + }) + + expect((await run()).toCursor).toBe(NOW - 60_000_000) + expect((await run()).toCursor).toBe(NOW - 60_000_000) + expect((await run()).toCursor).toBe(NOW - 40_000_000) + }) + + it('starts over from the lookback when Jetstream sends nothing, e.g. a cursor past retention', async () => { + const store = await newStore(7) + + const result = await drainJetstream({ + store, + subscribe: async function* (_url, signal) { + await new Promise(resolve => signal.addEventListener('abort', resolve)) + yield* [] + }, + maxDurationMs: 20, + }) + + expect(result).toMatchObject({ events: 0, toCursor: 7, caughtUp: false }) + expect(await store.getCursor()).toBeNull() + }) +}) From 1d9da33632bd423ef5617e329062622e424bc6c3 Mon Sep 17 00:00:00 2001 From: Philippe Serhal Date: Sun, 8 Nov 2026 18:49:52 -0500 Subject: [PATCH 2/5] feat(social): run the collections indexer on a cron route and serve browse /api/cron/collections-index runs the indexer task when called with CRON_SECRET, since the Vercel preset ignores Nitro's scheduledTasks. The schedule itself ships with the browse UI. /api/social/collections/browse pages the index by recent, most liked or most followed, hiding labeled accounts. --- nuxt.config.ts | 4 +++ server/api/cron/collections-index.get.ts | 14 +++++++++ server/api/social/collections/browse.get.ts | 33 +++++++++++++++++++++ server/tasks/collections/index.ts | 18 +++++++++++ server/utils/atproto/utils/collections.ts | 6 ++-- 5 files changed, 72 insertions(+), 3 deletions(-) create mode 100644 server/api/cron/collections-index.get.ts create mode 100644 server/api/social/collections/browse.get.ts create mode 100644 server/tasks/collections/index.ts diff --git a/nuxt.config.ts b/nuxt.config.ts index 5c7f2fec77..414b8e9ca2 100644 --- a/nuxt.config.ts +++ b/nuxt.config.ts @@ -186,6 +186,7 @@ export default defineNuxtConfig({ // never cache '/api/auth/**': { isr: false, cache: false }, '/api/social/**': { isr: false, cache: false }, + '/api/cron/**': { isr: false, cache: false }, '/api/atproto/bluesky-comments': { isr: { expiration: 60 * 60 /* one hour */, @@ -278,6 +279,9 @@ export default defineNuxtConfig({ compatibilityDate: '2026-01-31', nitro: { + experimental: { + tasks: true, + }, externals: { inline: [ 'shiki', diff --git a/server/api/cron/collections-index.get.ts b/server/api/cron/collections-index.get.ts new file mode 100644 index 0000000000..5b700bd448 --- /dev/null +++ b/server/api/cron/collections-index.get.ts @@ -0,0 +1,14 @@ +/** + * GET /api/cron/collections-index + * + * Runs the Jetstream indexer. Vercel Cron calls this (nuxt.config.ts) with `CRON_SECRET` as a + * bearer token; Nitro's `scheduledTasks` doesn't run on Vercel. + */ +export default defineEventHandler(async event => { + const secret = process.env.CRON_SECRET + if (!secret || getHeader(event, 'authorization') !== `Bearer ${secret}`) { + throw createError({ status: 401, message: 'Unauthorized' }) + } + const { result } = await runTask('collections:index') + return result +}) diff --git a/server/api/social/collections/browse.get.ts b/server/api/social/collections/browse.get.ts new file mode 100644 index 0000000000..80afd0e025 --- /dev/null +++ b/server/api/social/collections/browse.get.ts @@ -0,0 +1,33 @@ +import type { BrowseSort } from '#server/utils/atproto/utils/collections-index' +import type { BrowseCollection } from '#shared/types/social' + +/** + * GET /api/social/collections/browse?sort=recent|liked|followed&cursor=&limit= + * + * Collections across the network, from the Jetstream index. `cursor` is an offset. + */ +export default defineEventHandler(async event => { + const query = getQuery(event) + const sort: BrowseSort = + query.sort === 'liked' || query.sort === 'followed' ? query.sort : 'recent' + const { cursor, limit } = getPaginationQuery(event, 24, 60) + const offset = Math.max(Number.parseInt(cursor ?? '0', 10) || 0, 0) + + try { + const { entries, scanned } = await getCollectionsIndexStore().list(sort, offset, limit) + const items = await new CollectionsUtils().hydrate( + entries, + (miniDoc, entry): BrowseCollection => ({ + ...entry, + handle: miniDoc.handle, + }), + ) + return { + items, + // Hidden or stale entries can shorten a page without ending it + cursor: scanned === limit ? String(offset + limit) : undefined, + } + } catch (error: unknown) { + handleApiError(error, { statusCode: 502, message: 'Failed to browse collections' }) + } +}) diff --git a/server/tasks/collections/index.ts b/server/tasks/collections/index.ts new file mode 100644 index 0000000000..8e012d6e89 --- /dev/null +++ b/server/tasks/collections/index.ts @@ -0,0 +1,18 @@ +/** + * Indexes dev.npmx.collection records, and likes and follows of them, from Jetstream for browse. + * Vercel Cron runs it through /api/cron/collections-index. In dev, run it from + * Nuxt devtools or GET /_nitro/tasks/collections:index. + */ +export default defineTask({ + meta: { + name: 'collections:index', + description: 'Index collections and their likes and follows from Jetstream', + }, + async run() { + const result = await drainJetstream({ + store: getCollectionsIndexStore(), + subscribe: webSocketMessages, + }) + return { result } + }, +}) diff --git a/server/utils/atproto/utils/collections.ts b/server/utils/atproto/utils/collections.ts index 36221aa2af..1be63c8f83 100644 --- a/server/utils/atproto/utils/collections.ts +++ b/server/utils/atproto/utils/collections.ts @@ -443,10 +443,10 @@ export class CollectionsUtils { * Turns entries into summaries via `build`, dropping hidden authors and any entry whose author * no longer resolves (deleted, deactivated) or that `build` rejects. */ - async hydrate( + async hydrate( entries: T[], - build: (miniDoc: MiniDoc, entry: T) => Promise | CollectionSummary, - ): Promise { + build: (miniDoc: MiniDoc, entry: T) => Promise | S, + ): Promise { const hidden = await this.deps.moderation.getHiddenDids(entries.map(entry => entry.did)) const summaries = await Promise.all( entries From 747e2c977c416020929244f3bfa838faa6c1b01a Mon Sep 17 00:00:00 2001 From: Philippe Serhal Date: Sun, 8 Nov 2026 18:50:12 -0500 Subject: [PATCH 3/5] feat(ui): browse collections /collections lists collections across the network by recent, most liked or most followed, and it's in the nav and the command palette. Vercel Cron now runs the indexer every 5 minutes, so a new collection's page says it may take that long to show up in browse. --- app/components/AppHeader.vue | 8 ++ app/components/Collection/Card.stories.ts | 10 +++ app/components/Collection/Card.vue | 48 ++++++++++-- app/components/Collection/Grid.vue | 4 +- .../useCommandPaletteGlobalCommands.ts | 10 +++ app/pages/collections/[identifier]/[rkey].vue | 15 ++++ app/pages/collections/index.vue | 75 +++++++++++++++---- app/pages/collections/new.vue | 3 +- i18n/locales/en.json | 14 +++- i18n/locales/fr-FR.json | 14 +++- i18n/schema.json | 27 +++++++ nuxt.config.ts | 9 +++ shared/utils/constants.ts | 2 + test/nuxt/a11y.spec.ts | 13 ++++ 14 files changed, 222 insertions(+), 30 deletions(-) diff --git a/app/components/AppHeader.vue b/app/components/AppHeader.vue index 223a2be7ef..b12dd25831 100644 --- a/app/components/AppHeader.vue +++ b/app/components/AppHeader.vue @@ -36,6 +36,14 @@ const desktopLinks = computed(() => [ external: false, iconClass: 'i-lucide:git-compare', }, + { + name: 'Collections', + label: $t('nav.collections'), + to: { name: 'collections' }, + type: 'link', + external: false, + iconClass: 'i-lucide:library', + }, { name: 'Settings', label: $t('nav.settings'), diff --git a/app/components/Collection/Card.stories.ts b/app/components/Collection/Card.stories.ts index 1ba304a672..94818a3799 100644 --- a/app/components/Collection/Card.stories.ts +++ b/app/components/Collection/Card.stories.ts @@ -17,3 +17,13 @@ export const Default: Story = {} export const WithoutDescription: Story = { args: { collection: { ...mockCollectionSummary, description: undefined } }, } + +/** Browse adds counts and when the collection was last edited. */ +export const WithStats: Story = { + args: { + collection: { + ...mockCollectionSummary, + stats: { likes: 12, follows: 5, updatedAt: '2026-11-03T00:00:00.000Z' }, + }, + }, +} diff --git a/app/components/Collection/Card.vue b/app/components/Collection/Card.vue index c1734969dc..b360fb0cde 100644 --- a/app/components/Collection/Card.vue +++ b/app/components/Collection/Card.vue @@ -1,11 +1,14 @@