forked from
nekomimi.pet/wisp.place-monorepo
Monorepo for wisp.place. A static site hosting service built on top of the AT Protocol.
webhooks uses jetstream
+532
-122
+1
apps/webhook-service/package.json
+1
apps/webhook-service/package.json
+1
-1
apps/webhook-service/src/config.ts
+1
-1
apps/webhook-service/src/config.ts
···
1
1
export const config = {
2
-
firehoseService: process.env.FIREHOSE_SERVICE || 'wss://relay.fire.hose.cam',
2
+
jetstreamUrl: process.env.JETSTREAM_URL || 'wss://jetstream2.us-east.bsky.network/subscribe',
3
3
healthPort: parseInt(process.env.HEALTH_PORT || '3003', 10),
4
4
deliveryTimeoutMs: parseInt(process.env.DELIVERY_TIMEOUT_MS || '10000', 10),
5
5
deliveryMaxRetries: parseInt(process.env.DELIVERY_MAX_RETRIES || '3', 10),
+29
-3
apps/webhook-service/src/index.ts
+29
-3
apps/webhook-service/src/index.ts
···
1
1
import { createLogger } from '@wispplace/observability'
2
2
import { config } from './config'
3
-
import { closeDatabase, db } from './lib/db'
4
-
import { getFirehoseHealth, startFirehose, stopFirehose } from './lib/firehose'
3
+
import { closeDatabase, db, loadAllWebhooks } from './lib/db'
4
+
import { runStartupBackfill } from './lib/backfill'
5
+
import { getFirehoseHealth, initScopeDids, startFirehose, stopFirehose } from './lib/firehose'
5
6
import { closeRedisPublisher } from './lib/redis'
6
7
7
8
const logger = createLogger('webhook-service')
···
104
105
105
106
async function main() {
106
107
logger.info('Starting webhook-service')
107
-
logger.info(`Firehose: ${config.firehoseService}`)
108
+
logger.info(`Jetstream: ${config.jetstreamUrl}`)
108
109
logger.info(`Health endpoint: http://localhost:${config.healthPort}/health`)
109
110
111
+
const webhooks = await loadAllWebhooks()
112
+
if (webhooks.length === 0) {
113
+
logger.info('[registry] No webhook records in DB')
114
+
} else {
115
+
logger.info(`[registry] Tracking ${webhooks.length} webhook(s) across ${new Set(webhooks.map((w) => w.record.scope.aturi.replace(/^at:\/\//, '').split('/')[0])).size} DID(s)`)
116
+
for (const w of webhooks) {
117
+
logger.info(
118
+
`[registry] ${w.did}/${w.rkey}` +
119
+
` scope=${w.record.scope.aturi}` +
120
+
` events=${w.record.events?.join(',') ?? 'all'}` +
121
+
` backlinks=${w.record.scope.backlinks ?? false}` +
122
+
` enabled=${w.record.enabled ?? true}` +
123
+
` url=${w.record.url}`,
124
+
)
125
+
}
126
+
}
127
+
128
+
// Populate sync pre-filter sets before starting the firehose
129
+
initScopeDids(webhooks)
130
+
110
131
startFirehose()
132
+
133
+
// Backfill any place.wisp.v2.wh records that existed before this run
134
+
await runStartupBackfill()
135
+
// Re-init after backfill in case it added new webhooks
136
+
initScopeDids(await loadAllWebhooks())
111
137
}
112
138
113
139
main().catch((err) => {
+92
apps/webhook-service/src/lib/backfill.ts
+92
apps/webhook-service/src/lib/backfill.ts
···
1
+
import { getPdsForDid } from '@wispplace/atproto-utils'
2
+
import type { Main as WhRecord } from '@wispplace/lexicons/types/place/wisp/v2/wh'
3
+
import { createLogger } from '@wispplace/observability'
4
+
import { listAllKnownDids, upsertWebhookRecord } from './db'
5
+
6
+
const logger = createLogger('webhook-service:backfill')
7
+
8
+
interface ListRecordsResponse {
9
+
records: Array<{
10
+
uri: string
11
+
cid: string
12
+
value: WhRecord
13
+
}>
14
+
cursor?: string
15
+
}
16
+
17
+
/**
18
+
* Fetch all place.wisp.v2.wh records for a DID from their PDS.
19
+
* Pages through all results using the cursor.
20
+
*/
21
+
async function fetchWhRecordsForDid(did: string): Promise<Array<{ rkey: string; record: WhRecord }>> {
22
+
const pdsUrl = await getPdsForDid(did)
23
+
if (!pdsUrl) return []
24
+
25
+
const results: Array<{ rkey: string; record: WhRecord }> = []
26
+
let cursor: string | undefined
27
+
28
+
do {
29
+
const params = new URLSearchParams({
30
+
repo: did,
31
+
collection: 'place.wisp.v2.wh',
32
+
limit: '100',
33
+
})
34
+
if (cursor) params.set('cursor', cursor)
35
+
36
+
const res = await fetch(`${pdsUrl}/xrpc/com.atproto.repo.listRecords?${params}`, {
37
+
signal: AbortSignal.timeout(10_000),
38
+
})
39
+
40
+
if (!res.ok) {
41
+
if (res.status === 404) return results // DID has no records of this type
42
+
logger.warn(`[backfill] PDS returned ${res.status} for ${did}`)
43
+
return results
44
+
}
45
+
46
+
const data = (await res.json()) as ListRecordsResponse
47
+
cursor = data.cursor
48
+
49
+
for (const r of data.records) {
50
+
const rkey = r.uri.split('/').at(-1)
51
+
if (!rkey) continue
52
+
if (!r.value.scope?.aturi || !r.value.url) continue
53
+
results.push({ rkey, record: r.value })
54
+
}
55
+
} while (cursor)
56
+
57
+
return results
58
+
}
59
+
60
+
/**
61
+
* On startup, scan all known DIDs for existing place.wisp.v2.wh records
62
+
* and populate the local DB. This recovers webhook registrations that were
63
+
* created while the service was offline.
64
+
*/
65
+
export async function runStartupBackfill(): Promise<void> {
66
+
const dids = await listAllKnownDids()
67
+
if (dids.length === 0) {
68
+
logger.info('[backfill] No known DIDs to scan')
69
+
return
70
+
}
71
+
72
+
logger.info(`[backfill] Scanning ${dids.length} known DIDs for place.wisp.v2.wh records`)
73
+
74
+
let found = 0
75
+
let failed = 0
76
+
77
+
for (const did of dids) {
78
+
try {
79
+
const records = await fetchWhRecordsForDid(did)
80
+
for (const { rkey, record } of records) {
81
+
await upsertWebhookRecord(did, rkey, record)
82
+
found++
83
+
logger.info(`[backfill] Imported ${did}/${rkey}`)
84
+
}
85
+
} catch (err) {
86
+
failed++
87
+
logger.warn(`[backfill] Failed to scan ${did}`, { err: String(err) })
88
+
}
89
+
}
90
+
91
+
logger.info(`[backfill] Done — ${found} webhook record(s) imported, ${failed} DID(s) failed`)
92
+
}
+50
apps/webhook-service/src/lib/db.ts
+50
apps/webhook-service/src/lib/db.ts
···
43
43
`
44
44
45
45
await db`
46
+
CREATE TABLE IF NOT EXISTS firehose_cursor (
47
+
id TEXT PRIMARY KEY DEFAULT 'singleton',
48
+
seq BIGINT NOT NULL,
49
+
saved_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
50
+
)
51
+
`
52
+
53
+
await db`
46
54
CREATE TABLE IF NOT EXISTS webhook_event_logs (
47
55
id BIGSERIAL PRIMARY KEY,
48
56
owner_did TEXT NOT NULL,
···
233
241
status: r.status as 'ok' | 'failed',
234
242
deliveredAt: r.delivered_at,
235
243
}))
244
+
}
245
+
246
+
/** Persist the current firehose sequence number so restarts can resume. */
247
+
export async function saveCursor(seq: number): Promise<void> {
248
+
await db`
249
+
INSERT INTO firehose_cursor (id, seq, saved_at)
250
+
VALUES ('singleton', ${seq}, NOW())
251
+
ON CONFLICT (id) DO UPDATE SET seq = EXCLUDED.seq, saved_at = NOW()
252
+
`
253
+
}
254
+
255
+
/** Load the last saved firehose sequence number, or undefined if none. */
256
+
export async function loadCursor(): Promise<number | undefined> {
257
+
const rows = await db<Array<{ seq: number }>>`
258
+
SELECT seq FROM firehose_cursor WHERE id = 'singleton'
259
+
`
260
+
return rows[0]?.seq
261
+
}
262
+
263
+
/**
264
+
* Collect all DIDs we know about from the local DB.
265
+
* Used during startup backfill to find repos that may already have
266
+
* place.wisp.v2.wh records that arrived before the service started.
267
+
*/
268
+
export async function listAllKnownDids(): Promise<string[]> {
269
+
const dids = new Set<string>()
270
+
271
+
const sites = await db<Array<{ did: string }>>`
272
+
SELECT DISTINCT did FROM sites WHERE did IS NOT NULL AND did <> ''
273
+
`
274
+
for (const r of sites) dids.add(r.did)
275
+
276
+
try {
277
+
const sessions = await db<Array<{ sub: string }>>`
278
+
SELECT DISTINCT sub FROM oauth_sessions WHERE sub IS NOT NULL AND sub <> ''
279
+
`
280
+
for (const r of sessions) dids.add(r.sub)
281
+
} catch {
282
+
// oauth_sessions schema may differ; skip gracefully
283
+
}
284
+
285
+
return [...dids].sort()
236
286
}
237
287
238
288
/** Close all database connections gracefully. */
+187
-88
apps/webhook-service/src/lib/firehose.ts
+187
-88
apps/webhook-service/src/lib/firehose.ts
···
1
-
import { IdResolver } from '@atproto/identity'
2
-
import { Firehose } from '@atproto/sync'
3
-
import { BunFirehose, type CommitEvt, type Event, isBun } from '@wispplace/bun-firehose'
4
1
import type { Main as WhRecord } from '@wispplace/lexicons/types/place/wisp/v2/wh'
5
2
import { createLogger } from '@wispplace/observability'
6
3
import { config } from '../config'
7
-
import { deleteWebhookRecord, findBacklinkWebhooks, findWebhooksForDid, upsertWebhookRecord } from './db'
4
+
import {
5
+
deleteWebhookRecord,
6
+
findBacklinkWebhooks,
7
+
findWebhooksForDid,
8
+
loadAllWebhooks,
9
+
upsertWebhookRecord,
10
+
type WebhookEntry,
11
+
} from './db'
8
12
import { deliverWebhook } from './delivery'
13
+
import { JetstreamClient, type JetstreamEvent } from './jetstream'
9
14
import { matchWebhooks } from './matcher'
10
15
import { getCached, invalidate, setCached } from './registry'
11
16
12
17
const logger = createLogger('webhook-service:firehose')
13
-
const idResolver = new IdResolver()
14
18
15
19
let lastEventTime = Date.now()
16
20
let isConnected = false
21
+
let totalEvents = 0
22
+
let totalMatched = 0
17
23
18
24
export function getFirehoseHealth() {
19
25
return {
20
26
connected: isConnected,
21
27
lastEventTime,
22
28
timeSinceLastEvent: Date.now() - lastEventTime,
23
-
healthy: isConnected && Date.now() - lastEventTime < 60000,
29
+
healthy: isConnected && Date.now() - lastEventTime < 60_000,
30
+
}
31
+
}
32
+
33
+
export function getEventStats() {
34
+
return { events: totalEvents, matched: totalMatched }
35
+
}
36
+
37
+
let directScopeDids = new Set<string>()
38
+
let backlinkScopeDids = new Set<string>()
39
+
40
+
export function initScopeDids(webhooks: Array<{ record: { scope: { aturi: string; backlinks?: boolean } } }>): void {
41
+
directScopeDids = new Set()
42
+
backlinkScopeDids = new Set()
43
+
for (const w of webhooks) {
44
+
const did = w.record.scope.aturi.replace(/^at:\/\//, '').split('/')[0]
45
+
if (!did) continue
46
+
directScopeDids.add(did)
47
+
if (w.record.scope.backlinks) backlinkScopeDids.add(did)
48
+
}
49
+
logger.info(`[registry] tracking ${directScopeDids.size} scope DID(s), ${backlinkScopeDids.size} with backlinks`)
50
+
restartDirectJetstream()
51
+
if (backlinkScopeDids.size > 0 && !backlinkJetstream) {
52
+
startBacklinkJetstream()
53
+
} else if (backlinkScopeDids.size === 0 && backlinkJetstream) {
54
+
stopBacklinkJetstream()
24
55
}
25
56
}
26
57
27
-
async function getWebhooksForEvent(eventDid: string) {
28
-
// Direct scope matches: cached by eventDid
58
+
function extractAtUriDids(obj: unknown, found: Set<string>): void {
59
+
if (typeof obj === 'string') {
60
+
if (obj.startsWith('at://')) {
61
+
const rest = obj.slice(5)
62
+
const slash = rest.indexOf('/')
63
+
const did = slash === -1 ? rest : rest.slice(0, slash)
64
+
if (did) found.add(did)
65
+
}
66
+
return
67
+
}
68
+
if (Array.isArray(obj)) {
69
+
for (const v of obj) extractAtUriDids(v, found)
70
+
return
71
+
}
72
+
if (obj !== null && typeof obj === 'object') {
73
+
for (const v of Object.values(obj)) extractAtUriDids(v, found)
74
+
}
75
+
}
76
+
77
+
function recordReferencesAnyOf(record: unknown, dids: Set<string>): boolean {
78
+
if (record == null || dids.size === 0) return false
79
+
const found = new Set<string>()
80
+
extractAtUriDids(record, found)
81
+
for (const did of found) {
82
+
if (dids.has(did)) return true
83
+
}
84
+
return false
85
+
}
86
+
87
+
async function getWebhooksForEvent(eventDid: string, eventRecord: unknown) {
29
88
let direct = getCached(eventDid)
30
89
if (!direct) {
31
90
direct = await findWebhooksForDid(eventDid)
32
91
setCached(eventDid, direct)
33
92
}
34
93
35
-
// Backlink matches: cached under a fixed key
36
94
let backlink = getCached('__backlinks__')
37
95
if (!backlink) {
38
96
backlink = await findBacklinkWebhooks()
39
97
setCached('__backlinks__', backlink)
40
98
}
41
99
42
-
// Combine, deduplicate by ownerDid/rkey
100
+
const includeBacklinks = backlink.length > 0 && recordReferencesAnyOf(eventRecord, backlinkScopeDids)
101
+
if (!includeBacklinks) return direct
102
+
43
103
const seen = new Set(direct.map((e) => `${e.ownerDid}/${e.rkey}`))
44
104
const combined = [...direct]
45
105
for (const entry of backlink) {
···
52
112
return combined
53
113
}
54
114
55
-
async function handleEvent(evt: Event | CommitEvt): Promise<void> {
115
+
async function deliver(did: string, collection: string, rkey: string, op: string, cid: string | undefined, record: unknown): Promise<void> {
116
+
const candidates = await getWebhooksForEvent(did, record)
117
+
if (candidates.length === 0) return
118
+
119
+
const matched = matchWebhooks(candidates, did, collection, rkey, op as any, record)
120
+
121
+
if (process.env.FILTER_DEBUG) {
122
+
for (const c of candidates) {
123
+
logger.debug(
124
+
matched.includes(c)
125
+
? `[filter] ✓ ${c.ownerDid}/${c.rkey} scope=${c.record.scope.aturi}`
126
+
: `[filter] ✗ ${c.ownerDid}/${c.rkey} scope=${c.record.scope.aturi}`,
127
+
)
128
+
}
129
+
}
130
+
131
+
if (matched.length === 0) return
132
+
totalMatched += matched.length
133
+
logger.info(`[deliver] ${op} ${did}/${collection}/${rkey} → ${matched.length} webhook(s)`)
134
+
await Promise.allSettled(matched.map((entry) => deliverWebhook(entry, did, collection, rkey, op as any, cid, record)))
135
+
}
136
+
137
+
async function handleWhRecord(op: string, did: string, rkey: string, record: unknown): Promise<void> {
138
+
logger.info(`[wh] ${op} ${did}/${rkey}`)
139
+
if (op === 'delete') {
140
+
deleteWebhookRecord(did, rkey).catch((err) => logger.error(`[DB] delete ${did}/${rkey}`, err))
141
+
} else if (record) {
142
+
const wh = record as WhRecord
143
+
if (!wh.scope?.aturi || !wh.url) {
144
+
logger.error(`[wh] Skipping ${did}/${rkey} — invalid record`, { record })
145
+
} else {
146
+
logger.info(`[wh] scope=${wh.scope.aturi} url=${wh.url} enabled=${wh.enabled ?? true}`)
147
+
upsertWebhookRecord(did, rkey, wh).catch((err) => logger.error(`[DB] upsert ${did}/${rkey}`, err))
148
+
}
149
+
} else {
150
+
logger.warn(`[wh] ${op} ${did}/${rkey} — record missing`)
151
+
}
152
+
invalidate(did)
153
+
invalidate('__backlinks__')
154
+
loadAllWebhooks().then(initScopeDids).catch(() => {})
155
+
}
156
+
157
+
let directJetstream: JetstreamClient | null = null
158
+
159
+
async function handleDirectEvent(event: JetstreamEvent): Promise<void> {
56
160
try {
161
+
if (event.kind !== 'commit' || !event.commit) return
57
162
lastEventTime = Date.now()
58
-
59
-
if (!('event' in evt)) return
60
-
if (evt.event !== 'create' && evt.event !== 'update' && evt.event !== 'delete') return
61
-
const { did, collection, rkey, record, cid, event } = evt as CommitEvt
163
+
const { did } = event
164
+
const { operation: op, collection, rkey, record, cid } = event.commit
165
+
if (op !== 'create' && op !== 'update' && op !== 'delete') return
166
+
totalEvents++
62
167
63
-
// Keep DB up to date and invalidate cache when webhook records change
64
168
if (collection === 'place.wisp.v2.wh') {
65
-
logger.info(`[wh] Received ${event} for ${did}/${rkey}`)
66
-
if (event === 'delete') {
67
-
deleteWebhookRecord(did, rkey).catch((err) => logger.error(`[DB] Failed to delete webhook ${did}/${rkey}`, err))
68
-
} else if (record) {
69
-
logger.debug(`[wh] raw record: ${JSON.stringify(record)}`)
70
-
const wh = record as WhRecord
71
-
if (!wh.scope?.aturi || !wh.url) {
72
-
logger.error(`[wh] Skipping ${did}/${rkey} — record failed validation`, { record })
73
-
} else {
74
-
logger.info(`[wh] scope=${wh.scope.aturi} url=${wh.url} enabled=${wh.enabled ?? true}`)
75
-
upsertWebhookRecord(did, rkey, wh).catch((err) =>
76
-
logger.error(`[DB] Failed to upsert webhook ${did}/${rkey}`, err),
77
-
)
78
-
}
79
-
} else {
80
-
logger.warn(`[wh] ${event} ${did}/${rkey} — record missing from commit`)
81
-
}
82
-
invalidate(did)
83
-
invalidate('__backlinks__')
169
+
await handleWhRecord(op, did, rkey, record)
84
170
return
85
171
}
86
172
87
-
// Lookup webhooks for this event (cache-first)
88
-
const candidates = await getWebhooksForEvent(did)
89
-
if (candidates.length === 0) return
173
+
await deliver(did, collection, rkey, op, cid, record)
174
+
} catch (err) {
175
+
logger.error('Direct Jetstream event error', err)
176
+
}
177
+
}
178
+
179
+
function restartDirectJetstream(): void {
180
+
const cursor = directJetstream?.cursor
181
+
directJetstream?.destroy()
182
+
183
+
if (directScopeDids.size === 0) {
184
+
directJetstream = null
185
+
return
186
+
}
187
+
188
+
directJetstream = new JetstreamClient({
189
+
url: config.jetstreamUrl,
190
+
wantedDids: [...directScopeDids],
191
+
cursor,
192
+
onEvent: handleDirectEvent,
193
+
onError: (err) => logger.error('Direct Jetstream error', err),
194
+
onConnect: () => { isConnected = true; logger.info('Direct Jetstream connected') },
195
+
onDisconnect: () => { isConnected = false },
196
+
})
197
+
directJetstream.start()
198
+
}
90
199
91
-
const matched = matchWebhooks(candidates, did, collection, rkey, event, record)
92
-
if (matched.length === 0) return
200
+
let backlinkJetstream: JetstreamClient | null = null
93
201
94
-
logger.info(`[deliver] ${event} ${did}/${collection}/${rkey} → ${matched.length} webhook(s)`)
202
+
async function handleBacklinkEvent(event: JetstreamEvent): Promise<void> {
203
+
try {
204
+
if (event.kind !== 'commit' || !event.commit) return
205
+
lastEventTime = Date.now()
206
+
const { did } = event
207
+
const { operation: op, collection, rkey, record, cid } = event.commit
208
+
if (op !== 'create' && op !== 'update' && op !== 'delete') return
95
209
96
-
await Promise.allSettled(
97
-
matched.map((entry) => deliverWebhook(entry, did, collection, rkey, event, cid?.toString(), record)),
98
-
)
210
+
if (collection === 'place.wisp.v2.wh' && !directScopeDids.has(did)) {
211
+
await handleWhRecord(op, did, rkey, record)
212
+
return
213
+
}
214
+
215
+
if (!recordReferencesAnyOf(record, backlinkScopeDids)) return
216
+
217
+
await deliver(did, collection, rkey, op, cid, record)
99
218
} catch (err) {
100
-
logger.error('Unexpected error in handleEvent', err)
219
+
logger.error('Backlink Jetstream event error', err)
101
220
}
102
221
}
103
222
104
-
function handleError(err: Error): void {
105
-
logger.error('Firehose error', err)
223
+
function startBacklinkJetstream(): void {
224
+
backlinkJetstream = new JetstreamClient({
225
+
url: config.jetstreamUrl,
226
+
onEvent: handleBacklinkEvent,
227
+
onError: (err) => logger.error('Backlink Jetstream error', err),
228
+
onConnect: () => logger.info('Backlink Jetstream connected'),
229
+
onDisconnect: () => logger.warn('Backlink Jetstream disconnected, reconnecting'),
230
+
})
231
+
backlinkJetstream.start()
106
232
}
107
233
108
-
let firehoseHandle: { destroy: () => void } | null = null
234
+
function stopBacklinkJetstream(): void {
235
+
backlinkJetstream?.destroy()
236
+
backlinkJetstream = null
237
+
}
109
238
110
239
export function startFirehose(): void {
111
-
logger.info(`Starting firehose (runtime: ${isBun ? 'Bun' : 'Node.js'})`)
112
-
113
-
if (isBun) {
114
-
const f = new BunFirehose({
115
-
idResolver,
116
-
service: config.firehoseService,
117
-
unauthenticatedCommits: true,
118
-
handleEvent,
119
-
onError: handleError,
120
-
onConnect: () => {
121
-
isConnected = true
122
-
logger.info('Firehose connected')
123
-
},
124
-
onDisconnect: () => {
125
-
isConnected = false
126
-
logger.warn('Firehose disconnected, will reconnect')
127
-
},
128
-
})
129
-
f.start()
130
-
firehoseHandle = { destroy: () => f.destroy() }
131
-
} else {
132
-
isConnected = true
133
-
const f = new Firehose({
134
-
idResolver,
135
-
service: config.firehoseService,
136
-
handleEvent: handleEvent as any,
137
-
onError: handleError,
138
-
})
139
-
f.start()
140
-
firehoseHandle = { destroy: () => f.destroy() }
141
-
}
240
+
logger.info(`Jetstream: ${config.jetstreamUrl}`)
241
+
restartDirectJetstream()
242
+
if (backlinkScopeDids.size > 0) startBacklinkJetstream()
142
243
143
244
setInterval(() => {
144
-
const health = getFirehoseHealth()
145
-
if (health.timeSinceLastEvent > 30000) {
146
-
logger.warn(`No firehose events for ${Math.round(health.timeSinceLastEvent / 1000)}s`)
147
-
} else {
148
-
logger.info(`Firehose alive, last event ${Math.round(health.timeSinceLastEvent / 1000)}s ago`)
245
+
if (Date.now() - lastEventTime > 30_000) {
246
+
logger.warn(`No events for ${Math.round((Date.now() - lastEventTime) / 1000)}s`)
149
247
}
150
-
}, 30000)
248
+
}, 30_000)
151
249
}
152
250
153
251
export function stopFirehose(): void {
154
-
logger.info('Stopping firehose')
252
+
logger.info('Stopping Jetstream consumers')
155
253
isConnected = false
156
-
firehoseHandle?.destroy()
157
-
firehoseHandle = null
254
+
directJetstream?.destroy()
255
+
directJetstream = null
256
+
stopBacklinkJetstream()
158
257
}
+109
apps/webhook-service/src/lib/jetstream.ts
+109
apps/webhook-service/src/lib/jetstream.ts
···
1
+
import { createLogger } from '@wispplace/observability'
2
+
3
+
const logger = createLogger('webhook-service:jetstream')
4
+
5
+
export interface JetstreamCommit {
6
+
rev: string
7
+
operation: 'create' | 'update' | 'delete'
8
+
collection: string
9
+
rkey: string
10
+
record?: unknown
11
+
cid?: string
12
+
}
13
+
14
+
export interface JetstreamEvent {
15
+
did: string
16
+
time_us: number
17
+
kind: 'commit' | 'identity' | 'account'
18
+
commit?: JetstreamCommit
19
+
}
20
+
21
+
export interface JetstreamOptions {
22
+
url: string
23
+
wantedDids?: string[]
24
+
wantedCollections?: string[]
25
+
cursor?: number
26
+
onEvent: (event: JetstreamEvent) => void | Promise<void>
27
+
onConnect?: () => void
28
+
onDisconnect?: () => void
29
+
onError?: (err: Error) => void
30
+
31
+
}
32
+
33
+
export class JetstreamClient {
34
+
private ws: WebSocket | null = null
35
+
private destroyed = false
36
+
private reconnectTimer: ReturnType<typeof setTimeout> | null = null
37
+
private lastCursor: number | undefined
38
+
39
+
constructor(private opts: JetstreamOptions) {
40
+
this.lastCursor = opts.cursor
41
+
}
42
+
43
+
start(): void {
44
+
this.connect()
45
+
}
46
+
47
+
private buildUrl(cursor?: number): string {
48
+
const url = new URL(this.opts.url)
49
+
50
+
for (const did of this.opts.wantedDids ?? []) {
51
+
url.searchParams.append('wantedDids', did)
52
+
}
53
+
for (const col of this.opts.wantedCollections ?? []) {
54
+
url.searchParams.append('wantedCollections', col)
55
+
}
56
+
if (cursor !== undefined) {
57
+
url.searchParams.set('cursor', String(cursor))
58
+
}
59
+
60
+
return url.toString()
61
+
}
62
+
63
+
private connect(): void {
64
+
if (this.destroyed) return
65
+
66
+
const url = this.buildUrl(this.lastCursor)
67
+
logger.info(`Connecting to Jetstream: ${url}`)
68
+
69
+
const ws = new WebSocket(url)
70
+
this.ws = ws
71
+
72
+
ws.onopen = () => {
73
+
logger.info('Jetstream connected')
74
+
this.opts.onConnect?.()
75
+
}
76
+
77
+
ws.onmessage = async (e) => {
78
+
try {
79
+
const event = JSON.parse(e.data as string) as JetstreamEvent
80
+
this.lastCursor = event.time_us
81
+
await this.opts.onEvent(event)
82
+
} catch (err) {
83
+
this.opts.onError?.(err instanceof Error ? err : new Error(String(err)))
84
+
}
85
+
}
86
+
87
+
ws.onclose = () => {
88
+
if (this.destroyed) return
89
+
logger.warn('Jetstream disconnected, reconnecting in 3s')
90
+
this.opts.onDisconnect?.()
91
+
this.reconnectTimer = setTimeout(() => this.connect(), 3_000)
92
+
}
93
+
94
+
ws.onerror = () => {
95
+
this.opts.onError?.(new Error('Jetstream WebSocket error'))
96
+
}
97
+
}
98
+
99
+
get cursor(): number | undefined {
100
+
return this.lastCursor
101
+
}
102
+
103
+
destroy(): void {
104
+
this.destroyed = true
105
+
if (this.reconnectTimer) clearTimeout(this.reconnectTimer)
106
+
this.ws?.close()
107
+
this.ws = null
108
+
}
109
+
}
+60
-28
apps/webhook-service/src/lib/matcher.ts
+60
-28
apps/webhook-service/src/lib/matcher.ts
···
8
8
rkey?: string
9
9
}
10
10
11
+
/*Compiled regexes keyed by glob pattern*/
12
+
const globCache = new Map<string, RegExp>()
13
+
14
+
/**Parsed AT-URIs keyed by the raw aturi string*/
15
+
const atUriCache = new Map<string, ParsedAtUri | null>()
16
+
11
17
function parseAtUri(aturi: string): ParsedAtUri | null {
18
+
const cached = atUriCache.get(aturi)
19
+
if (cached !== undefined) return cached
20
+
12
21
const withoutScheme = aturi.replace(/^at:\/\//, '')
13
22
const parts = withoutScheme.split('/')
14
23
const did = parts[0]
15
-
if (!did) return null
16
-
return {
17
-
did,
18
-
collection: parts[1] || undefined,
19
-
rkey: parts[2] || undefined,
24
+
const result = did ? { did, collection: parts[1] || undefined, rkey: parts[2] || undefined } : null
25
+
atUriCache.set(aturi, result)
26
+
return result
27
+
}
28
+
29
+
function compileGlob(pattern: string): RegExp {
30
+
let re = globCache.get(pattern)
31
+
if (!re) {
32
+
const escaped = pattern.split('*').map((s) => s.replace(/[.+?^${}()|[\]\\]/g, '\\$&'))
33
+
re = new RegExp(`^${escaped.join('.*')}$`)
34
+
globCache.set(pattern, re)
20
35
}
36
+
return re
21
37
}
22
38
23
-
/** Matches a collection segment against a glob pattern */
24
39
function matchesGlob(pattern: string, value: string): boolean {
25
40
if (!pattern.includes('*')) return pattern === value
26
-
const escaped = pattern.split('*').map((s) => s.replace(/[.+?^${}()|[\]\\]/g, '\\$&'))
27
-
return new RegExp(`^${escaped.join('.*')}$`).test(value)
41
+
return compileGlob(pattern).test(value)
28
42
}
29
43
30
44
/**
31
-
* Checks whether a serialised record body contains a reference to the given DID/collection.
32
-
* When collection contains a glob, scans for any `at://did/<collection>` URI that matches.
45
+
* Recursively walk a parsed record object checking whether any string value
46
+
* starts with `prefix` and has a collection segment matching `collectionRe`.
33
47
*/
34
-
function containsReference(record: unknown, did: string, collection?: string): boolean {
35
-
const json = JSON.stringify(record)
36
-
37
-
if (!collection) {
38
-
return json.includes(`at://${did}`) || json.includes(`"${did}"`)
48
+
function walkForReference(obj: unknown, prefix: string, collectionRe: RegExp | null, exact: string | null): boolean {
49
+
if (typeof obj === 'string') {
50
+
const idx = obj.indexOf(prefix)
51
+
if (idx === -1) return false
52
+
const rest = obj.slice(idx + prefix.length)
53
+
if (collectionRe === null && exact === null) return true // at://did — any reference
54
+
const end = rest.search(/[/"\\]/)
55
+
const col = end === -1 ? rest : rest.slice(0, end)
56
+
if (!col) return false
57
+
return exact !== null ? col === exact : collectionRe!.test(col)
39
58
}
40
-
41
-
if (!collection.includes('*')) {
42
-
return json.includes(`at://${did}/${collection}`)
59
+
if (Array.isArray(obj)) {
60
+
for (const v of obj) {
61
+
if (walkForReference(v, prefix, collectionRe, exact)) return true
62
+
}
63
+
return false
43
64
}
65
+
if (obj !== null && typeof obj === 'object') {
66
+
for (const v of Object.values(obj)) {
67
+
if (walkForReference(v, prefix, collectionRe, exact)) return true
68
+
}
69
+
}
70
+
return false
71
+
}
44
72
45
-
// Glob collection: scan for all at://did/... URIs and match the collection segment
73
+
/**
74
+
* Checks whether a record contains a reference to the given DID/collection.
75
+
* Uses a recursive walk and pre-compiled regex — no JSON.stringify.
76
+
*/
77
+
function containsReference(record: unknown, did: string, collection?: string): boolean {
46
78
const prefix = `at://${did}/`
47
-
let idx = json.indexOf(prefix)
48
-
while (idx !== -1) {
49
-
const rest = json.slice(idx + prefix.length)
50
-
const end = rest.search(/[/"\\]/)
51
-
const col = end === -1 ? rest : rest.slice(0, end)
52
-
if (col && matchesGlob(collection, col)) return true
53
-
idx = json.indexOf(prefix, idx + prefix.length)
79
+
80
+
if (!collection) {
81
+
// Any reference to this DID at all
82
+
return walkForReference(record, `at://${did}`, null, null)
54
83
}
55
-
return false
84
+
85
+
const collectionRe = collection.includes('*') ? compileGlob(collection) : null
86
+
const exact = collectionRe ? null : collection
87
+
return walkForReference(record, prefix, collectionRe, exact)
56
88
}
57
89
58
90
/**
···
63
95
* - The event kind is in its `events` filter (or no filter is set)
64
96
* - **Direct match**: the event DID/collection/rkey falls within the webhook's scope AT-URI
65
97
* (collection supports glob patterns, e.g. `app.bsky.*`)
66
-
* - **Backlink match**: `scope.backlinks` is true and the serialised record body contains
98
+
* - **Backlink match**: `scope.backlinks` is true and the record body contains
67
99
* a reference to the scope DID/collection
68
100
*/
69
101
export function matchWebhooks(
+1
bun.lock
+1
bun.lock
···
145
145
"@atproto/identity": "^0.4.10",
146
146
"@atproto/sync": "^0.1.40",
147
147
"@atproto/syntax": "^0.5.0",
148
+
"@wispplace/atproto-utils": "workspace:*",
148
149
"@wispplace/bun-firehose": "workspace:*",
149
150
"@wispplace/lexicons": "workspace:*",
150
151
"@wispplace/observability": "workspace:*",