diff --git a/src/core/minions/queue.ts b/src/core/minions/queue.ts index 004aa2fe0..359756530 100644 --- a/src/core/minions/queue.ts +++ b/src/core/minions/queue.ts @@ -9,13 +9,28 @@ */ import type { BrainEngine } from '../engine.ts'; -import type { MinionJob, MinionJobInput, MinionJobStatus, InboxMessage, TokenUpdate } from './types.ts'; -import { rowToMinionJob, rowToInboxMessage } from './types.ts'; +import type { + MinionJob, MinionJobInput, MinionJobStatus, InboxMessage, TokenUpdate, + MinionQueueOpts, ChildDoneMessage, Attachment, AttachmentInput, +} from './types.ts'; +import { rowToMinionJob, rowToInboxMessage, rowToAttachment } from './types.ts'; +import { validateAttachment } from './attachments.ts'; -const MIGRATION_VERSION = 6; +const MIGRATION_VERSION = 7; + +const DEFAULT_MAX_SPAWN_DEPTH = 5; +const DEFAULT_MAX_ATTACHMENT_BYTES = 5 * 1024 * 1024; // 5 MiB + +const TERMINAL_STATUSES = ['completed', 'failed', 'dead', 'cancelled'] as const; export class MinionQueue { - constructor(private engine: BrainEngine) {} + readonly maxSpawnDepth: number; + readonly maxAttachmentBytes: number; + + constructor(private engine: BrainEngine, opts: MinionQueueOpts = {}) { + this.maxSpawnDepth = opts.maxSpawnDepth ?? DEFAULT_MAX_SPAWN_DEPTH; + this.maxAttachmentBytes = opts.maxAttachmentBytes ?? DEFAULT_MAX_ATTACHMENT_BYTES; + } /** Verify minion_jobs table exists (migration v5+). Call before first operation. */ async ensureSchema(): Promise { @@ -28,25 +43,89 @@ export class MinionQueue { } } - /** Submit a new job. Returns the created job record. */ + /** + * Submit a new job. + * + * Wrapped in engine.transaction(): when parent_job_id is set, takes + * SELECT ... FOR UPDATE on the parent so concurrent submissions serialize + * on the cap check. Without this, two concurrent submissions could both + * see count = N-1 and both insert, blowing max_children. + * + * Child status is 'waiting' (or 'delayed') — claimable. Parent is flipped + * to 'waiting-children' atomically. Idempotency_key dedups via PG unique + * partial index; same key returns the existing row (no second insert). + */ async add(name: string, data?: Record, opts?: Partial): Promise { if (!name || name.trim().length === 0) { throw new Error('Job name cannot be empty'); } await this.ensureSchema(); - const status: MinionJobStatus = opts?.delay ? 'delayed' : (opts?.parent_job_id ? 'waiting-children' : 'waiting'); + const childStatus: MinionJobStatus = opts?.delay ? 'delayed' : 'waiting'; const delayUntil = opts?.delay ? new Date(Date.now() + opts.delay) : null; + const maxSpawnDepth = opts?.max_spawn_depth ?? this.maxSpawnDepth; - const rows = await this.engine.executeRaw>( - `INSERT INTO minion_jobs (name, queue, status, priority, data, max_attempts, backoff_type, - backoff_delay, backoff_jitter, delay_until, parent_job_id, on_child_fail) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) - RETURNING *`, - [ + return this.engine.transaction(async (tx) => { + // 1. Idempotency fast path — if a row already exists for this key, return it + // without doing any other work. The unique partial index guarantees + // no second row can be inserted with the same non-null key. + if (opts?.idempotency_key) { + const existing = await tx.executeRaw>( + `SELECT * FROM minion_jobs WHERE idempotency_key = $1`, + [opts.idempotency_key] + ); + if (existing.length > 0) return rowToMinionJob(existing[0]); + } + + // 2. Parent lock + depth/cap validation + let depth = 0; + if (opts?.parent_job_id) { + const parentRows = await tx.executeRaw>( + `SELECT * FROM minion_jobs WHERE id = $1 FOR UPDATE`, + [opts.parent_job_id] + ); + if (parentRows.length === 0) { + throw new Error(`parent_job_id ${opts.parent_job_id} not found`); + } + const parent = rowToMinionJob(parentRows[0]); + + depth = parent.depth + 1; + if (depth > maxSpawnDepth) { + throw new Error(`spawn depth ${depth} exceeds maxSpawnDepth ${maxSpawnDepth}`); + } + + if (parent.max_children !== null) { + const countRows = await tx.executeRaw<{ count: string }>( + `SELECT count(*)::text as count FROM minion_jobs + WHERE parent_job_id = $1 AND status NOT IN ('completed','failed','dead','cancelled')`, + [opts.parent_job_id] + ); + const live = parseInt(countRows[0]?.count ?? '0', 10); + if (live >= parent.max_children) { + throw new Error(`parent ${opts.parent_job_id} already has ${live} live children (max_children=${parent.max_children})`); + } + } + } + + // 3. Insert child. Use ON CONFLICT for idempotency; if a concurrent submit + // raced past the fast-path SELECT, the unique index catches it here. + const insertSql = opts?.idempotency_key + ? `INSERT INTO minion_jobs (name, queue, status, priority, data, max_attempts, backoff_type, + backoff_delay, backoff_jitter, delay_until, parent_job_id, on_child_fail, + depth, max_children, timeout_ms, remove_on_complete, remove_on_fail, idempotency_key) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18) + ON CONFLICT (idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING + RETURNING *` + : `INSERT INTO minion_jobs (name, queue, status, priority, data, max_attempts, backoff_type, + backoff_delay, backoff_jitter, delay_until, parent_job_id, on_child_fail, + depth, max_children, timeout_ms, remove_on_complete, remove_on_fail, idempotency_key) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18) + RETURNING *`; + + const params = [ name.trim(), opts?.queue ?? 'default', - status, + childStatus, opts?.priority ?? 0, JSON.stringify(data ?? {}), opts?.max_attempts ?? 3, @@ -56,9 +135,43 @@ export class MinionQueue { delayUntil?.toISOString() ?? null, opts?.parent_job_id ?? null, opts?.on_child_fail ?? 'fail_parent', - ] - ); - return rowToMinionJob(rows[0]); + depth, + opts?.max_children ?? null, + opts?.timeout_ms ?? null, + opts?.remove_on_complete ?? false, + opts?.remove_on_fail ?? false, + opts?.idempotency_key ?? null, + ]; + + const inserted = await tx.executeRaw>(insertSql, params); + + // ON CONFLICT DO NOTHING returns 0 rows — fall back to SELECT to fetch the + // existing row that won the race. + if (inserted.length === 0 && opts?.idempotency_key) { + const existing = await tx.executeRaw>( + `SELECT * FROM minion_jobs WHERE idempotency_key = $1`, + [opts.idempotency_key] + ); + if (existing.length === 0) { + throw new Error(`idempotency_key ${opts.idempotency_key} insert returned no row and no existing row found`); + } + return rowToMinionJob(existing[0]); + } + + const child = rowToMinionJob(inserted[0]); + + // 4. Flip parent to waiting-children if this is a fresh child insert. + // Only transition from non-terminal, non-already-waiting-children states. + if (opts?.parent_job_id) { + await tx.executeRaw( + `UPDATE minion_jobs SET status = 'waiting-children', updated_at = now() + WHERE id = $1 AND status IN ('waiting','active','delayed')`, + [opts.parent_job_id] + ); + } + + return child; + }); } /** Get a job by ID. Returns null if not found. */ @@ -115,16 +228,45 @@ export class MinionQueue { return rows.length > 0; } - /** Cancel a waiting, active, or delayed job. */ + /** + * Cancel a job and cascade-kill all descendants in one statement. + * + * Honest scope: this is BullMQ-style best-effort cancel. The recursive CTE + * snapshots the parent_job_id chain at statement start. A descendant + * re-parented BEFORE the cancel call is excluded; one re-parented DURING + * the call may still get cancelled (cancel wins if seen in the snapshot). + * Re-parented descendants whose parent_job_id is NULL'd by + * removeChildDependency naturally fall out of the recursive walk. + * + * Active descendants get lock_token = NULL — same path pause uses, so the + * worker's renewLock will fail next tick and AbortController fires. + * + * Returns the *root* (the job matching id), not an arbitrary descendant. + */ async cancelJob(id: number): Promise { const rows = await this.engine.executeRaw>( - `UPDATE minion_jobs SET status = 'cancelled', lock_token = NULL, lock_until = NULL, - finished_at = now(), updated_at = now() - WHERE id = $1 AND status IN ('waiting', 'active', 'delayed', 'waiting-children', 'paused') + `WITH RECURSIVE descendants AS ( + SELECT id, 0 AS d FROM minion_jobs WHERE id = $1 + UNION ALL + SELECT m.id, descendants.d + 1 + FROM minion_jobs m + JOIN descendants ON m.parent_job_id = descendants.id + WHERE descendants.d < 100 + ) + UPDATE minion_jobs SET + status = 'cancelled', + lock_token = NULL, + lock_until = NULL, + finished_at = now(), + updated_at = now() + WHERE id IN (SELECT id FROM descendants) + AND status IN ('waiting','active','delayed','waiting-children','paused') RETURNING *`, [id] ); - return rows.length > 0 ? rowToMinionJob(rows[0]) : null; + if (rows.length === 0) return null; + const root = rows.find(r => (r.id as number) === id); + return root ? rowToMinionJob(root) : null; } /** Re-queue a failed or dead job for retry. */ @@ -210,7 +352,12 @@ export class MinionQueue { }; } - /** Claim the next waiting job for a worker. Token-fenced, filters by registered names. */ + /** + * Claim the next waiting job for a worker. Token-fenced, filters by registered names. + * + * Sets timeout_at = now() + timeout_ms when the job has a per-job deadline, + * so handleTimeouts() can dead-letter expired jobs without rereading timeout_ms. + */ async claim(lockToken: string, lockDurationMs: number, queue: string, registeredNames: string[]): Promise { if (registeredNames.length === 0) return null; @@ -219,6 +366,9 @@ export class MinionQueue { status = 'active', lock_token = $1, lock_until = now() + ($2::double precision * interval '1 millisecond'), + timeout_at = CASE WHEN timeout_ms IS NOT NULL + THEN now() + (timeout_ms::double precision * interval '1 millisecond') + ELSE NULL END, attempts_started = attempts_started + 1, started_at = COALESCE(started_at, now()), updated_at = now() @@ -235,36 +385,155 @@ export class MinionQueue { return rows.length > 0 ? rowToMinionJob(rows[0]) : null; } - /** Complete a job (token-fenced). Rolls up token counts to parent if applicable. Returns null if token mismatch. */ - async completeJob(id: number, lockToken: string, result?: Record): Promise { + /** + * Dead-letter active jobs whose timeout_at has passed. + * + * The lock_until > now() guard is critical: a stalled job (lock_until < now) + * is being requeued by handleStalled, NOT timed out terminally. Stall → + * retry, timeout → dead. Order in worker loop: handleStalled() before + * handleTimeouts() to give stall recovery first crack. + * + * Honest scope: 1-tick TOCTOU window remains. A job whose lock_until + * expires between handleStalled and handleTimeouts may miss this tick + * but will be caught the next one (after re-claim). Never double-handled. + */ + async handleTimeouts(): Promise { const rows = await this.engine.executeRaw>( - `UPDATE minion_jobs SET status = 'completed', result = $1, - finished_at = now(), lock_token = NULL, lock_until = NULL, updated_at = now() - WHERE id = $2 AND status = 'active' AND lock_token = $3 - RETURNING *`, - [result ? JSON.stringify(result) : null, id, lockToken] + `UPDATE minion_jobs SET + status = 'dead', + error_text = 'timeout exceeded', + lock_token = NULL, + lock_until = NULL, + finished_at = now(), + updated_at = now() + WHERE status = 'active' + AND timeout_at IS NOT NULL + AND timeout_at < now() + AND lock_until > now() + RETURNING *` ); - if (rows.length === 0) return null; - - const completed = rowToMinionJob(rows[0]); - - // Roll up token counts to parent (if parent exists and is not in a terminal state) - if (completed.parent_job_id && (completed.tokens_input > 0 || completed.tokens_output > 0 || completed.tokens_cache_read > 0)) { - await this.engine.executeRaw( - `UPDATE minion_jobs SET - tokens_input = tokens_input + $1, - tokens_output = tokens_output + $2, - tokens_cache_read = tokens_cache_read + $3, - updated_at = now() - WHERE id = $4 AND status NOT IN ('completed', 'dead', 'cancelled')`, - [completed.tokens_input, completed.tokens_output, completed.tokens_cache_read, completed.parent_job_id] - ); - } - - return completed; + return rows.map(rowToMinionJob); } - /** Fail a job (token-fenced). Sets delayed for retry or dead/failed for terminal. */ + /** + * Complete a job (token-fenced). All side effects atomic in one transaction: + * 1. UPDATE child to 'completed' with result + * 2. Roll up token counts to parent (skipped if parent is terminal) + * 3. Insert child_done message into parent's inbox (skipped if parent terminal) + * 4. Resolve parent (flip waiting-children → waiting if all kids done) + * 5. If remove_on_complete, DELETE the child row (cascades inbox + attachments) + * + * Returns the completed job (the in-memory snapshot before any delete), or + * null if the lock_token mismatched (e.g., reclaimed mid-completion). + * + * The fold-in of resolveParent eliminates the crash window where a process + * died between completeJob and worker's prior post-call resolveParent, + * stranding the parent in waiting-children forever. + */ + async completeJob(id: number, lockToken: string, result?: Record): Promise { + return this.engine.transaction(async (tx) => { + // Peek at parent_job_id before the UPDATE so we can lock the parent row + // FIRST. Without this SELECT FOR UPDATE, two siblings completing + // concurrently each see the other as still active (pre-commit snapshot + // under read-committed), neither flips the parent, and the parent is + // stuck in waiting-children forever. + const peek = await tx.executeRaw<{ parent_job_id: number | null }>( + `SELECT parent_job_id FROM minion_jobs WHERE id = $1`, + [id] + ); + const parentId = peek[0]?.parent_job_id ?? null; + if (parentId) { + await tx.executeRaw( + `SELECT id FROM minion_jobs WHERE id = $1 FOR UPDATE`, + [parentId] + ); + } + + const rows = await tx.executeRaw>( + `UPDATE minion_jobs SET status = 'completed', result = $1, + finished_at = now(), lock_token = NULL, lock_until = NULL, updated_at = now() + WHERE id = $2 AND status = 'active' AND lock_token = $3 + RETURNING *`, + [result ? JSON.stringify(result) : null, id, lockToken] + ); + if (rows.length === 0) return null; + + const completed = rowToMinionJob(rows[0]); + + if (completed.parent_job_id) { + // Roll up token counts. Guarded against parent already being terminal. + if (completed.tokens_input > 0 || completed.tokens_output > 0 || completed.tokens_cache_read > 0) { + await tx.executeRaw( + `UPDATE minion_jobs SET + tokens_input = tokens_input + $1, + tokens_output = tokens_output + $2, + tokens_cache_read = tokens_cache_read + $3, + updated_at = now() + WHERE id = $4 AND status NOT IN ('completed', 'failed', 'dead', 'cancelled')`, + [completed.tokens_input, completed.tokens_output, completed.tokens_cache_read, completed.parent_job_id] + ); + } + + // Auto-post child_done into parent's inbox. EXISTS guard skips if parent + // was deleted or hit a terminal state mid-flight (no FK violation, no + // contradiction with the token rollup guard). + const childDone: ChildDoneMessage = { + type: 'child_done', + child_id: completed.id, + job_name: completed.name, + result: result ?? null, + }; + await tx.executeRaw( + `INSERT INTO minion_inbox (job_id, sender, payload) + SELECT $1, 'minions', $2::jsonb + WHERE EXISTS ( + SELECT 1 FROM minion_jobs + WHERE id = $1 AND status NOT IN ('completed','failed','dead','cancelled') + )`, + [completed.parent_job_id, childDone] + ); + + // Fold-in resolveParent: flip parent to waiting once all children done. + await tx.executeRaw( + `UPDATE minion_jobs SET status = 'waiting', updated_at = now() + WHERE id = $1 AND status = 'waiting-children' + AND NOT EXISTS ( + SELECT 1 FROM minion_jobs + WHERE parent_job_id = $1 + AND status NOT IN ('completed', 'dead', 'cancelled') + )`, + [completed.parent_job_id] + ); + } + + // remove_on_complete cleanup AFTER all parent-side bookkeeping. + // The child_done we just inserted lives in the *parent's* inbox row, + // so it survives the child cascade-delete. + if (completed.remove_on_complete) { + await tx.executeRaw( + `DELETE FROM minion_jobs WHERE id = $1`, + [completed.id] + ); + } + + return completed; + }); + } + + /** + * Fail a job (token-fenced). All side effects atomic in one transaction: + * 1. UPDATE child to 'delayed' (retry) | 'failed' | 'dead' + * 2. If terminal AND parent_job_id, run on_child_fail policy: + * - 'fail_parent' → mark parent 'failed' (via failParent SQL) + * - 'remove_dep' → null out parent_job_id (via removeChildDependency SQL) + * - 'ignore' / 'continue' → no parent action + * 3. If remove_on_fail AND terminal, DELETE the child row (parent hook + * already ran in this txn using in-memory state, so child deletion is safe) + * + * Folding the parent hook into this transaction eliminates the crash window + * where a process died between failJob and worker's prior post-call hook, + * leaving the parent stuck in waiting-children. + */ async failJob( id: number, lockToken: string, @@ -272,18 +541,76 @@ export class MinionQueue { newStatus: 'delayed' | 'failed' | 'dead', backoffMs?: number ): Promise { - const rows = await this.engine.executeRaw>( - `UPDATE minion_jobs SET - status = $1, error_text = $2, attempts_made = attempts_made + 1, - stacktrace = COALESCE(stacktrace, '[]'::jsonb) || to_jsonb($3::text), - delay_until = CASE WHEN $1 = 'delayed' THEN now() + ($4::double precision * interval '1 millisecond') ELSE NULL END, - finished_at = CASE WHEN $1 IN ('failed', 'dead') THEN now() ELSE NULL END, - lock_token = NULL, lock_until = NULL, updated_at = now() - WHERE id = $5 AND status = 'active' AND lock_token = $6 - RETURNING *`, - [newStatus, errorText, errorText, backoffMs ?? 0, id, lockToken] - ); - return rows.length > 0 ? rowToMinionJob(rows[0]) : null; + return this.engine.transaction(async (tx) => { + // Lock the parent row first so concurrent sibling completions/failures + // serialize on the parent — same race fix as completeJob. + const peek = await tx.executeRaw<{ parent_job_id: number | null }>( + `SELECT parent_job_id FROM minion_jobs WHERE id = $1`, + [id] + ); + const parentId = peek[0]?.parent_job_id ?? null; + if (parentId) { + await tx.executeRaw( + `SELECT id FROM minion_jobs WHERE id = $1 FOR UPDATE`, + [parentId] + ); + } + + const rows = await tx.executeRaw>( + `UPDATE minion_jobs SET + status = $1, error_text = $2, attempts_made = attempts_made + 1, + stacktrace = COALESCE(stacktrace, '[]'::jsonb) || to_jsonb($3::text), + delay_until = CASE WHEN $1 = 'delayed' THEN now() + ($4::double precision * interval '1 millisecond') ELSE NULL END, + finished_at = CASE WHEN $1 IN ('failed', 'dead') THEN now() ELSE NULL END, + lock_token = NULL, lock_until = NULL, updated_at = now() + WHERE id = $5 AND status = 'active' AND lock_token = $6 + RETURNING *`, + [newStatus, errorText, errorText, backoffMs ?? 0, id, lockToken] + ); + if (rows.length === 0) return null; + + const failed = rowToMinionJob(rows[0]); + const terminal = newStatus === 'failed' || newStatus === 'dead'; + + // Parent hook on terminal failure. + if (terminal && failed.parent_job_id) { + if (failed.on_child_fail === 'fail_parent') { + await tx.executeRaw( + `UPDATE minion_jobs SET status = 'failed', + error_text = $1, finished_at = now(), updated_at = now() + WHERE id = $2 AND status = 'waiting-children'`, + [`child job ${failed.id} failed: ${errorText}`, failed.parent_job_id] + ); + } else if (failed.on_child_fail === 'remove_dep') { + await tx.executeRaw( + `UPDATE minion_jobs SET parent_job_id = NULL, updated_at = now() WHERE id = $1`, + [failed.id] + ); + // After dropping the dep, try to resolve the parent if all OTHER kids are done. + await tx.executeRaw( + `UPDATE minion_jobs SET status = 'waiting', updated_at = now() + WHERE id = $1 AND status = 'waiting-children' + AND NOT EXISTS ( + SELECT 1 FROM minion_jobs + WHERE parent_job_id = $1 + AND status NOT IN ('completed', 'dead', 'cancelled') + )`, + [failed.parent_job_id] + ); + } + // 'ignore' / 'continue' → parent stays in waiting-children waiting on siblings + } + + // remove_on_fail cleanup AFTER parent hook. + if (terminal && failed.remove_on_fail) { + await tx.executeRaw( + `DELETE FROM minion_jobs WHERE id = $1`, + [failed.id] + ); + } + + return failed; + }); } /** Update job progress (token-fenced). */ @@ -292,7 +619,7 @@ export class MinionQueue { `UPDATE minion_jobs SET progress = $1, updated_at = now() WHERE id = $2 AND status = 'active' AND lock_token = $3 RETURNING id`, - [JSON.stringify(progress), id, lockToken] + [progress, id, lockToken] ); return rows.length > 0; } @@ -425,7 +752,7 @@ export class MinionQueue { `INSERT INTO minion_inbox (job_id, sender, payload) VALUES ($1, $2, $3) RETURNING *`, - [jobId, sender, JSON.stringify(payload)] + [jobId, sender, payload] ); return rows.length > 0 ? rowToInboxMessage(rows[0]) : null; } @@ -490,4 +817,137 @@ export class MinionQueue { [childId] ); } + + /** + * Read child_done messages from a parent's inbox. Token-fenced (the parent + * job must currently hold lockToken — same fence as readInbox to prevent a + * stale process polling completions for jobs it no longer owns). + * + * Does NOT mark messages read (parent may want to poll repeatedly with a + * cursor). Use `since` to fetch only newer entries. + */ + async readChildCompletions( + parentId: number, + lockToken: string, + opts?: { since?: Date } + ): Promise { + // Verify the caller holds the parent's lock. + const lockCheck = await this.engine.executeRaw<{ id: number }>( + `SELECT id FROM minion_jobs WHERE id = $1 AND lock_token = $2 AND status = 'active'`, + [parentId, lockToken] + ); + if (lockCheck.length === 0) return []; + + const params: unknown[] = [parentId]; + let sinceClause = ''; + if (opts?.since) { + sinceClause = ` AND sent_at > $2`; + params.push(opts.since.toISOString()); + } + + const rows = await this.engine.executeRaw>( + `SELECT payload FROM minion_inbox + WHERE job_id = $1 AND (payload->>'type') = 'child_done'${sinceClause} + ORDER BY sent_at ASC`, + params + ); + + return rows.map(r => { + const p = typeof r.payload === 'string' ? JSON.parse(r.payload) : r.payload; + return p as ChildDoneMessage; + }); + } + + /** + * Attach a file to a job. Validates size, base64, filename safety, and + * duplicate filename. Returns the persisted attachment metadata (not the + * bytes — use getAttachment to fetch). + * + * The DB UNIQUE (job_id, filename) constraint is the authoritative duplicate + * fence; the in-memory check just gives a faster error. + */ + async addAttachment(jobId: number, input: AttachmentInput): Promise { + await this.ensureSchema(); + + // Verify job exists (FK guarantees this on insert too, but explicit error is clearer) + const exists = await this.engine.executeRaw<{ id: number }>( + `SELECT id FROM minion_jobs WHERE id = $1`, + [jobId] + ); + if (exists.length === 0) { + throw new Error(`job ${jobId} not found`); + } + + const existingRows = await this.engine.executeRaw<{ filename: string }>( + `SELECT filename FROM minion_attachments WHERE job_id = $1`, + [jobId] + ); + const existingFilenames = new Set(existingRows.map(r => r.filename)); + + const result = validateAttachment(input, { + maxBytes: this.maxAttachmentBytes, + existingFilenames, + }); + if (!result.ok) { + throw new Error(`attachment validation failed: ${result.error}`); + } + const { filename, content_type, bytes, size_bytes, sha256 } = result.normalized; + + const rows = await this.engine.executeRaw>( + `INSERT INTO minion_attachments (job_id, filename, content_type, content, size_bytes, sha256) + VALUES ($1, $2, $3, $4, $5, $6) + RETURNING id, job_id, filename, content_type, storage_uri, size_bytes, sha256, created_at`, + [jobId, filename, content_type, bytes, size_bytes, sha256] + ); + return rowToAttachment(rows[0]); + } + + /** List attachments for a job (metadata only, no bytes). */ + async listAttachments(jobId: number): Promise { + const rows = await this.engine.executeRaw>( + `SELECT id, job_id, filename, content_type, storage_uri, size_bytes, sha256, created_at + FROM minion_attachments + WHERE job_id = $1 + ORDER BY created_at ASC, id ASC`, + [jobId] + ); + return rows.map(rowToAttachment); + } + + /** + * Fetch a single attachment with bytes. Returns null if not found. + * The bytes are returned as a Buffer (Uint8Array under the hood). + */ + async getAttachment(jobId: number, filename: string): Promise<{ meta: Attachment; bytes: Buffer } | null> { + const rows = await this.engine.executeRaw>( + `SELECT id, job_id, filename, content_type, storage_uri, size_bytes, sha256, created_at, content + FROM minion_attachments + WHERE job_id = $1 AND filename = $2`, + [jobId, filename] + ); + if (rows.length === 0) return null; + const row = rows[0]; + const meta = rowToAttachment(row); + const raw = row.content; + let bytes: Buffer; + if (raw == null) { + bytes = Buffer.alloc(0); + } else if (Buffer.isBuffer(raw)) { + bytes = raw; + } else if (raw instanceof Uint8Array) { + bytes = Buffer.from(raw); + } else { + bytes = Buffer.from(raw as ArrayBuffer); + } + return { meta, bytes }; + } + + /** Delete an attachment by job + filename. Returns true if a row was removed. */ + async deleteAttachment(jobId: number, filename: string): Promise { + const rows = await this.engine.executeRaw<{ id: number }>( + `DELETE FROM minion_attachments WHERE job_id = $1 AND filename = $2 RETURNING id`, + [jobId, filename] + ); + return rows.length > 0; + } }