first commit

api/app/services/job_runner.ts

@@ -0,0 +1,278 @@
+import { spawn } from 'node:child_process'
+import { createWriteStream } from 'node:fs'
+import { EventEmitter } from 'node:events'
+import { createGzip } from 'node:zlib'
+import { pipeline } from 'node:stream/promises'
+import { DateTime } from 'luxon'
+import logger from '@adonisjs/core/services/logger'
+import AuditLog, { type AuditStatus } from '#models/audit_log'
+import UserError from '#exceptions/user_error'
+
+export interface JobLine {
+  ts: number
+  stream: 'out' | 'err' | 'info'
+  text: string
+}
+
+export interface RunOptions {
+  cwd?: string
+  env?: Record<string, string>
+
+  /**
+   * Exit codes considered as a success (default: [0]).
+   */
+  okCodes?: number[]
+
+  /**
+   * Text shown in the log instead of the real command (to hide secrets).
+   */
+  display?: string
+
+  /**
+   * Writes stdout to this file instead of the log.
+   */
+  stdoutFile?: string
+  gzipStdout?: boolean
+
+  timeoutMs?: number
+}
+
+export interface JobContext {
+  log(text: string, stream?: JobLine['stream']): void
+  run(command: string, args: string[], options?: RunOptions): Promise<void>
+}
+
+export interface JobMeta {
+  action: string
+  target: string | null
+  username: string | null
+  ip: string | null
+
+  /**
+   * Two jobs with the same lock key cannot run at the same time
+   * (for example two deployments of the same project).
+   */
+  lockKey?: string
+}
+
+/**
+ * Output kept in the action log, to keep the SQLite file small.
+ */
+const MAX_STORED_OUTPUT = 200_000
+
+/**
+ * Finished jobs stay in memory for a while so that a page opened late
+ * can still replay the whole output.
+ */
+const KEEP_FINISHED_MS = 15 * 60_000
+
+export class Job extends EventEmitter {
+  lines: JobLine[] = []
+  status: AuditStatus = 'running'
+
+  constructor(
+    readonly id: number,
+    readonly meta: JobMeta
+  ) {
+    super()
+    this.setMaxListeners(50)
+  }
+
+  push(line: JobLine) {
+    this.lines.push(line)
+    this.emit('line', line)
+  }
+
+  get output() {
+    return this.lines.map((line) => line.text).join('\n')
+  }
+}
+
+/**
+ * Runs the long actions (deployments, backups, project creation) in the
+ * background. Each job is recorded in the action log and its output is
+ * streamed to the browser.
+ */
+class JobRunner {
+  #jobs = new Map<number, Job>()
+  #locks = new Set<string>()
+
+  get(id: number) {
+    return this.#jobs.get(id)
+  }
+
+  isLocked(lockKey: string) {
+    return this.#locks.has(lockKey)
+  }
+
+  async start(meta: JobMeta, task: (ctx: JobContext) => Promise<void>): Promise<Job> {
+    if (meta.lockKey && this.#locks.has(meta.lockKey)) {
+      throw new UserError('Une action est déjà en cours sur cet élément, attends la fin.', 409)
+    }
+    if (meta.lockKey) this.#locks.add(meta.lockKey)
+
+    const entry = await AuditLog.create({
+      action: meta.action,
+      target: meta.target,
+      username: meta.username,
+      ip: meta.ip,
+      status: 'running',
+    })
+
+    const job = new Job(entry.id, meta)
+    this.#jobs.set(job.id, job)
+
+    const ctx: JobContext = {
+      log: (text, stream = 'info') => {
+        for (const line of text.split('\n')) job.push({ ts: Date.now(), stream, text: line })
+      },
+      run: (command, args, options) => this.#run(job, command, args, options),
+    }
+
+    /**
+     * Not awaited: the HTTP request returns the job id right away.
+     */
+    void (async () => {
+      try {
+        await task(ctx)
+        job.status = 'success'
+        ctx.log('✔ Terminé')
+      } catch (error) {
+        job.status = 'error'
+        ctx.log(`✖ ${error instanceof Error ? error.message : String(error)}`, 'err')
+      } finally {
+        if (meta.lockKey) this.#locks.delete(meta.lockKey)
+        await this.#finish(job, entry)
+        job.emit('end', job.status)
+        setTimeout(() => this.#jobs.delete(job.id), KEEP_FINISHED_MS).unref()
+      }
+    })()
+
+    return job
+  }
+
+  async #finish(job: Job, entry: AuditLog) {
+    try {
+      const output = job.output
+      entry.status = job.status
+      entry.output =
+        output.length > MAX_STORED_OUTPUT
+          ? `[…] ${output.slice(output.length - MAX_STORED_OUTPUT)}`
+          : output
+      entry.finishedAt = DateTime.now()
+      await entry.save()
+    } catch (error) {
+      logger.error({ err: error }, 'cannot save the job output')
+    }
+  }
+
+  #run(job: Job, command: string, args: string[], options: RunOptions = {}) {
+    const okCodes = options.okCodes ?? [0]
+    job.push({
+      ts: Date.now(),
+      stream: 'info',
+      text: `$ ${options.display ?? [command, ...args.map(quote)].join(' ')}`,
+    })
+
+    return new Promise<void>((resolve, reject) => {
+      const child = spawn(command, args, {
+        cwd: options.cwd,
+        env: cleanEnv(options.env),
+        stdio: ['ignore', 'pipe', 'pipe'],
+      })
+
+      const timeout = setTimeout(
+        () => {
+          job.push({ ts: Date.now(), stream: 'err', text: 'Délai dépassé, commande arrêtée.' })
+          child.kill('SIGKILL')
+        },
+        options.timeoutMs ?? 15 * 60_000
+      )
+
+      let fileWritten: Promise<void> = Promise.resolve()
+      if (options.stdoutFile) {
+        const file = createWriteStream(options.stdoutFile)
+        fileWritten = options.gzipStdout
+          ? pipeline(child.stdout, createGzip(), file)
+          : pipeline(child.stdout, file)
+      } else {
+        lineReader(child.stdout, (text) => job.push({ ts: Date.now(), stream: 'out', text }))
+      }
+      lineReader(child.stderr, (text) => job.push({ ts: Date.now(), stream: 'err', text }))
+
+      child.on('error', (error) => {
+        clearTimeout(timeout)
+        reject(new Error(`Impossible de lancer "${command}" : ${error.message}`))
+      })
+
+      child.on('close', async (code) => {
+        clearTimeout(timeout)
+        try {
+          await fileWritten
+        } catch (error) {
+          return reject(error)
+        }
+        if (code !== null && okCodes.includes(code)) return resolve()
+        reject(new Error(`La commande "${command}" a échoué (code ${code ?? 'tué'}).`))
+      })
+    })
+  }
+}
+
+/**
+ * Variables passed to the commands. The dashboard's own variables (PORT,
+ * APP_KEY…) are left out: "docker compose" would use them to fill the
+ * ${VAR} of the projects' compose files.
+ */
+const INHERITED_ENV = [
+  'PATH',
+  'HOME',
+  'TZ',
+  'DOCKER_HOST',
+  'DOCKER_CONFIG',
+  // Windows (local development)
+  'USERPROFILE',
+  'APPDATA',
+  'LOCALAPPDATA',
+  'ProgramData',
+  'ProgramFiles',
+  'SystemRoot',
+  'TEMP',
+  'TMP',
+]
+
+export function cleanEnv(extra: Record<string, string> = {}) {
+  const env: Record<string, string> = {}
+  for (const key of INHERITED_ENV) {
+    if (process.env[key] !== undefined) env[key] = process.env[key]!
+  }
+  /**
+   * Never wait for a password: fail right away instead.
+   */
+  env.GIT_TERMINAL_PROMPT = '0'
+  return { ...env, ...extra }
+}
+
+/**
+ * Calls the callback for each complete line of the stream.
+ */
+export function lineReader(stream: NodeJS.ReadableStream, onLine: (line: string) => void) {
+  let rest = ''
+  stream.setEncoding('utf8')
+  stream.on('data', (chunk: string) => {
+    const lines = (rest + chunk).split(/\r?\n/)
+    rest = lines.pop() ?? ''
+    for (const line of lines) onLine(line)
+  })
+  stream.on('end', () => {
+    if (rest) onLine(rest)
+    rest = ''
+  })
+}
+
+function quote(arg: string) {
+  return /^[\w@%+=:,./-]+$/.test(arg) ? arg : `'${arg.replace(/'/g, `'\\''`)}'`
+}
+
+const jobRunner = new JobRunner()
+export default jobRunner