Merge pull request backstage#26730 from drodil/scaffolder_events_service

feat(scaffolder): add support for EventsService

.changeset/orange-moles-kick.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-scaffolder-backend': minor
+---
+
+Emit scaffolder events using the optional `EventsService`

plugins/scaffolder-backend/package.json

@@ -73,6 +73,7 @@
     "@backstage/plugin-bitbucket-cloud-common": "workspace:^",
     "@backstage/plugin-catalog-backend-module-scaffolder-entity-model": "workspace:^",
     "@backstage/plugin-catalog-node": "workspace:^",
+    "@backstage/plugin-events-node": "workspace:^",
     "@backstage/plugin-permission-common": "workspace:^",
     "@backstage/plugin-permission-node": "workspace:^",
     "@backstage/plugin-scaffolder-backend-module-azure": "workspace:^",

plugins/scaffolder-backend/report.api.md

@@ -19,6 +19,7 @@ import { Config } from '@backstage/config';
 import { DatabaseService } from '@backstage/backend-plugin-api';
 import { DiscoveryService } from '@backstage/backend-plugin-api';
 import { Duration } from 'luxon';
+import { EventsService } from '@backstage/plugin-events-node';
 import { executeShellCommand as executeShellCommand_2 } from '@backstage/plugin-scaffolder-node';
 import { ExecuteShellCommandOptions } from '@backstage/plugin-scaffolder-node';
 import express from 'express';
@@ -520,6 +521,7 @@ export class DatabaseTaskStore implements TaskStore {
 // @public
 export type DatabaseTaskStoreOptions = {
   database: PluginDatabaseManager | Knex;
+  events?: EventsService;
 };
 
 // @public @deprecated
@@ -552,6 +554,8 @@ export interface RouterOptions {
   // (undocumented)
   discovery?: DiscoveryService;
   // (undocumented)
+  events?: EventsService;
+  // (undocumented)
   httpAuth?: HttpAuthService;
   // (undocumented)
   identity?: IdentityApi;

plugins/scaffolder-backend/src/ScaffolderPlugin.ts

@@ -15,8 +15,8 @@
  */
 
 import {
-  createBackendPlugin,
   coreServices,
+  createBackendPlugin,
 } from '@backstage/backend-plugin-api';
 import { loggerToWinstonLogger } from '@backstage/backend-common';
 import { ScmIntegrations } from '@backstage/integration';
@@ -51,6 +51,7 @@ import {
   createWaitAction,
 } from './scaffolder';
 import { createRouter } from './service/router';
+import { eventsServiceRef } from '@backstage/plugin-events-node';
 
 /**
  * Scaffolder plugin
@@ -115,6 +116,7 @@ export const scaffolderPlugin = createBackendPlugin({
         httpRouter: coreServices.httpRouter,
         httpAuth: coreServices.httpAuth,
         catalogClient: catalogServiceRef,
+        events: eventsServiceRef,
       },
       async init({
         logger,
@@ -128,6 +130,7 @@ export const scaffolderPlugin = createBackendPlugin({
         httpAuth,
         catalogClient,
         permissions,
+        events,
       }) {
         const log = loggerToWinstonLogger(logger);
         const integrations = ScmIntegrations.fromConfig(config);
@@ -191,6 +194,7 @@ export const scaffolderPlugin = createBackendPlugin({
           permissions,
           autocompleteHandlers,
           additionalWorkspaceProviders,
+          events,
         });
         httpRouter.use(router);
       },

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

@@ -21,8 +21,9 @@ import { TaskSpec } from '@backstage/plugin-scaffolder-common';
 import { ConflictError } from '@backstage/errors';
 import { createMockDirectory } from '@backstage/backend-test-utils';
 import fs from 'fs-extra';
+import { EventsService } from '@backstage/plugin-events-node';
 
-const createStore = async () => {
+const createStore = async (events?: EventsService) => {
   const manager = DatabaseManager.fromConfig(
     new ConfigReader({
       backend: {
@@ -35,6 +36,7 @@ const createStore = async () => {
   ).forPlugin('scaffolder');
   const store = await DatabaseTaskStore.create({
     database: manager,
+    events,
   });
   return { store, manager };
 };
@@ -52,6 +54,14 @@ const workspaceDir = createMockDirectory({
 });
 
 describe('DatabaseTaskStore', () => {
+  const eventsService = {
+    publish: jest.fn(),
+  } as unknown as EventsService;
+
+  beforeEach(() => {
+    jest.resetAllMocks();
+  });
+
   it('should create the database store and run migration', async () => {
     const { store, manager } = await createStore();
     expect(store).toBeDefined();
@@ -222,7 +232,7 @@ describe('DatabaseTaskStore', () => {
   });
 
   it('should sent an event to start cancelling the task', async () => {
-    const { store } = await createStore();
+    const { store } = await createStore(eventsService);
 
     const { taskId } = await store.createTask({
       spec: {} as TaskSpec,
@@ -244,10 +254,24 @@ describe('DatabaseTaskStore', () => {
     const event = events[0];
     expect(event.taskId).toBe(taskId);
     expect(event.body.status).toBe('cancelled');
+
+    expect(eventsService.publish).toHaveBeenCalledWith({
+      topic: 'scaffolder.task',
+      eventPayload: {
+        id: 1,
+        taskId,
+        status: 'cancelled',
+        body: {
+          message: `Step 2 has been cancelled.`,
+          stepId: 2,
+          status: 'cancelled',
+        },
+      },
+    });
   });
 
   it('should emit a log event', async () => {
-    const { store } = await createStore();
+    const { store } = await createStore(eventsService);
     const { taskId } = await store.createTask({
       spec: {} as TaskSpec,
       createdBy: 'me',
@@ -310,7 +334,7 @@ describe('DatabaseTaskStore', () => {
   });
 
   it('should complete the task', async () => {
-    const { store } = await createStore();
+    const { store } = await createStore(eventsService);
     const { taskId } = await store.createTask({
       spec: {} as TaskSpec,
       createdBy: 'me',
@@ -327,10 +351,21 @@ describe('DatabaseTaskStore', () => {
 
     const taskAfterCompletion = await store.getTask(taskId);
     expect(taskAfterCompletion.status).toBe('cancelled');
+
+    expect(eventsService.publish).toHaveBeenCalledWith({
+      topic: 'scaffolder.task',
+      eventPayload: {
+        id: taskId,
+        status: 'cancelled',
+        createdAt: expect.any(String),
+        lastHeartbeatAt: null,
+        createdBy: 'me',
+      },
+    });
   });
 
   it('should claim a new task', async () => {
-    const { store } = await createStore();
+    const { store } = await createStore(eventsService);
     const { taskId } = await store.createTask({
       spec: {} as TaskSpec,
       createdBy: 'me',
@@ -341,10 +376,22 @@ describe('DatabaseTaskStore', () => {
 
     const claimedTask = await store.getTask(taskId);
     expect(claimedTask.status).toBe('processing');
+
+    expect(eventsService.publish).toHaveBeenCalledWith({
+      topic: 'scaffolder.task',
+      eventPayload: {
+        id: taskId,
+        status: 'processing',
+        createdAt: expect.any(String),
+        lastHeartbeatAt: null,
+        createdBy: 'me',
+        spec: {},
+      },
+    });
   });
 
   it('should restore the state of the task after the task recovery', async () => {
-    const { store } = await createStore();
+    const { store } = await createStore(eventsService);
     const { taskId } = await store.createTask({
       spec: {} as TaskSpec,
       createdBy: 'me',
@@ -379,10 +426,22 @@ describe('DatabaseTaskStore', () => {
 
     const claimedTask = await store.getTask(taskId);
     expect(claimedTask.state).toEqual({ state: state.state });
+
+    expect(eventsService.publish).toHaveBeenCalledWith({
+      topic: 'scaffolder.task',
+      eventPayload: {
+        id: 1,
+        taskId,
+        body: {
+          recoverStrategy: 'none',
+        },
+        status: 'recovered',
+      },
+    });
   });
 
   it('should shutdown the running task', async () => {
-    const { store } = await createStore();
+    const { store } = await createStore(eventsService);
     const { taskId } = await store.createTask({
       spec: {} as TaskSpec,
       createdBy: 'me',
@@ -394,6 +453,17 @@ describe('DatabaseTaskStore', () => {
 
     const claimedTask = await store.getTask(taskId);
     expect(claimedTask.status).toBe('failed');
+
+    expect(eventsService.publish).toHaveBeenCalledWith({
+      topic: 'scaffolder.task',
+      eventPayload: {
+        id: taskId,
+        status: 'failed',
+        createdAt: expect.any(String),
+        lastHeartbeatAt: expect.any(String),
+        createdBy: 'me',
+      },
+    });
   });
 
   it('should be not possible to shutdown not running task', async () => {