Files
pikasTech-HWLAB/internal/tasktree/dispatcher.ts
T

182 lines
10 KiB
TypeScript

import { Client, Connection } from "@temporalio/client";
import type { TaskItem, TaskTreeCommand, TaskTreeResult } from "./contracts.ts";
import type { TaskTreeBackupService } from "./backup.ts";
import { parseMdtodoImport } from "./mdtodo-import.ts";
import { numberTaskTree, type NumberedTaskItem } from "./outline-ref.ts";
import { TaskTreeStore } from "./store.ts";
export type TaskTreeDispatcherOptions = {
store: TaskTreeStore;
backup?: TaskTreeBackupService;
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(),
...(options.backup ? { backup: await options.backup.status() } : {}),
};
else if (operation === "group.list") data = { groups: await store.listGroups() };
else if (operation === "group.overview") data = { groups: await store.groupOverview() };
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: requiredMutation(await store.deleteGroup(command.groupId), "group_not_found", "taskgroup was not found"),
mutation: true
};
else if (operation === "group.stats") {
required(await store.getGroup(command.groupId), "group_not_found", "taskgroup was not found");
data = await store.groupStats(command.groupId);
}
else if (operation === "group.import-markdown") {
const plan = parseMdtodoImport(command);
data = command.dryRun ? { dryRun: true, ...plan, tasks: undefined } : await store.importMdtodo(plan);
}
else if (operation === "task.list") {
required(await store.getGroup(command.groupId), "group_not_found", "taskgroup was not found");
data = { tasks: taskTree(numberTaskTree(await store.listTasks(command.groupId))) };
}
else if (operation === "task.get") {
const task = required(await store.getTask(command.taskId), "task_not_found", "task was not found");
const tasks = numberTaskTree(await store.listTasks(task.groupId));
const numberedTask = required(tasks.find((item) => item.id === task.id) ?? null, "task_not_found", "task was not found");
data = { ...taskNode(numberedTask, tasks), reports: await store.listReports(task.id) };
}
else if (operation === "task.create") {
const startAt = optionalDate(command.startAt, "startAt");
const dueAt = optionalDate(command.dueAt, "dueAt");
validTimeRange(startAt, dueAt);
data = await store.createTask({
...command,
title: validTitle(command.title),
startAt,
dueAt
});
}
else if (operation === "task.create-batch") {
const titles = command.titles.map(validTitle);
if (titles.length < 1) throw codedError("invalid_input", "at least one title is required");
const startAt = optionalDate(command.startAt, "startAt");
const dueAt = optionalDate(command.dueAt, "dueAt");
validTimeRange(startAt, dueAt);
data = { tasks: await store.createTasks({
...command,
titles,
startAt,
dueAt
}) };
}
else if (operation === "task.update") {
if (command.status === "completed") throw codedError("completion_command_required", "use task complete so the execution report requirement is enforced");
data = required(await store.updateTask(command.taskId, {
...command,
title: command.title === undefined ? undefined : validTitle(command.title),
status: command.status === undefined ? undefined : validStatus(command.status),
startAt: nullableDate(command.startAt, "startAt"),
dueAt: nullableDate(command.dueAt, "dueAt")
}), "task_not_found", "task was not found");
}
else if (operation === "task.delete") data = {
deleted: requiredMutation(await store.deleteTask(command.taskId), "task_not_found", "task was not found"),
mutation: true
};
else if (operation === "task.start") data = required(await store.updateTask(command.taskId, { status: "in_progress" }), "task_not_found", "task was not found");
else if (operation === "task.complete") data = required(await store.completeTask(command.taskId), "task_not_found", "task was not found");
else if (operation === "milestone.create") data = await store.createMilestone({ ...command, title: requiredText(command.title, "title"), occursAt: validDate(command.occursAt, "occursAt") });
else if (operation === "report.list") {
required(await store.getTask(command.taskId), "task_not_found", "task was not found");
data = { reports: await store.listReports(command.taskId) };
}
else if (operation === "report.get") data = required(await store.getReport(command.reportId), "report_not_found", "execution report was not found");
else if (operation === "report.write") data = await store.writeReport({ ...command, title: requiredText(command.title, "title"), body: requiredText(command.body, "body") });
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") {
const timeline = required(await store.timeline(command.groupId), "group_not_found", "taskgroup was not found");
data = { ...timeline, tasks: numberTaskTree(timeline.tasks) };
}
else if (operation === "backup.create") data = await requiredBackup(options.backup).create({
dryRun: command.dryRun,
reason: command.reason,
});
else if (operation === "backup.status") data = await requiredBackup(options.backup).status();
else if (operation === "backup.verify") data = await requiredBackup(options.backup).verify();
else if (operation === "backup.restore") data = await requiredBackup(options.backup).restore({
confirm: command.confirm === true,
expectedSha256: command.expectedSha256,
});
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) {
const migrationDetail = error?.tasktreeMigrationId
? { migrationId: String(error.tasktreeMigrationId), sqlState: error?.code ? String(error.code) : undefined }
: undefined;
return {
ok: false,
operation: command.operation,
error: { code: error?.code ?? "tasktree_error", message: error?.message ?? String(error), ...(migrationDetail ? { detail: migrationDetail } : {}) },
};
}
};
}
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 requiredBackup(value: TaskTreeBackupService | undefined): TaskTreeBackupService {
if (!value) throw codedError("backup_unavailable", "TaskTree backup service is unavailable");
return value;
}
function requiredMutation(value: boolean, code: string, message: string): true { if (!value) throw codedError(code, message); return true; }
function requiredText(value: unknown, field: string): string { const text = String(value ?? "").trim(); if (!text) throw codedError("invalid_input", `${field} is required`); return text; }
function validTitle(value: unknown): string {
const title = requiredText(value, "title");
if (/[\r\n\u0000]/u.test(title)) throw codedError("invalid_title", "title must be non-empty single-line text");
return title;
}
function validStatus(value: unknown): "pending" | "in_progress" | "completed" | "blocked" {
if (value === "pending" || value === "in_progress" || value === "completed" || value === "blocked") return value;
throw codedError("invalid_status", "status must be pending, in_progress, completed, or blocked");
}
function optionalDate(value: string | undefined, field: string): string | undefined { return value === undefined ? undefined : validDate(value, field); }
function nullableDate(value: string | null | undefined, field: string): string | null | undefined { return value === null || value === undefined ? value : validDate(value, field); }
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 }); }
function validTimeRange(startAt?: string, dueAt?: string) {
if (startAt !== undefined && dueAt !== undefined && new Date(startAt).valueOf() > new Date(dueAt).valueOf()) {
throw codedError("invalid_time_range", "startAt must be before or equal to dueAt");
}
}
type TaskNode = NumberedTaskItem & { children: TaskNode[] };
function taskTree(tasks: NumberedTaskItem[]): TaskNode[] {
return tasks.filter((task) => task.parentId === null).map((task) => taskNode(task, tasks));
}
function taskNode(task: NumberedTaskItem, tasks: NumberedTaskItem[]): TaskNode {
return {
...task,
children: tasks.filter((candidate) => candidate.parentId === task.id).map((child) => taskNode(child, tasks))
};
}