Import your Last.fm and Spotify listening history to the AT Protocol network using the
fm.teal.alpha.feed.play lexicon.
refactor: consolidate publisher modules
Remove publisher-applywrites.ts in favor of unified publisher.ts.
The functionality has been merged into src/lib/publisher.ts with enhanced
adaptive rate limiting capabilities.
-360
-360
src/lib/publisher-applywrites.ts
-360
src/lib/publisher-applywrites.ts
···
1
-
import type { AtpAgent } from '@atproto/api';
2
-
import { formatDuration } from '../utils/helpers.js';
3
-
import { isImportCancelled } from '../utils/killswitch.js';
4
-
import {
5
-
calculateDailySchedule,
6
-
displayRateLimitWarning,
7
-
displayRateLimitInfo,
8
-
calculateRateLimitedBatches,
9
-
} from '../utils/rate-limiter.js';
10
-
import type { PlayRecord, Config, PublishResult } from '../types.js';
11
-
12
-
/**
13
-
* Maximum operations allowed per applyWrites call
14
-
* See: https://github.com/bluesky-social/atproto/pull/1571
15
-
*/
16
-
const MAX_APPLY_WRITES_OPS = 10;
17
-
18
-
/**
19
-
* Publish records using com.atproto.repo.applyWrites for efficient batching
20
-
*/
21
-
export async function publishRecordsWithApplyWrites(
22
-
agent: AtpAgent | null,
23
-
records: PlayRecord[],
24
-
batchSize: number,
25
-
batchDelay: number,
26
-
config: Config,
27
-
dryRun = false,
28
-
syncMode = false
29
-
): Promise<PublishResult> {
30
-
const { RECORD_TYPE } = config;
31
-
const totalRecords = records.length;
32
-
33
-
if (dryRun) {
34
-
return handleDryRun(records, batchSize, batchDelay, config, syncMode);
35
-
}
36
-
37
-
if (!agent) {
38
-
throw new Error('Agent is required for publishing');
39
-
}
40
-
41
-
// Calculate rate-limited batch parameters
42
-
const rateLimitParams = calculateRateLimitedBatches(totalRecords, config);
43
-
44
-
// Override with calculated parameters if rate limiting is needed
45
-
if (rateLimitParams.needsRateLimiting) {
46
-
displayRateLimitWarning();
47
-
batchSize = rateLimitParams.batchSize;
48
-
batchDelay = rateLimitParams.batchDelay;
49
-
}
50
-
51
-
// Ensure batch size doesn't exceed applyWrites limit
52
-
batchSize = Math.min(batchSize, MAX_APPLY_WRITES_OPS);
53
-
54
-
displayRateLimitInfo(
55
-
totalRecords,
56
-
batchSize,
57
-
batchDelay,
58
-
rateLimitParams.estimatedDays,
59
-
rateLimitParams.recordsPerDay
60
-
);
61
-
62
-
// Calculate daily schedule if multi-day import
63
-
const dailySchedule =
64
-
rateLimitParams.estimatedDays > 1
65
-
? calculateDailySchedule(
66
-
totalRecords,
67
-
batchSize,
68
-
batchDelay,
69
-
rateLimitParams.recordsPerDay
70
-
)
71
-
: null;
72
-
73
-
let successCount = 0;
74
-
let errorCount = 0;
75
-
const startTime = Date.now();
76
-
77
-
const totalBatches = Math.ceil(totalRecords / batchSize);
78
-
const estimatedTime = formatDuration(totalBatches * batchDelay);
79
-
80
-
console.log(`Publishing ${totalRecords} records using applyWrites in batches of ${batchSize}...`);
81
-
console.log(`Total batches: ${totalBatches}`);
82
-
if (!dailySchedule) {
83
-
console.log(`Estimated time: ${estimatedTime}`);
84
-
}
85
-
console.log(`\n🚨 Press Ctrl+C to stop gracefully after current batch\n`);
86
-
87
-
// If multi-day, process day by day
88
-
if (dailySchedule) {
89
-
for (const day of dailySchedule) {
90
-
console.log(`\n╔═══════════════════════════════════════════════════════════════╗`);
91
-
console.log(`║ DAY ${day.day} of ${rateLimitParams.estimatedDays}`);
92
-
console.log(`║ Records: ${day.recordsStart + 1}-${day.recordsEnd} (${day.recordsCount} total)`);
93
-
console.log(`╚═══════════════════════════════════════════════════════════════╝\n`);
94
-
95
-
const dayRecords = records.slice(day.recordsStart, day.recordsEnd);
96
-
const result = await processDayBatchWithApplyWrites(
97
-
agent,
98
-
dayRecords,
99
-
batchSize,
100
-
batchDelay,
101
-
RECORD_TYPE,
102
-
day.recordsStart,
103
-
totalRecords,
104
-
startTime
105
-
);
106
-
107
-
successCount += result.successCount;
108
-
errorCount += result.errorCount;
109
-
110
-
if (result.cancelled) {
111
-
return { successCount, errorCount, cancelled: true };
112
-
}
113
-
114
-
// Pause between days
115
-
if (day.pauseAfter) {
116
-
console.log(`\n⏸️ Pausing for 24 hours before continuing...`);
117
-
console.log(` Next batch will start at: ${new Date(Date.now() + day.pauseDuration).toLocaleString()}`);
118
-
console.log(` Progress: ${successCount}/${totalRecords} records completed\n`);
119
-
console.log(` 💡 You can safely stop (Ctrl+C) and restart later.\n`);
120
-
121
-
await new Promise((resolve) => setTimeout(resolve, day.pauseDuration));
122
-
}
123
-
}
124
-
} else {
125
-
// Single day import - process normally
126
-
const result = await processDayBatchWithApplyWrites(
127
-
agent,
128
-
records,
129
-
batchSize,
130
-
batchDelay,
131
-
RECORD_TYPE,
132
-
0,
133
-
totalRecords,
134
-
startTime
135
-
);
136
-
137
-
successCount = result.successCount;
138
-
errorCount = result.errorCount;
139
-
140
-
if (result.cancelled) {
141
-
return { successCount, errorCount, cancelled: true };
142
-
}
143
-
}
144
-
145
-
return { successCount, errorCount, cancelled: false };
146
-
}
147
-
148
-
/**
149
-
* Process a batch of records using applyWrites (for a single day or entire import)
150
-
*/
151
-
async function processDayBatchWithApplyWrites(
152
-
agent: AtpAgent,
153
-
records: PlayRecord[],
154
-
batchSize: number,
155
-
batchDelay: number,
156
-
recordType: string,
157
-
globalOffset: number,
158
-
totalRecords: number,
159
-
startTime: number
160
-
): Promise<PublishResult> {
161
-
let successCount = 0;
162
-
let errorCount = 0;
163
-
164
-
for (let i = 0; i < records.length; i += batchSize) {
165
-
// Check killswitch before processing batch
166
-
if (isImportCancelled()) {
167
-
return handleCancellation(successCount, errorCount, totalRecords);
168
-
}
169
-
170
-
const batch = records.slice(i, Math.min(i + batchSize, records.length));
171
-
const globalIndex = globalOffset + i;
172
-
const batchNum = Math.floor(globalIndex / batchSize) + 1;
173
-
const progress = (((globalOffset + i) / totalRecords) * 100).toFixed(1);
174
-
175
-
console.log(
176
-
`[${progress}%] Batch ${batchNum} (records ${globalOffset + i + 1}-${Math.min(globalOffset + i + batchSize, globalOffset + records.length)})`
177
-
);
178
-
179
-
// Process batch using applyWrites
180
-
const batchStartTime = Date.now();
181
-
182
-
// Build writes array for applyWrites
183
-
const writes = batch.map((record) => ({
184
-
$type: 'com.atproto.repo.applyWrites#create',
185
-
collection: recordType,
186
-
value: record,
187
-
}));
188
-
189
-
try {
190
-
// Call applyWrites with the batch
191
-
const response = await agent.com.atproto.repo.applyWrites({
192
-
repo: agent.session?.did || '',
193
-
writes: writes as any, // Type assertion needed due to @atproto/api typing
194
-
});
195
-
196
-
// Count successful operations
197
-
const batchSuccessCount = response.data.results?.length || batch.length;
198
-
successCount += batchSuccessCount;
199
-
200
-
// Report if any operations in the batch failed
201
-
if (batchSuccessCount < batch.length) {
202
-
const batchFailCount = batch.length - batchSuccessCount;
203
-
errorCount += batchFailCount;
204
-
console.error(` ⚠️ ${batchFailCount} records failed in batch`);
205
-
}
206
-
} catch (error) {
207
-
// Entire batch failed
208
-
errorCount += batch.length;
209
-
const err = error as Error;
210
-
console.error(` ✗ Batch failed: ${err.message}`);
211
-
212
-
// Log which records were in the failed batch
213
-
batch.forEach((record) => {
214
-
console.error(` - ${record.trackName} by ${record.artists[0]?.artistName}`);
215
-
});
216
-
}
217
-
218
-
const batchDuration = Date.now() - batchStartTime;
219
-
const elapsed = formatDuration(Date.now() - startTime);
220
-
const remaining = formatDuration(
221
-
((totalRecords - (globalOffset + i + batchSize)) / batchSize) * batchDelay
222
-
);
223
-
224
-
console.log(
225
-
` ✓ Complete in ${batchDuration}ms (${successCount} successful, ${errorCount} failed)`
226
-
);
227
-
228
-
// Only show time estimates if not cancelled
229
-
if (!isImportCancelled()) {
230
-
console.log(` ⏱ Elapsed: ${elapsed} | Remaining: ~${remaining}\n`);
231
-
}
232
-
233
-
// Check again before waiting
234
-
if (isImportCancelled()) {
235
-
return handleCancellation(successCount, errorCount, totalRecords);
236
-
}
237
-
238
-
// Wait before next batch (except for last batch)
239
-
if (i + batchSize < records.length) {
240
-
await new Promise((resolve) => setTimeout(resolve, batchDelay));
241
-
}
242
-
}
243
-
244
-
return { successCount, errorCount, cancelled: false };
245
-
}
246
-
247
-
/**
248
-
* Handle dry run mode
249
-
*/
250
-
function handleDryRun(
251
-
records: PlayRecord[],
252
-
batchSize: number,
253
-
batchDelay: number,
254
-
config: Config,
255
-
syncMode: boolean
256
-
): PublishResult {
257
-
const totalRecords = records.length;
258
-
259
-
// Calculate rate limiting info
260
-
const rateLimitParams = calculateRateLimitedBatches(totalRecords, config);
261
-
262
-
if (rateLimitParams.needsRateLimiting) {
263
-
displayRateLimitWarning();
264
-
batchSize = rateLimitParams.batchSize;
265
-
batchDelay = rateLimitParams.batchDelay;
266
-
267
-
// Ensure batch size doesn't exceed applyWrites limit
268
-
batchSize = Math.min(batchSize, MAX_APPLY_WRITES_OPS);
269
-
270
-
displayRateLimitInfo(
271
-
totalRecords,
272
-
batchSize,
273
-
batchDelay,
274
-
rateLimitParams.estimatedDays,
275
-
rateLimitParams.recordsPerDay
276
-
);
277
-
278
-
if (rateLimitParams.estimatedDays > 1) {
279
-
const dailySchedule = calculateDailySchedule(
280
-
totalRecords,
281
-
batchSize,
282
-
batchDelay,
283
-
rateLimitParams.recordsPerDay
284
-
);
285
-
286
-
console.log('📅 Multi-Day Import Schedule:\n');
287
-
dailySchedule.forEach((day) => {
288
-
console.log(` Day ${day.day}:`);
289
-
console.log(` Records ${day.recordsStart + 1}-${day.recordsEnd} (${day.recordsCount} total)`);
290
-
if (day.pauseAfter) {
291
-
console.log(` → Pause 24h after completion`);
292
-
}
293
-
});
294
-
console.log('');
295
-
}
296
-
}
297
-
298
-
console.log(`\n=== DRY RUN MODE ${syncMode ? '(SYNC)' : ''} ===`);
299
-
if (syncMode) {
300
-
console.log(`Sync mode enabled: Only new records will be published`);
301
-
}
302
-
console.log(`Would publish ${totalRecords} records using applyWrites`);
303
-
console.log(`Batch size: ${Math.min(batchSize, MAX_APPLY_WRITES_OPS)} records per applyWrites call`);
304
-
305
-
if (rateLimitParams.estimatedDays > 1) {
306
-
console.log(
307
-
`Import would span ${rateLimitParams.estimatedDays} days with automatic pauses\n`
308
-
);
309
-
} else {
310
-
console.log(`Estimated time: ${formatDuration(Math.ceil(totalRecords / batchSize) * batchDelay)}\n`);
311
-
}
312
-
313
-
// Show first 5 records as preview
314
-
const previewCount = Math.min(5, totalRecords);
315
-
console.log(`Preview of first ${previewCount} records (in processing order):\n`);
316
-
317
-
for (let i = 0; i < previewCount; i++) {
318
-
const record = records[i];
319
-
console.log(`${i + 1}. ${record.artists[0]?.artistName} - ${record.trackName}`);
320
-
console.log(` Album: ${record.releaseName || 'N/A'}`);
321
-
console.log(` Played: ${record.playedTime}`);
322
-
console.log(` URL: ${record.originUrl}`);
323
-
324
-
// Show MusicBrainz IDs if available
325
-
const mbids = [];
326
-
if (record.artists[0]?.artistMbId)
327
-
mbids.push(`Artist: ${record.artists[0].artistMbId}`);
328
-
if (record.recordingMbId) mbids.push(`Recording: ${record.recordingMbId}`);
329
-
if (record.releaseMbId) mbids.push(`Release: ${record.releaseMbId}`);
330
-
331
-
if (mbids.length > 0) {
332
-
console.log(` MBIDs: ${mbids.join(', ')}`);
333
-
}
334
-
console.log('');
335
-
}
336
-
337
-
if (totalRecords > previewCount) {
338
-
console.log(`... and ${totalRecords - previewCount} more records\n`);
339
-
}
340
-
341
-
console.log('=== DRY RUN COMPLETE ===');
342
-
console.log('No records were actually published.');
343
-
console.log('Remove --dry-run flag to publish for real.\n');
344
-
345
-
return { successCount: totalRecords, errorCount: 0, cancelled: false };
346
-
}
347
-
348
-
/**
349
-
* Handle cancellation
350
-
*/
351
-
function handleCancellation(
352
-
successCount: number,
353
-
errorCount: number,
354
-
totalRecords: number
355
-
): PublishResult {
356
-
console.log(`\n🛑 Import cancelled by user`);
357
-
console.log(` Processed: ${successCount}/${totalRecords} records`);
358
-
console.log(` Remaining: ${totalRecords - successCount} records\n`);
359
-
return { successCount, errorCount, cancelled: true };
360
-
}