diff --git a/backend/package.json b/backend/package.json index 536889d..c990e23 100644 --- a/backend/package.json +++ b/backend/package.json @@ -11,12 +11,13 @@ "dev": "tsx watch src/server.ts", "build": "tsc -p tsconfig.json", "start": "node dist/server.js", + "start:worker": "node dist/tasks/worker.js", "db:migrate:plan": "npm run build && node dist/db/migrate-cli.js plan", "db:migrate:up": "npm run build && node dist/db/migrate-cli.js up", "db:migrate:verify": "npm run build && node dist/db/migrate-cli.js verify", "db:migrate:down": "npm run build && node dist/db/migrate-cli.js down", "test:mysql:migration": "npm run build && node tests/mysql-migration-roundtrip.test.mjs", - "test": "npm run build && node tests/backend-contract.test.mjs && node tests/migration-contract.test.mjs && node tests/mysql-pool-contract.test.mjs && node tests/migration-runner.test.mjs && node tests/legacy-money.test.mjs && node tests/legacy-read-repository.test.mjs" + "test": "npm run build && node tests/backend-contract.test.mjs && node tests/migration-contract.test.mjs && node tests/mysql-pool-contract.test.mjs && node tests/migration-runner.test.mjs && node tests/legacy-money.test.mjs && node tests/legacy-read-repository.test.mjs && node tests/task-repository.test.mjs" }, "dependencies": { "@fastify/cors": "^11.2.0", diff --git a/backend/src/db/migration-runner.ts b/backend/src/db/migration-runner.ts index 81919fb..c63cec3 100644 --- a/backend/src/db/migration-runner.ts +++ b/backend/src/db/migration-runner.ts @@ -19,10 +19,19 @@ export interface MigrationExecutionResult extends MigrationPlan { } const repoRoot = resolve(dirname(fileURLToPath(import.meta.url)), '../../..'); -const migrationFiles: Record = { - up: 'database/migrations/2026061601_m01b_core_schema.up.sql', - verify: 'database/migrations/2026061601_m01b_core_schema.verify.sql', - down: 'database/migrations/2026061601_m01b_core_schema.down.sql' +const migrationFiles: Record = { + up: [ + 'database/migrations/2026061601_m01b_core_schema.up.sql', + 'database/migrations/2026061802_m01c_async_tasks.up.sql' + ], + verify: [ + 'database/migrations/2026061601_m01b_core_schema.verify.sql', + 'database/migrations/2026061802_m01c_async_tasks.verify.sql' + ], + down: [ + 'database/migrations/2026061802_m01c_async_tasks.down.sql', + 'database/migrations/2026061601_m01b_core_schema.down.sql' + ] }; export function splitSqlStatements(sql: string): string[] { @@ -110,12 +119,15 @@ export function splitSqlStatements(sql: string): string[] { } export async function loadMigrationPlan(direction: MigrationDirection): Promise { - const relativeFile = migrationFiles[direction]; - const sql = await readFile(resolve(repoRoot, relativeFile), 'utf8'); + const relativeFiles = migrationFiles[direction]; + const sqlParts = await Promise.all( + relativeFiles.map((relativeFile) => readFile(resolve(repoRoot, relativeFile), 'utf8')) + ); + const sql = sqlParts.join('\n'); return { direction, - file: relativeFile, + file: relativeFiles.join(','), checksum: createHash('sha256').update(sql).digest('hex'), statements: splitSqlStatements(sql) }; @@ -134,7 +146,7 @@ export async function executeMigrationPlan( for (const [index, statement] of plan.statements.entries()) { const [result] = await pool.query(statement); if (plan.direction === 'verify') { - const minimumRows = [10, 26, 1][index] ?? 1; + const minimumRows = [10, 26, 1, 2, 5, 1][index] ?? 1; if (!Array.isArray(result) || result.length < minimumRows) { throw new Error( `Migration verification statement ${index + 1} returned fewer than ${minimumRows} rows.` diff --git a/backend/src/tasks/outbox-repository.ts b/backend/src/tasks/outbox-repository.ts new file mode 100644 index 0000000..79828a8 --- /dev/null +++ b/backend/src/tasks/outbox-repository.ts @@ -0,0 +1,69 @@ +import type { ResultSetHeader, RowDataPacket } from 'mysql2/promise'; +import type { MySqlPool } from '../db/mysql.js'; + +export interface AppendOutboxEventInput { + tenantId: string; + aggregateType: string; + aggregateId: string; + eventType: string; + idempotencyKey: string; + payload: unknown; + availableAt?: Date; +} + +type SqlExecutor = Pick; + +export class OutboxRepository { + constructor(private readonly executor: SqlExecutor) {} + + async append(input: AppendOutboxEventInput): Promise<{ id: string; created: boolean }> { + const [result] = await this.executor.execute( + `INSERT IGNORE INTO qipai_outbox_events + (tenant_id, aggregate_type, aggregate_id, event_type, + idempotency_key, payload, available_at) + VALUES (?, ?, ?, ?, ?, CAST(? AS JSON), ?)`, + [ + input.tenantId, + input.aggregateType, + input.aggregateId, + input.eventType, + input.idempotencyKey, + JSON.stringify(input.payload), + input.availableAt ?? new Date() + ] + ); + if (result.insertId > 0) { + return { id: String(result.insertId), created: true }; + } + const [rows] = await this.executor.execute>( + `SELECT id FROM qipai_outbox_events + WHERE tenant_id = ? AND idempotency_key = ?`, + [input.tenantId, input.idempotencyKey] + ); + if (!rows[0]) throw new Error('Idempotent outbox lookup failed after duplicate insert.'); + return { id: String(rows[0].id), created: false }; + } + + async markPublished(eventId: string): Promise { + const [result] = await this.executor.execute( + `UPDATE qipai_outbox_events + SET status = 'PUBLISHED', published_at = UTC_TIMESTAMP(3), + attempts = attempts + 1, last_error = NULL + WHERE id = ? AND status = 'PENDING'`, + [eventId] + ); + return result.affectedRows === 1; + } + + async markFailed(eventId: string, error: unknown, delayMs: number): Promise { + const message = error instanceof Error ? error.message : String(error); + const [result] = await this.executor.execute( + `UPDATE qipai_outbox_events + SET available_at = DATE_ADD(UTC_TIMESTAMP(3), INTERVAL ? MICROSECOND), + attempts = attempts + 1, last_error = ? + WHERE id = ? AND status = 'PENDING'`, + [delayMs * 1000, message.slice(0, 1000), eventId] + ); + return result.affectedRows === 1; + } +} diff --git a/backend/src/tasks/task-repository.ts b/backend/src/tasks/task-repository.ts new file mode 100644 index 0000000..a9f2fd2 --- /dev/null +++ b/backend/src/tasks/task-repository.ts @@ -0,0 +1,177 @@ +import type { PoolConnection, ResultSetHeader, RowDataPacket } from 'mysql2/promise'; +import type { MySqlPool } from '../db/mysql.js'; + +export const taskTypes = [ + 'notification.dispatch', + 'order.advance', + 'device.command', + 'refund.query', + 'statistics.aggregate', + 'outbox.publish' +] as const; + +export type TaskType = (typeof taskTypes)[number]; +export type TaskStatus = + | 'PENDING' + | 'RUNNING' + | 'RETRY' + | 'SUCCEEDED' + | 'FAILED' + | 'COMPENSATION_REQUIRED' + | 'CANCELLED'; + +export interface AsyncTask { + id: string; + tenantId: string; + taskType: TaskType; + idempotencyKey: string; + payload: unknown; + status: TaskStatus; + attempts: number; + maxAttempts: number; +} + +interface TaskRow extends RowDataPacket { + id: string; + tenantId: string; + taskType: TaskType; + idempotencyKey: string; + payload: string | object; + status: TaskStatus; + attempts: number; + maxAttempts: number; +} + +export interface EnqueueTaskInput { + tenantId: string; + taskType: TaskType; + idempotencyKey: string; + payload: unknown; + priority?: number; + availableAt?: Date; + maxAttempts?: number; +} + +export class TaskRepository { + constructor(private readonly pool: MySqlPool) {} + + async enqueue(input: EnqueueTaskInput): Promise<{ id: string; created: boolean }> { + const [result] = await this.pool.execute( + `INSERT IGNORE INTO qipai_async_tasks + (tenant_id, task_type, idempotency_key, payload, priority, available_at, max_attempts) + VALUES (?, ?, ?, CAST(? AS JSON), ?, ?, ?)`, + [ + input.tenantId, + input.taskType, + input.idempotencyKey, + JSON.stringify(input.payload), + input.priority ?? 0, + input.availableAt ?? new Date(), + input.maxAttempts ?? 8 + ] + ); + if (result.insertId > 0) { + return { id: String(result.insertId), created: true }; + } + const [rows] = await this.pool.execute>( + `SELECT id FROM qipai_async_tasks + WHERE tenant_id = ? AND task_type = ? AND idempotency_key = ?`, + [input.tenantId, input.taskType, input.idempotencyKey] + ); + if (!rows[0]) throw new Error('Idempotent task lookup failed after duplicate insert.'); + return { id: String(rows[0].id), created: false }; + } + + async claimNext(workerId: string, leaseMs: number): Promise { + const connection = await this.pool.getConnection(); + try { + await connection.beginTransaction(); + await this.recoverExpiredLease(connection); + const [rows] = await connection.execute( + `SELECT id, tenant_id AS tenantId, task_type AS taskType, + idempotency_key AS idempotencyKey, payload, status, + attempts, max_attempts AS maxAttempts + FROM qipai_async_tasks + WHERE status IN ('PENDING', 'RETRY') + AND available_at <= UTC_TIMESTAMP(3) + ORDER BY priority DESC, available_at ASC, id ASC + LIMIT 1 + FOR UPDATE SKIP LOCKED` + ); + const row = rows[0]; + if (!row) { + await connection.commit(); + return null; + } + + await connection.execute( + `UPDATE qipai_async_tasks + SET status = 'RUNNING', + lease_owner = ?, + lease_expires_at = DATE_ADD(UTC_TIMESTAMP(3), INTERVAL ? MICROSECOND), + attempts = attempts + 1, + last_error = NULL + WHERE id = ?`, + [workerId, leaseMs * 1000, row.id] + ); + await connection.commit(); + return { + ...row, + id: String(row.id), + tenantId: String(row.tenantId), + status: 'RUNNING', + attempts: row.attempts + 1, + payload: typeof row.payload === 'string' ? JSON.parse(row.payload) : row.payload + }; + } catch (error) { + await connection.rollback(); + throw error; + } finally { + connection.release(); + } + } + + async complete(taskId: string, workerId: string): Promise { + const [result] = await this.pool.execute( + `UPDATE qipai_async_tasks + SET status = 'SUCCEEDED', completed_at = UTC_TIMESTAMP(3), + lease_owner = NULL, lease_expires_at = NULL + WHERE id = ? AND status = 'RUNNING' AND lease_owner = ?`, + [taskId, workerId] + ); + return result.affectedRows === 1; + } + + async fail(task: AsyncTask, workerId: string, error: unknown): Promise { + const terminal = task.attempts >= task.maxAttempts; + const nextStatus: TaskStatus = terminal ? 'COMPENSATION_REQUIRED' : 'RETRY'; + const delayMs = retryDelayMs(task.attempts); + const message = error instanceof Error ? error.message : String(error); + const [result] = await this.pool.execute( + `UPDATE qipai_async_tasks + SET status = ?, available_at = DATE_ADD(UTC_TIMESTAMP(3), INTERVAL ? MICROSECOND), + lease_owner = NULL, lease_expires_at = NULL, last_error = ? + WHERE id = ? AND status = 'RUNNING' AND lease_owner = ?`, + [nextStatus, delayMs * 1000, message.slice(0, 1000), task.id, workerId] + ); + if (result.affectedRows !== 1) { + throw new Error(`Task ${task.id} lease was lost before failure could be recorded.`); + } + return nextStatus; + } + + private async recoverExpiredLease(connection: PoolConnection): Promise { + await connection.execute( + `UPDATE qipai_async_tasks + SET status = 'RETRY', lease_owner = NULL, lease_expires_at = NULL, + available_at = UTC_TIMESTAMP(3), + last_error = COALESCE(last_error, 'worker lease expired') + WHERE status = 'RUNNING' AND lease_expires_at < UTC_TIMESTAMP(3)` + ); + } +} + +export function retryDelayMs(attempts: number): number { + const exponent = Math.max(0, Math.min(attempts - 1, 20)); + return Math.min(60 * 60 * 1000, 1000 * 2 ** exponent); +} diff --git a/backend/src/tasks/worker.ts b/backend/src/tasks/worker.ts new file mode 100644 index 0000000..6e3144e --- /dev/null +++ b/backend/src/tasks/worker.ts @@ -0,0 +1,76 @@ +import { hostname } from 'node:os'; +import { randomUUID } from 'node:crypto'; +import { resolve } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import type { AsyncTask, TaskType } from './task-repository.js'; +import { TaskRepository } from './task-repository.js'; +import { closeMySqlPool, createMySqlPool } from '../db/mysql.js'; +import { loadConfig } from '../config.js'; + +type TaskHandler = (task: AsyncTask) => Promise; + +export interface WorkerOptions { + repository: TaskRepository; + handlers: ReadonlyMap; + workerId: string; + leaseMs?: number; +} + +export class TaskWorker { + private stopping = false; + + constructor(private readonly options: WorkerOptions) {} + + stop(): void { + this.stopping = true; + } + + async runOnce(): Promise { + const task = await this.options.repository.claimNext( + this.options.workerId, + this.options.leaseMs ?? 60_000 + ); + if (!task) return false; + + const handler = this.options.handlers.get(task.taskType); + try { + if (!handler) throw new Error(`No handler registered for task type ${task.taskType}.`); + await handler(task); + if (!(await this.options.repository.complete(task.id, this.options.workerId))) { + throw new Error(`Task ${task.id} lease was lost before completion.`); + } + } catch (error) { + await this.options.repository.fail(task, this.options.workerId, error); + } + return true; + } + + async run(pollIntervalMs = 1000): Promise { + while (!this.stopping) { + const processed = await this.runOnce(); + if (!processed) await sleep(pollIntervalMs); + } + } +} + +export function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +if (process.argv[1] && fileURLToPath(import.meta.url) === resolve(process.argv[1])) { + const config = loadConfig(); + const pool = createMySqlPool(config); + const worker = new TaskWorker({ + repository: new TaskRepository(pool), + handlers: new Map(), + workerId: `${hostname()}:${process.pid}:${randomUUID()}` + }); + const shutdown = () => worker.stop(); + process.on('SIGINT', shutdown); + process.on('SIGTERM', shutdown); + try { + await worker.run(); + } finally { + await closeMySqlPool(pool); + } +} diff --git a/backend/tests/migration-contract.test.mjs b/backend/tests/migration-contract.test.mjs index f98e937..7f6274c 100644 --- a/backend/tests/migration-contract.test.mjs +++ b/backend/tests/migration-contract.test.mjs @@ -12,6 +12,9 @@ const downSql = read('database/migrations/2026061601_m01b_core_schema.down.sql') const verifySql = read('database/migrations/2026061601_m01b_core_schema.verify.sql'); const seedSql = read('database/seeds/2026061601_m01b_minimal_seed.sql'); const legacyFixtureSql = read('database/fixtures/2026061801_m01b_legacy_schema.sql'); +const asyncUpSql = read('database/migrations/2026061802_m01c_async_tasks.up.sql'); +const asyncDownSql = read('database/migrations/2026061802_m01c_async_tasks.down.sql'); +const asyncVerifySql = read('database/migrations/2026061802_m01c_async_tasks.verify.sql'); const coreTables = [ 'qipai_schema_migrations', @@ -73,4 +76,14 @@ assert.match(legacyFixtureSql, /DECIMAL\(10,\s*2\)/); assert.match(legacyFixtureSql, /LEGACY-SANITIZED-001/); assert.doesNotMatch(legacyFixtureSql, /(?:https?:\/\/|@|password|secret|token)/i); -console.log('PASS: M01-B migration contract is present.'); +for (const table of ['qipai_outbox_events', 'qipai_async_tasks']) { + assert.match(asyncUpSql, new RegExp(`CREATE TABLE IF NOT EXISTS ${table}\\b`)); + assert.match(asyncDownSql, new RegExp(`DROP TABLE IF EXISTS ${table}\\b`)); + assert.match(asyncVerifySql, new RegExp(`'${table}'`)); +} +assert.match(asyncUpSql, /UNIQUE KEY uq_qipai_outbox_events_idempotency/); +assert.match(asyncUpSql, /UNIQUE KEY uq_qipai_async_tasks_idempotency/); +assert.match(asyncUpSql, /lease_expires_at DATETIME\(3\)/); +assert.match(asyncUpSql, /COMPENSATION_REQUIRED|status VARCHAR/); + +console.log('PASS: M01-B/M01-C migration contracts are present.'); diff --git a/backend/tests/migration-runner.test.mjs b/backend/tests/migration-runner.test.mjs index 1433ab7..5a43ae2 100644 --- a/backend/tests/migration-runner.test.mjs +++ b/backend/tests/migration-runner.test.mjs @@ -12,7 +12,8 @@ assert.deepEqual( const plan = await loadMigrationPlan('up'); assert.equal(plan.direction, 'up'); -assert.match(plan.file, /2026061601_m01b_core_schema\.up\.sql$/); +assert.match(plan.file, /2026061601_m01b_core_schema\.up\.sql/); +assert.match(plan.file, /2026061802_m01c_async_tasks\.up\.sql$/); assert.match(plan.checksum, /^[a-f0-9]{64}$/); assert.ok(plan.statements.length >= 11); diff --git a/backend/tests/mysql-migration-roundtrip.test.mjs b/backend/tests/mysql-migration-roundtrip.test.mjs index 6ed8847..2636b38 100644 --- a/backend/tests/mysql-migration-roundtrip.test.mjs +++ b/backend/tests/mysql-migration-roundtrip.test.mjs @@ -5,6 +5,7 @@ import { fileURLToPath } from 'node:url'; import { loadConfig } from '../dist/config.js'; import { closeMySqlPool, createMySqlPool } from '../dist/db/mysql.js'; import { LegacyReadRepository } from '../dist/db/legacy-read-repository.js'; +import { TaskRepository } from '../dist/tasks/task-repository.js'; import { executeMigrationPlan, loadMigrationPlan, @@ -12,11 +13,13 @@ import { } from '../dist/db/migration-runner.js'; const expectedTables = [ + 'qipai_async_tasks', 'qipai_audit_logs', 'qipai_devices', 'qipai_legacy_table_mappings', 'qipai_members', 'qipai_orders', + 'qipai_outbox_events', 'qipai_payments', 'qipai_rooms', 'qipai_schema_migrations', @@ -38,12 +41,13 @@ async function readCoreTables(pool) { return rows.map((row) => row.tableName); } -async function readMigrationVersion(pool) { +async function readMigrationVersions(pool) { const [rows] = await pool.query( `SELECT version, name FROM qipai_schema_migrations - WHERE version = ?`, - ['2026061601'] + WHERE version IN (?, ?) + ORDER BY version`, + ['2026061601', '2026061802'] ); return rows; } @@ -79,11 +83,48 @@ async function assertLegacyCompatibility(pool) { assert.equal(orders[0].refundAmountCents, null); } +async function assertTaskDurability(pool) { + await pool.query( + `INSERT INTO qipai_tenants (code, name) + VALUES ('M01C-TEST', 'M01C sanitized test tenant')` + ); + const firstRepository = new TaskRepository(pool); + const firstEnqueue = await firstRepository.enqueue({ + tenantId: '1', + taskType: 'order.advance', + idempotencyKey: 'order:501:paid', + payload: { orderId: '501' }, + maxAttempts: 3 + }); + const duplicateEnqueue = await firstRepository.enqueue({ + tenantId: '1', + taskType: 'order.advance', + idempotencyKey: 'order:501:paid', + payload: { orderId: '501' }, + maxAttempts: 3 + }); + assert.equal(firstEnqueue.created, true); + assert.equal(duplicateEnqueue.created, false); + assert.equal(duplicateEnqueue.id, firstEnqueue.id); + + const restartedRepository = new TaskRepository(pool); + const claimed = await restartedRepository.claimNext('worker-after-restart', 30_000); + assert.equal(claimed?.id, firstEnqueue.id); + assert.equal(claimed?.attempts, 1); + assert.equal(await restartedRepository.complete(claimed.id, 'worker-after-restart'), true); + + const [rows] = await pool.query( + `SELECT status, attempts FROM qipai_async_tasks WHERE id = ?`, + [firstEnqueue.id] + ); + assert.deepEqual(rows, [{ status: 'SUCCEEDED', attempts: 1 }]); +} + const config = loadConfig(); assert.equal(config.mysql.passwordConfigured, true, 'Live migration test requires a temporary password.'); assert.match( config.mysql.database, - /^qipai_m01b_test_[a-z0-9_]+$/, + /^qipai_m01c_test_[a-z0-9_]+$/, 'Live migration test refuses to use a non-temporary database.' ); @@ -102,25 +143,26 @@ try { await executeMigrationPlan(pool, plans.up); await executeMigrationPlan(pool, plans.verify); assert.deepEqual(await readCoreTables(pool), expectedTables); - assert.deepEqual(await readMigrationVersion(pool), [{ - version: '2026061601', - name: 'm01b_core_schema' - }]); + assert.deepEqual(await readMigrationVersions(pool), [ + { version: '2026061601', name: 'm01b_core_schema' }, + { version: '2026061802', name: 'm01c_async_tasks' } + ]); + await assertTaskDurability(pool); await assertLegacyCompatibility(pool); - console.log('PASS: first up and verify completed.'); + console.log('PASS: first up, verify and durable task restart check completed.'); await executeMigrationPlan(pool, plans.down); assert.deepEqual(await readCoreTables(pool), []); await assertLegacyCompatibility(pool); - console.log('PASS: down removed all M01-B core tables.'); + console.log('PASS: down removed all M01-B/M01-C tables.'); await executeMigrationPlan(pool, plans.up); await executeMigrationPlan(pool, plans.verify); assert.deepEqual(await readCoreTables(pool), expectedTables); - assert.deepEqual(await readMigrationVersion(pool), [{ - version: '2026061601', - name: 'm01b_core_schema' - }]); + assert.deepEqual(await readMigrationVersions(pool), [ + { version: '2026061601', name: 'm01b_core_schema' }, + { version: '2026061802', name: 'm01c_async_tasks' } + ]); await assertLegacyCompatibility(pool); console.log('PASS: second up and verify restored the schema.'); diff --git a/backend/tests/task-repository.test.mjs b/backend/tests/task-repository.test.mjs new file mode 100644 index 0000000..82654c6 --- /dev/null +++ b/backend/tests/task-repository.test.mjs @@ -0,0 +1,87 @@ +import assert from 'node:assert/strict'; +import { OutboxRepository } from '../dist/tasks/outbox-repository.js'; +import { retryDelayMs, taskTypes } from '../dist/tasks/task-repository.js'; +import { TaskWorker } from '../dist/tasks/worker.js'; + +assert.deepEqual(taskTypes, [ + 'notification.dispatch', + 'order.advance', + 'device.command', + 'refund.query', + 'statistics.aggregate', + 'outbox.publish' +]); +assert.equal(retryDelayMs(1), 1000); +assert.equal(retryDelayMs(4), 8000); +assert.equal(retryDelayMs(99), 60 * 60 * 1000); + +const outboxCalls = []; +const outbox = new OutboxRepository({ + async execute(sql, params) { + outboxCalls.push([sql, params]); + return [{ insertId: 7, affectedRows: 1 }, []]; + } +}); +assert.deepEqual( + await outbox.append({ + tenantId: '1', + aggregateType: 'order', + aggregateId: '88', + eventType: 'order.paid', + idempotencyKey: 'order:88:paid', + payload: { orderId: '88' } + }), + { id: '7', created: true } +); +assert.match(outboxCalls[0][0], /INSERT IGNORE INTO qipai_outbox_events/); + +const calls = []; +const task = { + id: '42', + tenantId: '1', + taskType: 'notification.dispatch', + idempotencyKey: 'notice:42', + payload: { channel: 'test' }, + status: 'RUNNING', + attempts: 1, + maxAttempts: 3 +}; +const repository = { + async claimNext() { + calls.push('claim'); + return task; + }, + async complete(id) { + calls.push(`complete:${id}`); + return true; + }, + async fail() { + calls.push('fail'); + } +}; +const worker = new TaskWorker({ + repository, + handlers: new Map([['notification.dispatch', async (claimed) => { + calls.push(`handle:${claimed.id}`); + }]]), + workerId: 'test-worker' +}); +assert.equal(await worker.runOnce(), true); +assert.deepEqual(calls, ['claim', 'handle:42', 'complete:42']); + +const failedCalls = []; +const failedWorker = new TaskWorker({ + repository: { + async claimNext() { return task; }, + async complete() { return true; }, + async fail(claimed, workerId, error) { + failedCalls.push([claimed.id, workerId, error.message]); + } + }, + handlers: new Map(), + workerId: 'test-worker' +}); +assert.equal(await failedWorker.runOnce(), true); +assert.deepEqual(failedCalls, [['42', 'test-worker', 'No handler registered for task type notification.dispatch.']]); + +console.log('PASS: M01-C task retry and worker contracts are present.'); diff --git a/database/migrations/2026061802_m01c_async_tasks.down.sql b/database/migrations/2026061802_m01c_async_tasks.down.sql new file mode 100644 index 0000000..090d46f --- /dev/null +++ b/database/migrations/2026061802_m01c_async_tasks.down.sql @@ -0,0 +1,5 @@ +-- Roll back M01-C asynchronous task tables. + +DELETE FROM qipai_schema_migrations WHERE version = '2026061802'; +DROP TABLE IF EXISTS qipai_async_tasks; +DROP TABLE IF EXISTS qipai_outbox_events; diff --git a/database/migrations/2026061802_m01c_async_tasks.up.sql b/database/migrations/2026061802_m01c_async_tasks.up.sql new file mode 100644 index 0000000..0173cbf --- /dev/null +++ b/database/migrations/2026061802_m01c_async_tasks.up.sql @@ -0,0 +1,48 @@ +-- M01-C lightweight asynchronous task foundation. +-- Tasks and outbox events are durable in MySQL and safe to resume after worker restarts. + +CREATE TABLE IF NOT EXISTS qipai_outbox_events ( + id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, + tenant_id BIGINT UNSIGNED NOT NULL, + aggregate_type VARCHAR(64) NOT NULL, + aggregate_id VARCHAR(64) NOT NULL, + event_type VARCHAR(128) NOT NULL, + idempotency_key VARCHAR(191) NOT NULL, + payload JSON NOT NULL, + status VARCHAR(32) NOT NULL DEFAULT 'PENDING', + available_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + published_at DATETIME(3) NULL, + attempts INT UNSIGNED NOT NULL DEFAULT 0, + last_error VARCHAR(1000) NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + CONSTRAINT fk_qipai_outbox_events_tenant FOREIGN KEY (tenant_id) REFERENCES qipai_tenants(id), + UNIQUE KEY uq_qipai_outbox_events_idempotency (tenant_id, idempotency_key), + KEY idx_qipai_outbox_events_dispatch (status, available_at, id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS qipai_async_tasks ( + id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, + tenant_id BIGINT UNSIGNED NOT NULL, + task_type VARCHAR(128) NOT NULL, + idempotency_key VARCHAR(191) NOT NULL, + payload JSON NOT NULL, + status VARCHAR(32) NOT NULL DEFAULT 'PENDING', + priority INT NOT NULL DEFAULT 0, + available_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + lease_owner VARCHAR(128) NULL, + lease_expires_at DATETIME(3) NULL, + attempts INT UNSIGNED NOT NULL DEFAULT 0, + max_attempts INT UNSIGNED NOT NULL DEFAULT 8, + last_error VARCHAR(1000) NULL, + completed_at DATETIME(3) NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + CONSTRAINT fk_qipai_async_tasks_tenant FOREIGN KEY (tenant_id) REFERENCES qipai_tenants(id), + UNIQUE KEY uq_qipai_async_tasks_idempotency (tenant_id, task_type, idempotency_key), + KEY idx_qipai_async_tasks_claim (status, available_at, priority, id), + KEY idx_qipai_async_tasks_lease (status, lease_expires_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +INSERT IGNORE INTO qipai_schema_migrations (version, name) +VALUES ('2026061802', 'm01c_async_tasks'); diff --git a/database/migrations/2026061802_m01c_async_tasks.verify.sql b/database/migrations/2026061802_m01c_async_tasks.verify.sql new file mode 100644 index 0000000..ccfc090 --- /dev/null +++ b/database/migrations/2026061802_m01c_async_tasks.verify.sql @@ -0,0 +1,29 @@ +-- Verify M01-C asynchronous task foundation. + +SELECT table_name +FROM information_schema.tables +WHERE table_schema = DATABASE() + AND table_name IN ('qipai_outbox_events', 'qipai_async_tasks') +ORDER BY table_name; + +SELECT table_name, index_name +FROM information_schema.statistics +WHERE table_schema = DATABASE() + AND ( + (table_name = 'qipai_outbox_events' AND index_name IN ( + 'uq_qipai_outbox_events_idempotency', + 'idx_qipai_outbox_events_dispatch' + )) + OR + (table_name = 'qipai_async_tasks' AND index_name IN ( + 'uq_qipai_async_tasks_idempotency', + 'idx_qipai_async_tasks_claim', + 'idx_qipai_async_tasks_lease' + )) + ) +GROUP BY table_name, index_name +ORDER BY table_name, index_name; + +SELECT version, name +FROM qipai_schema_migrations +WHERE version = '2026061802'; diff --git a/deploy/pm2/ecosystem.config.cjs b/deploy/pm2/ecosystem.config.cjs index af208a4..bf5eb47 100644 --- a/deploy/pm2/ecosystem.config.cjs +++ b/deploy/pm2/ecosystem.config.cjs @@ -11,6 +11,18 @@ module.exports = { QIPAI_API_HOST: '127.0.0.1', QIPAI_API_PORT: '3001' } + }, + { + name: 'qipai-worker', + cwd: '/opt/apps/qipai-backend/current/backend', + script: 'dist/tasks/worker.js', + instances: 1, + exec_mode: 'fork', + restart_delay: 3000, + kill_timeout: 10000, + env: { + NODE_ENV: 'production' + } } ] }; diff --git a/scripts/dev/windows/check-backend.ps1 b/scripts/dev/windows/check-backend.ps1 index 381cef5..9d8fbd0 100644 --- a/scripts/dev/windows/check-backend.ps1 +++ b/scripts/dev/windows/check-backend.ps1 @@ -11,6 +11,9 @@ $requiredFiles = @( "backend/src/db/migration-runner.ts", "backend/src/db/migrate-cli.ts", "backend/src/db/legacy-read-repository.ts", + "backend/src/tasks/task-repository.ts", + "backend/src/tasks/outbox-repository.ts", + "backend/src/tasks/worker.ts", "backend/src/routes/health.ts", "backend/src/server.ts", "backend/tests/backend-contract.test.mjs", @@ -19,10 +22,14 @@ $requiredFiles = @( "backend/tests/migration-runner.test.mjs", "backend/tests/mysql-migration-roundtrip.test.mjs", "backend/tests/legacy-read-repository.test.mjs", + "backend/tests/task-repository.test.mjs", "scripts/dev/wsl/mysql-migration-roundtrip.sh", "database/migrations/2026061601_m01b_core_schema.up.sql", "database/migrations/2026061601_m01b_core_schema.down.sql", "database/migrations/2026061601_m01b_core_schema.verify.sql", + "database/migrations/2026061802_m01c_async_tasks.up.sql", + "database/migrations/2026061802_m01c_async_tasks.down.sql", + "database/migrations/2026061802_m01c_async_tasks.verify.sql", "database/seeds/2026061601_m01b_minimal_seed.sql", "deploy/pm2/ecosystem.config.cjs" ) diff --git a/scripts/dev/wsl/mysql-migration-roundtrip.sh b/scripts/dev/wsl/mysql-migration-roundtrip.sh index ed7a259..824a534 100644 --- a/scripts/dev/wsl/mysql-migration-roundtrip.sh +++ b/scripts/dev/wsl/mysql-migration-roundtrip.sh @@ -33,7 +33,7 @@ fi mysql --batch --skip-column-names -e 'SELECT CURRENT_USER()' >/dev/null run_id="$(date -u +%Y%m%d%H%M%S)_$$_$(openssl rand -hex 3)" -database="qipai_m01b_test_${run_id}" +database="qipai_m01c_test_${run_id}" username="qmb_$$_$(openssl rand -hex 4)" password="$(openssl rand -hex 24)" created=0 @@ -63,7 +63,7 @@ if [[ "${1:-}" == "--force-failure" ]]; then fi if [[ "${1:-}" == "--cleanup-probe" ]]; then - before_database_count="$(mysql --batch --skip-column-names -e "SELECT COUNT(*) FROM information_schema.SCHEMATA WHERE SCHEMA_NAME LIKE 'qipai_m01b_test_%'")" + before_database_count="$(mysql --batch --skip-column-names -e "SELECT COUNT(*) FROM information_schema.SCHEMATA WHERE SCHEMA_NAME LIKE 'qipai_m01c_test_%'")" before_user_count="$(mysql --batch --skip-column-names -e "SELECT COUNT(*) FROM mysql.user WHERE User LIKE 'qmb_%' AND Host = '127.0.0.1'")" set +e bash "${BASH_SOURCE[0]}" --force-failure @@ -73,7 +73,7 @@ if [[ "${1:-}" == "--cleanup-probe" ]]; then echo "FAIL: cleanup probe did not produce the expected interrupted status." >&2 exit 1 } - after_database_count="$(mysql --batch --skip-column-names -e "SELECT COUNT(*) FROM information_schema.SCHEMATA WHERE SCHEMA_NAME LIKE 'qipai_m01b_test_%'")" + after_database_count="$(mysql --batch --skip-column-names -e "SELECT COUNT(*) FROM information_schema.SCHEMATA WHERE SCHEMA_NAME LIKE 'qipai_m01c_test_%'")" after_user_count="$(mysql --batch --skip-column-names -e "SELECT COUNT(*) FROM mysql.user WHERE User LIKE 'qmb_%' AND Host = '127.0.0.1'")" if [[ "${before_database_count}" != "${after_database_count}" || "${before_user_count}" != "${after_user_count}" ]]; then echo "FAIL: temporary MySQL resources survived an interrupted test." >&2 @@ -93,6 +93,6 @@ export QIPAI_MYSQL_USER="${username}" export QIPAI_MYSQL_PASSWORD="${password}" export QIPAI_MYSQL_CONNECTION_LIMIT=2 -echo "INFO: MySQL ${mysql_version}; running M01-B migration roundtrip in a temporary database." +echo "INFO: MySQL ${mysql_version}; running M01-B/M01-C migration roundtrip in a temporary database." npm --prefix backend run test:mysql:migration -echo "PASS: M01-B live MySQL migration roundtrip completed." +echo "PASS: M01-B/M01-C live MySQL migration roundtrip completed."