rhedyn/src/run.js

import { performance } from "node:perf_hooks"
import { getTaskKey, getTaskName, processTask } from "./lib.js"
import { getLogger } from "./logging.js"

export async function runWithConfig(config) {
  if (!config || typeof config !== "object") {
    throw new Error("Rhedyn config must export an object with `opts` and `tasks`.")
  }

  const startTime = performance.now()
  const { opts, tasks } = config

  if (!opts || !tasks) {
    throw new Error("Rhedyn config must define both `opts` and `tasks`.")
  }

  const log = getLogger(opts.logLevel, "main")
  const flatTasks = tasks.flatMap(step => (Array.isArray(step) ? step : [step]))
  const duplicateTaskKeys = []
  const seenTaskKeys = new Set()

  for (const task of flatTasks) {
    const taskKey = getTaskKey(task)
    if (!taskKey) {
      throw new Error("Each task must define `key` (or legacy `name`).")
    }
    if (seenTaskKeys.has(taskKey)) {
      duplicateTaskKeys.push(taskKey)
    } else {
      seenTaskKeys.add(taskKey)
    }
  }

  if (duplicateTaskKeys.length > 0) {
    const uniqueDuplicates = [...new Set(duplicateTaskKeys)]
    throw new Error(`Duplicate task keys found: ${uniqueDuplicates.join(", ")}`)
  }

  log.info(`Processing ${tasks.length} steps`)
  log.debug(`Running directory: ${opts.runDir}`)
  log.debug(`Output directory: ${opts.outDir}`)

  if (opts.cacheDir) {
    log.debug(`Cache directory: ${opts.cacheDir}`)
  } else {
    log.warn("Cache disabled")
  }

  const taskRunner = tasks.reduce(
    async (metaPromise, step) => {
      const stepTasks = Array.isArray(step) ? step : [step]
      const { meta, filesWritten } = await metaPromise
      const stepTaskNames = stepTasks.map(task => getTaskName(task))
      log.info(`Starting tasks: ${stepTaskNames.join(", ")}`)
      const stepResults = await Promise.all(stepTasks.map(async task => {
        const taskLog = getLogger(opts.logLevel, getTaskName(task))
        const taskResult = await processTask(meta, task)
        taskLog.trace(`taskResult: ${JSON.stringify(taskResult)}`)
        return taskResult
      }))
      return stepResults.reduce((newState, taskResult) => {
        const resources = Object.keys(taskResult.resources).length > 0
          ? {
            ...newState.meta.resources,
            [taskResult.key]: taskResult.resources,
          }
          : { ...newState.meta.resources }

        return {
          meta: {
            ...newState.meta,
            resources,
          },
          filesWritten: newState.filesWritten + taskResult.filesWritten,
        }
      }, { meta, filesWritten })
    },
    Promise.resolve({ meta: { opts }, filesWritten: 0 }),
  )

  const finalState = await taskRunner
  log.trace(`Final state: ${JSON.stringify(finalState, null, 2)}`)

  const endTime = performance.now()
  const timeTaken = endTime - startTime
  const hrTime = timeTaken > 1000
    ? `${Number.parseFloat(timeTaken / 1000).toFixed(2)}s`
    : `${Number.parseFloat(timeTaken).toFixed(2)}ms`

  log.info(
    `Completed ${tasks.length} steps in ${hrTime}, wrote ${finalState.filesWritten} files.`,
  )

  return finalState
}