forked from
nekomimi.pet/wisp.place-monorepo
Monorepo for wisp.place. A static site hosting service built on top of the AT Protocol.
webhook fixes
+123
-38
+6
-2
apps/webhook-service/src/index.ts
+6
-2
apps/webhook-service/src/index.ts
···
1
1
import { createLogger } from '@wispplace/observability'
2
2
import { config } from './config'
3
3
import { runStartupBackfill } from './lib/backfill'
4
-
import { closeDatabase, db, loadAllWebhooks } from './lib/db'
4
+
import { closeDatabase, db, loadAllWebhooks, loadCursor } from './lib/db'
5
5
import { getFirehoseHealth, initScopeDids, startFirehose, stopFirehose } from './lib/firehose'
6
6
import { closeRedisPublisher } from './lib/redis'
7
7
···
130
130
// Populate sync pre-filter sets before starting the firehose
131
131
initScopeDids(webhooks)
132
132
133
-
startFirehose()
133
+
const cursor = await loadCursor(config.jetstreamUrl)
134
+
if (cursor !== undefined) {
135
+
logger.info(`[cursor] Resuming from ${cursor}`)
136
+
}
137
+
startFirehose(cursor)
134
138
135
139
// Backfill any place.wisp.v2.wh records that existed before this run
136
140
await runStartupBackfill()
+42
-16
apps/webhook-service/src/lib/db.ts
+42
-16
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()
46
+
CREATE TABLE IF NOT EXISTS jetstream_cursor (
47
+
id TEXT PRIMARY KEY DEFAULT 'singleton',
48
+
seq BIGINT NOT NULL,
49
+
jetstream_url TEXT,
50
+
saved_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
50
51
)
51
52
`
52
53
···
129
130
* Insert or update a webhook record in both tables.
130
131
* `webhooks` holds structured columns for quick filtering; `webhook_records` holds the full JSONB record.
131
132
* Key is `did/rkey`.
133
+
* Returns true if the record was inserted or its content changed, false if it was already up-to-date.
132
134
*/
133
-
export async function upsertWebhookRecord(did: string, rkey: string, record: WhRecord): Promise<void> {
135
+
export async function upsertWebhookRecord(did: string, rkey: string, record: WhRecord): Promise<boolean> {
134
136
const k = `${did}/${rkey}`
135
137
try {
136
138
await db`
···
143
145
enabled = EXCLUDED.enabled,
144
146
updated_at = EXTRACT(EPOCH FROM NOW())
145
147
`
146
-
await db`
148
+
const rows = await db<Array<{ changed: boolean }>>`
149
+
WITH existing AS (
150
+
SELECT v FROM webhook_records WHERE k = ${k}
151
+
)
147
152
INSERT INTO webhook_records (k, v, updated_at)
148
153
VALUES (${k}, ${record}, EXTRACT(EPOCH FROM NOW()))
149
154
ON CONFLICT (k) DO UPDATE SET
150
155
v = EXCLUDED.v,
151
156
updated_at = EXTRACT(EPOCH FROM NOW())
157
+
RETURNING (
158
+
NOT EXISTS (SELECT 1 FROM existing)
159
+
OR (SELECT v FROM existing) IS DISTINCT FROM webhook_records.v
160
+
) AS changed
152
161
`
162
+
return rows[0]?.changed ?? true
153
163
} catch (err) {
154
164
logger.error(`[DB] upsertWebhookRecord error for ${k}`, err)
155
165
throw err
···
243
253
}))
244
254
}
245
255
246
-
/** Persist the current firehose sequence number so restarts can resume. */
247
-
export async function saveCursor(seq: number): Promise<void> {
256
+
const CURSOR_REWIND_US = 2_000_000 // 2 seconds in microseconds
257
+
258
+
/** Persist the current Jetstream cursor (time_us) and the URL it came from. */
259
+
export async function saveCursor(seq: number, jetstreamUrl: string): Promise<void> {
248
260
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()
261
+
INSERT INTO jetstream_cursor (id, seq, jetstream_url, saved_at)
262
+
VALUES ('singleton', ${seq}, ${jetstreamUrl}, NOW())
263
+
ON CONFLICT (id) DO UPDATE SET seq = EXCLUDED.seq, jetstream_url = EXCLUDED.jetstream_url, saved_at = NOW()
252
264
`
253
265
}
254
266
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'
267
+
/**
268
+
* Load the saved Jetstream cursor, rewound by 2 seconds to ensure gapless playback.
269
+
* Returns undefined if no cursor is saved or the saved cursor is from a different Jetstream instance
270
+
* (time_us is not comparable across instances).
271
+
*/
272
+
export async function loadCursor(jetstreamUrl: string): Promise<number | undefined> {
273
+
const rows = await db<Array<{ seq: number; jetstream_url: string | null }>>`
274
+
SELECT seq, jetstream_url FROM jetstream_cursor WHERE id = 'singleton'
259
275
`
260
-
return rows[0]?.seq
276
+
const row = rows[0]
277
+
if (!row) return undefined
278
+
if (row.jetstream_url && row.jetstream_url !== jetstreamUrl) return undefined
279
+
return Math.max(0, Number(row.seq) - CURSOR_REWIND_US)
261
280
}
262
281
263
282
/**
···
281
300
} catch {
282
301
// oauth_sessions schema may differ; skip gracefully
283
302
}
303
+
304
+
// Include any DIDs that already have webhook records — they may not be wisp.place
305
+
// users but still have place.wisp.v2.wh records we need to stay in sync with.
306
+
const webhookDids = await db<Array<{ did: string }>>`
307
+
SELECT DISTINCT did FROM webhooks WHERE did IS NOT NULL AND did <> ''
308
+
`
309
+
for (const r of webhookDids) dids.add(r.did)
284
310
285
311
return [...dids].sort()
286
312
}
+75
-20
apps/webhook-service/src/lib/firehose.ts
+75
-20
apps/webhook-service/src/lib/firehose.ts
···
6
6
findBacklinkWebhooks,
7
7
findWebhooksForDid,
8
8
loadAllWebhooks,
9
+
saveCursor,
9
10
upsertWebhookRecord,
10
11
} from './db'
11
12
import { deliverWebhook } from './delivery'
···
35
36
36
37
let directScopeDids = new Set<string>()
37
38
let backlinkScopeDids = new Set<string>()
39
+
let firehoseStarted = false
40
+
41
+
function setsEqual(a: Set<string>, b: Set<string>): boolean {
42
+
if (a.size !== b.size) return false
43
+
for (const v of a) if (!b.has(v)) return false
44
+
return true
45
+
}
38
46
39
47
export function initScopeDids(webhooks: Array<{ record: { scope: { aturi: string; backlinks?: boolean } } }>): void {
40
-
directScopeDids = new Set()
41
-
backlinkScopeDids = new Set()
48
+
const newDirectDids = new Set<string>()
49
+
const newBacklinkDids = new Set<string>()
42
50
for (const w of webhooks) {
43
51
const did = w.record.scope.aturi.replace(/^at:\/\//, '').split('/')[0]
44
52
if (!did) continue
45
-
directScopeDids.add(did)
46
-
if (w.record.scope.backlinks) backlinkScopeDids.add(did)
53
+
newDirectDids.add(did)
54
+
if (w.record.scope.backlinks) newBacklinkDids.add(did)
47
55
}
56
+
57
+
const directChanged = !setsEqual(directScopeDids, newDirectDids)
58
+
const backlinkChanged = !setsEqual(backlinkScopeDids, newBacklinkDids)
59
+
60
+
directScopeDids = newDirectDids
61
+
backlinkScopeDids = newBacklinkDids
62
+
48
63
logger.info(`[registry] tracking ${directScopeDids.size} scope DID(s), ${backlinkScopeDids.size} with backlinks`)
49
-
restartDirectJetstream()
50
-
if (backlinkScopeDids.size > 0 && !backlinkJetstream) {
51
-
startBacklinkJetstream()
52
-
} else if (backlinkScopeDids.size === 0 && backlinkJetstream) {
53
-
stopBacklinkJetstream()
64
+
65
+
if (!firehoseStarted) return
66
+
67
+
if (directChanged) {
68
+
restartDirectJetstream()
69
+
}
70
+
71
+
if (backlinkChanged) {
72
+
if (backlinkScopeDids.size > 0 && !backlinkJetstream) {
73
+
startBacklinkJetstream()
74
+
} else if (backlinkScopeDids.size === 0 && backlinkJetstream) {
75
+
stopBacklinkJetstream()
76
+
}
54
77
}
55
78
}
56
79
···
91
114
}
92
115
93
116
let backlink = getCached('__backlinks__')
94
-
if (!backlink) {
117
+
if (!backlink || backlink.length === 0) {
95
118
backlink = await findBacklinkWebhooks()
96
-
setCached('__backlinks__', backlink)
119
+
if (backlink.length > 0) setCached('__backlinks__', backlink)
97
120
}
98
121
99
122
const includeBacklinks = backlink.length > 0 && recordReferencesAnyOf(eventRecord, backlinkScopeDids)
···
142
165
143
166
async function handleWhRecord(op: string, did: string, rkey: string, record: unknown): Promise<void> {
144
167
logger.info(`[wh] ${op} ${did}/${rkey}`)
168
+
let changed = true
145
169
if (op === 'delete') {
146
170
deleteWebhookRecord(did, rkey).catch((err) => logger.error(`[DB] delete ${did}/${rkey}`, err))
147
171
} else if (record) {
148
172
const wh = record as WhRecord
149
173
if (!wh.scope?.aturi || !wh.url) {
150
174
logger.error(`[wh] Skipping ${did}/${rkey} — invalid record`, { record })
151
-
} else {
152
-
logger.info(`[wh] scope=${wh.scope.aturi} url=${wh.url} enabled=${wh.enabled ?? true}`)
153
-
upsertWebhookRecord(did, rkey, wh).catch((err) => logger.error(`[DB] upsert ${did}/${rkey}`, err))
175
+
return
154
176
}
177
+
logger.info(`[wh] scope=${wh.scope.aturi} url=${wh.url} enabled=${wh.enabled ?? true}`)
178
+
changed = await upsertWebhookRecord(did, rkey, wh).catch((err) => {
179
+
logger.error(`[DB] upsert ${did}/${rkey}`, err)
180
+
return false
181
+
})
155
182
} else {
156
183
logger.warn(`[wh] ${op} ${did}/${rkey} — record missing`)
184
+
return
185
+
}
186
+
if (!changed) {
187
+
logger.debug(`[wh] ${did}/${rkey} unchanged, skipping reload`)
188
+
return
157
189
}
158
190
invalidate(did)
159
191
invalidate('__backlinks__')
···
184
216
}
185
217
}
186
218
187
-
function restartDirectJetstream(): void {
188
-
const cursor = directJetstream?.cursor
219
+
function restartDirectJetstream(overrideCursor?: number): void {
220
+
const cursor = overrideCursor ?? directJetstream?.cursor
189
221
directJetstream?.destroy()
190
222
191
223
if (directScopeDids.size === 0) {
···
225
257
return
226
258
}
227
259
260
+
// Skip events from scoped DIDs — the direct jetstream already handles those
261
+
if (directScopeDids.has(did)) return
262
+
228
263
if (!recordReferencesAnyOf(record, backlinkScopeDids)) return
229
264
230
265
await deliver(did, collection, rkey, op, cid, record)
···
233
268
}
234
269
}
235
270
236
-
function startBacklinkJetstream(): void {
271
+
function startBacklinkJetstream(cursor?: number): void {
237
272
backlinkJetstream = new JetstreamClient({
238
273
url: config.jetstreamUrl,
274
+
cursor,
239
275
onEvent: handleBacklinkEvent,
240
276
onError: (err) => logger.error('Backlink Jetstream error', err),
241
277
onConnect: () => logger.info('Backlink Jetstream connected'),
···
249
285
backlinkJetstream = null
250
286
}
251
287
252
-
export function startFirehose(): void {
288
+
export function startFirehose(initialCursor?: number): void {
253
289
logger.info(`Jetstream: ${config.jetstreamUrl}`)
254
-
restartDirectJetstream()
255
-
if (backlinkScopeDids.size > 0) startBacklinkJetstream()
290
+
if (initialCursor !== undefined) {
291
+
logger.info(`Resuming from cursor ${initialCursor}`)
292
+
}
293
+
firehoseStarted = true
294
+
restartDirectJetstream(initialCursor)
295
+
if (backlinkScopeDids.size > 0) startBacklinkJetstream(initialCursor)
256
296
257
297
setInterval(() => {
258
298
if (Date.now() - lastEventTime > 30_000) {
259
299
logger.warn(`No events for ${Math.round((Date.now() - lastEventTime) / 1000)}s`)
260
300
}
261
301
}, 30_000)
302
+
303
+
let lastSavedCursor: number | undefined
304
+
setInterval(() => {
305
+
const direct = directJetstream?.cursor
306
+
const backlink = backlinkJetstream?.cursor
307
+
const cursor = direct !== undefined && backlink !== undefined
308
+
? Math.max(direct, backlink)
309
+
: (direct ?? backlink ?? (isConnected ? Date.now() * 1000 : undefined))
310
+
if (cursor !== undefined && cursor !== lastSavedCursor) {
311
+
lastSavedCursor = cursor
312
+
saveCursor(cursor, config.jetstreamUrl)
313
+
.then(() => logger.debug(`[cursor] Saved ${cursor}`))
314
+
.catch((err) => logger.error('[cursor] Failed to save cursor', err))
315
+
}
316
+
}, 5_000)
262
317
}
263
318
264
319
export function stopFirehose(): void {