Use InMemoryDatabase

plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.test.ts

@@ -20,7 +20,9 @@ describe('MemoryTaskBroker', () => {
   it('should claim a dispatched work item', async () => {
     const broker = new MemoryTaskBroker();
 
-    await broker.dispatch({});
+    await broker.dispatch({
+      metadata: '',
+    });
     await expect(broker.claim()).resolves.toEqual(expect.any(TaskAgent));
   });
 
@@ -31,16 +33,16 @@ describe('MemoryTaskBroker', () => {
 
     await expect(Promise.race([promise, 'waiting'])).resolves.toBe('waiting');
 
-    await broker.dispatch({});
+    await broker.dispatch({ metadata: 'foo' });
     await expect(promise).resolves.toEqual(expect.any(TaskAgent));
   });
 
   it('should dispatch multiple items and claim them in order', async () => {
     const broker = new MemoryTaskBroker();
 
-    await broker.dispatch({ name: 'a' });
-    await broker.dispatch({ name: 'b' });
-    await broker.dispatch({ name: 'c' });
+    await broker.dispatch({ metadata: 'a' });
+    await broker.dispatch({ metadata: 'b' });
+    await broker.dispatch({ metadata: 'c' });
 
     const taskA = await broker.claim();
     const taskB = await broker.claim();
@@ -48,9 +50,9 @@ describe('MemoryTaskBroker', () => {
     await expect(taskA).toEqual(expect.any(TaskAgent));
     await expect(taskB).toEqual(expect.any(TaskAgent));
     await expect(taskC).toEqual(expect.any(TaskAgent));
-    await expect(taskA.spec.name).toBe('a');
-    await expect(taskB.spec.name).toBe('b');
-    await expect(taskC.spec.name).toBe('c');
+    await expect(taskA.spec.metadata).toBe('a');
+    await expect(taskB.spec.metadata).toBe('b');
+    await expect(taskC.spec.metadata).toBe('c');
   });
 
   it('should complete a task', async () => {

plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.ts

@@ -14,14 +14,7 @@
  * limitations under the License.
  */
 
-import {
-  CompletedTaskState,
-  Task,
-  TaskSpec,
-  TaskBroker,
-  Status,
-} from './types';
-import { v4 as uuid } from 'uuid';
+import { CompletedTaskState, Task, TaskSpec, TaskBroker } from './types';
 import { InMemoryDatabase } from './database';
 
 export class TaskAgent implements Task {
@@ -48,20 +41,25 @@ export class TaskAgent implements Task {
   }
 
   async complete(result: CompletedTaskState): Promise<void> {
-    this.state.status = result === 'FAILED' ? 'COMPLETED' : 'FAILED';
+    this.db.setStatus(
+      this.state.taskId,
+      result === 'FAILED' ? 'COMPLETED' : 'FAILED',
+    );
   }
 
   private start() {
     this.heartbeartInterval = setInterval(() => {
-      const runId = 'iiiid';
-      this.db.heartBeat(runId);
-    }, 4269);
+      if (!this.state.runId) {
+        throw new Error('no run id provided');
+      }
+      this.db.heartBeat(this.state.runId);
+    }, 1000);
   }
 }
 
 interface TaskState {
   spec: TaskSpec;
-  status: Status;
+  taskId: string;
   runId: string | undefined;
 }
 
@@ -75,14 +73,20 @@ function defer() {
 
 export class MemoryTaskBroker implements TaskBroker {
   private readonly db = new InMemoryDatabase();
-  private readonly tasks = new Array<TaskState>();
   private deferredDispatch = defer();
 
   async claim(): Promise<Task> {
     for (;;) {
       const pendingTask = await this.db.claimTask();
       if (pendingTask) {
-        return TaskAgent.create(pendingTask, this.db);
+        return TaskAgent.create(
+          {
+            runId: pendingTask.runId,
+            taskId: pendingTask.taskId,
+            spec: pendingTask.spec,
+          },
+          this.db,
+        );
       }
 
       await this.waitForDispatch();
@@ -90,11 +94,7 @@ export class MemoryTaskBroker implements TaskBroker {
   }
 
   async dispatch(spec: TaskSpec): Promise<void> {
-    this.tasks.push({
-      spec,
-      status: 'OPEN',
-      runId: undefined,
-    });
+    await this.db.createTask(spec);
     this.signalDispatch();
   }