fix(api): honour default TTLs and expire stale job ledger rows (#2887)

fluxer_api/src/api/worker/BlocklistFeedStartup.ts

@@ -0,0 +1,52 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {Logger} from '@app/api/Logger';
+import type {WorkerTaskName} from '@app/api/worker/WorkerLaneConfig';
+import {WorkerQueueOverflowError} from '@app/api/worker/WorkerQueueOverflowError';
+import type {WorkerService} from '@app/api/worker/WorkerService';
+import type {IKVProvider} from '@pkgs/kv_client/src/IKVProvider';
+import {ms} from 'itty-time';
+
+const INITIAL_SYNC_KEY = 'sync:email_domains:initialized';
+const PURGE_KEY = 'sync:blocklist_feeds:purged';
+const CLAIM_TTL_SECONDS = ms('6 hours') / 1000;
+const FEED_TASKS = [
+	'syncDisposableEmailDomains',
+	'syncUrlBlocklists',
+	'syncFileShaBlocklists',
+] as const satisfies ReadonlyArray<WorkerTaskName>;
+
+export async function queueBlocklistFeedStartupJobs(
+	kvClient: Pick<IKVProvider, 'setnx' | 'del'>,
+	workerService: Pick<WorkerService, 'addJob'>,
+	enabled: boolean,
+): Promise<void> {
+	if (enabled) {
+		if (await kvClient.setnx(INITIAL_SYNC_KEY, '1', CLAIM_TTL_SECONDS)) {
+			Logger.info('Triggering initial disposable email domain sync');
+			await queueJobs(workerService, ['syncDisposableEmailDomains']);
+		}
+		return;
+	}
+	const wasEnabled = (await kvClient.del(INITIAL_SYNC_KEY)) > 0;
+	const claimed = await kvClient.setnx(PURGE_KEY, '1', CLAIM_TTL_SECONDS);
+	if (!wasEnabled && !claimed) return;
+	Logger.info('Removing blocklist feed data, blocklist feeds are disabled');
+	await queueJobs(workerService, FEED_TASKS);
+}
+
+async function queueJobs(
+	workerService: Pick<WorkerService, 'addJob'>,
+	tasks: ReadonlyArray<WorkerTaskName>,
+): Promise<void> {
+	for (const task of tasks) {
+		try {
+			await workerService.addJob(task, {});
+		} catch (error) {
+			if (!(error instanceof WorkerQueueOverflowError)) {
+				throw error;
+			}
+			Logger.warn({task}, 'Dropped blocklist feed job, jobs stream is at its limit');
+		}
+	}
+}

fluxer_api/src/api/worker/JetStreamWorkerQueue.ts

@@ -17,13 +17,13 @@ import {
 	StorageType,
 	type StreamConfig,
 } from '@nats-io/jetstream';
-import {nanos} from '@nats-io/transport-node';
+import {millis, nanos} from '@nats-io/transport-node';
 import type {JetStreamConnectionManager} from '@pkgs/nats/src/JetStreamConnectionManager';
 import type {WorkerJobPayload} from '@pkgs/worker/src/contracts/WorkerTypes';
 
 const STREAM_NAME = 'JOBS';
 const SUBJECT_PREFIX = 'jobs.';
-const MAX_AGE_MS = 7 * 24 * 60 * 60 * 1000;
+export const JOBS_STREAM_MAX_AGE_MS = 7 * 24 * 60 * 60 * 1000;
 const LEGACY_CONSUMER_NAME = 'workers';
 const DLQ_STREAM_NAME = 'JOBS_DLQ';
 const DLQ_SUBJECT_PREFIX = 'dlq.';
@@ -59,7 +59,7 @@ const JOBS_STREAM: WorkerStreamDefinition = {
 	name: STREAM_NAME,
 	subject: `${SUBJECT_PREFIX}>`,
 	retention: RetentionPolicy.Workqueue,
-	maxAgeMs: MAX_AGE_MS,
+	maxAgeMs: JOBS_STREAM_MAX_AGE_MS,
 	minBytes: STREAM_MIN_BYTES,
 	maxMessages: STREAM_MAX_MSGS,
 	maxMessagesPerSubject: STREAM_MAX_MSGS_PER_SUBJECT,
@@ -106,6 +106,7 @@ export class JetStreamWorkerQueue {
 	private consumersReady = false;
 	private streamSetup: Promise<void> | null = null;
 	private dlqStreamSetup: Promise<void> | null = null;
+	private jobsStreamMaxAgeMs = JOBS_STREAM_MAX_AGE_MS;
 
 	constructor(connectionManager: JetStreamConnectionManager) {
 		this.connectionManager = connectionManager;
@@ -125,11 +126,16 @@ export class JetStreamWorkerQueue {
 		if (existingConfig === null) {
 			await this.addStream(jsm);
 		} else {
+			this.jobsStreamMaxAgeMs = millis(existingConfig.max_age);
 			await this.applyStreamLimits(jsm, existingConfig);
 		}
 		this.streamReady = true;
 	}
 
+	getJobsStreamMaxAgeMs(): number {
+		return this.jobsStreamMaxAgeMs;
+	}
+
 	private async oversizedSubjects(jsm: JetStreamManager): Promise<Array<[string, number]> | null> {
 		try {
 			const info = await jsm.streams.info(STREAM_NAME, {subjects_filter: JOBS_STREAM.subject});
@@ -462,7 +468,7 @@ export class JetStreamWorkerQueue {
 			priority?: number;
 			jobKey?: string;
 		},
-	): Promise<string> {
+	): Promise<{seq: string; duplicate: boolean}> {
 		const js = this.connectionManager.getJetStreamClient();
 		const subject = `${SUBJECT_PREFIX}${taskType}`;
 		const body = JSON.stringify({
@@ -477,8 +483,7 @@ export class JetStreamWorkerQueue {
 			const ack = await js.publish(subject, body, {
 				msgID,
 			});
-			const jobId = `${ack.seq}`;
-			return jobId;
+			return {seq: `${ack.seq}`, duplicate: ack.duplicate === true};
 		} catch (error) {
 			const rejection = describeStreamRejection(error);
 			if (rejection === null) {

fluxer_api/src/api/worker/WorkerLaneConfig.ts

@@ -67,6 +67,7 @@ const LANE_CONFIG = {
 		consumerName: 'workers_batch',
 		tasks: [
 			'expireAttachments',
+			'expireStaleJobs',
 			'indexChannelMessages',
 			'indexGuildMembers',
 			'processAssetDeletionQueue',

fluxer_api/src/api/worker/WorkerMain.ts

@@ -23,8 +23,9 @@ import {
 } from '@app/api/middleware/ServiceSingletons';
 import {initializeSearch, shutdownSearch} from '@app/api/SearchFactory';
 import {awaitAll} from '@app/api/utils/ConcurrencyUtils';
+import {queueBlocklistFeedStartupJobs} from '@app/api/worker/BlocklistFeedStartup';
 import {CronScheduler} from '@app/api/worker/CronScheduler';
-import {JetStreamWorkerQueue} from '@app/api/worker/JetStreamWorkerQueue';
+import {JetStreamWorkerQueue, JOBS_STREAM_MAX_AGE_MS} from '@app/api/worker/JetStreamWorkerQueue';
 import {clearWorkerDependencies, setWorkerDependencies} from '@app/api/worker/WorkerContext';
 import {initializeWorkerDependencies, type WorkerDependencies} from '@app/api/worker/WorkerDependencies';
 import {WorkerHeartbeat} from '@app/api/worker/WorkerHeartbeat';
@@ -34,7 +35,6 @@ import {
 	validateLaneCompleteness,
 } from '@app/api/worker/WorkerLaneConfig';
 import {createWorkerProcessErrorHandler} from '@app/api/worker/WorkerProcessErrorHandler';
-import {WorkerQueueOverflowError} from '@app/api/worker/WorkerQueueOverflowError';
 import {WorkerRunner} from '@app/api/worker/WorkerRunner';
 import {WorkerService} from '@app/api/worker/WorkerService';
 import {workerTasks} from '@app/api/worker/WorkerTaskRegistry';
@@ -43,9 +43,8 @@ import {BACKGROUND_READ_TIMEOUT_MS, initCassandra, shutdownCassandra} from '@pkg
 import {JetStreamConnectionManager} from '@pkgs/nats/src/JetStreamConnectionManager';
 import {getDefaultPostgresClient, initPostgres, shutdownPostgres} from '@pkgs/postgres/src/Client';
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
-import {ms} from 'itty-time';
 
-function registerCronJobs(cron: CronScheduler): void {
+function registerCronJobs(cron: CronScheduler, jobsStreamMaxAgeMs: number): void {
 	cron.upsert('processAssetDeletionQueue', 'processAssetDeletionQueue', {}, '0 */5 * * * *', {ledger: false});
 	if (Config.cachePurge.adapter !== 'none') {
 		cron.upsert('processCachePurgeQueue', 'processCachePurgeQueue', {}, '*/10 * * * * *', {ledger: false});
@@ -62,6 +61,14 @@ function registerCronJobs(cron: CronScheduler): void {
 	}
 	cron.upsert('processInactivityDeletions', 'processInactivityDeletions', {}, '0 0 */6 * * *', {ledger: false});
 	cron.upsert('expireAttachments', 'expireAttachments', {}, '0 0 */12 * * *', {ledger: false});
+	if (jobsStreamMaxAgeMs > 0 && jobsStreamMaxAgeMs <= JOBS_STREAM_MAX_AGE_MS) {
+		cron.upsert('expireStaleJobs', 'expireStaleJobs', {}, '0 45 3 * * *', {ledger: false});
+	} else {
+		Logger.warn(
+			{jobsStreamMaxAgeMs},
+			'Jobs stream keeps jobs past 7 days, stale jobs stay active until their ledger rows expire',
+		);
+	}
 	cron.upsert('prunePostgresKvTtl', 'prunePostgresKvTtl', {}, '0 */5 * * * *', {ledger: false});
 	cron.upsert('syncDiscoveryIndex', 'syncDiscoveryIndex', {}, '0 */15 * * * *', {ledger: false});
 	if (Config.blocklistFeeds.enabled) {
@@ -240,26 +247,9 @@ export async function startWorkerMain(): Promise<void> {
 		}
 		dependencies = await initializeWorkerDependencies(snowflakeService);
 		setWorkerDependencies(dependencies);
-		if (Config.blocklistFeeds.enabled) {
-			const didClaimEmailSync = await dependencies.kvClient.setnx(
-				'sync:email_domains:initialized',
-				'1',
-				ms('6 hours') / 1000,
-			);
-			if (didClaimEmailSync) {
-				Logger.info('Triggering initial disposable email domain sync');
-				try {
-					await workerService.addJob('syncDisposableEmailDomains', {});
-				} catch (error) {
-					if (!(error instanceof WorkerQueueOverflowError)) {
-						throw error;
-					}
-					Logger.warn('Dropped initial disposable email domain sync, jobs stream is at its limit');
-				}
-			}
-		}
+		await queueBlocklistFeedStartupJobs(dependencies.kvClient, workerService, Config.blocklistFeeds.enabled);
 		cron = new CronScheduler(workerService, Logger, dependencies.kvClient, heartbeat);
-		registerCronJobs(cron);
+		registerCronJobs(cron, queue.getJobsStreamMaxAgeMs());
 		for (const lane of activeWorkerLanes) {
 			const laneTasks: Record<string, WorkerTaskHandler> = {};
 			for (const taskType of lane.taskTypes) {

fluxer_api/src/api/worker/WorkerService.ts

@@ -29,10 +29,10 @@ export class WorkerService implements IWorkerService<WorkerTaskName> {
 		const skipLedger = options?.skipLedger === true;
 		const requireLedger = options?.requireLedger === true;
 		const payloadRecord = payload as Record<string, unknown>;
-		let ledgerWritten = false;
+		let ledgerCreatedAt: Date | null = null;
 		if (!skipLedger) {
 			try {
-				await this.ledger.createJob({
+				ledgerCreatedAt = await this.ledger.createJob({
 					jobId,
 					taskType,
 					payload: payloadRecord,
@@ -43,29 +43,32 @@ export class WorkerService implements IWorkerService<WorkerTaskName> {
 					jetStreamLane: findLaneForTask(taskType),
 					jetStreamSeq: null,
 				});
-				ledgerWritten = true;
 			} catch (ledgerErr) {
 				Logger.error({err: ledgerErr, jobId: jobId.toString(), taskType}, 'Failed to write ledger row for job');
 				if (requireLedger) throw ledgerErr;
 			}
 		}
-		const enrichedPayload = ledgerWritten ? {...payloadRecord, __jobId: jobId.toString()} : payloadRecord;
+		const enrichedPayload = ledgerCreatedAt !== null ? {...payloadRecord, __jobId: jobId.toString()} : payloadRecord;
 		try {
-			const seq = await this.queue.enqueue(taskType, enrichedPayload, {
+			const {seq, duplicate} = await this.queue.enqueue(taskType, enrichedPayload, {
 				...(options?.runAt !== undefined && {runAt: options.runAt}),
 				...(options?.maxAttempts !== undefined && {maxAttempts: options.maxAttempts}),
 				...(options?.priority !== undefined && {priority: options.priority}),
 				...(options?.jobKey !== undefined && {jobKey: options.jobKey}),
 			});
-			if (ledgerWritten) {
+			if (ledgerCreatedAt !== null && duplicate) {
+				await this.ledger
+					.discardJob(jobId, ledgerCreatedAt)
+					.catch((err) => Logger.warn({err, jobId: jobId.toString()}, 'Ledger discardJob failed'));
+			} else if (ledgerCreatedAt !== null) {
 				await this.ledger
 					.setJetStreamSeq(jobId, seq)
 					.catch((err) => Logger.warn({err, jobId: jobId.toString()}, 'Ledger setJetStreamSeq failed'));
 			}
-			Logger.debug({taskType, jobId: jobId.toString(), seq}, 'Job queued successfully');
+			Logger.debug({taskType, jobId: jobId.toString(), seq, duplicate}, 'Job queued successfully');
 			return jobId;
 		} catch (error) {
-			if (ledgerWritten) {
+			if (ledgerCreatedAt !== null) {
 				await this.ledger
 					.markDeadletter(jobId, error instanceof Error ? error.message : String(error))
 					.catch((err) => Logger.warn({err, jobId: jobId.toString()}, 'Ledger markDeadletter failed'));

fluxer_api/src/api/worker/WorkerTaskRegistry.ts

@@ -14,6 +14,7 @@ import bulkDeleteUserMessages from '@app/api/worker/tasks/BulkDeleteUserMessages
 import bulkDeleteUserMessagesScoped from '@app/api/worker/tasks/BulkDeleteUserMessagesScoped';
 import deleteUserMessagesInGuildByTime from '@app/api/worker/tasks/DeleteUserMessagesInGuildByTime';
 import expireAttachments from '@app/api/worker/tasks/ExpireAttachments';
+import expireStaleJobs from '@app/api/worker/tasks/ExpireStaleJobs';
 import extractEmbeds from '@app/api/worker/tasks/ExtractEmbeds';
 import finalizeNcmecAttachmentReport from '@app/api/worker/tasks/FinalizeNcmecAttachmentReport';
 import flushUserActivityBuffer from '@app/api/worker/tasks/FlushUserActivityBuffer';
@@ -59,6 +60,7 @@ export const workerTasks: Record<WorkerTaskName, WorkerTaskHandler> = {
 	bulkUpdateUserFlags: bulkUpdateUserFlags,
 	deleteUserMessagesInGuildByTime,
 	expireAttachments,
+	expireStaleJobs,
 	extractEmbeds,
 	finalizeNcmecAttachmentReport,
 	handleMentions,

fluxer_api/src/api/worker/tasks/ExpireStaleJobs.ts

@@ -0,0 +1,22 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {JOB_STALE_AFTER_MS, JobLedgerRepository} from '@app/api/jobs/JobLedgerRepository';
+import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+
+const PAGE_SIZE = 500;
+const MAX_CLEARED_PER_RUN = 1000;
+
+const expireStaleJobs: WorkerTaskHandler = async (_payload, helpers) => {
+	const result = await new JobLedgerRepository().expireStaleActiveJobs({
+		staleBeforeMs: Date.now() - JOB_STALE_AFTER_MS,
+		pageSize: PAGE_SIZE,
+		maxCleared: MAX_CLEARED_PER_RUN,
+	});
+	if (!result.complete) {
+		helpers.logger.warn({...result}, 'Stale job sweep reached its per-run cap');
+	} else if (result.cleared > 0) {
+		helpers.logger.info({...result}, 'Expired stale jobs');
+	}
+};
+
+export default expireStaleJobs;

fluxer_api/src/api/worker/tasks/PrunePostgresKvTtl.ts

@@ -1,18 +1,34 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 import {Config} from '@app/api/Config';
+import {expireLegacyDefaultTtlRows} from '@app/api/database/PostgresKvDefaultTtlExpiry';
 import {pruneExpiredPostgresKvRows} from '@app/api/database/PostgresKvQueryExecutor';
+import {expireLegacyJobLedgerRows} from '@app/api/jobs/PostgresJobLedgerExpiry';
 import {getDefaultPostgresClient} from '@pkgs/postgres/src/Client';
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+import {ms} from 'itty-time';
 
 const PRUNE_BATCH_SIZE = 5000;
 const MAX_PRUNE_BATCHES_PER_RUN = 20;
+const LEGACY_EXPIRY_BUDGET_MS = ms('2 minutes');
 
 const prunePostgresKvTtl: WorkerTaskHandler = async (_payload, helpers) => {
 	if (Config.database.backend !== 'postgres') {
 		return;
 	}
 	const client = getDefaultPostgresClient();
+	const deadlineMs = Date.now() + LEGACY_EXPIRY_BUDGET_MS;
+	const legacyJobs = await expireLegacyJobLedgerRows(client, deadlineMs);
+	if (legacyJobs !== null && (legacyJobs.deleted > 0 || legacyJobs.expiring > 0 || !legacyJobs.complete)) {
+		helpers.logger.info({...legacyJobs}, 'Expired legacy job ledger rows');
+	}
+	const legacyDefaults = await expireLegacyDefaultTtlRows(client, deadlineMs);
+	if (
+		legacyDefaults !== null &&
+		(legacyDefaults.deleted > 0 || legacyDefaults.expiring > 0 || !legacyDefaults.complete)
+	) {
+		helpers.logger.info({...legacyDefaults}, 'Expired rows written without their table default TTL');
+	}
 	let deleted = 0;
 	for (let batch = 0; batch < MAX_PRUNE_BATCHES_PER_RUN; batch += 1) {
 		const batchDeleted = await pruneExpiredPostgresKvRows(client, PRUNE_BATCH_SIZE);

fluxer_api/src/api/worker/tasks/SyncDisposableEmailDomains.ts

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 import {domainToASCII} from 'node:url';
+import {Config} from '@app/api/Config';
 import {isAccountPolicyContactDomainReputationExempt} from '@app/api/risk/AccountPolicyService';
 import {EXTERNAL_RESPONSE_LIMITS} from '@app/api/utils/ExternalResponseLimits';
 import * as FetchUtils from '@app/api/utils/FetchUtils';
@@ -131,10 +132,7 @@ async function throwIfCancelled(helpers: WorkerTaskHelpers): Promise<void> {
 	}
 }
 
-const syncDisposableEmailDomains: WorkerTaskHandler = async (_payload, helpers) => {
-	helpers.logger.info('Starting disposable email domain sync');
-	await helpers.setContextLink('/suspicious-email-domains');
-	const {adminRepository} = getWorkerDependencies();
+async function fetchFeedDomains(helpers: WorkerTaskHelpers): Promise<Set<string>> {
 	const freshSet = new Set<string>();
 	const perSourceCounts: Record<string, number> = {};
 	const perSourceRawCounts: Record<string, number> = {};
@@ -164,6 +162,14 @@ const syncDisposableEmailDomains: WorkerTaskHandler = async (_payload, helpers)
 		},
 		'Fetched disposable email domains from all sources',
 	);
+	return freshSet;
+}
+
+const syncDisposableEmailDomains: WorkerTaskHandler = async (_payload, helpers) => {
+	helpers.logger.info('Starting disposable email domain sync');
+	await helpers.setContextLink('/suspicious-email-domains');
+	const {adminRepository} = getWorkerDependencies();
+	const freshSet = Config.blocklistFeeds.enabled ? await fetchFeedDomains(helpers) : new Set<string>();
 	const currentSet = await loadCurrentDisposableEmailDomains();
 	let addCount = 0;
 	for (const domain of freshSet) {

fluxer_api/src/api/worker/tasks/SyncFileShaBlocklists.ts

@@ -1,16 +1,39 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-import {BANNED_FILE_SHAS_REFRESH_CHANNEL} from '@app/api/constants/ContentModeration';
+import {Config} from '@app/api/Config';
+import {
+	BANNED_FILE_SHAS_REFRESH_CHANNEL,
+	ContentBlocklistCategory,
+	isBlocklistFeedFileSha,
+} from '@app/api/constants/ContentModeration';
 import {EXTERNAL_RESPONSE_LIMITS} from '@app/api/utils/ExternalResponseLimits';
 import * as FetchUtils from '@app/api/utils/FetchUtils';
 import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
-import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+import type {WorkerTaskHandler, WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
 
 const MALWARE_BAZAAR_SHA256_URL = 'https://bazaar.abuse.ch/export/txt/sha256/recent/';
 const SHA256_RE = /^[0-9a-fA-F]{64}$/;
+
+async function removeFeedFileShas(helpers: WorkerTaskHelpers): Promise<void> {
+	const {adminRepository, kvClient} = getWorkerDependencies();
+	let removed = 0;
+	for (const row of await adminRepository.loadAllBannedFileShas()) {
+		if (!isBlocklistFeedFileSha(row)) continue;
+		if (await adminRepository.unbanFeedFileSha(row.sha256_hex)) removed++;
+	}
+	if (removed > 0) {
+		await kvClient.publish(BANNED_FILE_SHAS_REFRESH_CHANNEL, 'refresh');
+	}
+	helpers.logger.info({removed}, 'Removed file-SHA blocklist feed rows');
+}
+
 const syncFileShaBlocklists: WorkerTaskHandler = async (_payload, helpers) => {
 	helpers.logger.info('Starting file-SHA blocklist sync');
 	await helpers.setContextLink('/file-sha-bans');
+	if (!Config.blocklistFeeds.enabled) {
+		await removeFeedFileShas(helpers);
+		return;
+	}
 	const {adminRepository, kvClient} = getWorkerDependencies();
 	let added = 0;
 	try {
@@ -38,7 +61,7 @@ const syncFileShaBlocklists: WorkerTaskHandler = async (_payload, helpers) => {
 			if (existingSet.has(sha)) continue;
 			await adminRepository.banFileSha({
 				sha256_hex: sha,
-				category: 'malware_bazaar',
+				category: ContentBlocklistCategory.MALWARE_BAZAAR,
 				severity: 2,
 				content_type: null,
 				source_url: MALWARE_BAZAAR_SHA256_URL,

fluxer_api/src/api/worker/tasks/SyncUrlBlocklists.ts

@@ -1,7 +1,8 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import {Config} from '@app/api/Config';
 import {BANNED_URLS_REFRESH_CHANNEL} from '@app/api/constants/ContentModeration';
-import {RISK_S3_KEYS, writeLinesToS3} from '@app/api/risk/RiskBlocklistS3';
+import {deleteRiskS3Object, RISK_S3_KEYS, writeLinesToS3} from '@app/api/risk/RiskBlocklistS3';
 import {EXTERNAL_RESPONSE_LIMITS} from '@app/api/utils/ExternalResponseLimits';
 import * as FetchUtils from '@app/api/utils/FetchUtils';
 import {canonicalizeUrl} from '@app/api/utils/UrlNormalizer';
@@ -62,10 +63,23 @@ async function fetchFeed(source: FeedSource): Promise<Array<string>> {
 	return source.parse(text);
 }
 
+const MISSING_FEED_FILE_ERRORS = new Set(['NoSuchBucket', 'NoSuchKey', 'NotFound']);
+
 const syncUrlBlocklists: WorkerTaskHandler = async (_payload, helpers) => {
 	helpers.logger.info('Starting URL blocklist sync');
 	await helpers.setContextLink('/url-domain-bans');
 	const {storageService, kvClient} = getWorkerDependencies();
+	if (!Config.blocklistFeeds.enabled) {
+		try {
+			await deleteRiskS3Object(storageService, RISK_S3_KEYS.feedUrls);
+		} catch (error) {
+			if (!(error instanceof Error && MISSING_FEED_FILE_ERRORS.has(error.name))) {
+				helpers.logger.warn({error}, 'Failed to delete the URL blocklist feed file');
+			}
+		}
+		await kvClient.publish(BANNED_URLS_REFRESH_CHANNEL, 'refresh');
+		return;
+	}
 	const results = await Promise.allSettled(
 		FEED_SOURCES.map(async (source) => ({source, rawUrls: await fetchFeed(source)})),
 	);

fluxer_api/src/api/worker/tests/BlocklistFeedStartup.test.ts

@@ -0,0 +1,93 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {MockKVProvider} from '@app/api/test/mocks/MockKVProvider';
+import {queueBlocklistFeedStartupJobs} from '@app/api/worker/BlocklistFeedStartup';
+import type {WorkerTaskName} from '@app/api/worker/WorkerLaneConfig';
+import {WorkerQueueOverflowError} from '@app/api/worker/WorkerQueueOverflowError';
+import type {WorkerJobPayload} from '@pkgs/worker/src/contracts/WorkerTypes';
+import {describe, expect, it, vi} from 'vitest';
+
+const INITIAL_SYNC_KEY = 'sync:email_domains:initialized';
+const FEED_TASKS = ['syncDisposableEmailDomains', 'syncUrlBlocklists', 'syncFileShaBlocklists'];
+
+function createWorkerService() {
+	return {addJob: vi.fn(async (_task: WorkerTaskName, _payload: WorkerJobPayload) => 1n)};
+}
+
+function queuedTasks(workerService: ReturnType<typeof createWorkerService>): Array<string> {
+	return workerService.addJob.mock.calls.map(([task]) => task);
+}
+
+describe('queueBlocklistFeedStartupJobs', () => {
+	it('with feeds on, a fresh start queues only the disposable sync and claims it for six hours', async () => {
+		const kv = new MockKVProvider();
+		const workerService = createWorkerService();
+
+		await queueBlocklistFeedStartupJobs(kv, workerService, true);
+
+		expect(workerService.addJob.mock.calls).toEqual([['syncDisposableEmailDomains', {}]]);
+		expect(kv.setnxSpy.mock.calls).toEqual([[INITIAL_SYNC_KEY, '1', 21600]]);
+		expect(kv.delSpy).not.toHaveBeenCalled();
+	});
+
+	it('with feeds on, a start inside the claim queues nothing', async () => {
+		const kv = new MockKVProvider();
+		const workerService = createWorkerService();
+
+		await queueBlocklistFeedStartupJobs(kv, workerService, true);
+		await queueBlocklistFeedStartupJobs(kv, workerService, true);
+
+		expect(queuedTasks(workerService)).toEqual(['syncDisposableEmailDomains']);
+	});
+
+	it('with feeds off, the first start queues all three feed tasks and a second start queues none', async () => {
+		const kv = new MockKVProvider();
+		const workerService = createWorkerService();
+
+		await queueBlocklistFeedStartupJobs(kv, workerService, false);
+		expect(workerService.addJob.mock.calls).toEqual(FEED_TASKS.map((task) => [task, {}]));
+
+		await queueBlocklistFeedStartupJobs(kv, workerService, false);
+		expect(queuedTasks(workerService)).toEqual(FEED_TASKS);
+	});
+
+	it('with feeds off, a start after a feeds-on start cleans up again inside the claim', async () => {
+		const kv = new MockKVProvider();
+		const workerService = createWorkerService();
+
+		await queueBlocklistFeedStartupJobs(kv, workerService, false);
+		await queueBlocklistFeedStartupJobs(kv, workerService, true);
+		await queueBlocklistFeedStartupJobs(kv, workerService, false);
+
+		expect(queuedTasks(workerService)).toEqual([...FEED_TASKS, 'syncDisposableEmailDomains', ...FEED_TASKS]);
+		expect(await kv.exists(INITIAL_SYNC_KEY)).toBe(0);
+	});
+
+	it('a legacy initial sync key without expiry is cleared, so re-enabling runs the initial sync', async () => {
+		const kv = new MockKVProvider();
+		await kv.set(INITIAL_SYNC_KEY, '1');
+		expect(await kv.ttl(INITIAL_SYNC_KEY)).toBe(-1);
+
+		await queueBlocklistFeedStartupJobs(kv, createWorkerService(), false);
+		expect(await kv.exists(INITIAL_SYNC_KEY)).toBe(0);
+
+		const workerService = createWorkerService();
+		await queueBlocklistFeedStartupJobs(kv, workerService, true);
+		expect(queuedTasks(workerService)).toEqual(['syncDisposableEmailDomains']);
+	});
+
+	it('a full jobs stream drops the job without failing startup', async () => {
+		const overflowing = createWorkerService();
+		overflowing.addJob.mockImplementation(async (task) => {
+			throw new WorkerQueueOverflowError(task, 'maximum messages exceeded');
+		});
+		await expect(queueBlocklistFeedStartupJobs(new MockKVProvider(), overflowing, false)).resolves.toBeUndefined();
+		expect(queuedTasks(overflowing)).toEqual(FEED_TASKS);
+
+		const failing = createWorkerService();
+		failing.addJob.mockRejectedValue(new Error('jetstream unavailable'));
+		await expect(queueBlocklistFeedStartupJobs(new MockKVProvider(), failing, false)).rejects.toThrow(
+			'jetstream unavailable',
+		);
+	});
+});

fluxer_api/src/api/worker/tests/BlocklistFeedsDisabled.test.ts

@@ -0,0 +1,228 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {Config} from '@app/api/Config';
+import {
+	BANNED_FILE_SHAS_REFRESH_CHANNEL,
+	BANNED_URLS_REFRESH_CHANNEL,
+	ContentBlocklistCategory,
+} from '@app/api/constants/ContentModeration';
+import type {BannedFileShaRow} from '@app/api/database/types/AdminArchiveTypes';
+import {fileShaCache} from '@app/api/middleware/FileShaCache';
+import {getAdminRepository} from '@app/api/middleware/ServiceSingletons';
+import {urlBlocklistCache} from '@app/api/middleware/UrlBlocklistCache';
+import {type ApiTestHarness, createApiTestHarness} from '@app/api/test/ApiTestHarness';
+import type {MockKVProvider} from '@app/api/test/mocks/MockKVProvider';
+import {NoopLogger} from '@app/api/test/mocks/NoopLogger';
+import {canonicalizeUrl} from '@app/api/utils/UrlNormalizer';
+import syncDisposableEmailDomains from '@app/api/worker/tasks/SyncDisposableEmailDomains';
+import syncFileShaBlocklists from '@app/api/worker/tasks/SyncFileShaBlocklists';
+import syncUrlBlocklists from '@app/api/worker/tasks/SyncUrlBlocklists';
+import {clearWorkerDependencies, setWorkerDependenciesForTest} from '@app/api/worker/WorkerContext';
+import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
+import {afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, type MockInstance, vi} from 'vitest';
+
+const FEED_BUCKET = 'fluxer-geoip';
+const FEED_KEY = 'blocklists/feed-urls.txt';
+const FEED_SHA = 'a1'.repeat(32);
+const ADMIN_BAZAAR_SHA = 'b2'.repeat(32);
+const ADMIN_MANUAL_SHA = 'c3'.repeat(32);
+const UNOWNED_NCMEC_SHA = 'd4'.repeat(32);
+const ADMIN_USER_ID = 42n;
+
+function createHelpers(overrides: Partial<WorkerTaskHelpers> = {}): WorkerTaskHelpers {
+	return {
+		logger: new NoopLogger(),
+		jobId: 4242n,
+		addJob: async () => 0n,
+		reportProgress: async () => {},
+		shouldCancel: async () => false,
+		setContextLink: async () => {},
+		...overrides,
+	};
+}
+
+function fileShaRow(sha256Hex: string, category: string, addedBy: bigint | null): BannedFileShaRow {
+	return {
+		sha256_hex: sha256Hex,
+		category,
+		severity: 2,
+		content_type: null,
+		source_url: null,
+		added_at: new Date(),
+		added_by: addedBy,
+		notes: null,
+	};
+}
+
+async function seedFileShas(): Promise<void> {
+	const adminRepository = getAdminRepository();
+	await adminRepository.banFileSha(fileShaRow(FEED_SHA, ContentBlocklistCategory.MALWARE_BAZAAR, null));
+	await adminRepository.banFileSha(
+		fileShaRow(ADMIN_BAZAAR_SHA, ContentBlocklistCategory.MALWARE_BAZAAR, ADMIN_USER_ID),
+	);
+	await adminRepository.banFileSha(fileShaRow(ADMIN_MANUAL_SHA, ContentBlocklistCategory.MANUAL, ADMIN_USER_ID));
+	await adminRepository.banFileSha(fileShaRow(UNOWNED_NCMEC_SHA, ContentBlocklistCategory.NCMEC, null));
+}
+
+describe('blocklist feeds turned off', () => {
+	let harness: ApiTestHarness;
+	let previousFeedsEnabled: boolean;
+	let fetchSpy: MockInstance<typeof fetch>;
+
+	beforeAll(async () => {
+		harness = await createApiTestHarness();
+	});
+
+	beforeEach(async () => {
+		previousFeedsEnabled = Config.blocklistFeeds.enabled;
+		await harness.reset();
+		harness.storageService.reset();
+		setWorkerDependenciesForTest({
+			adminRepository: getAdminRepository(),
+			kvClient: harness.kvProvider,
+			storageService: harness.storageService,
+		});
+		fetchSpy = vi.spyOn(globalThis, 'fetch');
+	});
+
+	afterEach(() => {
+		Config.blocklistFeeds.enabled = previousFeedsEnabled;
+		fetchSpy.mockRestore();
+	});
+
+	afterAll(async () => {
+		clearWorkerDependencies();
+		await harness?.shutdown();
+	});
+
+	function publishCalls(): Array<Array<string>> {
+		return (harness.kvProvider as MockKVProvider).publishSpy.mock.calls;
+	}
+
+	it('removes every stored disposable domain without fetching a feed', async () => {
+		const adminRepository = getAdminRepository();
+		for (const domain of ['mailinator.com', 'guerrillamail.com', 'tempmail.dev']) {
+			await adminRepository.addDisposableEmailDomain(domain);
+		}
+		Config.blocklistFeeds.enabled = false;
+		const reportProgress = vi.fn(async () => {});
+
+		await syncDisposableEmailDomains({}, createHelpers({reportProgress}));
+
+		expect(await adminRepository.listDisposableEmailDomains()).toEqual([]);
+		expect(fetchSpy).not.toHaveBeenCalled();
+		expect(reportProgress).toHaveBeenLastCalledWith(3, 3, '+0 added, -3 removed');
+	});
+
+	it('reports a stored domain as disposable only while feeds are on', async () => {
+		const adminRepository = getAdminRepository();
+		await adminRepository.addDisposableEmailDomain('mailinator.com');
+
+		Config.blocklistFeeds.enabled = true;
+		expect(await adminRepository.isEmailDomainDisposable('mailinator.com')).toBe(true);
+		Config.blocklistFeeds.enabled = false;
+		expect(await adminRepository.isEmailDomainDisposable('mailinator.com')).toBe(false);
+	});
+
+	it('removes MalwareBazaar feed rows and keeps every other file-SHA ban', async () => {
+		await seedFileShas();
+		Config.blocklistFeeds.enabled = false;
+
+		await syncFileShaBlocklists({}, createHelpers());
+
+		const remaining = (await getAdminRepository().loadAllBannedFileShas()).map((row) => row.sha256_hex).sort();
+		expect(remaining).toEqual([ADMIN_BAZAAR_SHA, ADMIN_MANUAL_SHA, UNOWNED_NCMEC_SHA]);
+		expect(publishCalls()).toEqual([[BANNED_FILE_SHAS_REFRESH_CHANNEL, 'refresh']]);
+		expect(fetchSpy).not.toHaveBeenCalled();
+	});
+
+	it('keeps a file-SHA ban an Admin took over after the purge read the table', async () => {
+		const adminRepository = getAdminRepository();
+		await adminRepository.banFileSha(fileShaRow(FEED_SHA, ContentBlocklistCategory.MALWARE_BAZAAR, null));
+		const staleRows = await adminRepository.loadAllBannedFileShas();
+		await adminRepository.banFileSha(fileShaRow(FEED_SHA, ContentBlocklistCategory.MALWARE_BAZAAR, ADMIN_USER_ID));
+		const staleRead = vi.spyOn(adminRepository, 'loadAllBannedFileShas').mockResolvedValueOnce(staleRows);
+		Config.blocklistFeeds.enabled = false;
+
+		await syncFileShaBlocklists({}, createHelpers());
+
+		staleRead.mockRestore();
+		expect(await adminRepository.isFileShaBanned(FEED_SHA)).toBe(true);
+	});
+
+	it('file-SHA cache skips feed rows only while feeds are off', async () => {
+		await seedFileShas();
+
+		Config.blocklistFeeds.enabled = false;
+		await fileShaCache.refresh();
+		expect(fileShaCache.isBanned(FEED_SHA)).toBe(false);
+		expect(fileShaCache.isBanned(ADMIN_BAZAAR_SHA)).toBe(true);
+		expect(fileShaCache.isBanned(ADMIN_MANUAL_SHA)).toBe(true);
+		expect(fileShaCache.isBanned(UNOWNED_NCMEC_SHA)).toBe(true);
+
+		Config.blocklistFeeds.enabled = true;
+		await fileShaCache.refresh();
+		expect(fileShaCache.isBanned(FEED_SHA)).toBe(true);
+	});
+
+	it('deletes the URL feed file and never fetches while feeds are off', async () => {
+		await harness.storageService.uploadObject({
+			bucket: FEED_BUCKET,
+			key: FEED_KEY,
+			body: Buffer.from('https://feed.example/x\n'),
+		});
+		Config.blocklistFeeds.enabled = false;
+
+		await syncUrlBlocklists({}, createHelpers());
+
+		expect(harness.storageService.deleteObjectSpy).toHaveBeenCalledWith(FEED_BUCKET, FEED_KEY);
+		expect(harness.storageService.hasObject(FEED_BUCKET, FEED_KEY)).toBe(false);
+		expect(publishCalls()).toEqual([[BANNED_URLS_REFRESH_CHANNEL, 'refresh']]);
+		expect(fetchSpy).not.toHaveBeenCalled();
+
+		const quietLogger = new NoopLogger();
+		const quietWarn = vi.spyOn(quietLogger, 'warn');
+		vi.spyOn(harness.storageService, 'deleteObject').mockRejectedValueOnce(
+			Object.assign(new Error('The specified bucket does not exist'), {name: 'NoSuchBucket'}),
+		);
+		await expect(syncUrlBlocklists({}, createHelpers({logger: quietLogger}))).resolves.toBeUndefined();
+		expect(quietWarn).not.toHaveBeenCalled();
+
+		const loudLogger = new NoopLogger();
+		const loudWarn = vi.spyOn(loudLogger, 'warn');
+		harness.storageService.configure({shouldFailDelete: true});
+		await expect(syncUrlBlocklists({}, createHelpers({logger: loudLogger}))).resolves.toBeUndefined();
+		expect(loudWarn).toHaveBeenCalledTimes(1);
+	});
+
+	it('URL cache ignores the feed file only while feeds are off', async () => {
+		const feedUrl = canonicalizeUrl('https://feed.example/x');
+		const adminUrl = canonicalizeUrl('https://admin.example/y');
+		expect(feedUrl).not.toBeNull();
+		expect(adminUrl).not.toBeNull();
+		await harness.storageService.uploadObject({
+			bucket: FEED_BUCKET,
+			key: FEED_KEY,
+			body: Buffer.from(`${feedUrl}\n`),
+		});
+		await getAdminRepository().banUrl({
+			url_canonical: adminUrl!,
+			category: ContentBlocklistCategory.MANUAL,
+			severity: 2,
+			source_url: null,
+			added_at: new Date(),
+			added_by: ADMIN_USER_ID,
+			notes: null,
+		});
+		urlBlocklistCache.setStorageService(harness.storageService);
+
+		Config.blocklistFeeds.enabled = false;
+		await urlBlocklistCache.refresh();
+		expect(urlBlocklistCache.isUrlBanned('https://feed.example/x')).toBe(false);
+		expect(urlBlocklistCache.isUrlBanned('https://admin.example/y')).toBe(true);
+
+		Config.blocklistFeeds.enabled = true;
+		await urlBlocklistCache.refresh();
+		expect(urlBlocklistCache.isUrlBanned('https://feed.example/x')).toBe(true);
+	});
+});

fluxer_api/src/api/worker/tests/BlocklistFeedsDisabledPostgres.test.ts

@@ -0,0 +1,206 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {spawnSync} from 'node:child_process';
+import {createServer} from 'node:net';
+import {AdminRepository} from '@app/api/admin/AdminRepository';
+import {Config} from '@app/api/Config';
+import {ContentBlocklistCategory} from '@app/api/constants/ContentModeration';
+import {setCassandraQueryExecutorForTesting} from '@app/api/database/CassandraQueryExecution';
+import {ensurePostgresKvSchema, PostgresKvQueryExecutor} from '@app/api/database/PostgresKvQueryExecutor';
+import type {BannedFileShaRow} from '@app/api/database/types/AdminArchiveTypes';
+import {startDockerContainer} from '@app/api/test/DockerTestContainer';
+import {InMemoryCassandraQueryExecutor} from '@app/api/test/InMemoryCassandraQueryExecutor';
+import {MockKVProvider} from '@app/api/test/mocks/MockKVProvider';
+import {MockStorageService} from '@app/api/test/mocks/MockStorageService';
+import {NoopLogger} from '@app/api/test/mocks/NoopLogger';
+import syncDisposableEmailDomains from '@app/api/worker/tasks/SyncDisposableEmailDomains';
+import syncFileShaBlocklists from '@app/api/worker/tasks/SyncFileShaBlocklists';
+import {
+	clearWorkerDependencies,
+	getWorkerDependencies,
+	setWorkerDependenciesForTest,
+} from '@app/api/worker/WorkerContext';
+import {
+	getDefaultPostgresClient,
+	type IPostgresClient,
+	initPostgres,
+	shutdownPostgres,
+} from '@pkgs/postgres/src/Client';
+import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
+import {afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi} from 'vitest';
+
+const KV_TABLE = 'kv_blocklist_feeds_disabled';
+const CONTAINER = `fluxer-feeds-${process.pid.toString(36)}-${Date.now().toString(36)}`;
+const dockerAvailable = spawnSync('docker', ['version'], {stdio: 'ignore'}).status === 0;
+const FEED_SHA = 'a1'.repeat(32);
+const ADMIN_BAZAAR_SHA = 'b2'.repeat(32);
+const ADMIN_MANUAL_SHA = 'c3'.repeat(32);
+const ADMIN_USER_ID = 1_234_567_890_123n;
+
+async function sleep(ms: number): Promise<void> {
+	await new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+async function freePort(): Promise<number> {
+	return new Promise((resolve, reject) => {
+		const server = createServer();
+		server.on('error', reject);
+		server.listen(0, '127.0.0.1', () => {
+			const address = server.address();
+			if (typeof address === 'string' || address === null) {
+				reject(new Error('no port'));
+				return;
+			}
+			const port = address.port;
+			server.close(() => resolve(port));
+		});
+	});
+}
+
+function createHelpers(): WorkerTaskHelpers {
+	return {
+		logger: new NoopLogger(),
+		jobId: 4242n,
+		addJob: async () => 0n,
+		reportProgress: async () => {},
+		shouldCancel: async () => false,
+		setContextLink: async () => {},
+	};
+}
+
+function fileShaRow(sha256Hex: string, category: string, addedBy: bigint | null): BannedFileShaRow {
+	return {
+		sha256_hex: sha256Hex,
+		category,
+		severity: 2,
+		content_type: null,
+		source_url: null,
+		added_at: new Date(),
+		added_by: addedBy,
+		notes: null,
+	};
+}
+
+describe.skipIf(!dockerAvailable)('blocklist feeds turned off against postgres', () => {
+	let raw: IPostgresClient;
+	let executor: PostgresKvQueryExecutor;
+	let previousFeedsEnabled: boolean;
+
+	beforeAll(async () => {
+		const port = await freePort();
+		startDockerContainer([
+			'run',
+			'-d',
+			'--name',
+			CONTAINER,
+			'-e',
+			'POSTGRES_USER=fluxer',
+			'-e',
+			'POSTGRES_PASSWORD=fluxer',
+			'-e',
+			'POSTGRES_DB=fluxer',
+			'-p',
+			`127.0.0.1:${port}:5432`,
+			'postgres:16-alpine',
+			'-c',
+			'fsync=off',
+		]);
+		let ready = false;
+		for (let attempt = 0; attempt < 180 && !ready; attempt += 1) {
+			await sleep(500);
+			const probe = spawnSync('docker', ['exec', CONTAINER, 'pg_isready', '-U', 'fluxer', '-d', 'fluxer'], {
+				stdio: 'ignore',
+			});
+			if (probe.status !== 0) continue;
+			try {
+				await initPostgres({
+					url: `postgres://fluxer:[email protected]:${port}/fluxer`,
+					maxConnections: 4,
+					kvTable: KV_TABLE,
+				});
+				await getDefaultPostgresClient().query('SELECT 1');
+				ready = true;
+			} catch {
+				await shutdownPostgres().catch(() => {});
+			}
+		}
+		if (!ready) throw new Error('postgres never came up');
+		raw = getDefaultPostgresClient();
+		await ensurePostgresKvSchema(raw);
+		executor = new PostgresKvQueryExecutor(raw);
+	}, 900_000);
+
+	beforeEach(async () => {
+		previousFeedsEnabled = Config.blocklistFeeds.enabled;
+		await raw.query(`DELETE FROM ${KV_TABLE}`);
+		setCassandraQueryExecutorForTesting(executor);
+		setWorkerDependenciesForTest({
+			adminRepository: new AdminRepository(),
+			kvClient: new MockKVProvider(),
+			storageService: new MockStorageService(),
+		});
+		Config.blocklistFeeds.enabled = false;
+	});
+
+	afterEach(() => {
+		Config.blocklistFeeds.enabled = previousFeedsEnabled;
+	});
+
+	afterAll(async () => {
+		clearWorkerDependencies();
+		setCassandraQueryExecutorForTesting(new InMemoryCassandraQueryExecutor());
+		await shutdownPostgres().catch(() => {});
+		spawnSync('docker', ['rm', '-f', CONTAINER], {stdio: 'ignore'});
+	});
+
+	it('removes every disposable_email_domains row from the KV table', async () => {
+		const repository = new AdminRepository();
+		for (let i = 0; i < 1200; i++) {
+			await repository.addDisposableEmailDomain(`disposable-${i}.example`);
+		}
+		const seeded = await raw.query<{count: number}>(
+			`SELECT count(*)::int AS count FROM ${KV_TABLE} WHERE table_name = 'disposable_email_domains'`,
+		);
+		expect(seeded.rows[0]?.count).toBe(1200);
+
+		await syncDisposableEmailDomains({}, createHelpers());
+
+		const remaining = await raw.query<{count: number}>(
+			`SELECT count(*)::int AS count FROM ${KV_TABLE} WHERE table_name = 'disposable_email_domains'`,
+		);
+		expect(remaining.rows[0]?.count).toBe(0);
+	});
+
+	it('keeps file-SHA rows with added_by set and removes feed rows stored with a JSON null added_by', async () => {
+		const repository = new AdminRepository();
+		await repository.banFileSha(fileShaRow(FEED_SHA, ContentBlocklistCategory.MALWARE_BAZAAR, null));
+		await repository.banFileSha(fileShaRow(ADMIN_BAZAAR_SHA, ContentBlocklistCategory.MALWARE_BAZAAR, ADMIN_USER_ID));
+		await repository.banFileSha(fileShaRow(ADMIN_MANUAL_SHA, ContentBlocklistCategory.MANUAL, ADMIN_USER_ID));
+		const feedRow = await raw.query<{added_by_type: string}>(
+			`SELECT jsonb_typeof(row_data->'added_by') AS added_by_type FROM ${KV_TABLE}
+WHERE table_name = 'banned_file_shas' AND row_data->>'sha256_hex' = $1`,
+			[FEED_SHA],
+		);
+		expect(feedRow.rows).toEqual([{added_by_type: 'null'}]);
+
+		await syncFileShaBlocklists({}, createHelpers());
+
+		const remaining = await raw.query<{sha: string}>(
+			`SELECT row_data->>'sha256_hex' AS sha FROM ${KV_TABLE} WHERE table_name = 'banned_file_shas' ORDER BY 1`,
+		);
+		expect(remaining.rows.map((row) => row.sha)).toEqual([ADMIN_BAZAAR_SHA, ADMIN_MANUAL_SHA]);
+	});
+
+	it('keeps a file-SHA ban an Admin took over after the purge read the table', async () => {
+		const {adminRepository} = getWorkerDependencies();
+		await adminRepository.banFileSha(fileShaRow(FEED_SHA, ContentBlocklistCategory.MALWARE_BAZAAR, null));
+		const staleRows = await adminRepository.loadAllBannedFileShas();
+		await adminRepository.banFileSha(fileShaRow(FEED_SHA, ContentBlocklistCategory.MALWARE_BAZAAR, ADMIN_USER_ID));
+		const staleRead = vi.spyOn(adminRepository, 'loadAllBannedFileShas').mockResolvedValueOnce(staleRows);
+
+		await syncFileShaBlocklists({}, createHelpers());
+
+		staleRead.mockRestore();
+		expect(await adminRepository.isFileShaBanned(FEED_SHA)).toBe(true);
+	});
+});

fluxer_api/src/api/worker/tests/JetStreamWorkerQueue.test.ts

@@ -94,7 +94,7 @@ function createQueue(params: {
 	dlqExists?: boolean;
 	reject?: (config: Partial<StreamConfig>) => Error | null;
 	updateError?: Error;
-	publish?: (subject: string) => {seq: number};
+	publish?: (subject: string, body: string, options: {msgID: string}) => {seq: number; duplicate?: boolean};
 	subjectCounts?: Record<string, number>;
 	subjectCountsError?: Error;
 }): {
@@ -171,9 +171,9 @@ function createQueue(params: {
 				},
 			}),
 		getJetStreamClient: () => ({
-			publish: (subject: string) => {
+			publish: (subject: string, body: string, options: {msgID: string}) => {
 				const publish = params.publish ?? (() => ({seq: 1}));
-				return Promise.resolve(publish(subject));
+				return Promise.resolve(publish(subject, body, options));
 			},
 		}),
 	} as unknown as JetStreamConnectionManager;
@@ -281,6 +281,27 @@ describe('jobs stream limits', () => {
 	});
 });
 
+describe('jobs stream max age', () => {
+	it('reports the max age of a jobs stream it creates', async () => {
+		const {queue, added} = createQueue({existing: null});
+		await queue.ensureStream();
+		expect(queue.getJobsStreamMaxAgeMs()).toBe(7 * 24 * 60 * 60 * 1000);
+		expect(added[0]?.max_age).toBe(queue.getJobsStreamMaxAgeMs() * 1_000_000);
+	});
+
+	it('reports the max age an existing jobs stream really has and never changes it', async () => {
+		for (const [maxAgeNanos, expectedMs] of [
+			[30 * 24 * 60 * 60 * 1_000_000_000, 30 * 24 * 60 * 60 * 1000],
+			[0, 0],
+		] as const) {
+			const {queue, updated} = createQueue({existing: {...LEGACY_CONFIG, max_age: maxAgeNanos} as StreamConfig});
+			await queue.ensureStream();
+			expect(queue.getJobsStreamMaxAgeMs()).toBe(expectedMs);
+			expect(updated.some((config) => 'max_age' in config)).toBe(false);
+		}
+	});
+});
+
 describe('dead-letter stream', () => {
 	it('keeps startup alive when the dead-letter stream does not fit', async () => {
 		const {queue, dlqAdded} = createQueue({dlqExists: false, reject: () => noStorageError()});
@@ -298,16 +319,16 @@ describe('dead-letter stream', () => {
 describe('jobs stream enqueue shedding', () => {
 	it('rejects enqueues once the stream is at its cap', async () => {
 		const {queue} = createQueue({existing: LEGACY_CONFIG, publish: boundedPublisher(2)});
-		await expect(queue.enqueue('extractEmbeds', {})).resolves.toBe('1');
-		await expect(queue.enqueue('extractEmbeds', {})).resolves.toBe('2');
+		await expect(queue.enqueue('extractEmbeds', {})).resolves.toEqual({seq: '1', duplicate: false});
+		await expect(queue.enqueue('extractEmbeds', {})).resolves.toEqual({seq: '2', duplicate: false});
 		await expect(queue.enqueue('extractEmbeds', {})).rejects.toBeInstanceOf(WorkerQueueOverflowError);
 	});
 
 	it('caps each task type independently', async () => {
 		const {queue} = createQueue({existing: LEGACY_CONFIG, publish: boundedPublisher(1)});
-		await expect(queue.enqueue('extractEmbeds', {})).resolves.toBe('1');
+		await expect(queue.enqueue('extractEmbeds', {})).resolves.toEqual({seq: '1', duplicate: false});
 		await expect(queue.enqueue('extractEmbeds', {})).rejects.toBeInstanceOf(WorkerQueueOverflowError);
-		await expect(queue.enqueue('handleMentions', {})).resolves.toBe('2');
+		await expect(queue.enqueue('handleMentions', {})).resolves.toEqual({seq: '2', duplicate: false});
 	});
 
 	it('sheds enqueues the server refuses for lack of resources', async () => {
@@ -320,6 +341,29 @@ describe('jobs stream enqueue shedding', () => {
 		await expect(queue.enqueue('handleMentions', {})).rejects.toBeInstanceOf(WorkerQueueOverflowError);
 	});
 
+	it('reports a publish the stream deduplicated under the same job key', async () => {
+		const seen = new Map<string, number>();
+		const {queue} = createQueue({
+			existing: LEGACY_CONFIG,
+			publish: (_subject, _body, options) => {
+				const existing = seen.get(options.msgID);
+				if (existing !== undefined) return {seq: existing, duplicate: true};
+				seen.set(options.msgID, seen.size + 1);
+				return {seq: seen.size, duplicate: false};
+			},
+		});
+		const options = {jobKey: 'batch-audit-log-message-deletes:1'};
+		await expect(queue.enqueue('batchGuildAuditLogMessageDeletes', {}, options)).resolves.toEqual({
+			seq: '1',
+			duplicate: false,
+		});
+		await expect(queue.enqueue('batchGuildAuditLogMessageDeletes', {}, options)).resolves.toEqual({
+			seq: '1',
+			duplicate: true,
+		});
+		await expect(queue.enqueue('batchGuildAuditLogMessageDeletes', {})).resolves.toEqual({seq: '2', duplicate: false});
+	});
+
 	it('rethrows publish failures that are not stream limits', async () => {
 		const failure = new Error('no responders');
 		const {queue} = createQueue({

fluxer_api/src/api/worker/tests/WorkerServiceLedger.test.ts

@@ -7,6 +7,7 @@ import {WorkerService} from '@app/api/worker/WorkerService';
 import {describe, expect, test} from 'vitest';
 
 const JOB_ID = 4242n;
+const CREATED_AT = new Date('2026-09-21T12:00:00.000Z');
 
 function createSnowflake(): ISnowflakeService {
 	return {
@@ -14,17 +15,24 @@ function createSnowflake(): ISnowflakeService {
 	} as unknown as ISnowflakeService;
 }
 
-function createHarness(options?: {createJobError?: Error; enqueueError?: Error}) {
+function createHarness(options?: {
+	createJobError?: Error;
+	enqueueError?: Error;
+	duplicate?: boolean;
+	discardError?: Error;
+}) {
 	const calls: Array<string> = [];
 	const createdJobs: Array<CreateJobInput> = [];
 	const enqueued: Array<{taskType: string; payload: Record<string, unknown>}> = [];
 	const seqUpdates: Array<{jobId: bigint; seq: string}> = [];
 	const deadletters: Array<{jobId: bigint; errorMessage: string}> = [];
+	const discarded: Array<{jobId: bigint; createdAt: Date}> = [];
 	const ledger = {
 		createJob: async (input: CreateJobInput) => {
 			calls.push('createJob');
 			if (options?.createJobError) throw options.createJobError;
 			createdJobs.push(input);
+			return CREATED_AT;
 		},
 		setJetStreamSeq: async (jobId: bigint, seq: string) => {
 			calls.push('setJetStreamSeq');
@@ -34,17 +42,22 @@ function createHarness(options?: {createJobError?: Error; enqueueError?: Error})
 			calls.push('markDeadletter');
 			deadletters.push({jobId, errorMessage});
 		},
+		discardJob: async (jobId: bigint, createdAt: Date) => {
+			calls.push('discardJob');
+			if (options?.discardError) throw options.discardError;
+			discarded.push({jobId, createdAt});
+		},
 	} as unknown as IJobLedgerRepository;
 	const queue = {
 		enqueue: async (taskType: string, payload: Record<string, unknown>) => {
 			calls.push('enqueue');
 			if (options?.enqueueError) throw options.enqueueError;
 			enqueued.push({taskType, payload});
-			return 'seq-9';
+			return {seq: 'seq-9', duplicate: options?.duplicate === true};
 		},
 	} as unknown as JetStreamWorkerQueue;
 	const service = new WorkerService(queue, createSnowflake(), ledger);
-	return {service, calls, createdJobs, enqueued, seqUpdates, deadletters};
+	return {service, calls, createdJobs, enqueued, seqUpdates, deadletters, discarded};
 }
 
 describe('WorkerService ledger ordering', () => {
@@ -88,6 +101,42 @@ describe('WorkerService ledger ordering', () => {
 		expect(harness.deadletters).toEqual([{jobId: JOB_ID, errorMessage: 'stream unreachable'}]);
 	});
 
+	test('discards the ledger row when the stream already holds a job under the same key', async () => {
+		const harness = createHarness({duplicate: true});
+
+		const jobId = await harness.service.addJob(
+			'batchGuildAuditLogMessageDeletes',
+			{guildId: '1'},
+			{jobKey: 'batch-audit-log-message-deletes:1'},
+		);
+
+		expect(jobId).toBe(JOB_ID);
+		expect(harness.calls).toEqual(['createJob', 'enqueue', 'discardJob']);
+		expect(harness.discarded).toEqual([{jobId: JOB_ID, createdAt: CREATED_AT}]);
+		expect(harness.seqUpdates).toEqual([]);
+	});
+
+	test('still returns the job id when discarding the duplicate ledger row fails', async () => {
+		const harness = createHarness({duplicate: true, discardError: new Error('write timeout')});
+
+		const jobId = await harness.service.addJob(
+			'batchGuildAuditLogMessageDeletes',
+			{guildId: '1'},
+			{jobKey: 'batch-audit-log-message-deletes:1'},
+		);
+
+		expect(jobId).toBe(JOB_ID);
+		expect(harness.calls).toEqual(['createJob', 'enqueue', 'discardJob']);
+	});
+
+	test('does not touch the ledger for a duplicate the caller never ledgered', async () => {
+		const harness = createHarness({duplicate: true});
+
+		await harness.service.addJob('handleMentions', {}, {skipLedger: true, jobKey: 'mentions:1'});
+
+		expect(harness.calls).toEqual(['enqueue']);
+	});
+
 	test('never touches the ledger when the caller skips it', async () => {
 		const harness = createHarness();