pikasTech-HWLAB/internal/harnessrl/temporal.ts

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

import type { HarnessRLTemporalGateway } from "./contracts.ts";

export function createHarnessRLTemporalGateway(options: { address: string; namespace: string; taskQueue: string; activity: Record<string, number> }): HarnessRLTemporalGateway {
  if (!options.address) throw codedError("harnessrl_temporal_address_required", "HARNESSRL_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;
  };
  return {
    async start(input) {
      const temporalClient = await getClient();
      try {
        const handle = await temporalClient.workflow.start("caseRunWorkflow", {
          taskQueue: options.taskQueue,
          workflowId: input.workflowId,
          args: [{ runId: input.runId, activity: options.activity }]
        });
        return { workflowRunId: handle.firstExecutionRunId, reused: false };
      } catch (error: any) {
        if (error?.name !== "WorkflowExecutionAlreadyStartedError") throw error;
        const description = await temporalClient.workflow.getHandle(input.workflowId).describe();
        return { workflowRunId: description.runId, reused: true };
      }
    },
    async requestCancel(workflowId) {
      await (await getClient()).workflow.getHandle(workflowId).signal("cancelCaseRun");
    },
    async close() {
      await connection?.close();
      connection = undefined;
      client = undefined;
    }
  };
}

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