Add listWorkflowRuns op to Backend

Author: jamescmartinez

packages/backend-postgres/backend.test.ts

@@ -87,6 +87,84 @@ describe("BackendPostgres", () => {
     });
   });
 
+  describe("listWorkflowRuns()", () => {
+    test("lists workflow runs ordered by creation time", async () => {
+      const backend = await BackendPostgres.connect(DEFAULT_DATABASE_URL, {
+        namespaceId: randomUUID(),
+      });
+      const first = await createPendingWorkflowRun(backend);
+      await sleep(10); // ensure timestamp difference
+      const second = await createPendingWorkflowRun(backend);
+
+      const listed = await backend.listWorkflowRuns({});
+      expect(listed.data.map((run) => run.id)).toEqual([first.id, second.id]);
+      await backend.stop();
+    });
+
+    test("paginates workflow runs", async () => {
+      const backend = await BackendPostgres.connect(DEFAULT_DATABASE_URL, {
+        namespaceId: randomUUID(),
+      });
+      const runs: WorkflowRun[] = [];
+      for (let i = 0; i < 5; i++) {
+        runs.push(await createPendingWorkflowRun(backend));
+        await sleep(10);
+      }
+
+      // p1
+      const page1 = await backend.listWorkflowRuns({ limit: 2 });
+      expect(page1.data).toHaveLength(2);
+      expect(page1.data[0]?.id).toBe(runs[0]?.id);
+      expect(page1.data[1]?.id).toBe(runs[1]?.id);
+      expect(page1.pagination.next).not.toBeNull();
+      expect(page1.pagination.prev).toBeNull();
+
+      // p2
+      const page2 = await backend.listWorkflowRuns({
+        limit: 2,
+        after: page1.pagination.next!, // eslint-disable-line @typescript-eslint/no-non-null-assertion
+      });
+      expect(page2.data).toHaveLength(2);
+      expect(page2.data[0]?.id).toBe(runs[2]?.id);
+      expect(page2.data[1]?.id).toBe(runs[3]?.id);
+      expect(page2.pagination.next).not.toBeNull();
+      expect(page2.pagination.prev).not.toBeNull();
+
+      // p3
+      const page3 = await backend.listWorkflowRuns({
+        limit: 2,
+        after: page2.pagination.next!, // eslint-disable-line @typescript-eslint/no-non-null-assertion
+      });
+      expect(page3.data).toHaveLength(1);
+      expect(page3.data[0]?.id).toBe(runs[4]?.id);
+      expect(page3.pagination.next).toBeNull();
+      expect(page3.pagination.prev).not.toBeNull();
+
+      // p2 again
+      const page2Back = await backend.listWorkflowRuns({
+        limit: 2,
+        before: page3.pagination.prev!, // eslint-disable-line @typescript-eslint/no-non-null-assertion
+      });
+      expect(page2Back.data).toHaveLength(2);
+      expect(page2Back.data[0]?.id).toBe(runs[2]?.id);
+      expect(page2Back.data[1]?.id).toBe(runs[3]?.id);
+      expect(page2Back.pagination.next).toEqual(page2.pagination.next);
+      expect(page2Back.pagination.prev).toEqual(page2.pagination.prev);
+      await backend.stop();
+    });
+
+    test("handles empty results", async () => {
+      const backend = await BackendPostgres.connect(DEFAULT_DATABASE_URL, {
+        namespaceId: randomUUID(),
+      });
+      const listed = await backend.listWorkflowRuns({});
+      expect(listed.data).toHaveLength(0);
+      expect(listed.pagination.next).toBeNull();
+      expect(listed.pagination.prev).toBeNull();
+      await backend.stop();
+    });
+  });
+
   describe("claimWorkflowRun()", () => {
     // because claims involve timing and leases, we create and teardown a new
     // namespaced backend instance for each test

packages/backend-postgres/backend.ts

@@ -16,6 +16,7 @@ import {
   GetWorkflowRunParams,
   HeartbeatWorkflowRunParams,
   ListStepAttemptsParams,
+  ListWorkflowRunsParams,
   PaginatedResponse,
   MarkStepAttemptFailedParams,
   MarkStepAttemptSucceededParams,
@@ -28,6 +29,8 @@ import {
   JsonValue,
 } from "openworkflow";
 
+export const DEFAULT_PAGINATION_PAGE_SIZE = 100;
+
 interface BackendPostgresOptions {
   namespaceId?: string;
   runMigrations?: boolean;
@@ -132,6 +135,61 @@ export class BackendPostgres implements Backend {
     return workflowRun ?? null;
   }
 
+  async listWorkflowRuns(
+    params: ListWorkflowRunsParams,
+  ): Promise<PaginatedResponse<WorkflowRun>> {
+    const limit = params.limit ?? DEFAULT_PAGINATION_PAGE_SIZE;
+    const { after, before } = params;
+
+    let cursor: Cursor | null = null;
+    if (after) {
+      cursor = decodeCursor(after);
+    } else if (before) {
+      cursor = decodeCursor(before);
+    }
+
+    const whereClause = this.buildListWorkflowRunsWhere(params, cursor);
+    const order = before
+      ? this.pg`ORDER BY "created_at" DESC, "id" DESC`
+      : this.pg`ORDER BY "created_at" ASC, "id" ASC`;
+
+    const rows = await this.pg<WorkflowRun[]>`
+      SELECT *
+      FROM "openworkflow"."workflow_runs"
+      WHERE ${whereClause}
+      ${order}
+      LIMIT ${limit + 1}
+    `;
+
+    return this.processPaginationResults(rows, limit, !!after, !!before);
+  }
+
+  private buildListWorkflowRunsWhere(
+    params: ListWorkflowRunsParams,
+    cursor: Cursor | null,
+  ) {
+    const { after } = params;
+    const conditions = [this.pg`"namespace_id" = ${this.namespaceId}`];
+
+    if (cursor) {
+      const op = after ? this.pg`>` : this.pg`<`;
+      conditions.push(
+        this.pg`("created_at", "id") ${op} (${cursor.createdAt}, ${cursor.id})`,
+      );
+    }
+
+    let whereClause = conditions[0];
+    if (!whereClause) throw new Error("No conditions");
+
+    for (let i = 1; i < conditions.length; i++) {
+      const condition = conditions[i];
+      if (condition) {
+        whereClause = this.pg`${whereClause} AND ${condition}`;
+      }
+    }
+    return whereClause;
+  }
+
   async claimWorkflowRun(
     params: ClaimWorkflowRunParams,
   ): Promise<WorkflowRun | null> {
@@ -416,7 +474,7 @@ export class BackendPostgres implements Backend {
   async listStepAttempts(
     params: ListStepAttemptsParams,
   ): Promise<PaginatedResponse<StepAttempt>> {
-    const limit = params.limit ?? 100;
+    const limit = params.limit ?? DEFAULT_PAGINATION_PAGE_SIZE;
     const { after, before } = params;
 
     let cursor: Cursor | null = null;

packages/openworkflow/backend.ts

@@ -7,6 +7,9 @@ export interface Backend {
   // Workflow Runs
   createWorkflowRun(params: CreateWorkflowRunParams): Promise<WorkflowRun>;
   getWorkflowRun(params: GetWorkflowRunParams): Promise<WorkflowRun | null>;
+  listWorkflowRuns(
+    params: ListWorkflowRunsParams,
+  ): Promise<PaginatedResponse<WorkflowRun>>;
   claimWorkflowRun(params: ClaimWorkflowRunParams): Promise<WorkflowRun | null>;
   heartbeatWorkflowRun(
     params: HeartbeatWorkflowRunParams,
@@ -49,6 +52,8 @@ export interface GetWorkflowRunParams {
   workflowRunId: string;
 }
 
+export type ListWorkflowRunsParams = PaginationOptions;
+
 export interface ClaimWorkflowRunParams {
   workerId: string;
   leaseDurationMs: number;