feat: add "github-workflow" mode

Author: Leksat

.idea/Silverback.iml

@@ -1,6 +1,7 @@
 import { defineConfig } from '@amazeelabs/publisher';
 
 export default defineConfig({
+  mode: 'local',
   commands: {
     build: {
       command: 'pnpm build:gatsby',

apps/silverback-gatsby/publisher.config.ts

@@ -1,23 +1,9 @@
-import { existsSync } from 'fs';
 import { HttpTerminator } from 'http-terminator/src/types';
-import { join } from 'path';
-import { loadSync } from 'ts-import';
 
 import { runServer } from './server';
-import { setConfig } from './tools/config';
 import { core } from './tools/core';
 import { initDatabase } from './tools/database';
 
-const configPath = join(process.cwd(), 'publisher.config.ts');
-if (!existsSync(configPath)) {
-  console.error(`Publisher config not found: ${configPath}`);
-  process.exit(1);
-}
-const config = loadSync(configPath, {
-  compiledJsExtension: '.cjs',
-}).default;
-setConfig(config);
-
 core.output$.subscribe((chunk) => {
   process.stdout.write(chunk);
 });

packages/npm/@amazeelabs/publisher/src/cli.ts

@@ -0,0 +1,183 @@
+import { execSync, SpawnSyncReturns } from 'node:child_process';
+
+import {
+  ApplicationState,
+  WorkflowPublisherPayload,
+} from '@amazeelabs/publisher-shared';
+import { pairwise } from 'rxjs';
+
+import { getConfigGithubWorkflow as config } from '../tools/config';
+import { saveBuildInfo } from '../tools/database';
+import { TaskController, TaskJob } from '../tools/queue';
+import { core } from './core';
+
+export const buildTask: (args?: { clean: boolean }) => TaskJob =
+  (args) => async (controller) => {
+    core.state.buildNumber++;
+    core.state.applicationState$.next(
+      core.state.buildNumber === 1
+        ? ApplicationState.Starting
+        : ApplicationState.Updating,
+    );
+
+    const startedAt = Date.now();
+
+    const output: Array<string> = [];
+    const outputSubscription = core.output$.subscribe((chunk) => {
+      output.push(
+        `${new Date().toISOString().substring(0, 19).replace('T', ' ')} ${chunk}`,
+      );
+    });
+
+    const finalizeBuild = (isSuccess: boolean): boolean => {
+      core.state.applicationState$.next(
+        isSuccess ? ApplicationState.Ready : ApplicationState.Error,
+      );
+      saveBuildInfo({
+        type: 'github-workflow',
+        startedAt,
+        finishedAt: Date.now(),
+        success: isSuccess,
+        logs: output.join(''),
+      });
+      outputSubscription.unsubscribe();
+      return isSuccess;
+    };
+
+    const attempts =
+      core.state.buildNumber === 1
+        ? 3 // The first build gets 3 attempts.
+        : 1;
+    for (let attempt = 1; attempt <= attempts; attempt++) {
+      const result =
+        attempt === 2
+          ? await runWorkflow({ controller, clean: true })
+          : await runWorkflow({ controller, clean: !!args?.clean });
+      if (result) {
+        return finalizeBuild(true);
+      }
+    }
+    return finalizeBuild(false);
+  };
+
+async function runWorkflow(args: {
+  clean: boolean;
+  controller: TaskController;
+}): Promise<boolean> {
+  return new Promise<boolean>((resolve) => {
+    core.output$.next('Starting the workflow', 'info');
+
+    const timeout = setTimeout(() => {
+      core.output$.next('Timeout reached', 'error');
+      args.controller.cancel();
+    }, config().workflowTimeout);
+
+    args.controller.onCancel(async () => {
+      core.output$.next('Cancelling the workflow', 'warning');
+      await cancelWorkflow();
+      clearTimeout(timeout);
+      return resolve(false);
+    });
+
+    try {
+      execSync(
+        `gh workflow run ${config().workflow} --repo ${config().repo} --ref ${config().ref} --json`,
+        {
+          input: JSON.stringify({
+            ...config().inputs,
+            publisher_payload: JSON.stringify({
+              callbackUrl:
+                config().publisherBaseUrl + '/github-workflow-status',
+              clearCache: args.clean,
+              environmentVariables: config().environmentVariables,
+            } satisfies WorkflowPublisherPayload),
+          }),
+        },
+      );
+    } catch (error) {
+      core.output$.next('Error starting the workflow', 'error');
+      logExecError(error);
+
+      clearTimeout(timeout);
+      return resolve(false);
+    }
+
+    const subscription = core.state.workflowState$
+      .pipe(pairwise())
+      .subscribe(([previous, current]) => {
+        if (current === 'started') {
+          core.output$.next('Workflow started', 'info');
+          core.output$.next('Logs: ' + core.state.workflowRunUrl);
+          return;
+        }
+        if (
+          previous === 'started' &&
+          (current === 'success' || current === 'failure')
+        ) {
+          subscription.unsubscribe();
+          current === 'success'
+            ? core.output$.next('Workflow succeeded', 'success')
+            : core.output$.next('Workflow failed or cancelled', 'error');
+          core.output$.next('Logs: ' + core.state.workflowRunUrl);
+
+          clearTimeout(timeout);
+          return resolve(current === 'success');
+        }
+      });
+  });
+}
+
+async function cancelWorkflow(): Promise<void> {
+  type Run = { name: string; conclusion: string; databaseId: number };
+
+  function matchesEnvironment(run: Run): boolean {
+    return run.name.includes(`[env: ${config().environment}]`);
+  }
+  function isCompleted(run: Run): boolean {
+    return !!run.conclusion;
+  }
+
+  const listCommand = `gh run list --workflow=${config().workflow} --repo ${config().repo} --json name,conclusion,databaseId --limit 100`;
+
+  try {
+    // Cancel the running workflows.
+    const result = execSync(listCommand).toString();
+    const runs = JSON.parse(result) as Array<Run>;
+    for (const run of runs) {
+      if (!isCompleted(run) && matchesEnvironment(run)) {
+        execSync(`gh run cancel ${run.databaseId} --repo ${config().repo}`);
+      }
+    }
+
+    // Wait for the workflows to stop. Give it a minute.
+    // This may slightly impact the GitHub API rate limits, but cancellations
+    // are quite rare operations.
+    const checkAttempts = 6;
+    const delay = 10_000;
+    for (let checkAttempt = 1; checkAttempt <= checkAttempts; checkAttempt++) {
+      await new Promise((resolve) => setTimeout(resolve, delay));
+      const result = execSync(listCommand).toString();
+      const runs = JSON.parse(result) as Array<Run>;
+      if (runs.every((run) => isCompleted(run) || !matchesEnvironment(run))) {
+        return;
+      }
+    }
+  } catch (error) {
+    core.output$.next('Error canceling the workflow', 'error');
+    logExecError(error);
+  }
+}
+
+function isSpawnError(error: unknown): error is SpawnSyncReturns<Buffer> {
+  return !!error && typeof error === 'object' && 'status' in error;
+}
+
+function logExecError(error: unknown): void {
+  if (isSpawnError(error)) {
+    core.output$.next(`Error: ${error}`);
+    core.output$.next(`Exit code: ${error.status}`);
+    core.output$.next(`Stdout: ${error.stdout?.toString()}`);
+    core.output$.next(`Stderr: ${error.stderr?.toString()}`);
+  }
+  console.error(error);
+}

packages/npm/@amazeelabs/publisher/src/mode-github-workflow/build.ts

@@ -1,3 +1,48 @@
+import { ApplicationState } from '@amazeelabs/publisher-shared';
+import { BehaviorSubject, Subject } from 'rxjs';
+
 import { Core } from '../tools/core';
+import { OutputSubject } from '../tools/output';
+import { Queue } from '../tools/queue';
+import { buildTask } from './build';
+
+type WorkflowState = 'unknown' | 'started' | 'success' | 'failure';
+
+class CoreGithubWorkflow implements Core {
+  state = {
+    applicationState$: new Subject<ApplicationState>(),
+    buildNumber: 0,
+    workflowState$: new BehaviorSubject<WorkflowState>('unknown'),
+    workflowRunUrl: '',
+  };
+
+  output$ = new OutputSubject();
+
+  queue = new Queue();
+
+  start = () => {
+    this.queue.add({ job: buildTask() });
+    this.queue.run();
+  };
+
+  stop = async () => {
+    await this.queue.clear();
+  };
+
+  build = (): void => {
+    // Consider any pending task a build task.
+    if (!this.queue.hasPendingTasks()) {
+      this.queue.add({ job: buildTask() });
+    }
+  };
+
+  clean = async (): Promise<void> => {
+    await this.queue.clear();
+    this.queue.add({ job: buildTask({ clean: true }) });
+  };
+}
+
+const core = new CoreGithubWorkflow();
+core.state.applicationState$.next(ApplicationState.Starting);
 
-export const core = {} as Core;
+export { core };

packages/npm/@amazeelabs/publisher/src/mode-github-workflow/core.ts

@@ -1,4 +1,7 @@
-import { ApplicationState } from '@amazeelabs/publisher-shared';
+import {
+  ApplicationState,
+  workflowStatusNotificationSchema,
+} from '@amazeelabs/publisher-shared';
 import cors from 'cors';
 import express from 'express';
 import expressWs from 'express-ws';
@@ -16,7 +19,7 @@ import {
   isSessionRequired,
 } from './tools/authentication';
 import { getConfig } from './tools/config';
-import { core } from './tools/core';
+import { core, CoreGithubWorkflow } from './tools/core';
 import { getDatabase } from './tools/database';
 import {
   getOAuth2AuthorizeUrl,
@@ -61,6 +64,8 @@ const runServer = async (): Promise<HttpTerminator> => {
   // @TODO see if we need to lock this down
   app.use(referrerPolicy());
 
+  app.use(express.json());
+
   app.use((req, res, next) => {
     res.set('Cache-control', 'no-cache');
     next();
@@ -253,6 +258,19 @@ const runServer = async (): Promise<HttpTerminator> => {
     res.redirect('/oauth/login');
   });
 
+  app.post('/github-workflow-status', async (req, res) => {
+    const result = workflowStatusNotificationSchema.safeParse(req.body);
+    if (!result.success) {
+      console.error(result.error);
+      res.status(400).send('Invalid request\n');
+      return;
+    }
+    const { status, workflowRunUrl } = result.data;
+    (core as CoreGithubWorkflow).state.workflowRunUrl = workflowRunUrl;
+    (core as CoreGithubWorkflow).state.workflowState$.next(status);
+    res.send();
+  });
+
   app.get('*', (req, res, next) => {
     if (!req.app.locals.isReady) {
       if (req.accepts('text/html')) {