Merge pull request #1095 from cardstack/cs-6459-review-home-grown-job-queue-implementation

Add Queue interface and BrowserQueue implementation

Author: habdelra

packages/host/app/lib/browser-queue.ts

@@ -0,0 +1,112 @@
+import debounce from 'lodash/debounce';
+
+import { v4 as uuidv4 } from 'uuid';
+
+import {
+  type Queue,
+  type PgPrimitive,
+  Job,
+  Deferred,
+} from '@cardstack/runtime-common';
+
+export class BrowserQueue implements Queue {
+  #isDestroyed = false;
+  #hasStarted = false;
+  #flush: Promise<void> | undefined;
+
+  // no need for "onAfterJob--that's just the Job.done promise
+  constructor(private onBeforeJob?: (jobId: string) => void) {}
+
+  private jobs: {
+    jobId: string;
+    category: string;
+    arg: PgPrimitive;
+    notifier: Deferred<any>;
+  }[] = [];
+  private categories: Map<string, (arg: any) => Promise<any>> = new Map();
+
+  get isDestroyed() {
+    return this.#isDestroyed;
+  }
+
+  get hasStarted() {
+    return this.#hasStarted;
+  }
+
+  async flush() {
+    await this.#flush;
+  }
+
+  start() {
+    this.#hasStarted = true;
+  }
+
+  register<A, T>(category: string, handler: (arg: A) => Promise<T>) {
+    if (!this.#hasStarted) {
+      throw new Error(`Cannot register category on unstarted Queue`);
+    }
+    if (this.isDestroyed) {
+      throw new Error(`Cannot register category on a destroyed Queue`);
+    }
+    this.categories.set(category, handler);
+    this.debouncedDrainJobs();
+  }
+
+  async publish<T>(category: string, arg: PgPrimitive): Promise<Job<T>> {
+    if (!this.#hasStarted) {
+      throw new Error(`Cannot publish job on unstarted Queue`);
+    }
+    if (this.isDestroyed) {
+      throw new Error(`Cannot publish job on a destroyed Queue`);
+    }
+    let jobId = uuidv4();
+    let notifier = new Deferred<T>();
+    let job = new Job(jobId, notifier);
+    this.jobs.push({
+      jobId,
+      notifier,
+      category,
+      arg,
+    });
+    this.debouncedDrainJobs();
+    return job;
+  }
+
+  private debouncedDrainJobs = debounce(() => {
+    this.drainJobs();
+  }, 250);
+
+  private async drainJobs() {
+    await this.flush();
+
+    let jobsDrained: () => void;
+    this.#flush = new Promise((res) => (jobsDrained = res));
+    let jobs = [...this.jobs];
+    this.jobs = [];
+    for (let workItem of jobs) {
+      let { jobId, category, notifier, arg } = workItem;
+      let handler = this.categories.get(category);
+      if (!handler) {
+        // no handler for this job, add it back to the queue
+        this.jobs.push(workItem);
+        continue;
+      }
+
+      if (this.onBeforeJob) {
+        this.onBeforeJob(jobId);
+      }
+      try {
+        notifier.fulfill(await handler(arg));
+      } catch (e: any) {
+        notifier.reject(e);
+      }
+    }
+
+    jobsDrained!();
+  }
+
+  async destroy() {
+    this.#isDestroyed = true;
+    await this.flush();
+  }
+}

packages/host/tests/unit/queue-test.ts

@@ -0,0 +1,55 @@
+import { module, test } from 'qunit';
+
+import { type Queue } from '@cardstack/runtime-common';
+
+import { BrowserQueue } from '@cardstack/host/lib/browser-queue';
+
+module('Unit | queue | browser implementation', function (hooks) {
+  let queue: Queue;
+
+  hooks.beforeEach(async function () {
+    queue = new BrowserQueue();
+    queue.start();
+  });
+
+  hooks.afterEach(async function () {
+    await queue.destroy();
+  });
+
+  test('it can run a job', async function (assert) {
+    let job = await queue.publish<number>('increment', 17, {
+      queueName: 'first-ephemeral-realm-incrementing',
+    });
+    queue.register('increment', async (a: number) => a + 1);
+    let result = await job.done;
+    assert.strictEqual(result, 18);
+  });
+
+  test(`a job can throw an exception`, async function (assert) {
+    queue.register('increment', async (a: number) => a + 1);
+    queue.register('boom', async () => {
+      throw new Error('boom!');
+    });
+    let [errorJob, nonErrorJob] = await Promise.all([
+      queue.publish<number>('boom', null),
+      queue.publish<number>('increment', 17),
+    ]);
+
+    // assert that the error that was thrown does not prevent subsequent jobs
+    // from running
+    let [errorResults, nonErrorResults] = await Promise.allSettled([
+      errorJob.done,
+      nonErrorJob.done,
+    ]);
+    if (errorResults.status === 'rejected') {
+      assert.strictEqual(errorResults.reason.message, 'boom!');
+    } else {
+      assert.ok(false, `expected 'errorJob' to be rejected`);
+    }
+    if (nonErrorResults.status === 'fulfilled') {
+      assert.strictEqual(nonErrorResults.value, 18);
+    } else {
+      assert.ok(false, `expected 'nonErrorJob' to be fulfilled`);
+    }
+  });
+});

packages/runtime-common/expression.ts

@@ -0,0 +1,16 @@
+import * as JSON from 'json-typescript';
+
+export type Expression = (string | Param)[];
+
+export type PgPrimitive =
+  | number
+  | string
+  | boolean
+  | JSON.Object
+  | JSON.Arr
+  | null;
+
+export interface Param {
+  param: PgPrimitive;
+  kind: 'param';
+}

packages/runtime-common/index.ts

@@ -37,6 +37,8 @@ import { RealmPaths, type LocalPath } from './paths';
 import { Query } from './query';
 import { Loader } from './loader';
 export * from './constants';
+export * from './queue';
+export * from './expression';
 export { makeLogDefinitions, logger } from './log';
 export { RealmPaths, Loader, type LocalPath, type Query };
 export { NotLoaded, isNotLoadedError } from './not-loaded';

packages/runtime-common/queue.ts

@@ -0,0 +1,36 @@
+import { PgPrimitive } from './expression';
+import { Deferred } from './deferred';
+
+export interface QueueOpts {
+  queueName?: string;
+}
+
+export interface Queue {
+  isDestroyed: boolean;
+  hasStarted: boolean;
+  // postgres needs time to initialize, so we only start our queue after
+  // postgres is running
+  start: () => void;
+  destroy: () => Promise<void>;
+  register: <A, T>(category: string, handler: (arg: A) => Promise<T>) => void;
+  publish: <T>(
+    category: string,
+    arg: PgPrimitive,
+    opts?: QueueOpts,
+  ) => Promise<Job<T>>;
+}
+
+export interface JobNotifier {
+  resolve: Function;
+  reject: Function;
+}
+
+export class Job<T> {
+  constructor(
+    public uuid: string,
+    private notifier: Deferred<T>,
+  ) {}
+  get done(): Promise<T> {
+    return this.notifier.promise;
+  }
+}