pikasTech-HWLAB/internal/workbench/temporal.ts

import { Client, Connection } from "@temporalio/client";

import type { WorkbenchActivityConfig, WorkbenchTemporalGateway, WorkbenchTurnInput } from "./contracts.ts";

export function createWorkbenchTemporalGateway(options: { address: string; namespace: string; taskQueue: string; activity: WorkbenchActivityConfig }): WorkbenchTemporalGateway {
  if (!options.address) throw codedError("workbench_temporal_address_required", "WORKBENCH_TEMPORAL_ADDRESS is required");
  let connection: Connection | undefined;
  let client: Client | undefined;
  const getClient = async () => {
    connection ??= await Connection.connect({ address: options.address });
    client ??= new Client({ connection, namespace: options.namespace });
    return client;
  };
  const start = async (workflowType: string, workflowId: string, input: unknown) => {
    const temporal = await getClient();
    try {
      const handle = await temporal.workflow.start(workflowType, { taskQueue: options.taskQueue, workflowId, args: [input] });
      return { workflowId, workflowRunId: handle.firstExecutionRunId, reused: false };
    } catch (error: any) {
      if (error?.name !== "WorkflowExecutionAlreadyStartedError") throw error;
      const description = await temporal.workflow.getHandle(workflowId).describe();
      return { workflowId, workflowRunId: description.runId, reused: true };
    }
  };
  return {
    submit(input: WorkbenchTurnInput) {
      return start("workbenchTurnWorkflow", `workbench-turn-${input.traceId}`, { ...input, activity: options.activity });
    },
    cancel(input) {
      return start("workbenchCancelWorkflow", `workbench-cancel-${input.traceId}`, { ...input, activity: options.activity });
    },
    async close() {
      await connection?.close();
      connection = undefined;
      client = undefined;
    }
  };
}

function codedError(code: string, message: string) {
  return Object.assign(new Error(message), { code });
}