fix: various fixes to sentry-reported errors and more

packages/api/src/worker/CronScheduler.tsx

@@ -0,0 +1,162 @@
+/*
+ * Copyright (C) 2026 Fluxer Contributors
+ *
+ * This file is part of Fluxer.
+ *
+ * Fluxer is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Fluxer is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with Fluxer. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+import type {JetStreamWorkerQueue} from '@fluxer/api/src/worker/JetStreamWorkerQueue';
+import type {LoggerInterface} from '@fluxer/logger/src/LoggerInterface';
+import type {WorkerJobPayload} from '@fluxer/worker/src/contracts/WorkerTypes';
+
+interface CronDefinition {
+	id: string;
+	taskType: string;
+	payload: WorkerJobPayload;
+	cronExpression: string;
+	lastFired: number;
+}
+
+function parseCronField(field: string, min: number, max: number): Array<number> {
+	if (field === '*') {
+		return [];
+	}
+
+	const values: Array<number> = [];
+
+	for (const part of field.split(',')) {
+		const stepMatch = part.match(/^(.+)\/(\d+)$/);
+		if (stepMatch) {
+			const [, range, stepStr] = stepMatch;
+			const step = Number.parseInt(stepStr!, 10);
+			let start = min;
+			let end = max;
+			if (range !== '*') {
+				const rangeParts = range!.split('-');
+				start = Number.parseInt(rangeParts[0]!, 10);
+				if (rangeParts.length > 1) {
+					end = Number.parseInt(rangeParts[1]!, 10);
+				}
+			}
+			for (let i = start; i <= end; i += step) {
+				values.push(i);
+			}
+		} else if (part.includes('-')) {
+			const [startStr, endStr] = part.split('-');
+			const start = Number.parseInt(startStr!, 10);
+			const end = Number.parseInt(endStr!, 10);
+			for (let i = start; i <= end; i++) {
+				values.push(i);
+			}
+		} else {
+			values.push(Number.parseInt(part, 10));
+		}
+	}
+
+	return values;
+}
+
+function matchesCronExpression(expression: string, date: Date): boolean {
+	const parts = expression.trim().split(/\s+/);
+	if (parts.length !== 6) {
+		return false;
+	}
+
+	const [secField, minField, hourField, domField, monField, dowField] = parts;
+	const second = date.getSeconds();
+	const minute = date.getMinutes();
+	const hour = date.getHours();
+	const dayOfMonth = date.getDate();
+	const month = date.getMonth() + 1;
+	const dayOfWeek = date.getDay();
+
+	function matches(field: string, value: number, min: number, max: number): boolean {
+		const allowed = parseCronField(field, min, max);
+		return allowed.length === 0 || allowed.includes(value);
+	}
+
+	return (
+		matches(secField!, second, 0, 59) &&
+		matches(minField!, minute, 0, 59) &&
+		matches(hourField!, hour, 0, 23) &&
+		matches(domField!, dayOfMonth, 1, 31) &&
+		matches(monField!, month, 1, 12) &&
+		matches(dowField!, dayOfWeek, 0, 6)
+	);
+}
+
+export class CronScheduler {
+	private readonly queue: JetStreamWorkerQueue;
+	private readonly logger: LoggerInterface;
+	private readonly definitions: Map<string, CronDefinition> = new Map();
+	private intervalId: ReturnType<typeof setInterval> | null = null;
+
+	constructor(queue: JetStreamWorkerQueue, logger: LoggerInterface) {
+		this.queue = queue;
+		this.logger = logger;
+	}
+
+	upsert(id: string, taskType: string, payload: WorkerJobPayload, cronExpression: string): void {
+		this.definitions.set(id, {
+			id,
+			taskType,
+			payload,
+			cronExpression,
+			lastFired: 0,
+		});
+	}
+
+	start(): void {
+		if (this.intervalId !== null) {
+			return;
+		}
+
+		this.intervalId = setInterval(() => {
+			this.tick().catch((error) => {
+				this.logger.error({err: error}, 'Cron scheduler tick failed');
+			});
+		}, 1000);
+
+		this.logger.info(`Cron scheduler started with ${this.definitions.size} definitions`);
+	}
+
+	stop(): void {
+		if (this.intervalId !== null) {
+			clearInterval(this.intervalId);
+			this.intervalId = null;
+		}
+	}
+
+	private async tick(): Promise<void> {
+		const now = new Date();
+		const nowSeconds = Math.floor(now.getTime() / 1000);
+
+		for (const def of this.definitions.values()) {
+			if (def.lastFired === nowSeconds) {
+				continue;
+			}
+
+			if (matchesCronExpression(def.cronExpression, now)) {
+				def.lastFired = nowSeconds;
+				try {
+					await this.queue.enqueue(def.taskType, def.payload);
+					this.logger.debug({cronId: def.id, taskType: def.taskType}, 'Cron job fired');
+				} catch (error) {
+					this.logger.error({err: error, cronId: def.id, taskType: def.taskType}, 'Failed to enqueue cron job');
+				}
+			}
+		}
+	}
+}

packages/api/src/worker/JetStreamWorkerQueue.tsx

@@ -0,0 +1,136 @@
+/*
+ * Copyright (C) 2026 Fluxer Contributors
+ *
+ * This file is part of Fluxer.
+ *
+ * Fluxer is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Fluxer is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with Fluxer. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+import {randomUUID} from 'node:crypto';
+import {addSpanEvent, setSpanAttributes, withSpan} from '@fluxer/api/src/telemetry/Tracing';
+import type {JetStreamConnectionManager} from '@fluxer/nats/src/JetStreamConnectionManager';
+import type {WorkerJobPayload} from '@fluxer/worker/src/contracts/WorkerTypes';
+import {AckPolicy, nanos, RetentionPolicy, StorageType} from 'nats';
+
+const STREAM_NAME = 'JOBS';
+const CONSUMER_NAME = 'workers';
+const SUBJECT_PREFIX = 'jobs.';
+const MAX_AGE_MS = 7 * 24 * 60 * 60 * 1000;
+const MAX_DELIVER = 5;
+const ACK_WAIT_MS = 30_000;
+const MAX_ACK_PENDING = 100;
+
+export class JetStreamWorkerQueue {
+	private readonly connectionManager: JetStreamConnectionManager;
+	private streamReady = false;
+	private consumerReady = false;
+
+	constructor(connectionManager: JetStreamConnectionManager) {
+		this.connectionManager = connectionManager;
+	}
+
+	async ensureStream(): Promise<void> {
+		if (this.streamReady) {
+			return;
+		}
+		const jsm = await this.connectionManager.getJetStreamManager();
+		try {
+			await jsm.streams.info(STREAM_NAME);
+		} catch {
+			await jsm.streams.add({
+				name: STREAM_NAME,
+				subjects: [`${SUBJECT_PREFIX}>`],
+				retention: RetentionPolicy.Workqueue,
+				storage: StorageType.File,
+				max_age: nanos(MAX_AGE_MS),
+				num_replicas: 1,
+			});
+		}
+		this.streamReady = true;
+	}
+
+	async ensureConsumer(): Promise<void> {
+		if (this.consumerReady) {
+			return;
+		}
+		const jsm = await this.connectionManager.getJetStreamManager();
+		try {
+			await jsm.consumers.info(STREAM_NAME, CONSUMER_NAME);
+		} catch {
+			await jsm.consumers.add(STREAM_NAME, {
+				durable_name: CONSUMER_NAME,
+				ack_policy: AckPolicy.Explicit,
+				max_deliver: MAX_DELIVER,
+				ack_wait: nanos(ACK_WAIT_MS),
+				max_ack_pending: MAX_ACK_PENDING,
+			});
+		}
+		this.consumerReady = true;
+	}
+
+	async ensureInfrastructure(): Promise<void> {
+		await this.ensureStream();
+		await this.ensureConsumer();
+	}
+
+	async enqueue(
+		taskType: string,
+		payload: WorkerJobPayload,
+		options?: {runAt?: Date; maxAttempts?: number; priority?: number},
+	): Promise<string> {
+		return await withSpan(
+			{
+				name: 'queue.enqueue',
+				attributes: {
+					'queue.task_type': taskType,
+					'queue.priority': options?.priority ?? 0,
+					'queue.max_attempts': options?.maxAttempts ?? 5,
+					'queue.scheduled': options?.runAt !== undefined,
+				},
+			},
+			async () => {
+				const js = this.connectionManager.getJetStreamClient();
+				const subject = `${SUBJECT_PREFIX}${taskType}`;
+				const body = JSON.stringify({
+					payload,
+					run_at: options?.runAt?.toISOString(),
+					max_attempts: options?.maxAttempts ?? 5,
+					priority: options?.priority ?? 0,
+					created_at: new Date().toISOString(),
+				});
+
+				const ack = await js.publish(subject, body, {
+					msgID: randomUUID(),
+				});
+
+				const jobId = `${ack.seq}`;
+				setSpanAttributes({'queue.job_id': jobId});
+				addSpanEvent('enqueue.complete');
+				return jobId;
+			},
+		);
+	}
+
+	getStreamName(): string {
+		return STREAM_NAME;
+	}
+
+	getConsumerName(): string {
+		return CONSUMER_NAME;
+	}
+
+	getConnectionManager(): JetStreamConnectionManager {
+		return this.connectionManager;
+	}
+}

packages/api/src/worker/WorkerMain.tsx

@@ -17,61 +17,71 @@
  * along with Fluxer. If not, see <https://www.gnu.org/licenses/>.
  */
 
+import {Config} from '@fluxer/api/src/Config';
 import {getMetricsService, initializeMetricsService} from '@fluxer/api/src/infrastructure/MetricsService';
 import {SnowflakeService} from '@fluxer/api/src/infrastructure/SnowflakeService';
 import {Logger} from '@fluxer/api/src/Logger';
-import {getKVClient} from '@fluxer/api/src/middleware/ServiceRegistry';
+import {getKVClient, setInjectedWorkerService} from '@fluxer/api/src/middleware/ServiceRegistry';
 import {initializeSearch} from '@fluxer/api/src/SearchFactory';
-import {HttpWorkerQueue} from '@fluxer/api/src/worker/HttpWorkerQueue';
+import {CronScheduler} from '@fluxer/api/src/worker/CronScheduler';
+import {JetStreamWorkerQueue} from '@fluxer/api/src/worker/JetStreamWorkerQueue';
 import {setWorkerDependencies} from '@fluxer/api/src/worker/WorkerContext';
 import {initializeWorkerDependencies, shutdownWorkerDependencies} from '@fluxer/api/src/worker/WorkerDependencies';
 import {WorkerMetricsCollector} from '@fluxer/api/src/worker/WorkerMetricsCollector';
 import {WorkerRunner} from '@fluxer/api/src/worker/WorkerRunner';
+import {WorkerService} from '@fluxer/api/src/worker/WorkerService';
 import {workerTasks} from '@fluxer/api/src/worker/WorkerTaskRegistry';
 import {setupGracefulShutdown} from '@fluxer/hono/src/Server';
+import {JetStreamConnectionManager} from '@fluxer/nats/src/JetStreamConnectionManager';
 import {captureException, flushSentry as flush} from '@fluxer/sentry/src/Sentry';
 import {ms} from 'itty-time';
 
 const WORKER_CONCURRENCY = 20;
 
-async function registerCronJobs(queue: HttpWorkerQueue): Promise<void> {
-	try {
-		await queue.upsertCron('processAssetDeletionQueue', 'processAssetDeletionQueue', {}, '0 */5 * * * *');
-		await queue.upsertCron('processCloudflarePurgeQueue', 'processCloudflarePurgeQueue', {}, '0 */2 * * * *');
-		await queue.upsertCron(
-			'processPendingBulkMessageDeletions',
-			'processPendingBulkMessageDeletions',
-			{},
-			'0 */10 * * * *',
-		);
-		await queue.upsertCron('processInactivityDeletions', 'processInactivityDeletions', {}, '0 0 */6 * * *');
-		await queue.upsertCron('expireAttachments', 'expireAttachments', {}, '0 0 */12 * * *');
-		await queue.upsertCron('cleanupCsamEvidence', 'cleanupCsamEvidence', {}, '0 0 3 * * *');
-		await queue.upsertCron('csamScanConsumer', 'csamScanConsumer', {}, '* * * * * *');
-		await queue.upsertCron('syncDiscoveryIndex', 'syncDiscoveryIndex', {}, '0 */15 * * * *');
+function registerCronJobs(cron: CronScheduler): void {
+	cron.upsert('processAssetDeletionQueue', 'processAssetDeletionQueue', {}, '0 */5 * * * *');
+	cron.upsert('processCloudflarePurgeQueue', 'processCloudflarePurgeQueue', {}, '0 */2 * * * *');
+	cron.upsert('processPendingBulkMessageDeletions', 'processPendingBulkMessageDeletions', {}, '0 */10 * * * *');
+	cron.upsert('processInactivityDeletions', 'processInactivityDeletions', {}, '0 0 */6 * * *');
+	cron.upsert('expireAttachments', 'expireAttachments', {}, '0 0 */12 * * *');
+	// cron.upsert('cleanupCsamEvidence', 'cleanupCsamEvidence', {}, '0 0 3 * * *');
+	// cron.upsert('csamScanConsumer', 'csamScanConsumer', {}, '* * * * * *');
+	cron.upsert('syncDiscoveryIndex', 'syncDiscoveryIndex', {}, '0 */15 * * * *');
 
-		Logger.info('Cron jobs registered successfully');
-	} catch (error) {
-		Logger.error({error}, 'Failed to register cron jobs');
-	}
+	Logger.info('Cron jobs registered successfully');
 }
 
 export async function startWorkerMain(): Promise<void> {
 	Logger.info('Starting worker backend...');
 
 	initializeMetricsService();
-	Logger.info('MetricsService initialized');
+	Logger.info('MetricsService initialised');
 
 	const kvClient = getKVClient();
 	const snowflakeService = new SnowflakeService(kvClient);
 	await snowflakeService.initialize();
-	Logger.info('Shared SnowflakeService initialized');
+	Logger.info('Shared SnowflakeService initialised');
+
+	const jsConnectionManager = new JetStreamConnectionManager({
+		url: Config.nats.jetStreamUrl,
+		token: Config.nats.authToken || undefined,
+		name: 'fluxer-worker',
+	});
+	await jsConnectionManager.connect();
+	Logger.info('JetStream connection established');
+
+	const queue = new JetStreamWorkerQueue(jsConnectionManager);
+	await queue.ensureInfrastructure();
+	Logger.info('JetStream stream and consumer verified');
+
+	const workerService = new WorkerService(queue);
+	setInjectedWorkerService(workerService);
 
 	const dependencies = await initializeWorkerDependencies(snowflakeService);
 	setWorkerDependencies(dependencies);
 
-	const queue = new HttpWorkerQueue();
-	await registerCronJobs(queue);
+	const cron = new CronScheduler(queue, Logger);
+	registerCronJobs(cron);
 
 	const metricsCollector = new WorkerMetricsCollector({
 		kvClient: dependencies.kvClient,
@@ -84,6 +94,7 @@ export async function startWorkerMain(): Promise<void> {
 
 	const runner = new WorkerRunner({
 		tasks: workerTasks,
+		queue,
 		concurrency: WORKER_CONCURRENCY,
 	});
 
@@ -98,13 +109,18 @@ export async function startWorkerMain(): Promise<void> {
 		metricsCollector.start();
 		Logger.info('WorkerMetricsCollector started');
 
+		cron.start();
+		Logger.info('Cron scheduler started');
+
 		await runner.start();
 		Logger.info(`Worker runner started with ${WORKER_CONCURRENCY} workers`);
 
 		const shutdown = async (): Promise<void> => {
 			Logger.info('Shutting down worker backend...');
+			cron.stop();
 			metricsCollector.stop();
 			await runner.stop();
+			await jsConnectionManager.drain();
 			await shutdownWorkerDependencies(dependencies);
 			await snowflakeService.shutdown();
 		};

packages/api/src/worker/WorkerRunner.tsx

@@ -21,35 +21,32 @@ import {randomUUID} from 'node:crypto';
 import {Logger} from '@fluxer/api/src/Logger';
 import {getWorkerService} from '@fluxer/api/src/middleware/ServiceRegistry';
 import {addSpanEvent, setSpanAttributes, withSpan} from '@fluxer/api/src/telemetry/Tracing';
-import type {HttpWorkerQueue} from '@fluxer/api/src/worker/HttpWorkerQueue';
-import {HttpWorkerQueue as HttpWorkerQueueClass} from '@fluxer/api/src/worker/HttpWorkerQueue';
+import type {JetStreamWorkerQueue} from '@fluxer/api/src/worker/JetStreamWorkerQueue';
 import type {IWorkerService} from '@fluxer/worker/src/contracts/IWorkerService';
 import type {WorkerTaskHandler} from '@fluxer/worker/src/contracts/WorkerTask';
-import {ms} from 'itty-time';
+import type {ConsumerMessages, JsMsg} from 'nats';
 
 interface WorkerRunnerOptions {
 	tasks: Record<string, WorkerTaskHandler>;
+	queue: JetStreamWorkerQueue;
 	workerId?: string;
-	taskTypes?: Array<string>;
 	concurrency?: number;
 }
 
 export class WorkerRunner {
 	private readonly tasks: Record<string, WorkerTaskHandler>;
+	private readonly queue: JetStreamWorkerQueue;
 	private readonly workerId: string;
-	private readonly taskTypes: Array<string>;
 	private readonly concurrency: number;
-	private readonly queue: HttpWorkerQueue;
 	private readonly workerService: IWorkerService;
 	private running = false;
-	private abortController: AbortController | null = null;
+	private consumerMessages: ConsumerMessages | null = null;
 
 	constructor(options: WorkerRunnerOptions) {
 		this.tasks = options.tasks;
+		this.queue = options.queue;
 		this.workerId = options.workerId ?? `worker-${randomUUID()}`;
-		this.taskTypes = options.taskTypes ?? Object.keys(options.tasks);
 		this.concurrency = options.concurrency ?? 1;
-		this.queue = new HttpWorkerQueueClass();
 		this.workerService = getWorkerService();
 	}
 
@@ -60,14 +57,19 @@ export class WorkerRunner {
 		}
 
 		this.running = true;
-		this.abortController = new AbortController();
 
-		Logger.info({workerId: this.workerId, taskTypes: this.taskTypes, concurrency: this.concurrency}, 'Worker starting');
+		Logger.info({workerId: this.workerId, concurrency: this.concurrency}, 'Worker starting');
 
-		const workers = Array.from({length: this.concurrency}, (_, i) => this.workerLoop(i, this.abortController!.signal));
+		const js = this.queue.getConnectionManager().getJetStreamClient();
+		const consumer = await js.consumers.get(this.queue.getStreamName(), this.queue.getConsumerName());
 
-		Promise.all(workers).catch((error) => {
-			Logger.error({workerId: this.workerId, error}, 'Worker loop failed unexpectedly');
+		this.consumerMessages = await consumer.consume({
+			max_messages: this.concurrency,
+			idle_heartbeat: 5000,
+		});
+
+		this.processMessages().catch((error) => {
+			Logger.error({workerId: this.workerId, err: error}, 'Worker message processing failed unexpectedly');
 		});
 	}
 
@@ -77,88 +79,90 @@ export class WorkerRunner {
 		}
 
 		this.running = false;
-		this.abortController?.abort();
 
-		await new Promise((resolve) => setTimeout(resolve, ms('5 seconds')));
+		if (this.consumerMessages !== null) {
+			await this.consumerMessages.close();
+			this.consumerMessages = null;
+		}
 
 		Logger.info({workerId: this.workerId}, 'Worker stopped');
 	}
 
-	private async workerLoop(workerIndex: number, signal: AbortSignal): Promise<void> {
-		Logger.info({workerId: this.workerId, workerIndex}, 'Worker loop started');
+	private async processMessages(): Promise<void> {
+		if (this.consumerMessages === null) {
+			return;
+		}
 
-		while (!signal.aborted) {
-			try {
-				const leasedJobs = await this.queue.dequeue(this.taskTypes, 1);
+		for await (const msg of this.consumerMessages) {
+			if (!this.running) {
+				break;
+			}
 
-				if (!leasedJobs || leasedJobs.length === 0) {
-					continue;
-				}
+			const taskType = msg.subject.startsWith('jobs.') ? msg.subject.slice(5) : msg.subject;
 
-				const leasedJob = leasedJobs[0]!;
-				const job = leasedJob.job;
+			Logger.info(
+				{
+					workerId: this.workerId,
+					taskType,
+					seq: msg.seq,
+					redelivered: msg.redelivered,
+				},
+				'Processing job',
+			);
 
-				Logger.info(
-					{
-						workerId: this.workerId,
-						workerIndex,
-						jobId: job.id,
-						taskType: job.task_type,
-						attempts: job.attempts,
-						receipt: leasedJob.receipt,
-					},
-					'Processing job',
-				);
-
-				const succeeded = await this.processJob(leasedJob);
-				if (succeeded) {
-					Logger.info({workerId: this.workerId, workerIndex, jobId: job.id}, 'Job completed successfully');
-				}
-			} catch (error) {
-				Logger.error({workerId: this.workerId, workerIndex, error}, 'Worker loop error');
-
-				await this.sleep(ms('1 second'));
+			const succeeded = await this.processJob(taskType, msg);
+			if (succeeded) {
+				Logger.info({workerId: this.workerId, taskType, seq: msg.seq}, 'Job completed successfully');
 			}
 		}
 
-		Logger.info({workerId: this.workerId, workerIndex}, 'Worker loop stopped');
+		Logger.info({workerId: this.workerId}, 'Worker message iterator ended');
 	}
 
-	private async processJob(leasedJob: {
-		receipt: string;
-		job: {id: string; task_type: string; payload: unknown; attempts: number};
-	}): Promise<boolean> {
+	private async processJob(taskType: string, msg: JsMsg): Promise<boolean> {
 		return await withSpan(
 			{
 				name: 'worker.process_job',
 				attributes: {
 					'worker.id': this.workerId,
-					'job.id': leasedJob.job.id,
-					'job.task_type': leasedJob.job.task_type,
-					'job.attempts': leasedJob.job.attempts,
+					'job.seq': msg.seq,
+					'job.task_type': taskType,
+					'job.redelivered': msg.redelivered,
 				},
 			},
 			async () => {
-				const task = this.tasks[leasedJob.job.task_type];
+				const task = this.tasks[taskType];
 				if (!task) {
-					throw new Error(`Unknown task: ${leasedJob.job.task_type}`);
+					Logger.error({taskType, seq: msg.seq}, 'Unknown task type, terminating message');
+					msg.term(`unknown task type: ${taskType}`);
+					return false;
+				}
+
+				let jobPayload: Record<string, unknown> = {};
+				try {
+					const decoded = JSON.parse(new TextDecoder().decode(msg.data)) as {payload?: Record<string, unknown>};
+					jobPayload = decoded.payload ?? {};
+				} catch {
+					Logger.error({taskType, seq: msg.seq}, 'Failed to decode job payload, terminating message');
+					msg.term('invalid payload');
+					return false;
 				}
 
 				addSpanEvent('job.execution.start');
 
 				try {
-					await task(leasedJob.job.payload as never, {
-						logger: Logger.child({jobId: leasedJob.job.id}),
+					await task(jobPayload as never, {
+						logger: Logger.child({taskType, seq: msg.seq}),
 						addJob: this.workerService.addJob.bind(this.workerService),
 					});
 
 					addSpanEvent('job.execution.success');
 					setSpanAttributes({'job.status': 'success'});
 
-					await this.queue.complete(leasedJob.receipt);
+					msg.ack();
 					return true;
 				} catch (error) {
-					Logger.error({jobId: leasedJob.job.id, error}, 'Job failed');
+					Logger.error({taskType, seq: msg.seq, err: error}, 'Job failed');
 
 					setSpanAttributes({
 						'job.status': 'failed',
@@ -168,14 +172,10 @@ export class WorkerRunner {
 						error: error instanceof Error ? error.message : String(error),
 					});
 
-					await this.queue.fail(leasedJob.receipt, String(error));
+					msg.nak(5000);
 					return false;
 				}
 			},
 		);
 	}
-
-	private async sleep(ms: number): Promise<void> {
-		return new Promise((resolve) => setTimeout(resolve, ms));
-	}
 }

packages/api/src/worker/WorkerService.tsx

@@ -18,15 +18,15 @@
  */
 
 import {Logger} from '@fluxer/api/src/Logger';
-import {HttpWorkerQueue} from '@fluxer/api/src/worker/HttpWorkerQueue';
+import type {JetStreamWorkerQueue} from '@fluxer/api/src/worker/JetStreamWorkerQueue';
 import type {IWorkerService} from '@fluxer/worker/src/contracts/IWorkerService';
 import type {WorkerJobOptions, WorkerJobPayload} from '@fluxer/worker/src/contracts/WorkerTypes';
 
 export class WorkerService implements IWorkerService {
-	private readonly queue: HttpWorkerQueue;
+	private readonly queue: JetStreamWorkerQueue;
 
-	constructor() {
-		this.queue = new HttpWorkerQueue();
+	constructor(queue: JetStreamWorkerQueue) {
+		this.queue = queue;
 	}
 
 	async addJob<TPayload extends WorkerJobPayload = WorkerJobPayload>(
@@ -47,33 +47,11 @@ export class WorkerService implements IWorkerService {
 		}
 	}
 
-	async cancelJob(jobId: string): Promise<boolean> {
-		try {
-			const cancelled = await this.queue.cancelJob(jobId);
-			if (cancelled) {
-				Logger.info({jobId}, 'Job cancelled successfully');
-			} else {
-				Logger.debug({jobId}, 'Job not found (may have already been processed)');
-			}
-			return cancelled;
-		} catch (error) {
-			Logger.error({error, jobId}, 'Failed to cancel job');
-			throw error;
-		}
+	async cancelJob(_jobId: string): Promise<boolean> {
+		return false;
 	}
 
-	async retryDeadLetterJob(jobId: string): Promise<boolean> {
-		try {
-			const retried = await this.queue.retryDeadLetterJob(jobId);
-			if (retried) {
-				Logger.info({jobId}, 'Dead letter job retried successfully');
-			} else {
-				Logger.debug({jobId}, 'Job not found in dead letter queue');
-			}
-			return retried;
-		} catch (error) {
-			Logger.error({error, jobId}, 'Failed to retry dead letter job');
-			throw error;
-		}
+	async retryDeadLetterJob(_jobId: string): Promise<boolean> {
+		return false;
 	}
 }

packages/api/src/worker/WorkerTaskRegistry.tsx

@@ -20,8 +20,8 @@
 import applicationProcessDeletion from '@fluxer/api/src/worker/tasks/ApplicationProcessDeletion';
 import batchGuildAuditLogMessageDeletes from '@fluxer/api/src/worker/tasks/BatchGuildAuditLogMessageDeletes';
 import bulkDeleteUserMessages from '@fluxer/api/src/worker/tasks/BulkDeleteUserMessages';
-import cleanupCsamEvidence from '@fluxer/api/src/worker/tasks/CleanupCsamEvidence';
-import csamScanConsumer from '@fluxer/api/src/worker/tasks/CsamScanConsumerWorker';
+// import cleanupCsamEvidence from '@fluxer/api/src/worker/tasks/CleanupCsamEvidence';
+// import csamScanConsumer from '@fluxer/api/src/worker/tasks/CsamScanConsumerWorker';
 import deleteUserMessagesInGuildByTime from '@fluxer/api/src/worker/tasks/DeleteUserMessagesInGuildByTime';
 import expireAttachments from '@fluxer/api/src/worker/tasks/ExpireAttachments';
 import extractEmbeds from '@fluxer/api/src/worker/tasks/ExtractEmbeds';
@@ -48,7 +48,7 @@ export const workerTasks: Record<string, WorkerTaskHandler> = {
 	applicationProcessDeletion,
 	batchGuildAuditLogMessageDeletes,
 	bulkDeleteUserMessages,
-	csamScanConsumer,
+	// csamScanConsumer,
 	deleteUserMessagesInGuildByTime,
 	expireAttachments,
 	extractEmbeds,
@@ -59,7 +59,7 @@ export const workerTasks: Record<string, WorkerTaskHandler> = {
 	indexGuildMembers,
 	messageShred,
 	processAssetDeletionQueue,
-	cleanupCsamEvidence,
+	// cleanupCsamEvidence,
 	processCloudflarePurgeQueue,
 	processInactivityDeletions,
 	processPendingBulkMessageDeletions,

packages/api/src/worker/tasks/RefreshSearchIndex.tsx

@@ -19,6 +19,7 @@
 
 import type {GuildID, ReportID, UserID} from '@fluxer/api/src/BrandedTypes';
 import {createGuildID} from '@fluxer/api/src/BrandedTypes';
+import {GuildDiscoveryRepository} from '@fluxer/api/src/guild/repositories/GuildDiscoveryRepository';
 import {Logger} from '@fluxer/api/src/Logger';
 import type {User} from '@fluxer/api/src/models/User';
 import {
@@ -30,12 +31,21 @@ import {
 	getUserSearchService,
 } from '@fluxer/api/src/SearchFactory';
 import {getWorkerDependencies} from '@fluxer/api/src/worker/WorkerContext';
+import {DiscoveryApplicationStatus} from '@fluxer/constants/src/DiscoveryConstants';
 import type {IKVProvider} from '@fluxer/kv_client/src/IKVProvider';
 import type {WorkerTaskHandler, WorkerTaskHelpers} from '@fluxer/worker/src/contracts/WorkerTask';
 import {seconds} from 'itty-time';
 import {z} from 'zod';
 
-const INDEX_TYPES = ['guilds', 'users', 'reports', 'audit_logs', 'channel_messages', 'guild_members'] as const;
+const INDEX_TYPES = [
+	'guilds',
+	'users',
+	'reports',
+	'audit_logs',
+	'channel_messages',
+	'guild_members',
+	'discovery',
+] as const;
 type IndexType = (typeof INDEX_TYPES)[number];
 
 const PayloadSchema = z
@@ -259,6 +269,49 @@ const refreshGuildMembers: IndexHandler = async (payload, _helpers, kvClient, pr
 	return indexedCount;
 };
 
+const DISCOVERY_BATCH_SIZE = 50;
+
+const refreshDiscovery: IndexHandler = async (_payload, _helpers, kvClient, progressKey) => {
+	const {guildRepository} = getWorkerDependencies();
+	const searchService = requireSearchService(getGuildSearchService());
+	const discoveryRepository = new GuildDiscoveryRepository();
+
+	const approvedRows = await discoveryRepository.listByStatus(DiscoveryApplicationStatus.APPROVED, 1000);
+	if (approvedRows.length === 0) {
+		return 0;
+	}
+
+	const guildIds = approvedRows.map((row) => row.guild_id);
+
+	let synced = 0;
+	for (let i = 0; i < guildIds.length; i += DISCOVERY_BATCH_SIZE) {
+		const batch = guildIds.slice(i, i + DISCOVERY_BATCH_SIZE);
+		for (const guildId of batch) {
+			const guild = await guildRepository.findUnique(guildId);
+			if (!guild) continue;
+
+			const discoveryRow = await discoveryRepository.findByGuildId(guildId);
+			if (!discoveryRow) continue;
+
+			await searchService.updateGuild(guild, {
+				description: discoveryRow.description,
+				categoryId: discoveryRow.category_type,
+			});
+			synced++;
+		}
+
+		await setProgress(kvClient, progressKey, {
+			status: 'in_progress',
+			index_type: 'discovery',
+			total: guildIds.length,
+			indexed: synced,
+			started_at: new Date().toISOString(),
+		});
+	}
+
+	return synced;
+};
+
 const INDEX_HANDLERS: Record<IndexType, IndexHandler> = {
 	guilds: refreshGuilds,
 	users: refreshUsers,
@@ -266,6 +319,7 @@ const INDEX_HANDLERS: Record<IndexType, IndexHandler> = {
 	audit_logs: refreshAuditLogs,
 	channel_messages: refreshChannelMessages,
 	guild_members: refreshGuildMembers,
+	discovery: refreshDiscovery,
 };
 
 const refreshSearchIndex: WorkerTaskHandler = async (payload, helpers) => {

packages/api/src/worker/tasks/SyncDiscoveryIndex.tsx

@@ -17,9 +17,7 @@
  * along with Fluxer. If not, see <https://www.gnu.org/licenses/>.
  */
 
-import type {GuildID} from '@fluxer/api/src/BrandedTypes';
 import {GuildDiscoveryRepository} from '@fluxer/api/src/guild/repositories/GuildDiscoveryRepository';
-import {Logger} from '@fluxer/api/src/Logger';
 import {getGuildSearchService} from '@fluxer/api/src/SearchFactory';
 import {getWorkerDependencies} from '@fluxer/api/src/worker/WorkerContext';
 import {DiscoveryApplicationStatus} from '@fluxer/constants/src/DiscoveryConstants';
@@ -36,7 +34,7 @@ const syncDiscoveryIndex: WorkerTaskHandler = async (_payload, helpers) => {
 		return;
 	}
 
-	const {guildRepository, gatewayService} = getWorkerDependencies();
+	const {guildRepository} = getWorkerDependencies();
 	const discoveryRepository = new GuildDiscoveryRepository();
 
 	const approvedRows = await discoveryRepository.listByStatus(DiscoveryApplicationStatus.APPROVED, 1000);
@@ -46,13 +44,6 @@ const syncDiscoveryIndex: WorkerTaskHandler = async (_payload, helpers) => {
 	}
 
 	const guildIds = approvedRows.map((row) => row.guild_id);
-	let onlineCounts = new Map<GuildID, number>();
-
-	try {
-		onlineCounts = await gatewayService.getDiscoveryOnlineCounts(guildIds);
-	} catch (error) {
-		Logger.warn({err: error}, 'Failed to fetch online counts from gateway, proceeding with zero counts');
-	}
 
 	let synced = 0;
 	for (let i = 0; i < guildIds.length; i += BATCH_SIZE) {
@@ -68,7 +59,6 @@ const syncDiscoveryIndex: WorkerTaskHandler = async (_payload, helpers) => {
 			await guildSearchService.updateGuild(guild, {
 				description: discoveryRow.description,
 				categoryId: discoveryRow.category_type,
-				onlineCount: onlineCounts.get(guildId) ?? 0,
 			});
 
 			synced++;