Add eventType for closing observer stream

Co-authored-by: Patrik Oldsberg <poldsberg@gmail.com>

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

@@ -14,7 +14,13 @@
  * limitations under the License.
  */
 
-import { DbTaskRow, DbTaskEventRow, Status, TaskSpec } from './types';
+import {
+  DbTaskRow,
+  DbTaskEventRow,
+  Status,
+  TaskSpec,
+  TaskEventType,
+} from './types';
 import { v4 as uuid } from 'uuid';
 
 export interface Database {
@@ -28,7 +34,8 @@ export interface Database {
 type EmitOptions = {
   taskId: string;
   runId: string;
-  event: string;
+  body: string;
+  type: TaskEventType;
 };
 
 type ReadOptions = {
@@ -40,12 +47,13 @@ export class MemoryDatabase implements Database {
   private readonly store = new Map<string, DbTaskRow>();
   private readonly events = new Array<DbTaskEventRow>();
 
-  async emit({ taskId, runId, event }: EmitOptions) {
+  async emit({ taskId, runId, body, type }: EmitOptions) {
     this.events.push({
       id: this.events.length,
       taskId,
       runId,
-      event,
+      body,
+      type,
       createdAt: new Date().toISOString(),
     });
   }
@@ -89,10 +97,10 @@ export class MemoryDatabase implements Database {
 
   async claimTask(): Promise<DbTaskRow | undefined> {
     for (const t of this.store.values()) {
-      if (t.status === 'OPEN') {
+      if (t.status === 'open') {
         const task: DbTaskRow = {
           ...t,
-          status: 'PROCESSING',
+          status: 'processing',
           runId: uuid(),
         };
         this.store.set(t.taskId, task);
@@ -106,7 +114,7 @@ export class MemoryDatabase implements Database {
     const taskRow = {
       taskId: uuid(),
       spec,
-      status: 'OPEN' as Status,
+      status: 'open' as Status,
       retryCount: 0,
       createdAt: new Date().toISOString(),
     };

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

@@ -70,16 +70,16 @@ describe('MemoryTaskBroker', () => {
   it('should complete a task', async () => {
     const dispatchResult = await broker.dispatch(taskSpec);
     const task = await broker.claim();
-    await task.complete('COMPLETED');
+    await task.complete('completed');
     const taskRow = await storage.get(dispatchResult.taskId);
-    expect(taskRow.status).toBe('COMPLETED');
+    expect(taskRow.status).toBe('completed');
   });
 
   it('should fail a task', async () => {
     const dispatchResult = await broker.dispatch(taskSpec);
     const task = await broker.claim();
-    await task.complete('FAILED');
+    await task.complete('failed');
     const taskRow = await storage.get(dispatchResult.taskId);
-    expect(taskRow.status).toBe('FAILED');
+    expect(taskRow.status).toBe('failed');
   });
 });

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

@@ -47,16 +47,22 @@ export class TaskAgent implements Task {
     await this.storage.emit({
       taskId: this.state.taskId,
       runId: this.state.runId,
-      event: message,
+      body: message,
+      type: 'log',
     });
   }
 
   async complete(result: CompletedTaskState): Promise<void> {
     await this.storage.setStatus(
       this.state.taskId,
-      result === 'FAILED' ? 'FAILED' : 'COMPLETED',
+      result === 'failed' ? 'failed' : 'completed',
     );
-
+    this.storage.emit({
+      taskId: this.state.taskId,
+      runId: this.state.runId,
+      body: `Run completed with status: ${result}`,
+      type: 'completion',
+    });
     if (this.heartbeartInterval) {
       clearInterval(this.heartbeartInterval);
     }

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

@@ -102,9 +102,9 @@ export class TaskWorker {
 
       task.emitLog(`Completely done now!`);
 
-      await task.complete('COMPLETED');
+      await task.complete('completed');
     } catch (error) {
-      await task.complete('FAILED');
+      await task.complete('failed');
     }
   }
 }

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

@@ -18,13 +18,13 @@ import { TemplateEntityV1alpha1 } from '@backstage/catalog-model';
 import { TemplaterValues } from '..';
 
 export type Status =
-  | 'OPEN'
-  | 'PROCESSING'
-  | 'FAILED'
-  | 'CANCELLED'
-  | 'COMPLETED';
+  | 'open'
+  | 'processing'
+  | 'failed'
+  | 'cancelled'
+  | 'completed';
 
-export type CompletedTaskState = 'FAILED' | 'COMPLETED';
+export type CompletedTaskState = 'failed' | 'completed';
 
 export type DbTaskRow = {
   taskId: string;
@@ -36,11 +36,13 @@ export type DbTaskRow = {
   runId?: string;
 };
 
+export type TaskEventType = 'completion' | 'log';
 export type DbTaskEventRow = {
   id: number;
   runId: string;
   taskId: string;
-  event: string;
+  body: string;
+  type: TaskEventType;
   createdAt: string;
 };
 

plugins/scaffolder-backend/src/service/router.ts

@@ -150,6 +150,10 @@ export async function createRouter(
         ({ events }) => {
           for (const event of events) {
             res.write(`event:${JSON.stringify(event)}\n\n`);
+            if (event.type === 'completion') {
+              unsubscribe();
+              res.end();
+            }
           }
         },
       );