gigafibre-fsm/services/targo-hub/lib/flow-api.js

'use strict'
/**
 * flow-api.js — HTTP endpoints for the flow runtime.
 *
 *   POST /flow/start        { template, doctype?, docname?, customer?, variables? }
 *   POST /flow/advance      { run }                   — re-evaluate a run (idempotent)
 *   POST /flow/complete     { run, step_id, result? } — external completion (scheduler / webhook / manual)
 *   POST /flow/event        { event, ctx? }           — dispatch a trigger event to all listening templates
 *   GET  /flow/runs         ?customer=…&template=…&status=…
 *   GET  /flow/runs/:name   — fetch a run (with step_state parsed)
 *
 * Security: internal endpoints (start/advance/complete/event) optionally require
 * cfg.INTERNAL_TOKEN via `Authorization: Bearer …` so only the ERPNext scheduler
 * and trusted services can launch flows.
 */

const { log, json, parseBody, erpFetch } = require('./helpers')
const runtime = require('./flow-runtime')
const cfg = require('./config')

const ENC_FR = encodeURIComponent('Flow Run')

// ── Auth guard (soft — allows open in dev when INTERNAL_TOKEN is unset) ────

function checkInternalAuth (req) {
  if (!cfg.INTERNAL_TOKEN) return true
  const a = req.headers.authorization || ''
  return a === 'Bearer ' + cfg.INTERNAL_TOKEN
}

// ── Handlers ────────────────────────────────────────────────────────────────

async function startFlow (req, res) {
  if (!checkInternalAuth(req)) return json(res, 401, { error: 'Unauthorized' })
  const body = await parseBody(req)
  if (!body.template) return json(res, 400, { error: 'template required' })
  try {
    const result = await runtime.startFlow(body.template, {
      doctype:      body.doctype,
      docname:      body.docname,
      customer:     body.customer,
      variables:    body.variables || {},
      triggerEvent: body.trigger_event || body.triggerEvent,
    })
    return json(res, 201, result)
  } catch (e) {
    log('[flow-api] start failed:', e.message)
    return json(res, 500, { error: e.message })
  }
}

async function advanceFlow (req, res) {
  if (!checkInternalAuth(req)) return json(res, 401, { error: 'Unauthorized' })
  const body = await parseBody(req)
  if (!body.run) return json(res, 400, { error: 'run required' })
  try { return json(res, 200, await runtime.advanceFlow(body.run)) }
  catch (e) { return json(res, 500, { error: e.message }) }
}

async function completeStep (req, res) {
  if (!checkInternalAuth(req)) return json(res, 401, { error: 'Unauthorized' })
  const body = await parseBody(req)
  if (!body.run || !body.step_id) return json(res, 400, { error: 'run and step_id required' })
  try { return json(res, 200, await runtime.completeStep(body.run, body.step_id, body.result || {})) }
  catch (e) { return json(res, 500, { error: e.message }) }
}

async function dispatchEvent (req, res) {
  if (!checkInternalAuth(req)) return json(res, 401, { error: 'Unauthorized' })
  const body = await parseBody(req)
  if (!body.event) return json(res, 400, { error: 'event required' })
  try {
    const results = await runtime.dispatchEvent(body.event, body.ctx || {})
    return json(res, 200, { event: body.event, started: results.length, results })
  } catch (e) {
    return json(res, 500, { error: e.message })
  }
}

async function listRuns (req, res, urlObj) {
  const p = Object.fromEntries(urlObj.searchParams)
  const filters = []
  if (p.customer)  filters.push(['customer', '=', p.customer])
  if (p.template)  filters.push(['flow_template', '=', p.template])
  if (p.status)    filters.push(['status', '=', p.status])
  if (p.doctype && p.docname) {
    filters.push(['context_doctype', '=', p.doctype])
    filters.push(['context_docname', '=', p.docname])
  }
  const qs = new URLSearchParams({
    fields: JSON.stringify(['name', 'flow_template', 'status', 'customer', 'context_doctype', 'context_docname', 'started_at', 'completed_at', 'last_error']),
    filters: JSON.stringify(filters),
    limit_page_length: String(p.limit || 50),
    order_by: 'started_at desc',
  })
  const r = await erpFetch(`/api/resource/${ENC_FR}?${qs}`)
  if (r.status !== 200) return json(res, r.status, { error: 'Failed to list', detail: r.data })
  return json(res, 200, { runs: r.data.data || [] })
}

async function getRun (req, res, name) {
  const r = await erpFetch(`/api/resource/${ENC_FR}/${encodeURIComponent(name)}`)
  if (r.status !== 200) return json(res, r.status, { error: 'Not found' })
  const run = r.data.data
  try { run.variables = JSON.parse(run.variables || '{}') } catch { /* noop */ }
  try { run.step_state = JSON.parse(run.step_state || '{}') } catch { /* noop */ }
  return json(res, 200, { run })
}

// ── Router ──────────────────────────────────────────────────────────────────

async function handle (req, res, method, urlPath, urlObj) {
  if (urlPath === '/flow/start'    && method === 'POST') return startFlow(req, res)
  if (urlPath === '/flow/advance'  && method === 'POST') return advanceFlow(req, res)
  if (urlPath === '/flow/complete' && method === 'POST') return completeStep(req, res)
  if (urlPath === '/flow/event'    && method === 'POST') return dispatchEvent(req, res)

  const runMatch = urlPath.match(/^\/flow\/runs\/([^/]+)$/)
  if (runMatch && method === 'GET') return getRun(req, res, decodeURIComponent(runMatch[1]))

  if (urlPath === '/flow/runs' && method === 'GET') return listRuns(req, res, urlObj)

  return json(res, 404, { error: 'Flow endpoint not found' })
}

module.exports = { handle }