root
56 lines
3.9 KiB
TypeScript
56 lines
3.9 KiB
TypeScript
import { Client, Connection } from "@temporalio/client";
|
|
|
|
import type { TaskTreeCommand, TaskTreeResult } from "./contracts.ts";
|
|
import { TaskTreeStore } from "./store.ts";
|
|
|
|
export type TaskTreeDispatcherOptions = {
|
|
store: TaskTreeStore;
|
|
temporalAddress?: string;
|
|
temporalNamespace?: string;
|
|
taskQueue?: string;
|
|
};
|
|
|
|
export function createTaskTreeDispatcher(options: TaskTreeDispatcherOptions) {
|
|
const { store } = options;
|
|
return async function dispatch(command: TaskTreeCommand): Promise<TaskTreeResult> {
|
|
try {
|
|
const operation = command.operation;
|
|
let data: unknown;
|
|
if (operation === "health") data = await store.health();
|
|
else if (operation === "group.list") data = { groups: await store.listGroups() };
|
|
else if (operation === "group.get") data = required(await store.getGroup(command.groupId), "group_not_found", "taskgroup was not found");
|
|
else if (operation === "group.create") data = await store.createGroup(requiredText(command.name, "name"), command.description);
|
|
else if (operation === "group.delete") data = { deleted: await store.deleteGroup(command.groupId) };
|
|
else if (operation === "task.create") data = await store.createTask({ ...command, title: requiredText(command.title, "title") });
|
|
else if (operation === "task.update") data = required(await store.updateTask(command.taskId, command), "task_not_found", "task was not found");
|
|
else if (operation === "task.delete") data = { deleted: await store.deleteTask(command.taskId) };
|
|
else if (operation === "milestone.create") data = await store.createMilestone({ ...command, title: requiredText(command.title, "title"), occursAt: validDate(command.occursAt, "occursAt") });
|
|
else if (operation === "report.create") data = await store.createReport({ ...command, title: requiredText(command.title, "title"), body: requiredText(command.body, "body") });
|
|
else if (operation === "timeline.get") data = required(await store.timeline(command.groupId), "group_not_found", "taskgroup was not found");
|
|
else if (operation === "workflow.start") data = await startTaskWorkflow(options, command.taskId);
|
|
else throw codedError("unsupported_operation", `unsupported operation: ${String(operation)}`);
|
|
return { ok: true, operation, data };
|
|
} catch (error: any) {
|
|
return { ok: false, operation: command.operation, error: { code: error?.code ?? "tasktree_error", message: error?.message ?? String(error) } };
|
|
}
|
|
};
|
|
}
|
|
|
|
async function startTaskWorkflow(options: TaskTreeDispatcherOptions, taskId: string) {
|
|
if (!options.temporalAddress) throw codedError("temporal_address_required", "Temporal address is required for workflow.start");
|
|
const connection = await Connection.connect({ address: options.temporalAddress });
|
|
try {
|
|
const client = new Client({ connection, namespace: options.temporalNamespace ?? "unidesk" });
|
|
const workflowId = `tasktree-${taskId}-${Date.now()}`;
|
|
const handle = await client.workflow.start("taskExecutionWorkflow", { taskQueue: options.taskQueue ?? "hwlab-v03-tasktree", workflowId, args: [{ taskId }] });
|
|
return { workflowId, runId: handle.firstExecutionRunId, taskId, result: await handle.result() };
|
|
} finally {
|
|
await connection.close();
|
|
}
|
|
}
|
|
|
|
function required<T>(value: T | null, code: string, message: string): T { if (value === null) throw codedError(code, message); return value; }
|
|
function requiredText(value: unknown, field: string): string { const text = String(value ?? "").trim(); if (!text) throw codedError("invalid_input", `${field} is required`); return text; }
|
|
function validDate(value: unknown, field: string): string { const date = new Date(String(value ?? "")); if (Number.isNaN(date.valueOf())) throw codedError("invalid_input", `${field} must be an ISO date`); return date.toISOString(); }
|
|
function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }
|