See the best posts from any Bluesky account
add BackfillJob queue job with full test coverage
Implements Task 7: the @adonisjs/queue job class that paginates getAuthorFeed,
hydrates batches of 25 URIs via getPosts, inserts post_snapshots into ClickHouse,
tracks progress on backfill_jobs.fetched_posts, and marks users.backfilled_at on
completion. The failed() hook sets state='failed' with the error message.
Also adds the #jobs/* import alias to package.json (was missing — caused import
resolution errors in tests and would have caused the same at runtime).
7 new tests covering: happy path, pagination, BACKFILL_MAX_POSTS cap, empty user,
failed() hook, missing User row, and missing BackfillJob row.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
+535
-1
+152
apps/web/app/jobs/backfill_job.ts
+152
apps/web/app/jobs/backfill_job.ts
···
1
+
import { inject } from '@adonisjs/core'
2
+
import { Job } from '@adonisjs/queue'
3
+
import { AtprotoClient, parseGetPostsResponse } from '@skystar/atproto'
4
+
import { ClickHouseStore } from '@skystar/clickhouse'
5
+
import User from '#models/user'
6
+
import BackfillJobRow from '#models/backfill_job'
7
+
8
+
// ---------------------------------------------------------------------------
9
+
// Payload
10
+
// ---------------------------------------------------------------------------
11
+
12
+
export interface BackfillJobPayload {
13
+
did: string
14
+
}
15
+
16
+
// ---------------------------------------------------------------------------
17
+
// BackfillJob
18
+
// ---------------------------------------------------------------------------
19
+
20
+
/**
21
+
* Queue job that backfills a user's historical posts into ClickHouse.
22
+
*
23
+
* Dispatched by the web process when a new user is searched and their
24
+
* backfilled_at is NULL. The job:
25
+
*
26
+
* 1. Paginates getAuthorFeed until cursor is exhausted or BACKFILL_MAX_POSTS
27
+
* is reached (default 10,000).
28
+
* 2. For every batch of 25 URIs, calls getPosts, parses via
29
+
* parseGetPostsResponse, and inserts into ClickHouse via
30
+
* ClickHouseStore.insertPostSnapshots.
31
+
* 3. Updates backfill_jobs.fetched_posts so the loading page can display
32
+
* progress.
33
+
* 4. On completion, sets users.backfilled_at = now() and
34
+
* backfill_jobs.state = 'done'.
35
+
* 5. On permanent failure (after all retries), sets
36
+
* backfill_jobs.state = 'failed' and backfill_jobs.error = <message>.
37
+
*
38
+
* Spec: §6 Flow 2
39
+
*/
40
+
@inject()
41
+
export default class BackfillJob extends Job<BackfillJobPayload> {
42
+
static options = {
43
+
queue: 'default',
44
+
retry: {
45
+
maxRetries: 3,
46
+
},
47
+
}
48
+
49
+
constructor(
50
+
private readonly atprotoClient: AtprotoClient,
51
+
private readonly clickHouseStore: ClickHouseStore
52
+
) {
53
+
super()
54
+
}
55
+
56
+
async execute(): Promise<void> {
57
+
const { did } = this.payload
58
+
59
+
// Read BACKFILL_MAX_POSTS from env (default 10,000)
60
+
const maxPosts = Number(process.env['BACKFILL_MAX_POSTS'] ?? 10000)
61
+
62
+
// Fetch the User row — dispatcher must insert it before dispatching
63
+
const user = await User.find(did)
64
+
if (!user) {
65
+
throw new Error(
66
+
`BackfillJob: User row missing for DID ${did} — dispatcher should insert it first`
67
+
)
68
+
}
69
+
70
+
// Fetch the BackfillJob row (the Lucid row)
71
+
const jobRow = await BackfillJobRow.find(did)
72
+
if (!jobRow) {
73
+
throw new Error(
74
+
`BackfillJob: BackfillJob row missing for DID ${did} — dispatcher should insert it first`
75
+
)
76
+
}
77
+
78
+
// If state is not 'running', this job is already done or was superseded
79
+
if (jobRow.state !== 'running') {
80
+
return
81
+
}
82
+
83
+
// ---------------------------------------------------------------------------
84
+
// Main backfill loop
85
+
// ---------------------------------------------------------------------------
86
+
87
+
let cursor: string | undefined
88
+
let fetchedCount = 0
89
+
90
+
while (true) {
91
+
const page = await this.atprotoClient.getAuthorFeed(did, cursor, 100)
92
+
93
+
// Process the page in batches of 25 URIs
94
+
const uris = page.posts.map((feedItem) => feedItem.post.uri)
95
+
96
+
for (let i = 0; i < uris.length; i += 25) {
97
+
const uriBatch = uris.slice(i, i + 25)
98
+
99
+
// Hydrate with aggregate counts
100
+
const postsResponse = await this.atprotoClient.getPosts(uriBatch)
101
+
102
+
// Parse into PostSnapshot[] (snapshotTakenAt set to now() inside parser)
103
+
const snapshots = parseGetPostsResponse({ posts: postsResponse })
104
+
105
+
// Insert into ClickHouse
106
+
await this.clickHouseStore.insertPostSnapshots(snapshots)
107
+
108
+
fetchedCount += snapshots.length
109
+
110
+
// Update progress periodically
111
+
jobRow.fetchedPosts = fetchedCount
112
+
await jobRow.save()
113
+
114
+
// Check cap
115
+
if (fetchedCount >= maxPosts) {
116
+
break
117
+
}
118
+
}
119
+
120
+
// Check cap or cursor exhausted
121
+
if (fetchedCount >= maxPosts || page.cursor === undefined) {
122
+
break
123
+
}
124
+
125
+
cursor = page.cursor
126
+
}
127
+
128
+
// ---------------------------------------------------------------------------
129
+
// Mark completion
130
+
// ---------------------------------------------------------------------------
131
+
132
+
const now = Date.now()
133
+
134
+
user.backfilledAt = now
135
+
await user.save()
136
+
137
+
jobRow.state = 'done'
138
+
jobRow.finishedAt = now
139
+
jobRow.fetchedPosts = fetchedCount
140
+
await jobRow.save()
141
+
}
142
+
143
+
async failed(error: Error): Promise<void> {
144
+
const jobRow = await BackfillJobRow.find(this.payload.did)
145
+
if (!jobRow) return
146
+
147
+
jobRow.state = 'failed'
148
+
jobRow.error = error.message
149
+
jobRow.finishedAt = Date.now()
150
+
await jobRow.save()
151
+
}
152
+
}
+2
-1
apps/web/package.json
+2
-1
apps/web/package.json
+381
apps/web/tests/unit/jobs/backfill_job.spec.ts
+381
apps/web/tests/unit/jobs/backfill_job.spec.ts
···
1
+
import { test } from '@japa/runner'
2
+
import testUtils from '@adonisjs/core/services/test_utils'
3
+
import type { PostSnapshot } from '@skystar/atproto'
4
+
import type { AtprotoClient } from '@skystar/atproto'
5
+
import type { ClickHouseStore } from '@skystar/clickhouse'
6
+
import User from '#models/user'
7
+
import BackfillJobRow from '#models/backfill_job'
8
+
import BackfillJob from '#jobs/backfill_job'
9
+
10
+
// ---------------------------------------------------------------------------
11
+
// Helpers
12
+
// ---------------------------------------------------------------------------
13
+
14
+
const TEST_DID = 'did:plc:backfilljobspec'
15
+
16
+
/**
17
+
* Create a minimal FeedViewPost-shaped object for a given URI.
18
+
* The real shape from the AppView has a .post field — that's the gotcha.
19
+
*/
20
+
function makeFeedViewPost(uri: string) {
21
+
return {
22
+
post: {
23
+
uri,
24
+
cid: 'bafytest',
25
+
author: { did: TEST_DID, handle: 'test.bsky.social' },
26
+
record: { $type: 'app.bsky.feed.post', text: 'Hello', createdAt: '2024-01-01T00:00:00.000Z' },
27
+
likeCount: 1,
28
+
repostCount: 0,
29
+
quoteCount: 0,
30
+
indexedAt: '2024-01-01T00:00:00.000Z',
31
+
},
32
+
}
33
+
}
34
+
35
+
/**
36
+
* Minimal PostView for getPosts mock response.
37
+
*/
38
+
function makePostView(uri: string, index: number = 0) {
39
+
return {
40
+
uri,
41
+
cid: 'bafytest',
42
+
author: { did: TEST_DID, handle: 'test.bsky.social' },
43
+
record: {
44
+
$type: 'app.bsky.feed.post',
45
+
text: `Post ${index}`,
46
+
createdAt: '2024-01-01T00:00:00.000Z',
47
+
},
48
+
likeCount: 1,
49
+
repostCount: 0,
50
+
quoteCount: 0,
51
+
indexedAt: '2024-01-01T00:00:00.000Z',
52
+
}
53
+
}
54
+
55
+
type MockGetAuthorFeedPage = { posts: ReturnType<typeof makeFeedViewPost>[]; cursor?: string }
56
+
57
+
interface MockAtprotoClientConfig {
58
+
pages: MockGetAuthorFeedPage[]
59
+
}
60
+
61
+
/**
62
+
* Build a mock AtprotoClient. getPosts returns PostView objects matching the
63
+
* URIs given. getAuthorFeed pages through the pages array in order.
64
+
*/
65
+
function makeMockAtprotoClient(config: MockAtprotoClientConfig): AtprotoClient {
66
+
let pageIndex = 0
67
+
const getAuthorFeedCalls: Array<{ did: string; cursor?: string }> = []
68
+
const getPostsCalls: string[][] = []
69
+
70
+
const client = {
71
+
_getAuthorFeedCalls: getAuthorFeedCalls,
72
+
_getPostsCalls: getPostsCalls,
73
+
74
+
async getAuthorFeed(
75
+
did: string,
76
+
cursor?: string
77
+
): Promise<{ posts: (typeof config.pages)[0]['posts']; cursor?: string }> {
78
+
getAuthorFeedCalls.push({ did, cursor })
79
+
const page = config.pages[pageIndex] ?? { posts: [] }
80
+
pageIndex++
81
+
return page
82
+
},
83
+
84
+
async getPosts(uris: string[]): Promise<ReturnType<typeof makePostView>[]> {
85
+
getPostsCalls.push(uris)
86
+
return uris.map((uri, i) => makePostView(uri, i))
87
+
},
88
+
} as unknown as AtprotoClient & {
89
+
_getAuthorFeedCalls: typeof getAuthorFeedCalls
90
+
_getPostsCalls: typeof getPostsCalls
91
+
}
92
+
93
+
return client
94
+
}
95
+
96
+
interface MockClickHouseStoreConfig {
97
+
shouldThrow?: boolean
98
+
}
99
+
100
+
function makeMockClickHouseStore(config: MockClickHouseStoreConfig = {}): ClickHouseStore & {
101
+
_calls: PostSnapshot[][]
102
+
} {
103
+
const calls: PostSnapshot[][] = []
104
+
105
+
return {
106
+
_calls: calls,
107
+
async insertPostSnapshots(snapshots: PostSnapshot[]) {
108
+
if (config.shouldThrow) throw new Error('ClickHouse unavailable')
109
+
calls.push(snapshots)
110
+
},
111
+
} as unknown as ClickHouseStore & { _calls: PostSnapshot[][] }
112
+
}
113
+
114
+
/** Create a BackfillJob instance with mocked dependencies, bypassing the IoC container. */
115
+
function makeJob(
116
+
atprotoClient: AtprotoClient,
117
+
clickHouseStore: ClickHouseStore,
118
+
did: string = TEST_DID
119
+
): BackfillJob {
120
+
const job = new BackfillJob(atprotoClient, clickHouseStore)
121
+
// Hydrate with minimal context (payload is what we need)
122
+
job.$hydrate({ did }, { jobId: 'test-job-id', attempt: 1, queue: 'default' } as Parameters<
123
+
typeof job.$hydrate
124
+
>[1])
125
+
return job
126
+
}
127
+
128
+
// ---------------------------------------------------------------------------
129
+
// Test suites
130
+
// ---------------------------------------------------------------------------
131
+
132
+
test.group('BackfillJob — happy path: full backfill of small user', (group) => {
133
+
group.each.setup(() => testUtils.db().withGlobalTransaction())
134
+
135
+
test('inserts snapshots, sets backfilledAt, marks job done', async ({ assert }) => {
136
+
// Seed SQLite rows
137
+
await User.create({ did: TEST_DID, handle: 'test.bsky.social', firstSeenAt: Date.now() })
138
+
await BackfillJobRow.create({
139
+
did: TEST_DID,
140
+
startedAt: Date.now(),
141
+
state: 'running',
142
+
fetchedPosts: 0,
143
+
})
144
+
145
+
const uris = Array.from({ length: 5 }, (_, i) => `at://${TEST_DID}/app.bsky.feed.post/rkey${i}`)
146
+
const atproto = makeMockAtprotoClient({
147
+
pages: [{ posts: uris.map(makeFeedViewPost) }], // no cursor = single page
148
+
}) as ReturnType<typeof makeMockAtprotoClient> & { _getPostsCalls: string[][] }
149
+
150
+
const clickhouse = makeMockClickHouseStore()
151
+
const job = makeJob(atproto, clickhouse)
152
+
153
+
await job.execute()
154
+
155
+
// insertPostSnapshots called once with 5 snapshots (all fit in one batch of 25)
156
+
assert.equal(clickhouse._calls.length, 1)
157
+
assert.equal(clickhouse._calls[0].length, 5)
158
+
159
+
// users.backfilledAt is now set
160
+
const user = await User.findOrFail(TEST_DID)
161
+
assert.isNotNull(user.backfilledAt)
162
+
163
+
// backfill_jobs.state = 'done'
164
+
const jobRow = await BackfillJobRow.findOrFail(TEST_DID)
165
+
assert.equal(jobRow.state, 'done')
166
+
assert.isNotNull(jobRow.finishedAt)
167
+
assert.equal(jobRow.fetchedPosts, 5)
168
+
})
169
+
})
170
+
171
+
test.group('BackfillJob — pagination: backfills multiple pages', (group) => {
172
+
group.each.setup(() => testUtils.db().withGlobalTransaction())
173
+
174
+
test('calls getAuthorFeed twice, getPosts 8 times, final fetchedPosts is 200', async ({
175
+
assert,
176
+
}) => {
177
+
await User.create({ did: TEST_DID, handle: 'test.bsky.social', firstSeenAt: Date.now() })
178
+
await BackfillJobRow.create({
179
+
did: TEST_DID,
180
+
startedAt: Date.now(),
181
+
state: 'running',
182
+
fetchedPosts: 0,
183
+
})
184
+
185
+
// Two pages of 100 posts each
186
+
const makePage = (count: number, cursor?: string): MockGetAuthorFeedPage => ({
187
+
posts: Array.from({ length: count }, (_, i) =>
188
+
makeFeedViewPost(`at://${TEST_DID}/app.bsky.feed.post/p${i}`)
189
+
),
190
+
cursor,
191
+
})
192
+
193
+
const atproto = makeMockAtprotoClient({
194
+
pages: [
195
+
makePage(100, 'cursor-page-2'), // page 1 has a cursor
196
+
makePage(100), // page 2 has no cursor
197
+
],
198
+
}) as unknown as ReturnType<typeof makeMockAtprotoClient> & {
199
+
_getAuthorFeedCalls: Array<{ did: string; cursor?: string }>
200
+
_getPostsCalls: string[][]
201
+
}
202
+
203
+
const clickhouse = makeMockClickHouseStore()
204
+
const job = makeJob(atproto, clickhouse)
205
+
206
+
await job.execute()
207
+
208
+
const feedCalls = (
209
+
atproto as unknown as { _getAuthorFeedCalls: Array<{ did: string; cursor?: string }> }
210
+
)._getAuthorFeedCalls
211
+
const postsCalls = (atproto as unknown as { _getPostsCalls: string[][] })._getPostsCalls
212
+
213
+
// getAuthorFeed called twice
214
+
assert.equal(feedCalls.length, 2)
215
+
assert.isUndefined(feedCalls[0].cursor)
216
+
assert.equal(feedCalls[1].cursor, 'cursor-page-2')
217
+
218
+
// 200 posts / 25 per batch = 8 getPosts calls
219
+
assert.equal(postsCalls.length, 8)
220
+
221
+
// 8 insertPostSnapshots calls
222
+
assert.equal(clickhouse._calls.length, 8)
223
+
224
+
// Final state
225
+
const jobRow = await BackfillJobRow.findOrFail(TEST_DID)
226
+
assert.equal(jobRow.fetchedPosts, 200)
227
+
assert.equal(jobRow.state, 'done')
228
+
})
229
+
})
230
+
231
+
test.group('BackfillJob — cap at BACKFILL_MAX_POSTS', (group) => {
232
+
group.each.setup(() => testUtils.db().withGlobalTransaction())
233
+
234
+
test('stops after max posts, does not fetch more pages', async ({ assert }) => {
235
+
await User.create({ did: TEST_DID, handle: 'test.bsky.social', firstSeenAt: Date.now() })
236
+
await BackfillJobRow.create({
237
+
did: TEST_DID,
238
+
startedAt: Date.now(),
239
+
state: 'running',
240
+
fetchedPosts: 0,
241
+
})
242
+
243
+
// Lower the cap to 50 for this test
244
+
const originalMax = process.env['BACKFILL_MAX_POSTS']
245
+
process.env['BACKFILL_MAX_POSTS'] = '50'
246
+
247
+
try {
248
+
// Many pages of 100 posts each
249
+
const makePage = (cursor?: string): MockGetAuthorFeedPage => ({
250
+
posts: Array.from({ length: 100 }, (_, i) =>
251
+
makeFeedViewPost(`at://${TEST_DID}/app.bsky.feed.post/p${i}`)
252
+
),
253
+
cursor,
254
+
})
255
+
256
+
const pages: MockGetAuthorFeedPage[] = Array.from({ length: 10 }, (_, i) =>
257
+
makePage(`cursor-${i + 1}`)
258
+
)
259
+
260
+
const atproto = makeMockAtprotoClient({ pages }) as unknown as ReturnType<
261
+
typeof makeMockAtprotoClient
262
+
> & {
263
+
_getAuthorFeedCalls: Array<{ did: string; cursor?: string }>
264
+
}
265
+
const clickhouse = makeMockClickHouseStore()
266
+
const job = makeJob(atproto, clickhouse)
267
+
268
+
await job.execute()
269
+
270
+
const feedCalls = (
271
+
atproto as unknown as { _getAuthorFeedCalls: Array<{ did: string; cursor?: string }> }
272
+
)._getAuthorFeedCalls
273
+
274
+
// Should stop after 50 posts (2 batches of 25 from page 1)
275
+
// Only 1 getAuthorFeed call (stopped mid-page after 50 posts)
276
+
assert.equal(feedCalls.length, 1)
277
+
278
+
const jobRow = await BackfillJobRow.findOrFail(TEST_DID)
279
+
assert.equal(jobRow.fetchedPosts, 50)
280
+
assert.equal(jobRow.state, 'done')
281
+
} finally {
282
+
if (originalMax === undefined) {
283
+
delete process.env['BACKFILL_MAX_POSTS']
284
+
} else {
285
+
process.env['BACKFILL_MAX_POSTS'] = originalMax
286
+
}
287
+
}
288
+
})
289
+
})
290
+
291
+
test.group('BackfillJob — empty user', (group) => {
292
+
group.each.setup(() => testUtils.db().withGlobalTransaction())
293
+
294
+
test('no insertPostSnapshots call, backfilledAt still set, state done', async ({ assert }) => {
295
+
await User.create({ did: TEST_DID, handle: 'test.bsky.social', firstSeenAt: Date.now() })
296
+
await BackfillJobRow.create({
297
+
did: TEST_DID,
298
+
startedAt: Date.now(),
299
+
state: 'running',
300
+
fetchedPosts: 0,
301
+
})
302
+
303
+
const atproto = makeMockAtprotoClient({
304
+
pages: [{ posts: [] }], // empty — no cursor
305
+
})
306
+
307
+
const clickhouse = makeMockClickHouseStore()
308
+
const job = makeJob(atproto, clickhouse)
309
+
310
+
await job.execute()
311
+
312
+
// No snapshots inserted (empty user has no posts)
313
+
assert.equal(clickhouse._calls.length, 0)
314
+
315
+
// User still gets marked as backfilled
316
+
const user = await User.findOrFail(TEST_DID)
317
+
assert.isNotNull(user.backfilledAt)
318
+
319
+
// Job row marked done
320
+
const jobRow = await BackfillJobRow.findOrFail(TEST_DID)
321
+
assert.equal(jobRow.state, 'done')
322
+
assert.equal(jobRow.fetchedPosts, 0)
323
+
})
324
+
})
325
+
326
+
test.group('BackfillJob — failed() hook', (group) => {
327
+
group.each.setup(() => testUtils.db().withGlobalTransaction())
328
+
329
+
test('failed() sets state=failed and saves error message', async ({ assert }) => {
330
+
await User.create({ did: TEST_DID, handle: 'test.bsky.social', firstSeenAt: Date.now() })
331
+
await BackfillJobRow.create({
332
+
did: TEST_DID,
333
+
startedAt: Date.now(),
334
+
state: 'running',
335
+
fetchedPosts: 0,
336
+
})
337
+
338
+
const atproto = makeMockAtprotoClient({ pages: [] })
339
+
const clickhouse = makeMockClickHouseStore()
340
+
const job = makeJob(atproto, clickhouse)
341
+
342
+
const syntheticError = new Error('Bluesky AppView exploded')
343
+
await job.failed(syntheticError)
344
+
345
+
const jobRow = await BackfillJobRow.findOrFail(TEST_DID)
346
+
assert.equal(jobRow.state, 'failed')
347
+
assert.include(jobRow.error!, 'Bluesky AppView exploded')
348
+
assert.isNotNull(jobRow.finishedAt)
349
+
})
350
+
})
351
+
352
+
test.group('BackfillJob — error cases', (group) => {
353
+
group.each.setup(() => testUtils.db().withGlobalTransaction())
354
+
355
+
test('throws if User row is missing', async ({ assert }) => {
356
+
// No User.create — just the backfill job row
357
+
await BackfillJobRow.create({
358
+
did: TEST_DID,
359
+
startedAt: Date.now(),
360
+
state: 'running',
361
+
fetchedPosts: 0,
362
+
})
363
+
364
+
const atproto = makeMockAtprotoClient({ pages: [] })
365
+
const clickhouse = makeMockClickHouseStore()
366
+
const job = makeJob(atproto, clickhouse)
367
+
368
+
await assert.rejects(() => job.execute(), /User row missing/)
369
+
})
370
+
371
+
test('throws if BackfillJob row is missing', async ({ assert }) => {
372
+
// No BackfillJobRow.create — just the user
373
+
await User.create({ did: TEST_DID, handle: 'test.bsky.social', firstSeenAt: Date.now() })
374
+
375
+
const atproto = makeMockAtprotoClient({ pages: [] })
376
+
const clickhouse = makeMockClickHouseStore()
377
+
const job = makeJob(atproto, clickhouse)
378
+
379
+
await assert.rejects(() => job.execute(), /BackfillJob row missing/)
380
+
})
381
+
})