import { randomUUID } from 'node:crypto'; import { getRun, start } from 'workflow/api'; import type { WorkflowRunStatus } from '@workflow/world'; import type { Task, TaskStatus, TaskTimeline, TaskType } from '@/lib/types'; import { runTaskWorkflow } from '@/app/workflows/task-runner'; import { describeTaskFailure } from '@/lib/server/task-errors'; import { countTasksByStatus, createTaskRunRecord, findInFlightTaskByResourceKey, getTaskByIdForUser, listTaskStageEventsForTask, listRecentTasksForUser, markTaskFailure, setTaskStatusFromWorkflow, setTaskWorkflowRunId, updateTaskNotificationState } from '@/lib/server/repos/tasks'; import { buildTaskNotification } from '@/lib/server/task-notifications'; type EnqueueTaskInput = { userId: string; taskType: TaskType; payload?: Record; priority?: number; maxAttempts?: number; resourceKey?: string; }; type UpdateTaskNotificationInput = { read?: boolean; silenced?: boolean; }; function mapWorkflowStatus(status: WorkflowRunStatus): TaskStatus { switch (status) { case 'pending': return 'queued'; case 'running': return 'running'; case 'completed': return 'completed'; case 'failed': case 'cancelled': return 'failed'; default: return 'failed'; } } function isProjectionPendingSync(task: Task) { return task.status === 'queued' || task.status === 'running'; } async function reconcileTaskWithWorkflow(task: Task) { if (!task.workflow_run_id || !isProjectionPendingSync(task)) { return task; } try { const run = getRun(task.workflow_run_id); const workflowStatus = await run.status; const nextStatus = mapWorkflowStatus(workflowStatus); if (nextStatus === task.status) { return task; } const failure = nextStatus === 'failed' ? describeTaskFailure(task, workflowStatus === 'cancelled' ? 'Workflow run cancelled' : 'Workflow run failed') : null; const updated = await setTaskStatusFromWorkflow( task.id, nextStatus, failure?.detail ?? null, failure?.summary ?? null ); const fallbackTask = { ...task, status: nextStatus, stage: nextStatus, stage_detail: nextStatus === 'failed' ? (failure?.summary ?? 'The background workflow stopped unexpectedly.') : 'Workflow run completed.', stage_context: null, error: failure?.detail ?? null, finished_at: nextStatus === 'queued' || nextStatus === 'running' ? null : task.finished_at ?? new Date().toISOString() } satisfies Omit; return updated ?? { ...fallbackTask, notification: buildTaskNotification(fallbackTask) }; } catch { return task; } } export async function enqueueTask(input: EnqueueTaskInput) { const task = await createTaskRunRecord({ id: randomUUID(), user_id: input.userId, task_type: input.taskType, payload: input.payload ?? {}, priority: input.priority ?? 50, max_attempts: input.maxAttempts ?? 3, resource_key: input.resourceKey ?? null }); try { const run = await start(runTaskWorkflow, [task.id]); await setTaskWorkflowRunId(task.id, run.runId); return { ...task, workflow_run_id: run.runId } satisfies Task; } catch (error) { const failure = describeTaskFailure(task, 'Failed to start workflow'); await markTaskFailure(task.id, failure.detail, 'failed', { detail: failure.summary }); const wrapped = new Error(failure.detail); (wrapped as Error & { cause?: unknown }).cause = error; throw wrapped; } } export async function findInFlightTask(userId: string, taskType: TaskType, resourceKey: string) { const task = await findInFlightTaskByResourceKey(userId, taskType, resourceKey); if (!task) { return null; } return await reconcileTaskWithWorkflow(task); } export async function getTaskById(taskId: string, userId: string) { const task = await getTaskByIdForUser(taskId, userId); if (!task) { return null; } return await reconcileTaskWithWorkflow(task); } export async function listRecentTasks(userId: string, limit = 20, statuses?: TaskStatus[]) { const tasks = await listRecentTasksForUser(userId, limit, statuses); return await Promise.all(tasks.map((task) => reconcileTaskWithWorkflow(task))); } export async function updateTaskNotification( userId: string, taskId: string, input: UpdateTaskNotificationInput ) { const task = await updateTaskNotificationState(taskId, userId, input); if (!task) { return null; } return await reconcileTaskWithWorkflow(task); } export async function getTaskTimeline(taskId: string, userId: string): Promise { const task = await getTaskById(taskId, userId); if (!task) { return null; } const events = await listTaskStageEventsForTask(taskId, userId); return { task, events }; } export async function getTaskQueueSnapshot() { return await countTasksByStatus(); }