Something went wrong. Try again.
FlockOff is a mass-block tool for the Atmosphere flockoff.app
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306import 'server-only'import { Agent } from '@atproto/api'import { getOAuthClient, restoreAgent } from './oauth'import { getBlockEvent, updateBlockEvent, createBlock } from './lexicon'import { sleep } from './utils'import type { QueueJob, JobStatus, ProgressMessage } from '@/types'
// ─── In-Memory Job Store ──────────────────────────────────────────────────────// Anchored to globalThis so state survives Next.js HMR module re-evaluation in// dev. In production (single Docker process) this is equivalent to module scope.// For multi-instance deployments, swap for Redis.
declare global { // eslint-disable-next-line no-var var __activeJobs: Map<string, QueueJob> | undefined // eslint-disable-next-line no-var var __activeEventRkeys: Set<string> | undefined}
const activeJobs: Map<string, QueueJob> = (globalThis.__activeJobs ??= new Map())export const MAX_ACTIVE_JOBS_PER_USER = 3
// Track which event rkeys currently have a running job. Prevents concurrent// jobs on the same event, which causes persistProgress to stomp each other's state.const activeEventRkeys: Set<string> = (globalThis.__activeEventRkeys ??= new Set())
export function isEventJobRunning(eventRkey: string): boolean { return activeEventRkeys.has(eventRkey)}
export function countRunningJobsForDid(did: string): number { let count = 0 for (const job of activeJobs.values()) { if (job.did === did && job.status === 'running') count++ } return count}
export function canStartJobForDid(did: string): boolean { return countRunningJobsForDid(did) < MAX_ACTIVE_JOBS_PER_USER}
// ─── Public API ───────────────────────────────────────────────────────────────
export function getJob(jobId: string): QueueJob | undefined { return activeJobs.get(jobId)}
export function addSseClient( jobId: string, controller: ReadableStreamDefaultController): void { activeJobs.get(jobId)?.sseClients.add(controller)}
export function removeSseClient( jobId: string, controller: ReadableStreamDefaultController): void { activeJobs.get(jobId)?.sseClients.delete(controller)}
/** * Start a block job. Creates the in-memory job, kicks off background processing, * and returns immediately. Progress is streamed via SSE. */export async function startBlockJob( jobId: string, did: string, eventRkey: string, dids: string[]): Promise<QueueJob> { const job: QueueJob = { id: jobId, did, eventRkey, status: 'running' as JobStatus, total: dids.length, completed: 0, failed: 0, pendingDids: [...dids], blockedDids: [], failedDids: [], blockedRkeys: [], sseClients: new Set(), startedAt: new Date(), }
activeJobs.set(jobId, job) activeEventRkeys.add(eventRkey)
// Fire and forget — runs in background while caller returns processJob(job) .catch((err) => { console.error(`[queue] Job ${jobId} crashed unexpectedly:`, err) job.status = 'failed' broadcastProgress(job) }) .finally(() => { activeEventRkeys.delete(eventRkey) })
return job}
/** * Append new DIDs to an existing block event. Pre-seeds the job with the * event's current state so persistProgress writes a merged record (not a reset). * If listUri is provided, newly blocked accounts are also added to that list. */export async function startAppendJob( jobId: string, did: string, eventRkey: string, newDids: string[], existingState: { blockedDids: string[] blockedRkeys: string[] failedDids: string[] }, listUri?: string): Promise<QueueJob> { const job: QueueJob = { id: jobId, did, eventRkey, status: 'running' as JobStatus, // Total covers everything: existing + new total: existingState.blockedDids.length + existingState.failedDids.length + newDids.length, // Start counters from existing completed/failed counts completed: existingState.blockedDids.length, failed: existingState.failedDids.length, pendingDids: [...newDids], blockedDids: [...existingState.blockedDids], failedDids: [...existingState.failedDids], blockedRkeys: [...existingState.blockedRkeys], sseClients: new Set(), startedAt: new Date(), listUri, }
activeJobs.set(jobId, job) activeEventRkeys.add(eventRkey)
processJob(job) .catch((err) => { console.error(`[queue] Append job ${jobId} crashed:`, err) job.status = 'failed' broadcastProgress(job) }) .finally(() => { activeEventRkeys.delete(eventRkey) })
return job}
// ─── Job Processing ───────────────────────────────────────────────────────────
async function processJob(job: QueueJob): Promise<void> { // Restore once — the OAuthSession inside the Agent handles token refresh // transparently on every request via its fetchHandler. const agent = await restoreAgent(job.did) const retryAttempts = new Map<string, number>()
while (job.pendingDids.length > 0) { const targetDid = job.pendingDids[0] let succeeded = false
try { const rkey = await createBlock(agent, job.did, targetDid)
job.blockedDids.push(targetDid) job.blockedRkeys.push(rkey) job.completed++ succeeded = true
// If this job is appending to an existing block list, also create a listitem if (job.listUri) { try { await agent.com.atproto.repo.createRecord({ repo: job.did, collection: 'app.bsky.graph.listitem', record: { $type: 'app.bsky.graph.listitem', list: job.listUri, subject: targetDid, createdAt: new Date().toISOString(), }, }) } catch { // Non-fatal — block record is created; list item failure is logged only console.warn(`[queue] Failed to add ${targetDid} to list ${job.listUri}`) } } } catch (err: unknown) { const status = (err as { status?: number })?.status const headers = (err as { headers?: Record<string, string> })?.headers
const attempts = retryAttempts.get(targetDid) ?? 0 if (status === 429 && attempts < 1) { // Back off and retry this DID once const retryAfterSec = parseInt(headers?.['retry-after'] ?? '30') const backoff = Math.min(retryAfterSec * 1000, 60_000) console.warn(`[queue] Rate limited. Backing off ${backoff}ms before retry.`) await sleep(backoff) retryAttempts.set(targetDid, attempts + 1) continue // retry same DID }
// Any other error: mark as failed, move on job.failedDids.push(targetDid) job.failed++ retryAttempts.delete(targetDid) console.error(`[queue] Failed to block ${targetDid}:`, (err as Error)?.message ?? err) }
// Only advance the queue if we didn't loop back for a retry if (succeeded || job.failedDids[job.failedDids.length - 1] === targetDid) { retryAttempts.delete(targetDid) job.pendingDids.shift() }
broadcastProgress(job)
// Persist to PDS every 10 blocks, and always on completion const processed = job.completed + job.failed if (processed % 10 === 0 || job.pendingDids.length === 0) { await persistProgress(job, agent).catch((err) => console.error('[queue] PDS persist failed:', err) ) }
// Conservative rate limit: 1 block per second if (job.pendingDids.length > 0) { await sleep(1000) } }
job.status = 'complete' await persistProgress(job, agent).catch((err) => console.error('[queue] Final PDS persist failed:', err) ) broadcastProgress(job)
// Keep job in memory for 5 minutes so late SSE clients can get final status setTimeout(() => activeJobs.delete(job.id), 5 * 60_000)}
// ─── PDS Persistence ──────────────────────────────────────────────────────────
async function persistProgress(job: QueueJob, agent: Agent): Promise<void> { for (let attempt = 0; attempt < 3; attempt++) { const { record, cid } = await getBlockEvent(agent, job.did, job.eventRkey)
try { await updateBlockEvent(agent, job.did, job.eventRkey, { ...record, pendingDids: [...job.pendingDids], blockedDids: [...job.blockedDids], failedDids: [...job.failedDids], blockedRkeys: [...job.blockedRkeys], stats: { total: job.total, completed: job.completed, failed: job.failed, }, }, { swapRecord: cid }) return } catch (err: unknown) { const status = (err as { status?: number })?.status const errorName = (err as { error?: string })?.error if ((status === 409 || errorName === 'InvalidSwap') && attempt < 2) { await sleep(250) continue } throw err } }}
// ─── SSE Broadcasting ─────────────────────────────────────────────────────────
function broadcastProgress(job: QueueJob): void { const message: ProgressMessage = { jobId: job.id, status: job.status, total: job.total, completed: job.completed, failed: job.failed, }
const encoded = new TextEncoder().encode(`data: ${JSON.stringify(message)}\n\n`) const isDone = job.status !== 'running'
for (const controller of job.sseClients) { try { controller.enqueue(encoded) if (isDone) controller.close() } catch { // Client disconnected — clean up job.sseClients.delete(controller) } }}