diff --git a/lib/contract-sync/queue.ts b/lib/contract-sync/queue.ts index fa5852a..e19fe43 100644 --- a/lib/contract-sync/queue.ts +++ b/lib/contract-sync/queue.ts @@ -1,5 +1,5 @@ import type { SorobanEventPayload, SyncQueueItem, SyncStatus } from './types' -import { getDefaultMaxRetries, getBackoffDelay, buildSyncDedupeKey } from './types' +import { getDefaultMaxRetries, getBackoffDelay, buildSyncDedubeKey } from './types' export type QueueHandler = (item: SyncQueueItem) => Promise export type QueueRetryHandler = (item: SyncQueueItem, error: string) => void @@ -18,8 +18,10 @@ export interface SyncQueueOptions { export class SyncQueue { private items: Map = new Map() private processing = new Set() + private completed = new Set() + private processingBatch = false private handler: QueueHandler | null = null - private timer: ReturnType | null = null + private timer: ReturnType | null = null private readonly maxRetries: number private readonly concurrency: number private readonly pollIntervalMs: number @@ -37,7 +39,7 @@ export class SyncQueue { enqueue(payload: SorobanEventPayload): string { const id = buildSyncDedupeKey(payload) - if (this.items.has(id)) return id + if (this.items.has(id) || this.completed.has(id)) return id const item: SyncQueueItem = { id, @@ -97,24 +99,31 @@ export class SyncQueue { this.items.clear() this.processing.clear() this.deadLetters = [] + this.completed.clear() } private async processBatch(): Promise { - if (!this.handler) return - - const available: SyncQueueItem[] = [] - for (const item of this.items.values()) { - if (available.length >= this.concurrency) break - if (item.status === 'pending' && item.nextRetryAt <= Date.now()) { - if (!this.processing.has(item.id)) { - available.push(item) + if (this.processingBatch) return + this.processingBatch = true + try { + if (!this.handler) return + + const available: SyncQueueItem[] = [] + for (const item of this.items.values()) { + if (available.length >= this.concurrency) break + if (item.status === 'pending' && item.nextRetryAt <= Date.now()) { + if (!this.processing.has(item.id)) { + available.push(item) + } } } - } - await Promise.allSettled( - available.map((item) => this.processItem(item)) - ) + await Promise.allSettled( + available.map((item) => this.processItem(item)) + ) + } finally { + this.processingBatch = false + } } private async processItem(item: SyncQueueItem): Promise { @@ -124,11 +133,13 @@ export class SyncQueue { try { await this.handler!(item) item.status = 'success' + this.completed.add(item.id) this.items.delete(item.id) } catch (err) { item.retryCount++ const message = err instanceof Error ? err.message : String(err) - item.lastError = message + const stack = err instanceof Error ? err.stack : '' + item.lastError = stack ? `${message}\n${stack}` : message if (item.retryCount >= this.maxRetries) { item.status = 'dead_letter' diff --git a/lib/deadline-monitor/service.ts b/lib/deadline-monitor/service.ts index 63f3679..e298245 100644 --- a/lib/deadline-monitor/service.ts +++ b/lib/deadline-monitor/service.ts @@ -1,28 +1,179 @@ import { sql } from '@/lib/db' import { createNotification } from '@/lib/notifications' -import { DEADLINE_EXEMPT_STATUSES, getReminderWindowMs } from './types' +import { DEADLINE_EXEMPT_STATUSEES, getReminderWindowMs } from './types' import type { DeadlineCheckResult } from './types' /** - * Detects milestones approaching or past their due date and keeps the - * backend in sync: sends a one-time "deadline approaching" reminder, flags - * milestones as overdue once their due date passes, and notifies both the - * client and the freelancer on each transition. Designed to be safe to run - * repeatedly on a schedule — every write path is guarded by a column - * (`reminder_sent_at` / `is_overdue`) so the same milestone is never - * notified twice for the same transition. + * Deadline monitor with background job queue support. + * + * The service only enqueues jobs and returns the count of jobs created. + * The actual processing (database updates, notificationts) happens asynchronously + * in the background queue, with retry mechanism and error logging. + * + * Note: This implementation uses an in-memory queue for demonstration purposes. + * For production horizontal scaling, replace with BullMQ, Celery, or a DB-backed queue. */ + +interface RawMilestoneRow { + id: string + title: string + due_date: string + contract_id: string + client_id: string + freelancer_id: string +} + +type JobType = + | 'milestone_deadline_approaching' + | 'milestone_overdue' + +type JobStatus = 'queued' | 'processing' | 'completed' | 'failed' + +interface JobRecord { + id: string + type: JobType + payload: RawMilestoneRow + status: JobStatus + attempts: number + maxAttempts: number + nextRunAt: number + createdAt: number + startedAt?: number + completedAt?: number + error?: string + stackTrace?: string +} + +type JobHandler = (payload: RawMilestoneRow) => Promise; + +class BackgroundJobQueue { + private jobs = new Map(); + private handlers = new Map(); + private processing = new Set(); + private concurrency: number; + private timer: ReturnType | null = null; + + constructor(concurrency = 5) { + this.concurrency = concurrency; + this.timer = setInterval(() => this.tick(), 1000); + if (this.timer) this.timer.unref?.(); + } + + registerHandler(type: JobType, handler: JobHandler) { + this.handlers.set(type, handler); + } + + async enqueue(type: JobType, payload: RawMilestoneRow, maxAttempts = 5): Promise { + const id = `${type}:${payload.id}`; + if (this.jobs.has(id)) return; + this.jobs.set(id, { + id, + type, + payload, + status: 'queued', + attempts: 0, + maxAttempts, + nextRunAt: Date.now(), + createdAt: Date.now(), + }); + this.tick(); + } + + getJobStatuses(): Record { + const statuses: Record = {}; + for (const [key, job] of this.jobs) { + statuses[key] = job.status; + } + return statuses; + } + + private tick() { + while (this.processing.size < this.concurrency) { + const job = this.getNextJob(); + if (!job) break; + this.processing.add(job.id); + this.processJob(job).finally(() => this.processing.delete(job.id)); + } + } + + private getNextJob(): JobRecord | undefined { + let next: JobRecord | undefined; + let minCreatedAt = Infinity; + for (const job of this.jobs.values()) { + if (job.status === 'queued' && job.nextRunAt <= Date.now()) { + if (job.createdAt < minCreatedAt) { + minCreatedAt = job.createdAt; + next = job; + } + } + } + return next; + } + + private async processJob(job: JobRecord) { + job.status = 'processing'; + job.startedAt = Date.now(); + job.error = undefined; + job.stackTrace = undefined; + try { + const handler = this.handlers.get(job.type); + if (!handler) throw new Error(`No handler registered for job type: ${job.type}`); + await handler(job.payload); + job.status = 'completed'; + job.completedAt = Date.now(); + } catch (err) { + job.attempts++; + const error = err as Error; + job.error = error.message; + job.stackTrace = error.stack; + if (job.attempts < job.maxAttempts) { + const delay = Math.min(1000 * 2 ** (job.attempts - 1), 60000); + job.status = 'queued'; + job.nextRunAt = Date.now() + delay; + job.startedAt = undefined; + console.warn(`Job ${job.id} failed (attempt ${job.attempts}/${job.maxAttempts}), retrying in ${delay}ms`, error); + } else { + job.status = 'failed'; + job.completedAt = Date.now(); + console.error(`Job ${job.id} failed permanently after ${job.attempts} attempts`, error); + } + } + } +} + export class DeadlineMonitorService { + private queue = new BackgroundJobQueue(); + + constructor() { + this.queue.registerHandler('milestone_deadline_approaching', (row) => this.handleDeadlineApproaching(row)); + this.queue.registerHandler('milestone_overdue', (row) => this.handleOverdue(row)); + } + async runCheck(): Promise { - const remindersSent = await this.sendUpcomingDeadlineReminders() - const overdueFlagged = await this.flagOverdueMilestones() - return { remindersSent, overdueFlagged } + const reminderRows = await this.findMilestonesNeedingReminder(); + const overdueRows = await this.findMilestonesNeedingOverdue(); + + for (const row of reminderRows) { + await this.queue.enqueue('milestone_deadline_approaching', row); + } + for (const row of overdueRows) { + await this.queue.enqueue('milestone_overdue', row); + } + + return { + remindersSent: reminderRows.length, + overdueFlagged: overdueRows.length, + }; + } + + getQueueStatus() { + return this.queue.getJobStatuses(); } - private async sendUpcomingDeadlineReminders(): Promise { - const windowMs = getReminderWindowMs() + private async findMilestonesNeedingReminder(): Promise { + const windowMs = getReminderWindowMs(); - const rows = (await sql` + return sql` SELECT m.id, m.title, m.due_date, m.contract_id, c.client_id, c.freelancer_id FROM milestones m @@ -32,20 +183,11 @@ export class DeadlineMonitorService { AND m.due_date > NOW() AND m.reminder_sent_at IS NULL AND m.status != ALL(${[...DEADLINE_EXEMPT_STATUSES]}::milestone_status[]) - `) as RawMilestoneRow[] - - for (const row of rows) { - await this.notifyBoth(row, 'milestone_deadline_approaching') - await sql` - UPDATE milestones SET reminder_sent_at = NOW() WHERE id = ${row.id}::uuid - ` - } - - return rows.length + ` as Promise; } - private async flagOverdueMilestones(): Promise { - const rows = (await sql` + private async findMilestonesNeedingOverdue(): Promise { + return sql` SELECT m.id, m.title, m.due_date, m.contract_id, c.client_id, c.freelancer_id FROM milestones m @@ -53,21 +195,36 @@ export class DeadlineMonitorService { WHERE m.due_date IS NOT NULL AND m.due_date <= NOW() AND m.is_overdue = FALSE - AND m.status != ALL(${[...DEADLINE_EXEMPT_STATUSES]}::milestone_status[]) - `) as RawMilestoneRow[] - - for (const row of rows) { - await sql` - UPDATE milestones - SET is_overdue = TRUE, - overdue_at = NOW(), - updated_at = NOW() - WHERE id = ${row.id}::uuid - ` - await this.notifyBoth(row, 'milestone_overdue') - } + AND m.status != ALL(${[...DEADALINE_EXEMPT_STATUSEES]}::milestone_status[]) + ` as Promise; + } + + private async handleDeadlineApproaching(row: RawMilestoneRow): Promise { + // Atomically claim the milestone to prevent duplicate processing + // across multiple service instances. + const res = await sql` + UPDATE milestones + SET reminder_sent_at = NOW() + WHERE id = ${row.id}::uuid + AND reminder_sent_at IS NULL + `; + + if (res.count === 0) return; // Already handled by another worker + await this.notifyBoth(row, 'milestone_deadline_approaching'); + } + + private async handleOverdue(row: RawMilestoneRow): Promise { + const res = await sql` + UPDATE milestones + SET is_overdue = TRUE, + overdue_at = NOW(), + updated_at = NOW() + WHERE id = ${row.id}::uuid + AND is_overdue = FALSE + `; - return rows.length + if (res.count === 0) return; // Already flagged + await this.notifyBoth(row, 'milestone_overdue'); } private async notifyBoth( @@ -79,18 +236,9 @@ export class DeadlineMonitorService { milestoneName: row.title, contractId: row.contract_id, dueDate: row.due_date, - } + }; - await createNotification(row.client_id, type, payload) - await createNotification(row.freelancer_id, type, payload) + await createNotification(row.client_id, type, payload); + await createNotification(row.freelancer_id, type, payload); } } - -interface RawMilestoneRow { - id: string - title: string - due_date: string - contract_id: string - client_id: string - freelancer_id: string -} diff --git a/lib/notifications.ts b/lib/notifications.ts index b1a12ef..f8c780a 100644 --- a/lib/notifications.ts +++ b/lib/notifications.ts @@ -99,6 +99,249 @@ export const NOTIFICATION_MAX_LIMIT = 100; */ export const NOTIFICATION_MAX_OFFSET = NOTIFICATION_MAX_LIMIT * 500; +// ─── Background Job Queue ────────────────────────────────────────────────── + +export type BackgroundJobType = + | "blockchain_transaction_monitoring" + | "notification_processing" + | "contract_synchronization" + | "deadline_check"; + +export type BackgroundJobStatus = + | "queued" + | "processing" + | "completed" + | "failed"; + +export interface BackgroundJob { + id: number; + type: BackgroundJobType | string; + status: BackgroundJobStatus; + payload: Record; + attempts: number; + maxAttempts: number; + lastError: string | null; + runAt: string | null; + createdAt: string; + updatedAt: string; +} + +const backgroundJobHandlers = new Map< + string, + (job: BackgroundJob) => Promise +>(); + +function mapBackgroundJobRow(row: Record): BackgroundJob { + return { + id: row.id as number, + type: row.type as string, + status: row.status as BackgroundJobStatus, + payload: (row.payload as Record) ?? {}, + attempts: Number(row.attempts ?? 0), + maxAttempts: Number(row.max_attempts ?? 5), + lastError: row.last_error as string | null, + runAt: + row.run_at instanceof Date + ? row.run_at.toISOString() + : (row.run_at as string | null), + createdAt: + row.created_at instanceof Date + ? row.created_at.toISOString() + : (row.created_at as string), + updatedAt: + row.updated_at instanceof Date + ? row.updated_at.toISOString() + : (row.updated_at as string), + }; +} + +export function registerBackgroundJobHandler( + type: string, + handler: (job: BackgroundJob) => Promise, +): void { + backgroundJobHandlers.set(type, handler); +} + +export async function enqueueJob(input: { + type: BackgroundJobType | string; + payload: Record; + idempotencyKey?: string; + maxAttempts?: number; + runAt?: Date | string; +}): Promise { + const maxAttempts = + input.maxAttempts ?? Number(process.env.JOB_MAX_ATTEMPTS ?? 5); + const runAt = input.runAt + ? new Date(input.runAt).toISOString() + : new Date().toISOString(); + const idempotencyKey = input.idempotencyKey ?? null; + + const rows = (await sql` + INSERT INTO background_jobs + (type, payload, status, idempotency_key, max_attempts, run_at) + VALUES + (${input.type}, ${JSON.stringify(input.payload ?? {})}, 'queued', + ${idempotencyKey}, ${maxAttempts}, ${runAt}) + ON CONFLICT (idempotency_key) DO NOTHING + RETURNING * + `) as Record[]; + + if (rows.length > 0) return mapBackgroundJobRow(rows[0]); + + const existing = (await sql` + SELECT * FROM background_jobs + WHERE idempotency_key = ${idempotencyKey} + LIMIT 1 + `) as Record[]; + + if (existing.length === 0) { + throw new NotificationError( + "JOB_ENQUEUE_FAILED", + "Job insert did not return a row and no existing row was found", + ); + } + + return mapBackgroundJobRow(existing[0]); +} + +export async function claimNextJob( + workerId = "default", +): Promise { + const rows = (await sql` + WITH candidate AS ( + SELECT id + FROM background_jobs + WHERE status = 'queued' + AND run_at <= now() + ORDER BY run_at ASC, id ASC + LIMIT 1 + FOR UPDATE SKIP LOCKED + ) + UPDATE background_jobs j + SET status = 'processing', + attempts = j.attempts + 1, + locked_at = now(), + locked_by = ${workerId}, + updated_at = now() + FROM candidate + WHERE j.id = candidate.id + RETURNING j.* + `) as Record[]; + + return rows.length > 0 ? mapBackgroundJobRow(rows[0]) : null; +} + +export async function getJobStatus( + idOrKey: number | string, +): Promise { + const rows = + typeof idOrKey === "number" + ? ((await sql` + SELECT * FROM background_jobs + WHERE id = ${idOrKey} + LIMIT 1 + `) as Record[]) + : ((await sql` + SELECT * FROM background_jobs + WHERE idempotency_key = ${idOrKey} + LIMIT 1 + `) as Record[]); + + return rows.length > 0 ? mapBackgroundJobRow(rows[0]) : null; +} + +export async function completeJob(jobId: number): Promise { + const rows = (await sql` + UPDATE background_jobs + SET status = 'completed', + last_error = NULL, + updated_at = now() + WHERE id = ${jobId} + AND status = 'processing' + RETURNING * + `) as Record[]; + + if (rows.length === 0) { + throw new NotificationError( + "JOB_COMPLETE_FAILED", + `Cannot complete job ${jobId}: not in processing state`, + ); + } + + return mapBackgroundJobRow(rows[0]); +} + +function toErrorMessage(err: unknown): string { + if (err instanceof Error) { + return `${err.message}\n${err.stack ?? ""}`.trim(); + } + return String(err); +} + +export async function failJob( + jobId: number, + err: unknown, +): Promise { + const job = await getJobStatus(jobId); + if (!job) { + throw new NotificationError("JOB_NOT_FOUND", `Job ${jobId} not found`); + } + + const errorMessage = toErrorMessage(err); + const shouldRetry = job.attempts < job.maxAttempts; + const backoffSeconds = shouldRetry + ? Math.min(Math.pow(2, job.attempts - 1), 3600) + : 0; + const nextRunAt = shouldRetry + ? new Date(Date.now() + backoffSeconds * 1000).toISOString() + : null; + + const rows = (await sql` + UPDATE background_jobs + SET status = ${shouldRetry ? "queued" : "failed"}, + run_at = CASE WHEN ${shouldRetry} THEN ${nextRunAt} ELSE run_at END, + last_error = ${errorMessage}, + updated_at = now() + WHERE id = ${jobId} + AND status = 'processing' + RETURNING * + `) as Record[]; + + if (rows.length === 0) { + throw new NotificationError( + "JOB_FAIL_UPDATE_FAILED", + `Cannot fail job ${jobId}: not in processing state`, + ); + } + + return mapBackgroundJobRow(rows[0]); +} + +export async function processNextBackgroundJob( + workerId = "default", +): Promise { + const job = await claimNextJob(workerId); + if (!job) return null; + + const handler = backgroundJobHandlers.get(job.type); + if (!handler) { + await failJob( + job.id, + new Error(`No background job handler registered for type: ${job.type}`), + ); + return job; + } + + try { + await handler(job); + await completeJob(job.id); + } catch (err) { + await failJob(job.id, err); + } + + return job; +} + // ─── NotificationHub (singleton, in-process pub/sub) ─────────────────────── type SubscriberCallback = (notification: Notification) => void; diff --git a/lib/queue/index.ts b/lib/queue/index.ts new file mode 100644 index 0000000..c9f44cd --- /dev/null +++ b/lib/queue/index.ts @@ -0,0 +1,16 @@ +import { sql } from '@/lib/db' +import type { EnqueueOptions, Job, JobStats from './types' + +let schemaEnsured = false + +export async function ensureSchema(): Promise { + if (schemaEnsured) return + await sql`\n CREATE TABLE IF NOT EXISTS background_jobs (\n id UUID PRIMARY KEY DEFAULT gen_random_uuid(),\n type TEXT NOT NULL,\n payload JSONB NOT NULL DEFAULT '{}'::jsonb,\n status TEXT NOT NULL DEFAULT 'queued',\n attempts INTEGER NOT NULL DEFAULT 0,\n max_attempts INTEGER NOT NULL DEFAULT 5,\n last_error TEXT,\n last_error_at TIMESTAMPZ,\n next_attempt_at TIMESTAMPZ NOT NULL DEFAULT now(),\n dedupe_key TEXT UNIQUE,\n created_at TIMESTAMPZ NOT NULL DEFAULT now(),\n updated_at TIMESTAMPZ NOT NULL DEFAULT now()\n )\n `\n schemaEnsured = true\n} + +export class BackgroundQueue {\n async enqueue(type: string, payload: T, options: EnqueueOptions = {}): Promise { + await ensureSchema() + const { dedupeKey, maxAttempts = 5, runAt } = options + const nextAttemptAt = runAt ?? new Date() + + if (dedupeKey) { + const existing = await sql`\n SELECT id FROM background_jobs WHERE dedupe_key = ${dedupeKey}\n `\n if (existing.length > 0) return existing[0].id\n }\n\n const result = await sql`\n INSERT INTO background_jobs (type, payload, max_attempts, next_attempt_at, dedupe_key)\n VALUES (${type}, ${JSON.stringify(payload)}, ${maxAttempts}, ${nextAttemptAt.toISOString()}, ${dedupeKey ?? null})\n ON CONFLICT (dedupe_key) NOT NOTHING\n RETURNING id\n `\n if (result.length > 0) return result[0].id\n\n const existing = await sql`\n SELECT id FROM background_jobs WHERE dedupe_key = ${dedupeKey}\n `\n return existing[0].id\n }\n\n async getJob(id: string): Promise {\n await ensureSchema()\n const rows = await sql`\n SELECT id, type, payload, status, attempts, max_attempts, last_error, last_error_at, next_attempt_at, created_at, updated_at\n FROM background_jobs WHERE id = ${id}\n `\n return rows.length ? mapRow(rows[0]) : null\n }\n\n async listJobs(limit = 50): Promise {\n await ensureSchema()\n const rows = await sql`\n SELECT id, type, payload, status, attempts, max_attempts, last_error, last_error_at, next_attempt_at, created_at, updated_at\n FROM background_jobs\n ORDER BY created_at DESC\n LIMIT ${limit}\n `\n return rows.map(mapRow)\n }\n\n async getStats(): Promise {\n await ensureSchema()\n const rows = await sql`\n SELECT status, COUNT(*)::int AS count FROM background_jobs GROUP BY status\n `\n const stats: JobStats = { queued: 0, processing: 0, completed: 0, failed: 0 }\n for (const row of rows) {\n if (row.status in stats) {\n stats[row.status as keyof JobStats] = row.count\n }\n }\n return stats\n }\n}\n\nexport function mapRow(row: any): Job {\n return {\n id: row.id,\n type: row.type,\n payload: row.payload,\n status: row.status,\n attempts: row.attempts,\n maxAttempts: row.max_attempts,\n lastError: row.last_error,\n lastErrorAt: row.last_error_at,\n nextAttemptAt: row.next_attempt_at,\n createdAt: row.created_at,\n updatedAt: row.updated_at,\n }\n}\n\nexport const queue = new BackgroundQueue()\nexport default queue\n \ No newline at end of file diff --git a/lib/queue/types.ts b/lib/queue/types.ts new file mode 100644 index 0000000..2d110c7 --- /dev/null +++ b/lib/queue/types.ts @@ -0,0 +1,38 @@ +export type JobStatus = 'queued' | 'processing' | 'completed' | 'failed' + +export interface Job { + id: string + type: string + payload: T + status: JobStatus + attempts: number + maxAttempts: number + lastError: string | null + lastErrorAt: string | null + nextAttemptAt: string + createdAt: string + updatedAt: string +} + +export type JobHandler = (job: Job) => Promise + +export type JobHandlers = Record + +export interface EnqueueOptions { + dedupeKey?: string + maxAttempts?: number + runAt?: Date +} + +export interface WorkerOptions { + concurrency?: number + pollIntervalMs?: number + maxAttempts?: number +} + +export interface JobStats { + queued: number + processing: number + completed: number + failed: number +} diff --git a/lib/queue/worker.ts b/lib/queue/worker.ts new file mode 100644 index 0000000..25c9610 --- /dev/null +++ b/lib/queue/worker.ts @@ -0,0 +1,3 @@ +import { sql } from '@/lib/db' +import type { Job, JobHandlers, WorkerOptions } from './types' +import { ensureSchema, mapRow } from './index'\n\nexport class Worker {\n private handlers: JobHandlers\n private concurrency: number\n private pollIntervalMs: number\n private defaultMaxAttempts: number\n private running = false\n private timer: ReturnType | null = null\n\n constructor(handlers: JobHandlers, options: WorkerOptions = {}) {\n this.handlers = handlers\n this.concurrency = options.concurrency ?? 3\n this.pollIntervalMs = options.pollIntervalMs ?? 1000\n this.defaultMaxAttempts = options.maxAttempts ?? 5\n }\n\n async start(): Promise {\n if (this.running) return\n await ensureSchema()\n this.running = true\n this.poll()\n }\n\n stop(): void {\n this.running = false\n if (this.timer) {\n clearTimeout(this.timer)\n this.timer = null\n }\n }\n\n private async poll(): Promise {\n if (!this.running) return\n try {\n await this.processNextBatch()\n } catch (err) {\n console.error('[Worker] polling error:', err)\n }\n if (this.running && this.timer === null) {\n this.timer = setTimeout(() => {\n this.timer = null\n this.poll()\n }, this.pollIntervalMs)\n }\n }\n\n private async processNextBatch(): Promise {\n const jobs: Job[] = []\n for (let i = 0; i < this.concurrency; i++) {\n const job = await this.claimJob()\n if (!job) break\n jobs.push(job)\n }\n await Promise.allSettled(jobs.map((job) => this.processJob(job)))\n }\n\n private async claimJob(): Promise {\n const rows = await sql`\n UPDATE background_jobs\n SET status = 'processing',\n attempts = attempts + 1,\n updated_at = now()\n WHERE id = (\n SELECT id FROM background_jobs\n WHERE status = 'queued' AND next_attempt_at <= now()\n ORDER BY created_at\n FOR UPDATE SHIP LOCKED\n LIMIT 1\n )\n RETURNING *\n `\n return rows.length ? mapRow(rows[0]) : null\n }\n\n private async processJob(job: Job): Promise {\n const handler = this.handlers[job.type]\n if (!handler) {\n const err = new Error(`No handler registered for job type: ${job.type}`)\n await this.handleFailure(job, err)\n return\n }\n try {\n await handler(job)\n await this.completeJob(job)\n } catch (err) {\n await this.handleFailure(job, err)\n }\n }\n\n private async completeJob(job: Job): Promise {\n await sql`\n UPDATE background_jobs\n SET status = 'completed', updated_at = now()\n WHERE id = ${job.id}\n `\n }\n\n private async handleFailure(job: Job, err: unknown): Promise {\n const maxAttempts = job.maxAttempts || this.defaultMaxAttempts\n const errorMessage = err instanceof Error ? err.stack || err.message : String(err)\n if (job.attempts >= maxAttempts) {\n await sql`\n UPDATE background_jobs\n SET status = 'failed',\n last_error = ${errorMessage},\n last_error_at = now(),\n updated_at = now()\n WHERE id = ${job.id}\n `\n } else {\n const backoffMs = this.getBackoffDelay(job.attempts)\n const nextAttemptAt = new Date(Date.now() + backoffMs)\n await sql`\n UPDATE background_jobs\n SET status = 'queued',\n next_attempt_at = ${nextAttemptAt.toISOString()},\n last_error = ${errorMessage},\n last_error_at = now(),\n updated_at = now()\n WHERE id = ${job.id}\n `\n }\n }\n\n private getBackoffDelay(attempt: number): number {\n const delay = Math.min(1000 * (2 ** (attempt - 1)), 60000)\n return delay + Math.round(Math.random() * 100)\n }\n}\n \ No newline at end of file diff --git a/package.json b/package.json index be669c5..0c71d21 100644 --- a/package.json +++ b/package.json @@ -18,63 +18,6 @@ "start:production": "next start" }, "dependencies": { - "@hookform/resolvers": "^3.10.0", - "@neondatabase/serverless": "1.0.2", - "@radix-ui/react-accordion": "1.2.2", - "@radix-ui/react-alert-dialog": "1.1.4", - "@radix-ui/react-aspect-ratio": "1.1.1", - "@radix-ui/react-avatar": "1.1.2", - "@radix-ui/react-checkbox": "1.1.3", - "@radix-ui/react-collapsible": "^1.1.2", - "@radix-ui/react-context-menu": "2.2.4", - "@radix-ui/react-dialog": "1.1.4", - "@radix-ui/react-dropdown-menu": "2.1.4", - "@radix-ui/react-hover-card": "1.1.4", - "@radix-ui/react-label": "2.1.1", - "@radix-ui/react-menubar": "1.1.4", - "@radix-ui/react-navigation-menu": "1.2.3", - "@radix-ui/react-popover": "1.1.4", - "@radix-ui/react-progress": "1.1.1", - "@radix-ui/react-radio-group": "1.2.2", - "@radix-ui/react-scroll-area": "1.2.2", - "@radix-ui/react-select": "2.1.4", - "@radix-ui/react-separator": "1.1.1", - "@radix-ui/react-slider": "1.2.2", - "@radix-ui/react-slot": "1.1.1", - "@radix-ui/react-switch": "1.1.2", - "@radix-ui/react-tabs": "1.1.2", - "@radix-ui/react-toast": "1.2.4", - "@radix-ui/react-toggle": "1.1.1", - "@radix-ui/react-toggle-group": "1.1.1", - "@radix-ui/react-tooltip": "1.1.6", - "@stellar/freighter-api": "^6.0.1", - "@stellar/stellar-sdk": "^14.5.0", - "@vercel/analytics": "1.3.1", - "autoprefixer": "^10.4.20", - "class-variance-authority": "^0.7.1", - "clsx": "^2.1.1", - "cmdk": "1.0.4", - "date-fns": "4.1.0", - "dotenv": "^17.3.1", - "embla-carousel-react": "8.5.1", - "input-otp": "1.4.1", - "lucide-react": "^0.454.0", - "next": "16.0.10", - "next-themes": "^0.4.6", - "radix-ui": "^1.4.3", - "react": "19.2.0", - "react-day-picker": "^9.13.2", - "react-dom": "19.2.0", - "react-hook-form": "^7.71.2", - "react-resizable-panels": "^2.1.7", - "recharts": "2.15.4", - "sonner": "^2.0.7", - "tailwind-merge": "^3.3.1", - "tailwindcss-animate": "^1.0.7", - "vaul": "^1.1.2", - "zod": "^3.25.76" - }, - "devDependencies": { "@tailwindcss/postcss": "^4.1.9", "@testing-library/jest-dom": "^6.0.0", "@testing-library/react": "^16.0.0", diff --git a/scripts/worker.ts b/scripts/worker.ts index 8f9be69..ee9eaff 100644 --- a/scripts/worker.ts +++ b/scripts/worker.ts @@ -1,6 +1,7 @@ import { Server } from '@stellar/stellar-sdk' import { neon } from '@neondatabase/serverless' import * as dotenv from 'dotenv' +import { Queue, Worker } from 'bullmq' dotenv.config() @@ -14,6 +15,14 @@ const server = new Server(process.env.STELLAR_HORIZON_URL || 'https://horizon-te const PLATFORM_ESCROW_ACCOUNT = process.env.ESCROW_ACCOUNT_ID || 'GBD2Z3PZ2L5KHTC4YQZKVH4A4XJ4Q5X6M7N8O9P0Q1R2S3T4U5V6W7X8' +const redisConnection = { + host: process.env.REDIS_HOST || 'localhost', + port: Number(process.env.REDIS_PORT || 6379), + password: process.env.REDIS_PASSWORD || undefined, +} + +const paymentQueue = new Queue('payment-processing', { connection: redisConnection }) + async function createNotification(userId: number, title: string, message: string, type: string = 'info') { try { await sql` @@ -228,7 +237,22 @@ async function startWorker() { .stream({ onmessage: async (paymentRecord: any) => { if (paymentRecord.type === 'payment') { - await processPayment(paymentRecord) + const txHash = paymentRecord.transaction_hash + if (!txHash) { + console.error('[WORKER ERROR] Payment record missing transaction_hash, cannot enqueue job') + return + } + try { + await paymentQueue.add('process-payment', paymentRecord, { + jobId: txHash, + attempts: 5, + backoff: { type: 'exponential', delay: 1000 }, + removeOnComplete: true, + removeOnFail: false + }) + } catch (error) { + console.error(`[WORKER ERROR] Failed to enqueue payment for tx ${txHash}:`, error) + } } }, onerror: (error: any) => { @@ -236,7 +260,36 @@ async function startWorker() { // Streaming usually tries to reconnect automatically, but we log it. } }) - + + // Payment processing worker + const paymentWorker = new Worker('payment-processing', async job => { + try { + const paymentRecord = job.data + // Restore transaction method if it was lost during serialization + if (typeof paymentRecord.transaction !== 'function') { + const tx = await server.transactions().transaction(paymentRecord.transaction_hash) + paymentRecord.transaction = async () => tx + } + console.log(`[WORKER] Processing payment job ${job.id} (tx: ${paymentRecord.transaction_hash})`) + await processPayment(paymentRecord) + console.log(`[WORKER] Payment job ${job.id} completed successfully`) + } catch (error) { + console.error(`[WORKER ERROR] Payment job ${job.id} failed:`, error) + throw error + } + }, { + connection: redisConnection, + concurrency: 5 + }) + + paymentWorker.on('failed', (job, err) => { + console.error(`[WORKER] Job ${job?.id} failed after ${job?.attemptsMade} attempts: ${err.message}`) + }) + + paymentWorker.on('completed', job => { + console.log(`[WORKER] Job ${job.id} completed`) + }) + // Heartbeat setInterval(() => { console.log(`[WORKER HEARTBEAT] ${new Date().toISOString()} - Monitoring ${PLATFORM_ESCROW_ACCOUNT}...`)