Restructure external clients and add comments

StephanEwen · igalshilman · commit b67fcac8a15a · 2024-02-12T15:31:59.000+01:00
diff --git a/examples/workflow_example.ts b/examples/workflow_example.ts
@@ -1,5 +1,4 @@
 import * as restate from "../src/public_api";
-import * as restate_clients from "../src/clients/workflow_client";
 import { randomUUID } from "crypto";
 
 /* eslint-disable no-console */
@@ -67,24 +66,34 @@ const workflowApi = myworkflow.api;
 restate.createServer().bind(myworkflow).listen(9080);
 
 //
-// (2) Code to nteract with the workflow using an external client
+// (2) Code to interact with the workflow using an external client
 //
 // This submits a workflow and sends signals / queries to the workflow.
 //
-
 async function startWorkflowAndInteract(restateUrl: string) {
-  const restate = restate_clients.connectRestate(restateUrl);
+  const restateServer = restate.clients.connect(restateUrl);
 
   const args = { name: "Restatearius" };
   const workflowId = randomUUID();
 
   // Option a) we can create clients either with just the workflow service path
-  await restate.submitWorkflow("acme.myworkflow", workflowId, args);
+  const submit1 = await restateServer.submitWorkflow(
+    "acme.myworkflow",
+    workflowId,
+    args
+  );
+  console.log("Submitted workflow with result: " + submit1.status);
 
   // Option b) we can supply the API signature and get a typed interface for all the methods
   // Because the submit is idempotent, this call here will effectively attach to the
   // previous workflow
-  const client = await restate.submitWorkflow(workflowApi, workflowId, args);
+  const submit2 = await restateServer.submitWorkflow(
+    workflowApi,
+    workflowId,
+    args
+  );
+  console.log("Submitted workflow with result: " + submit2.status);
+  const client = submit2.client;
 
   // check the status (should be RUNNING)
   const status = await client.status();
diff --git a/src/clients/workflow_client.ts b/src/clients/workflow_client.ts
@@ -10,60 +10,173 @@
  */
 
 import * as restate from "../public_api";
-import {
-  LifecycleStatus,
-  WorkflowConnectedSignature,
-  WorkflowExternalSignature,
-  WorkflowRequest,
-} from "../workflows/workflow";
+import { ensureError } from "../types/errors";
 
-/* eslint-disable no-console */
+/**
+ * A client to interact with running workflows.
+ */
+export interface WorkflowClient<R, U> {
+  /**
+   * Gets the ID of the workflow that this client talks to.
+   */
+  workflowId(): string;
+
+  /**
+   * Gets the status of the workflow, as a {@link restate.workflow.LifecycleStatus}.
+   * This will take on the values "NOT_STARTED", "RUNNING", "FINISHED", "FAILED".
+   */
+  status(): Promise<restate.workflow.LifecycleStatus>;
+
+  /**
+   * Returns a promise completed with the result. This will resolve successfully on successful
+   * termination of the workflow, and will be rejected if the workflow throws an Error.
+   */
+  result(): Promise<R>;
+
+  /**
+   * Gets the interface to the workflow through which all the workflow's additional methods
+   * can be called.
+   *
+   * To get the proper typed client, use the {@link WorkflowConnection.submitWorkflow} or
+   * {@link WorkflowConnection.connectToWorkflow} functions that accpet a typed ServiceApi
+   * object, as in the example below.
+   *
+   * @example
+   * In the workflow definition:
+   * ```
+   * const myWorkflow = restate.workflow.workflow("acme.myworkflow", { ... });
+   * export const myWorkflowApi = myworkflow.api;
+   * ```
+   * In the client code:
+   * ```
+   * import { myWorkflowApi } from "../server/myWorkflow"
+   * ...
+   * const restate = connectWorkflows("https://restatehost:8080");
+   * restate.submitWorkflow(myWorkflowApi, workflowId, args);
+   * restate.connectToWorkflow(myWorkflowApi, workflowId);
+   * ```
+   */
+  workflowInterface(): restate.Client<
+    restate.workflow.WorkflowConnectedSignature<U>
+  >;
+}
 
-export interface Restate {
+/**
+ * A connection to Restate that let's you submit workflows or connect to workflows.
+ * This is a typed client that internally makes HTTP calls to Restate to launch trigger
+ * an execution of a workflow service, or to connect to an existing execution.
+ */
+export interface RestateClient {
   submitWorkflow<R, T>(
     path: string,
     workflowId: string,
     params: T
-  ): Promise<WorkflowClient<R, unknown>>;
+  ): Promise<{
+    status: restate.workflow.WorkflowStartResult;
+    client: WorkflowClient<R, unknown>;
+  }>;
 
   submitWorkflow<R, T, U>(
-    workflowApi: restate.ServiceApi<WorkflowExternalSignature<R, T, U>>,
+    workflowApi: restate.ServiceApi<
+      restate.workflow.WorkflowExternalSignature<R, T, U>
+    >,
     workflowId: string,
     params: T
-  ): Promise<WorkflowClient<R, U>>;
+  ): Promise<{
+    status: restate.workflow.WorkflowStartResult;
+    client: WorkflowClient<R, U>;
+  }>;
 
   connectToWorkflow<R = unknown>(
     path: string,
     workflowId: string
-  ): Promise<WorkflowClient<R, unknown>>;
+  ): Promise<{
+    status: restate.workflow.LifecycleStatus;
+    client: WorkflowClient<R, unknown>;
+  }>;
 
   connectToWorkflow<R, T, U>(
-    workflowApi: restate.ServiceApi<WorkflowExternalSignature<R, T, U>>,
+    workflowApi: restate.ServiceApi<
+      restate.workflow.WorkflowExternalSignature<R, T, U>
+    >,
     workflowId: string
-  ): Promise<WorkflowClient<R, U>>;
+  ): Promise<{
+    status: restate.workflow.LifecycleStatus;
+    client: WorkflowClient<R, U>;
+  }>;
 }
 
-export interface WorkflowClient<R, U> {
-  workflowId(): string;
-  status(): Promise<LifecycleStatus>; // RUNNING / FINISHED / FAILED
-
-  result(): Promise<R>;
-
-  workflowInterface(): restate.Client<WorkflowConnectedSignature<U>>; // call methods on workflow
-
-  // latestMessage(): Promise<StatusMessage>;
+/**
+ * Creates a typed client to start and interact with workflow executions.
+ * The specifiec URI must point to the Restate request endpoint (ingress).
+ *
+ * This function doesn't immediately verify the connection, it will not fail
+ * if Restate is unreachable. Connection failures will only manifest when
+ * attempting to submit or connect a specific workflow.
+ */
+export function connect(restateUri: string): RestateClient {
+  return {
+    submitWorkflow: async <R, T, U>(
+      pathOrApi:
+        | string
+        | restate.ServiceApi<
+            restate.workflow.WorkflowExternalSignature<R, T, U>
+          >,
+      workflowId: string,
+      params: T
+    ): Promise<{
+      status: restate.workflow.WorkflowStartResult;
+      client: WorkflowClient<R, U>;
+    }> => {
+      const path = typeof pathOrApi === "string" ? pathOrApi : pathOrApi.path;
+
+      let result: restate.workflow.WorkflowStartResult;
+      try {
+        result = await makeCall(restateUri, path, "start", workflowId, params);
+      } catch (err) {
+        const error = ensureError(err);
+        throw new Error("Cannot start workflow: " + error.message, {
+          cause: error,
+        });
+      }
 
-  // getMessages(
-  //   fromSeqNum: number
-  // ): AsyncGenerator<StatusMessage, void, undefined>;
-}
+      return {
+        status: result,
+        client: new WorkflowClientImpl(restateUri, path, workflowId),
+      };
+    },
 
-export function connectRestate(uri: string) {
-  return new RestateImpl(uri);
+    async connectToWorkflow<R, T, U>(
+      pathOrApi:
+        | string
+        | restate.ServiceApi<
+            restate.workflow.WorkflowExternalSignature<R, T, U>
+          >,
+      workflowId: string
+    ): Promise<{
+      status: restate.workflow.LifecycleStatus;
+      client: WorkflowClient<R, U>;
+    }> {
+      const path = typeof pathOrApi === "string" ? pathOrApi : pathOrApi.path;
+      const client: WorkflowClient<R, U> = new WorkflowClientImpl(
+        restateUri,
+        path,
+        workflowId
+      );
+      const status = await client.status();
+      if (status === restate.workflow.LifecycleStatus.NOT_STARTED) {
+        throw new Error(
+          "No workflow running/finished/failed with ID " + workflowId
+        );
+      }
+      return {
+        status,
+        client: new WorkflowClientImpl(restateUri, path, workflowId),
+      };
+    },
+  } satisfies RestateClient;
 }
 
-// ------------------------------ implementation ------------------------------
-
 class WorkflowClientImpl<R, U> implements WorkflowClient<R, U> {
   constructor(
     private readonly restateUri: string,
@@ -75,15 +188,17 @@ class WorkflowClientImpl<R, U> implements WorkflowClient<R, U> {
     return this.wfId;
   }
 
-  status(): Promise<LifecycleStatus> {
+  status(): Promise<restate.workflow.LifecycleStatus> {
     return this.makeCall("status", {});
   }
 
   result(): Promise<R> {
     return this.makeCall("waitForResult", {});
   }
 
-  workflowInterface(): restate.Client<WorkflowConnectedSignature<U>> {
+  workflowInterface(): restate.Client<
+    restate.workflow.WorkflowConnectedSignature<U>
+  > {
     const clientProxy = new Proxy(
       {},
       {
@@ -96,25 +211,11 @@ class WorkflowClientImpl<R, U> implements WorkflowClient<R, U> {
       }
     );
 
-    return clientProxy as restate.Client<WorkflowConnectedSignature<U>>;
+    return clientProxy as restate.Client<
+      restate.workflow.WorkflowConnectedSignature<U>
+    >;
   }
 
-  // latestMessage(): Promise<StatusMessage> {
-  //   return this.makeCall("getLatestMessage", {});
-  // }
-
-  // async *getMessages(fromSeqNum: number) {
-  //   while (true) {
-  //     const msgs: StatusMessage[] = await this.makeCall("pollNextMessages", {
-  //       from: fromSeqNum,
-  //     });
-  //     for (const msg of msgs) {
-  //       yield msg;
-  //     }
-  //     fromSeqNum += msgs.length;
-  //   }
-  // }
-
   private async makeCall<RR, TT>(method: string, args: TT): Promise<RR> {
     return await makeCall(
       this.restateUri,
@@ -126,36 +227,6 @@ class WorkflowClientImpl<R, U> implements WorkflowClient<R, U> {
   }
 }
 
-class RestateImpl implements Restate {
-  constructor(private readonly restateUri: string) {}
-
-  async submitWorkflow<R, T, U>(
-    pathOrApi: string | restate.ServiceApi<WorkflowExternalSignature<R, T, U>>,
-    workflowId: string,
-    params: T
-  ): Promise<WorkflowClient<R, U>> {
-    const path = typeof pathOrApi === "string" ? pathOrApi : pathOrApi.path;
-    const response = await makeCall(
-      this.restateUri,
-      path,
-      "start",
-      workflowId,
-      params
-    );
-    console.log("Start() call completed: Workflow is " + response);
-
-    return new WorkflowClientImpl(this.restateUri, path, workflowId);
-  }
-
-  async connectToWorkflow<R, T, U>(
-    pathOrApi: string | restate.ServiceApi<WorkflowExternalSignature<R, T, U>>,
-    workflowId: string
-  ): Promise<WorkflowClient<R, U>> {
-    const path = typeof pathOrApi === "string" ? pathOrApi : pathOrApi.path;
-    return new WorkflowClientImpl(this.restateUri, path, workflowId);
-  }
-}
-
 // ----------------------------------------------------------------------------
 //                                 Utils
 // ----------------------------------------------------------------------------
@@ -167,7 +238,7 @@ async function makeCall<R, T>(
   workflowId: string,
   params: T
 ): Promise<R> {
-  if (typeof workflowId !== "string" || workflowId.length === 0) {
+  if (!workflowId || typeof workflowId !== "string") {
     throw new Error("missing workflowId");
   }
   if (params === undefined) {
@@ -179,7 +250,10 @@ async function makeCall<R, T>(
 
   const url = `${restateUri}/${serviceName}/${method}`;
   const data = {
-    request: { workflowId, ...params } satisfies WorkflowRequest<T>,
+    request: {
+      workflowId,
+      ...params,
+    } satisfies restate.workflow.WorkflowRequest<T>,
   };
 
   let body: string;
@@ -189,7 +263,8 @@ async function makeCall<R, T>(
     throw new Error("Cannot encode request: " + err, { cause: err });
   }
 
-  console.log(`Making call to Restate workflow at ${url} with ${body}`);
+  // eslint-disable-next-line no-console
+  console.debug(`Making call to Restate at ${url}`);
 
   const httpResponse = await fetch(url, {
     method: "POST",
diff --git a/src/public_api.ts b/src/public_api.ts
@@ -48,3 +48,4 @@ export {
   RestateConnectionOptions,
 } from "./embedded/api";
 export * as workflow from "./workflows/workflow";
+export * as clients from "./clients/workflow_client";
