+118 -32

package-lock.json

··· 13 13 "@atproto/lexicon": "0.4.1-rc.0", 14 14 "@atproto/oauth-client-node": "0.0.2-rc.2", 15 15 "@atproto/repo": "0.4.2-rc.0", 16 16 + "@atproto/sync": "^0.1.0", 16 17 "@atproto/syntax": "^0.3.0", 17 18 "@atproto/xrpc-server": "0.5.4-rc.0", 18 19 "better-sqlite3": "^11.1.2", ··· 23 24 "kysely": "^0.27.4", 24 25 "multiformats": "^9.9.0", 25 26 "pino": "^9.3.2", 26 26 - "pino-http": "^10.0.0", 27 27 "uhtml": "^4.5.9" 28 28 }, 29 29 "devDependencies": { ··· 210 210 } 211 211 }, 212 212 "node_modules/@atproto/crypto": { 213 213 - "version": "0.4.0", 214 214 - "resolved": "https://registry.npmjs.org/@atproto/crypto/-/crypto-0.4.0.tgz", 215 215 - "integrity": "sha512-Kj/4VgJ7hzzXvE42L0rjzP6lM0tai+OfPnP1rxJ+UZg/YUDtuewL4uapnVoWXvlNceKgaLZH98g5n9gXBVTe5Q==", 213 213 + "version": "0.4.1", 214 214 + "resolved": "https://registry.npmjs.org/@atproto/crypto/-/crypto-0.4.1.tgz", 215 215 + "integrity": "sha512-7pQNHWYyx8jGhYdPbmcuPD9W73nd/5v3mfBlncO0sBzxnPbmA6aXAWOz+fNVZwHwBJPeb/Gzf/FT/uDx7/eYFg==", 216 216 "dependencies": { 217 217 "@noble/curves": "^1.1.0", 218 218 "@noble/hashes": "^1.3.1", ··· 228 228 } 229 229 }, 230 230 "node_modules/@atproto/identity": { 231 231 - "version": "0.4.0", 232 232 - "resolved": "https://registry.npmjs.org/@atproto/identity/-/identity-0.4.0.tgz", 233 233 - "integrity": "sha512-KKdVlqBgkFuTUx3KFiiQe0LuK9kopej1bhKm6SHRPEYbSEPFmRZQMY9TAjWJQrvQt8DpQzz6kVGjASFEjd3teQ==", 231 231 + "version": "0.4.1", 232 232 + "resolved": "https://registry.npmjs.org/@atproto/identity/-/identity-0.4.1.tgz", 233 233 + "integrity": "sha512-5AoPJDSD0rAay/6Sib+n/FjfwGulM/+xCNxwwDLR9QI4EoeUlvIH8g5BNdix812v312/Qd42kJrLpCNTZ5rvew==", 234 234 "dependencies": { 235 235 "@atproto/common-web": "^0.3.0", 236 236 - "@atproto/crypto": "^0.4.0", 236 236 + "@atproto/crypto": "^0.4.1", 237 237 "axios": "^0.27.2" 238 238 } 239 239 }, ··· 369 369 "zod": "^3.23.8" 370 370 } 371 371 }, 372 372 + "node_modules/@atproto/sync": { 373 373 + "version": "0.1.0", 374 374 + "resolved": "https://registry.npmjs.org/@atproto/sync/-/sync-0.1.0.tgz", 375 375 + "integrity": "sha512-2O1UPaeZfL0agitE9rp2mjYVezvZsao3DgJwWCSid1S0N7Y2pOdc7/fSLH/OHn96QhG7g0FGpWzTEcdekRuT0g==", 376 376 + "dependencies": { 377 377 + "@atproto/common": "^0.4.1", 378 378 + "@atproto/identity": "^0.4.1", 379 379 + "@atproto/lexicon": "^0.4.1", 380 380 + "@atproto/repo": "^0.5.0", 381 381 + "@atproto/syntax": "^0.3.0", 382 382 + "@atproto/xrpc-server": "^0.6.3", 383 383 + "multiformats": "^9.9.0", 384 384 + "p-queue": "^6.6.2" 385 385 + } 386 386 + }, 387 387 + "node_modules/@atproto/sync/node_modules/@atproto/lexicon": { 388 388 + "version": "0.4.1", 389 389 + "resolved": "https://registry.npmjs.org/@atproto/lexicon/-/lexicon-0.4.1.tgz", 390 390 + "integrity": "sha512-bzyr+/VHXLQWbumViX5L7h1NKQObfs8Z+XZJl43OUK8nYFUI4e/sW1IZKRNfw7Wvi5YVNK+J+yP3DWIBZhkCYA==", 391 391 + "dependencies": { 392 392 + "@atproto/common-web": "^0.3.0", 393 393 + "@atproto/syntax": "^0.3.0", 394 394 + "iso-datestring-validator": "^2.2.2", 395 395 + "multiformats": "^9.9.0", 396 396 + "zod": "^3.23.8" 397 397 + } 398 398 + }, 399 399 + "node_modules/@atproto/sync/node_modules/@atproto/repo": { 400 400 + "version": "0.5.0", 401 401 + "resolved": "https://registry.npmjs.org/@atproto/repo/-/repo-0.5.0.tgz", 402 402 + "integrity": "sha512-kZbj4wW5eFrDjkSTS9z+6bT4OTr5K4GrqWukWbfdBJtZPXsRDm75AV0C9ItoHDTdbBXn65TK6kqaJTrf89osCg==", 403 403 + "dependencies": { 404 404 + "@atproto/common": "^0.4.1", 405 405 + "@atproto/common-web": "^0.3.0", 406 406 + "@atproto/crypto": "^0.4.1", 407 407 + "@atproto/lexicon": "^0.4.1", 408 408 + "@ipld/car": "^3.2.3", 409 409 + "@ipld/dag-cbor": "^7.0.0", 410 410 + "multiformats": "^9.9.0", 411 411 + "uint8arrays": "3.0.0", 412 412 + "zod": "^3.23.8" 413 413 + } 414 414 + }, 415 415 + "node_modules/@atproto/sync/node_modules/@atproto/xrpc": { 416 416 + "version": "0.6.1", 417 417 + "resolved": "https://registry.npmjs.org/@atproto/xrpc/-/xrpc-0.6.1.tgz", 418 418 + "integrity": "sha512-Zy5ydXEdk6sY7FDUZcEVfCL1jvbL4tXu5CcdPqbEaW6LQtk9GLds/DK1bCX9kswTGaBC88EMuqQMfkxOhp2t4A==", 419 419 + "dependencies": { 420 420 + "@atproto/lexicon": "^0.4.1", 421 421 + "zod": "^3.23.8" 422 422 + } 423 423 + }, 424 424 + "node_modules/@atproto/sync/node_modules/@atproto/xrpc-server": { 425 425 + "version": "0.6.3", 426 426 + "resolved": "https://registry.npmjs.org/@atproto/xrpc-server/-/xrpc-server-0.6.3.tgz", 427 427 + "integrity": "sha512-0YXeBM9NjiIlR5eXWo8qzArRcBOKhwVimpH+ajKgZzlncPO53brVZ9+3BUnD5J1PG8mEQFRERi+Jt77QyF89qA==", 428 428 + "dependencies": { 429 429 + "@atproto/common": "^0.4.1", 430 430 + "@atproto/crypto": "^0.4.1", 431 431 + "@atproto/lexicon": "^0.4.1", 432 432 + "@atproto/xrpc": "^0.6.1", 433 433 + "cbor-x": "^1.5.1", 434 434 + "express": "^4.17.2", 435 435 + "http-errors": "^2.0.0", 436 436 + "mime-types": "^2.1.35", 437 437 + "rate-limiter-flexible": "^2.4.1", 438 438 + "uint8arrays": "3.0.0", 439 439 + "ws": "^8.12.0", 440 440 + "zod": "^3.23.8" 441 441 + } 442 442 + }, 372 443 "node_modules/@atproto/syntax": { 373 444 "version": "0.3.0", 374 445 "resolved": "https://registry.npmjs.org/@atproto/syntax/-/syntax-0.3.0.tgz", ··· 1712 1783 "node": ">=6" 1713 1784 } 1714 1785 }, 1786 1786 + "node_modules/eventemitter3": { 1787 1787 + "version": "4.0.7", 1788 1788 + "resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-4.0.7.tgz", 1789 1789 + "integrity": "sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw==" 1790 1790 + }, 1715 1791 "node_modules/events": { 1716 1792 "version": "3.3.0", 1717 1793 "resolved": "https://registry.npmjs.org/events/-/events-3.3.0.tgz", ··· 1943 2019 "version": "0.3.1", 1944 2020 "resolved": "https://registry.npmjs.org/gc-hook/-/gc-hook-0.3.1.tgz", 1945 2021 "integrity": "sha512-E5M+O/h2o7eZzGhzRZGex6hbB3k4NWqO0eA+OzLRLXxhdbYPajZnynPwAtphnh+cRHPwsj5Z80dqZlfI4eK55A==" 1946 1946 - }, 1947 1947 - "node_modules/get-caller-file": { 1948 1948 - "version": "2.0.5", 1949 1949 - "resolved": "https://registry.npmjs.org/get-caller-file/-/get-caller-file-2.0.5.tgz", 1950 1950 - "integrity": "sha512-DyFP3BM/3YHTQOCUL/w0OZHR0lpKeGrxotcHWcqNEdnltqFwXVfhEBQ94eIo34AfQpo0rGki4cyIiftY06h2Fg==", 1951 1951 - "engines": { 1952 1952 - "node": "6.* || 8.* || >= 10.*" 1953 1953 - } 1954 2022 }, 1955 2023 "node_modules/get-intrinsic": { 1956 2024 "version": "1.2.4", ··· 2642 2710 "wrappy": "1" 2643 2711 } 2644 2712 }, 2713 2713 + "node_modules/p-finally": { 2714 2714 + "version": "1.0.0", 2715 2715 + "resolved": "https://registry.npmjs.org/p-finally/-/p-finally-1.0.0.tgz", 2716 2716 + "integrity": "sha512-LICb2p9CB7FS+0eR1oqWnHhp0FljGLZCWBE9aix0Uye9W8LTQPwMTYVGWQWIw9RdQiDg4+epXQODwIYJtSJaow==", 2717 2717 + "engines": { 2718 2718 + "node": ">=4" 2719 2719 + } 2720 2720 + }, 2721 2721 + "node_modules/p-queue": { 2722 2722 + "version": "6.6.2", 2723 2723 + "resolved": "https://registry.npmjs.org/p-queue/-/p-queue-6.6.2.tgz", 2724 2724 + "integrity": "sha512-RwFpb72c/BhQLEXIZ5K2e+AhgNVmIejGlTgiB9MzZ0e93GRvqZ7uSi0dvRF7/XIXDeNkra2fNHBxTyPDGySpjQ==", 2725 2725 + "dependencies": { 2726 2726 + "eventemitter3": "^4.0.4", 2727 2727 + "p-timeout": "^3.2.0" 2728 2728 + }, 2729 2729 + "engines": { 2730 2730 + "node": ">=8" 2731 2731 + }, 2732 2732 + "funding": { 2733 2733 + "url": "https://github.com/sponsors/sindresorhus" 2734 2734 + } 2735 2735 + }, 2736 2736 + "node_modules/p-timeout": { 2737 2737 + "version": "3.2.0", 2738 2738 + "resolved": "https://registry.npmjs.org/p-timeout/-/p-timeout-3.2.0.tgz", 2739 2739 + "integrity": "sha512-rhIwUycgwwKcP9yTOOFK/AKsAopjjCakVqLHePO3CC6Mir1Z99xT+R63jZxAT5lFZLa2inS5h+ZS2GvR99/FBg==", 2740 2740 + "dependencies": { 2741 2741 + "p-finally": "^1.0.0" 2742 2742 + }, 2743 2743 + "engines": { 2744 2744 + "node": ">=8" 2745 2745 + } 2746 2746 + }, 2645 2747 "node_modules/package-json-from-dist": { 2646 2748 "version": "1.0.0", 2647 2749 "resolved": "https://registry.npmjs.org/package-json-from-dist/-/package-json-from-dist-1.0.0.tgz", ··· 2748 2850 "readable-stream": "^4.0.0", 2749 2851 "split2": "^4.0.0" 2750 2852 } 2751 2751 - }, 2752 2752 - "node_modules/pino-http": { 2753 2753 - "version": "10.2.0", 2754 2754 - "resolved": "https://registry.npmjs.org/pino-http/-/pino-http-10.2.0.tgz", 2755 2755 - "integrity": "sha512-am03BxnV3Ckx68OkbH0iZs3indsrH78wncQ6w1w51KroIbvJZNImBKX2X1wjdY8lSyaJ0UrX/dnO2DY3cTeCRw==", 2756 2756 - "dependencies": { 2757 2757 - "get-caller-file": "^2.0.5", 2758 2758 - "pino": "^9.0.0", 2759 2759 - "pino-std-serializers": "^7.0.0", 2760 2760 - "process-warning": "^3.0.0" 2761 2761 - } 2762 2762 - }, 2763 2763 - "node_modules/pino-http/node_modules/process-warning": { 2764 2764 - "version": "3.0.0", 2765 2765 - "resolved": "https://registry.npmjs.org/process-warning/-/process-warning-3.0.0.tgz", 2766 2766 - "integrity": "sha512-mqn0kFRl0EoqhnL0GQ0veqFHyIN1yig9RHh/InzORTUiZHFRAur+aMtRkELNwGs9aNwKS6tg/An4NYBPGwvtzQ==" 2767 2853 }, 2768 2854 "node_modules/pino-pretty": { 2769 2855 "version": "11.2.2",

+1

package.json

··· 18 18 "@atproto/lexicon": "0.4.1-rc.0", 19 19 "@atproto/oauth-client-node": "0.0.2-rc.2", 20 20 "@atproto/repo": "0.4.2-rc.0", 21 21 + "@atproto/sync": "^0.1.0", 21 22 "@atproto/syntax": "^0.3.0", 22 23 "@atproto/xrpc-server": "0.5.4-rc.0", 23 24 "better-sqlite3": "^11.1.2",

-194

src/firehose/firehose.ts

··· 1 1 - import type { RepoRecord } from '@atproto/lexicon' 2 2 - import { cborToLexRecord, readCar } from '@atproto/repo' 3 3 - import { AtUri } from '@atproto/syntax' 4 4 - import { Subscription } from '@atproto/xrpc-server' 5 5 - import type { CID } from 'multiformats/cid' 6 6 - import { 7 7 - type Account, 8 8 - type Commit, 9 9 - type Identity, 10 10 - type RepoEvent, 11 11 - isAccount, 12 12 - isCommit, 13 13 - isIdentity, 14 14 - isValidRepoEvent, 15 15 - } from './lexicons' 16 16 - 17 17 - type Opts = { 18 18 - service?: string 19 19 - getCursor?: () => Promise<number | undefined> 20 20 - setCursor?: (cursor: number) => Promise<void> 21 21 - subscriptionReconnectDelay?: number 22 22 - filterCollections?: string[] 23 23 - excludeIdentity?: boolean 24 24 - excludeAccount?: boolean 25 25 - excludeCommit?: boolean 26 26 - } 27 27 - 28 28 - export class Firehose { 29 29 - public sub: Subscription<RepoEvent> 30 30 - private abortController: AbortController 31 31 - 32 32 - constructor(public opts: Opts) { 33 33 - this.abortController = new AbortController() 34 34 - this.sub = new Subscription({ 35 35 - service: opts.service ?? 'https://bsky.network', 36 36 - method: 'com.atproto.sync.subscribeRepos', 37 37 - signal: this.abortController.signal, 38 38 - getParams: async () => { 39 39 - if (!opts.getCursor) return undefined 40 40 - const cursor = await opts.getCursor() 41 41 - return { cursor } 42 42 - }, 43 43 - validate: (value: unknown) => { 44 44 - try { 45 45 - return isValidRepoEvent(value) 46 46 - } catch (err) { 47 47 - console.error('repo subscription skipped invalid message', err) 48 48 - } 49 49 - }, 50 50 - }) 51 51 - } 52 52 - 53 53 - async *run(): AsyncGenerator<Event> { 54 54 - try { 55 55 - for await (const evt of this.sub) { 56 56 - try { 57 57 - if (isCommit(evt) && !this.opts.excludeCommit) { 58 58 - const parsed = await parseCommit(evt) 59 59 - for (const write of parsed) { 60 60 - if ( 61 61 - !this.opts.filterCollections || 62 62 - this.opts.filterCollections.includes(write.uri.collection) 63 63 - ) { 64 64 - yield write 65 65 - } 66 66 - } 67 67 - } else if (isAccount(evt) && !this.opts.excludeAccount) { 68 68 - const parsed = parseAccount(evt) 69 69 - if (parsed) { 70 70 - yield parsed 71 71 - } 72 72 - } else if (isIdentity(evt) && !this.opts.excludeIdentity) { 73 73 - yield parseIdentity(evt) 74 74 - } 75 75 - } catch (err) { 76 76 - console.error('repo subscription could not handle message', err) 77 77 - } 78 78 - if (this.opts.setCursor && typeof evt.seq === 'number') { 79 79 - await this.opts.setCursor(evt.seq) 80 80 - } 81 81 - } 82 82 - } catch (err) { 83 83 - console.error('repo subscription errored', err) 84 84 - setTimeout(() => this.run(), this.opts.subscriptionReconnectDelay ?? 3000) 85 85 - } 86 86 - } 87 87 - 88 88 - destroy() { 89 89 - this.abortController.abort() 90 90 - } 91 91 - } 92 92 - 93 93 - export const parseCommit = async (evt: Commit): Promise<CommitEvt[]> => { 94 94 - const car = await readCar(evt.blocks) 95 95 - 96 96 - const evts: CommitEvt[] = [] 97 97 - 98 98 - for (const op of evt.ops) { 99 99 - const uri = new AtUri(`at://${evt.repo}/${op.path}`) 100 100 - 101 101 - const meta: CommitMeta = { 102 102 - uri, 103 103 - author: uri.host, 104 104 - collection: uri.collection, 105 105 - rkey: uri.rkey, 106 106 - } 107 107 - 108 108 - if (op.action === 'create' || op.action === 'update') { 109 109 - if (!op.cid) continue 110 110 - const recordBytes = car.blocks.get(op.cid) 111 111 - if (!recordBytes) continue 112 112 - const record = cborToLexRecord(recordBytes) 113 113 - evts.push({ 114 114 - ...meta, 115 115 - event: op.action as 'create' | 'update', 116 116 - cid: op.cid, 117 117 - record, 118 118 - }) 119 119 - } 120 120 - 121 121 - if (op.action === 'delete') { 122 122 - evts.push({ 123 123 - ...meta, 124 124 - event: 'delete', 125 125 - }) 126 126 - } 127 127 - } 128 128 - 129 129 - return evts 130 130 - } 131 131 - 132 132 - export const parseIdentity = (evt: Identity): IdentityEvt => { 133 133 - return { 134 134 - event: 'identity', 135 135 - did: evt.did, 136 136 - handle: evt.handle, 137 137 - } 138 138 - } 139 139 - 140 140 - export const parseAccount = (evt: Account): AccountEvt | undefined => { 141 141 - if (evt.status && !isValidStatus(evt.status)) return 142 142 - return { 143 143 - event: 'account', 144 144 - did: evt.did, 145 145 - active: evt.active, 146 146 - status: evt.status as AccountStatus, 147 147 - } 148 148 - } 149 149 - 150 150 - const isValidStatus = (str: string): str is AccountStatus => { 151 151 - return ['takendown', 'suspended', 'deleted', 'deactivated'].includes(str) 152 152 - } 153 153 - 154 154 - type Event = CommitEvt | IdentityEvt | AccountEvt 155 155 - 156 156 - type CommitMeta = { 157 157 - uri: AtUri 158 158 - author: string 159 159 - collection: string 160 160 - rkey: string 161 161 - } 162 162 - 163 163 - type CommitEvt = Create | Update | Delete 164 164 - 165 165 - type Create = CommitMeta & { 166 166 - event: 'create' 167 167 - record: RepoRecord 168 168 - cid: CID 169 169 - } 170 170 - 171 171 - type Update = CommitMeta & { 172 172 - event: 'update' 173 173 - record: RepoRecord 174 174 - cid: CID 175 175 - } 176 176 - 177 177 - type Delete = CommitMeta & { 178 178 - event: 'delete' 179 179 - } 180 180 - 181 181 - type IdentityEvt = { 182 182 - event: 'identity' 183 183 - did: string 184 184 - handle?: string 185 185 - } 186 186 - 187 187 - type AccountEvt = { 188 188 - event: 'account' 189 189 - did: string 190 190 - active: boolean 191 191 - status?: AccountStatus 192 192 - } 193 193 - 194 194 - type AccountStatus = 'takendown' | 'suspended' | 'deleted' | 'deactivated'

+19 -18

src/firehose/ingester.ts src/ingester.ts

··· 1 1 + import pino from 'pino' 2 2 + import { IdResolver } from '@atproto/identity' 3 3 + import { Firehose } from '@atproto/sync' 1 4 import type { Database } from '#/db' 2 2 - import { Firehose } from '#/firehose/firehose' 3 5 import * as Status from '#/lexicon/types/com/example/status' 4 6 5 5 - export class Ingester { 6 6 - firehose: Firehose | undefined 7 7 - constructor(public db: Database) {} 8 8 - 9 9 - async start() { 10 10 - const firehose = new Firehose({}) 11 11 - 12 12 - for await (const evt of firehose.run()) { 7 7 + export function createIngester(db: Database, idResolver: IdResolver) { 8 8 + const logger = pino({ name: 'firehose ingestion' }) 9 9 + return new Firehose({ 10 10 + idResolver, 11 11 + handleEvent: async(evt) => { 13 12 // Watch for write events 14 13 if (evt.event === 'create' || evt.event === 'update') { 15 14 const record = evt.record ··· 21 20 Status.validateRecord(record).success 22 21 ) { 23 22 // Store the status in our SQLite 24 24 - await this.db 23 23 + await db 25 24 .insertInto('status') 26 25 .values({ 27 27 - authorDid: evt.author, 26 26 + authorDid: evt.did, 28 27 status: record.status, 29 28 updatedAt: record.updatedAt, 30 29 indexedAt: new Date().toISOString(), ··· 39 38 .execute() 40 39 } 41 40 } 42 42 - } 43 43 - } 44 44 - 45 45 - destroy() { 46 46 - this.firehose?.destroy() 47 47 - } 48 48 - } 41 41 + }, 42 42 + onError: (err) => { 43 43 + logger.error({err}, 'error on firehose ingestion') 44 44 + }, 45 45 + filterCollections: ['com.example.status'], 46 46 + excludeIdentity: true, 47 47 + excludeAccount: true, 48 48 + }) 49 49 + }

-355

src/firehose/lexicons.ts

··· 1 1 - import type { IncomingMessage } from 'node:http' 2 2 - 3 3 - import { type LexiconDoc, Lexicons } from '@atproto/lexicon' 4 4 - import type { ErrorFrame, HandlerAuth } from '@atproto/xrpc-server' 5 5 - import type { CID } from 'multiformats/cid' 6 6 - 7 7 - // @NOTE: this file is an ugly copy job of codegen output. I'd like to clean this whole thing up 8 8 - 9 9 - export function isObj(v: unknown): v is Record<string, unknown> { 10 10 - return typeof v === 'object' && v !== null 11 11 - } 12 12 - 13 13 - export function hasProp<K extends PropertyKey>(data: object, prop: K): data is Record<K, unknown> { 14 14 - return prop in data 15 15 - } 16 16 - 17 17 - export interface QueryParams { 18 18 - /** The last known event seq number to backfill from. */ 19 19 - cursor?: number 20 20 - } 21 21 - 22 22 - export type RepoEvent = 23 23 - | Commit 24 24 - | Identity 25 25 - | Account 26 26 - | Handle 27 27 - | Migrate 28 28 - | Tombstone 29 29 - | Info 30 30 - | { $type: string; [k: string]: unknown } 31 31 - export type HandlerError = ErrorFrame<'FutureCursor' | 'ConsumerTooSlow'> 32 32 - export type HandlerOutput = HandlerError | RepoEvent 33 33 - export type HandlerReqCtx<HA extends HandlerAuth = never> = { 34 34 - auth: HA 35 35 - params: QueryParams 36 36 - req: IncomingMessage 37 37 - signal: AbortSignal 38 38 - } 39 39 - export type Handler<HA extends HandlerAuth = never> = (ctx: HandlerReqCtx<HA>) => AsyncIterable<HandlerOutput> 40 40 - 41 41 - /** Represents an update of repository state. Note that empty commits are allowed, which include no repo data changes, but an update to rev and signature. */ 42 42 - export interface Commit { 43 43 - /** The stream sequence number of this message. */ 44 44 - seq: number 45 45 - /** DEPRECATED -- unused */ 46 46 - rebase: boolean 47 47 - /** Indicates that this commit contained too many ops, or data size was too large. Consumers will need to make a separate request to get missing data. */ 48 48 - tooBig: boolean 49 49 - /** The repo this event comes from. */ 50 50 - repo: string 51 51 - /** Repo commit object CID. */ 52 52 - commit: CID 53 53 - /** DEPRECATED -- unused. WARNING -- nullable and optional; stick with optional to ensure golang interoperability. */ 54 54 - prev?: CID | null 55 55 - /** The rev of the emitted commit. Note that this information is also in the commit object included in blocks, unless this is a tooBig event. */ 56 56 - rev: string 57 57 - /** The rev of the last emitted commit from this repo (if any). */ 58 58 - since: string | null 59 59 - /** CAR file containing relevant blocks, as a diff since the previous repo state. */ 60 60 - blocks: Uint8Array 61 61 - ops: RepoOp[] 62 62 - blobs: CID[] 63 63 - /** Timestamp of when this message was originally broadcast. */ 64 64 - time: string 65 65 - [k: string]: unknown 66 66 - } 67 67 - 68 68 - export function isCommit(v: unknown): v is Commit { 69 69 - return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#commit' 70 70 - } 71 71 - 72 72 - /** Represents a change to an account's identity. Could be an updated handle, signing key, or pds hosting endpoint. Serves as a prod to all downstream services to refresh their identity cache. */ 73 73 - export interface Identity { 74 74 - seq: number 75 75 - did: string 76 76 - time: string 77 77 - /** The current handle for the account, or 'handle.invalid' if validation fails. This field is optional, might have been validated or passed-through from an upstream source. Semantics and behaviors for PDS vs Relay may evolve in the future; see atproto specs for more details. */ 78 78 - handle?: string 79 79 - [k: string]: unknown 80 80 - } 81 81 - 82 82 - export function isIdentity(v: unknown): v is Identity { 83 83 - return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#identity' 84 84 - } 85 85 - 86 86 - /** Represents a change to an account's status on a host (eg, PDS or Relay). The semantics of this event are that the status is at the host which emitted the event, not necessarily that at the currently active PDS. Eg, a Relay takedown would emit a takedown with active=false, even if the PDS is still active. */ 87 87 - export interface Account { 88 88 - seq: number 89 89 - did: string 90 90 - time: string 91 91 - /** Indicates that the account has a repository which can be fetched from the host that emitted this event. */ 92 92 - active: boolean 93 93 - /** If active=false, this optional field indicates a reason for why the account is not active. */ 94 94 - status?: 'takendown' | 'suspended' | 'deleted' | 'deactivated' | (string & {}) 95 95 - [k: string]: unknown 96 96 - } 97 97 - 98 98 - export function isAccount(v: unknown): v is Account { 99 99 - return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#account' 100 100 - } 101 101 - 102 102 - /** DEPRECATED -- Use #identity event instead */ 103 103 - export interface Handle { 104 104 - seq: number 105 105 - did: string 106 106 - handle: string 107 107 - time: string 108 108 - [k: string]: unknown 109 109 - } 110 110 - 111 111 - export function isHandle(v: unknown): v is Handle { 112 112 - return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#handle' 113 113 - } 114 114 - 115 115 - /** DEPRECATED -- Use #account event instead */ 116 116 - export interface Migrate { 117 117 - seq: number 118 118 - did: string 119 119 - migrateTo: string | null 120 120 - time: string 121 121 - [k: string]: unknown 122 122 - } 123 123 - 124 124 - export function isMigrate(v: unknown): v is Migrate { 125 125 - return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#migrate' 126 126 - } 127 127 - 128 128 - /** DEPRECATED -- Use #account event instead */ 129 129 - export interface Tombstone { 130 130 - seq: number 131 131 - did: string 132 132 - time: string 133 133 - [k: string]: unknown 134 134 - } 135 135 - 136 136 - export function isTombstone(v: unknown): v is Tombstone { 137 137 - return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#tombstone' 138 138 - } 139 139 - 140 140 - export interface Info { 141 141 - name: 'OutdatedCursor' | (string & {}) 142 142 - message?: string 143 143 - [k: string]: unknown 144 144 - } 145 145 - 146 146 - export function isInfo(v: unknown): v is Info { 147 147 - return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#info' 148 148 - } 149 149 - 150 150 - /** A repo operation, ie a mutation of a single record. */ 151 151 - export interface RepoOp { 152 152 - action: 'create' | 'update' | 'delete' | (string & {}) 153 153 - path: string 154 154 - /** For creates and updates, the new record CID. For deletions, null. */ 155 155 - cid: CID | null 156 156 - [k: string]: unknown 157 157 - } 158 158 - 159 159 - export function isRepoOp(v: unknown): v is RepoOp { 160 160 - return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#repoOp' 161 161 - } 162 162 - 163 163 - export const ComAtprotoSyncSubscribeRepos: LexiconDoc = { 164 164 - lexicon: 1, 165 165 - id: 'com.atproto.sync.subscribeRepos', 166 166 - defs: { 167 167 - main: { 168 168 - type: 'subscription', 169 169 - description: 'Subscribe to repo updates', 170 170 - parameters: { 171 171 - type: 'params', 172 172 - properties: { 173 173 - cursor: { 174 174 - type: 'integer', 175 175 - description: 'The last known event to backfill from.', 176 176 - }, 177 177 - }, 178 178 - }, 179 179 - message: { 180 180 - schema: { 181 181 - type: 'union', 182 182 - refs: [ 183 183 - 'lex:com.atproto.sync.subscribeRepos#commit', 184 184 - 'lex:com.atproto.sync.subscribeRepos#handle', 185 185 - 'lex:com.atproto.sync.subscribeRepos#migrate', 186 186 - 'lex:com.atproto.sync.subscribeRepos#tombstone', 187 187 - 'lex:com.atproto.sync.subscribeRepos#info', 188 188 - ], 189 189 - }, 190 190 - }, 191 191 - errors: [ 192 192 - { 193 193 - name: 'FutureCursor', 194 194 - }, 195 195 - { 196 196 - name: 'ConsumerTooSlow', 197 197 - }, 198 198 - ], 199 199 - }, 200 200 - commit: { 201 201 - type: 'object', 202 202 - required: ['seq', 'rebase', 'tooBig', 'repo', 'commit', 'rev', 'since', 'blocks', 'ops', 'blobs', 'time'], 203 203 - nullable: ['prev', 'since'], 204 204 - properties: { 205 205 - seq: { 206 206 - type: 'integer', 207 207 - }, 208 208 - rebase: { 209 209 - type: 'boolean', 210 210 - }, 211 211 - tooBig: { 212 212 - type: 'boolean', 213 213 - }, 214 214 - repo: { 215 215 - type: 'string', 216 216 - format: 'did', 217 217 - }, 218 218 - commit: { 219 219 - type: 'cid-link', 220 220 - }, 221 221 - prev: { 222 222 - type: 'cid-link', 223 223 - }, 224 224 - rev: { 225 225 - type: 'string', 226 226 - description: 'The rev of the emitted commit', 227 227 - }, 228 228 - since: { 229 229 - type: 'string', 230 230 - description: 'The rev of the last emitted commit from this repo', 231 231 - }, 232 232 - blocks: { 233 233 - type: 'bytes', 234 234 - description: 'CAR file containing relevant blocks', 235 235 - maxLength: 1000000, 236 236 - }, 237 237 - ops: { 238 238 - type: 'array', 239 239 - items: { 240 240 - type: 'ref', 241 241 - ref: 'lex:com.atproto.sync.subscribeRepos#repoOp', 242 242 - }, 243 243 - maxLength: 200, 244 244 - }, 245 245 - blobs: { 246 246 - type: 'array', 247 247 - items: { 248 248 - type: 'cid-link', 249 249 - }, 250 250 - }, 251 251 - time: { 252 252 - type: 'string', 253 253 - format: 'datetime', 254 254 - }, 255 255 - }, 256 256 - }, 257 257 - handle: { 258 258 - type: 'object', 259 259 - required: ['seq', 'did', 'handle', 'time'], 260 260 - properties: { 261 261 - seq: { 262 262 - type: 'integer', 263 263 - }, 264 264 - did: { 265 265 - type: 'string', 266 266 - format: 'did', 267 267 - }, 268 268 - handle: { 269 269 - type: 'string', 270 270 - format: 'handle', 271 271 - }, 272 272 - time: { 273 273 - type: 'string', 274 274 - format: 'datetime', 275 275 - }, 276 276 - }, 277 277 - }, 278 278 - migrate: { 279 279 - type: 'object', 280 280 - required: ['seq', 'did', 'migrateTo', 'time'], 281 281 - nullable: ['migrateTo'], 282 282 - properties: { 283 283 - seq: { 284 284 - type: 'integer', 285 285 - }, 286 286 - did: { 287 287 - type: 'string', 288 288 - format: 'did', 289 289 - }, 290 290 - migrateTo: { 291 291 - type: 'string', 292 292 - }, 293 293 - time: { 294 294 - type: 'string', 295 295 - format: 'datetime', 296 296 - }, 297 297 - }, 298 298 - }, 299 299 - tombstone: { 300 300 - type: 'object', 301 301 - required: ['seq', 'did', 'time'], 302 302 - properties: { 303 303 - seq: { 304 304 - type: 'integer', 305 305 - }, 306 306 - did: { 307 307 - type: 'string', 308 308 - format: 'did', 309 309 - }, 310 310 - time: { 311 311 - type: 'string', 312 312 - format: 'datetime', 313 313 - }, 314 314 - }, 315 315 - }, 316 316 - info: { 317 317 - type: 'object', 318 318 - required: ['name'], 319 319 - properties: { 320 320 - name: { 321 321 - type: 'string', 322 322 - knownValues: ['OutdatedCursor'], 323 323 - }, 324 324 - message: { 325 325 - type: 'string', 326 326 - }, 327 327 - }, 328 328 - }, 329 329 - repoOp: { 330 330 - type: 'object', 331 331 - description: 332 332 - "A repo operation, ie a write of a single record. For creates and updates, cid is the record's CID as of this operation. For deletes, it's null.", 333 333 - required: ['action', 'path', 'cid'], 334 334 - nullable: ['cid'], 335 335 - properties: { 336 336 - action: { 337 337 - type: 'string', 338 338 - knownValues: ['create', 'update', 'delete'], 339 339 - }, 340 340 - path: { 341 341 - type: 'string', 342 342 - }, 343 343 - cid: { 344 344 - type: 'cid-link', 345 345 - }, 346 346 - }, 347 347 - }, 348 348 - }, 349 349 - } 350 350 - 351 351 - const lexicons = new Lexicons([ComAtprotoSyncSubscribeRepos]) 352 352 - 353 353 - export const isValidRepoEvent = (evt: unknown) => { 354 354 - return lexicons.assertValidXrpcMessage<RepoEvent>('com.atproto.sync.subscribeRepos', evt) 355 355 - }

+9 -6

src/firehose/resolver.ts src/id-resolver.ts

··· 3 3 const HOUR = 60e3 * 60 4 4 const DAY = HOUR * 24 5 5 6 6 - export interface Resolver { 6 6 + 7 7 + export function createIdResolver() { 8 8 + return new IdResolver({ 9 9 + didCache: new MemoryCache(HOUR, DAY), 10 10 + }) 11 11 + } 12 12 + 13 13 + export interface BidirectionalResolver { 7 14 resolveDidToHandle(did: string): Promise<string> 8 15 resolveDidsToHandles(dids: string[]): Promise<Record<string, string>> 9 16 } 10 17 11 11 - export function createResolver() { 12 12 - const resolver = new IdResolver({ 13 13 - didCache: new MemoryCache(HOUR, DAY), 14 14 - }) 15 15 - 18 18 + export function createBidirectionalResolver(resolver: IdResolver) { 16 19 return { 17 20 async resolveDidToHandle(did: string): Promise<string> { 18 21 const didDoc = await resolver.did.resolveAtprotoData(did)

+10 -7

src/index.ts

··· 3 3 import express, { type Express } from 'express' 4 4 import { pino } from 'pino' 5 5 import type { OAuthClient } from '@atproto/oauth-client-node' 6 6 + import { Firehose } from '@atproto/sync' 6 7 7 8 import { createDb, migrateToLatest } from '#/db' 8 9 import { env } from '#/lib/env' 9 9 - import { Ingester } from '#/firehose/ingester' 10 10 + import { createIngester } from '#/ingester' 10 11 import { createRouter } from '#/routes' 11 12 import { createClient } from '#/auth/client' 12 12 - import { createResolver, Resolver } from '#/firehose/resolver' 13 13 + import { createBidirectionalResolver, createIdResolver, BidirectionalResolver } from '#/id-resolver' 13 14 import type { Database } from '#/db' 15 15 + import { IdResolver, MemoryCache } from '@atproto/identity' 14 16 15 17 // Application state passed to the router and elsewhere 16 18 export type AppContext = { 17 19 db: Database 18 18 - ingester: Ingester 20 20 + ingester: Firehose 19 21 logger: pino.Logger 20 22 oauthClient: OAuthClient 21 21 - resolver: Resolver 23 23 + resolver: BidirectionalResolver 22 24 } 23 25 24 26 export class Server { ··· 38 40 39 41 // Create the atproto utilities 40 42 const oauthClient = await createClient(db) 41 41 - const ingester = new Ingester(db) 42 42 - const resolver = createResolver() 43 43 + const baseIdResolver = createIdResolver() 44 44 + const ingester = createIngester(db, baseIdResolver) 45 45 + const resolver = createBidirectionalResolver(baseIdResolver) 43 46 const ctx = { 44 47 db, 45 48 ingester, ··· 72 75 73 76 async close() { 74 77 this.ctx.logger.info('sigint received, shutting down') 75 75 - this.ctx.ingester.destroy() 78 78 + await this.ctx.ingester.destroy() 76 79 return new Promise<void>((resolve) => { 77 80 this.server.close(() => { 78 81 this.ctx.logger.info('server closed')