qipai/backend/src/tasks/task-repository.ts

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<ResultSetHeader>(
      `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<Array<RowDataPacket & { id: string }>>(
      `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<AsyncTask | null> {
    const connection = await this.pool.getConnection();
    try {
      await connection.beginTransaction();
      await this.recoverExpiredLease(connection);
      const [rows] = await connection.execute<TaskRow[]>(
        `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<boolean> {
    const [result] = await this.pool.execute<ResultSetHeader>(
      `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<TaskStatus> {
    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<ResultSetHeader>(
      `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<void> {
    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);
}