Initial support for job queue

ts/jobs/JobQueueDatabaseStore.ts

@@ -0,0 +1,107 @@
+// Copyright 2021 Signal Messenger, LLC
+// SPDX-License-Identifier: AGPL-3.0-only
+
+import { pick, noop } from 'lodash';
+import { AsyncQueue } from '../util/AsyncQueue';
+import { concat, wrapPromise } from '../util/asyncIterables';
+import { JobQueueStore, StoredJob } from './types';
+import databaseInterface from '../sql/Client';
+import * as log from '../logging/log';
+
+type Database = {
+  getJobsInQueue(queueType: string): Promise<Array<StoredJob>>;
+  insertJob(job: Readonly<StoredJob>): Promise<void>;
+  deleteJob(id: string): Promise<void>;
+};
+
+export class JobQueueDatabaseStore implements JobQueueStore {
+  private activeQueueTypes = new Set<string>();
+
+  private queues = new Map<string, AsyncQueue<StoredJob>>();
+
+  private initialFetchPromises = new Map<string, Promise<void>>();
+
+  constructor(private readonly db: Database) {}
+
+  async insert(job: Readonly<StoredJob>): Promise<void> {
+    log.info(
+      `JobQueueDatabaseStore adding job ${job.id} to queue ${JSON.stringify(
+        job.queueType
+      )}`
+    );
+
+    const initialFetchPromise = this.initialFetchPromises.get(job.queueType);
+    if (!initialFetchPromise) {
+      throw new Error(
+        `JobQueueDatabaseStore tried to add job for queue ${JSON.stringify(
+          job.queueType
+        )} but streaming had not yet started`
+      );
+    }
+    await initialFetchPromise;
+
+    await this.db.insertJob(
+      pick(job, ['id', 'timestamp', 'queueType', 'data'])
+    );
+
+    this.getQueue(job.queueType).add(job);
+  }
+
+  async delete(id: string): Promise<void> {
+    await this.db.deleteJob(id);
+  }
+
+  stream(queueType: string): AsyncIterable<StoredJob> {
+    if (this.activeQueueTypes.has(queueType)) {
+      throw new Error(
+        `Cannot stream queue type ${JSON.stringify(queueType)} more than once`
+      );
+    }
+    this.activeQueueTypes.add(queueType);
+
+    return concat([
+      wrapPromise(this.fetchJobsAtStart(queueType)),
+      this.getQueue(queueType),
+    ]);
+  }
+
+  private getQueue(queueType: string): AsyncQueue<StoredJob> {
+    const existingQueue = this.queues.get(queueType);
+    if (existingQueue) {
+      return existingQueue;
+    }
+
+    const result = new AsyncQueue<StoredJob>();
+    this.queues.set(queueType, result);
+    return result;
+  }
+
+  private async fetchJobsAtStart(queueType: string): Promise<Array<StoredJob>> {
+    log.info(
+      `JobQueueDatabaseStore fetching existing jobs for queue ${JSON.stringify(
+        queueType
+      )}`
+    );
+
+    // This is initialized to `noop` because TypeScript doesn't know that `Promise` calls
+    //   its callback synchronously, making sure `onFinished` is defined.
+    let onFinished: () => void = noop;
+    const initialFetchPromise = new Promise<void>(resolve => {
+      onFinished = resolve;
+    });
+    this.initialFetchPromises.set(queueType, initialFetchPromise);
+
+    const result = await this.db.getJobsInQueue(queueType);
+    log.info(
+      `JobQueueDatabaseStore finished fetching existing ${
+        result.length
+      } jobs for queue ${JSON.stringify(queueType)}`
+    );
+    onFinished();
+    return result;
+  }
+}
+
+export const jobQueueDatabaseStore = new JobQueueDatabaseStore(
+  databaseInterface
+);