memind/services/orchestrator/app.mjs

import express from 'express';

function errorStatus(error) {
  if (Number.isInteger(error?.status)) return error.status;
  if (['WORKFLOW_NOT_SUPPORTED', 'WORKFLOW_OBSERVE_ONLY_REQUIRED'].includes(error?.code)) {
    return 422;
  }
  return 500;
}

function authorize(serviceToken) {
  const expected = String(serviceToken ?? '').trim();
  return (request, response, next) => {
    if (!expected) return next();
    if (request.get('authorization') === `Bearer ${expected}`) return next();
    return response.status(401).json({
      error: {
        code: 'UNAUTHORIZED',
        message: 'Missing or invalid service token',
      },
    });
  };
}

function authorizeWorker(workerToken) {
  const expected = String(workerToken ?? '').trim();
  return (request, response, next) => {
    if (!expected) return next();
    if (request.get('x-orchestrator-worker-token') === expected) return next();
    return response.status(401).json({
      error: {
        code: 'WORKER_UNAUTHORIZED',
        message: 'Missing or invalid worker token',
      },
    });
  };
}

function prometheusMetrics(metrics) {
  const lines = [
    '# HELP memind_orchestrator_executor_jobs Executor jobs by state.',
    '# TYPE memind_orchestrator_executor_jobs gauge',
  ];
  for (const [status, count] of Object.entries(metrics.queue?.counts ?? {})) {
    lines.push(`memind_orchestrator_executor_jobs{status="${status}"} ${Number(count) || 0}`);
  }
  lines.push(
    '# HELP memind_orchestrator_executor_claimable Claimable executor jobs.',
    '# TYPE memind_orchestrator_executor_claimable gauge',
    `memind_orchestrator_executor_claimable ${Number(metrics.queue?.claimable) || 0}`,
    '# HELP memind_orchestrator_executor_expired_leases Expired executor leases.',
    '# TYPE memind_orchestrator_executor_expired_leases gauge',
    `memind_orchestrator_executor_expired_leases ${Number(metrics.queue?.expiredLeases) || 0}`,
    '# HELP memind_orchestrator_executor_workers Healthy registered workers.',
    '# TYPE memind_orchestrator_executor_workers gauge',
    `memind_orchestrator_executor_workers ${Number(metrics.queue?.healthyWorkers) || 0}`,
    '# HELP memind_orchestrator_execution_enabled Whether executor queue admission is enabled.',
    '# TYPE memind_orchestrator_execution_enabled gauge',
    `memind_orchestrator_execution_enabled ${metrics.gateway?.executionEnabled ? 1 : 0}`,
  );
  return `${lines.join('\n')}\n`;
}

export function createOrchestratorApp({
  runtime,
  serviceToken = process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN,
  workerToken = process.env.MEMIND_ORCHESTRATOR_WORKER_TOKEN,
} = {}) {
  if (!runtime) throw new Error('Orchestrator app requires runtime');
  const app = express();
  app.disable('x-powered-by');
  app.use(express.json({ limit: '256kb' }));

  app.get('/health', (_request, response) => response.json(runtime.health()));
  app.get('/ready', async (_request, response) => {
    try {
      return response.json(await runtime.ready());
    } catch {
      return response.status(503).json({
        ...runtime.health(),
        ready: false,
      });
    }
  });
  app.get('/metrics', async (_request, response, next) => {
    try {
      const metrics = await runtime.metrics();
      return response
        .type('text/plain; version=0.0.4; charset=utf-8')
        .send(prometheusMetrics(metrics));
    } catch (error) {
      return next(error);
    }
  });

  app.use('/v1', authorize(serviceToken));

  app.post('/v1/runs', async (request, response, next) => {
    try {
      response.status(202).json(await runtime.start(request.body));
    } catch (error) {
      next(error);
    }
  });

  app.post('/v1/executor-jobs', async (request, response, next) => {
    try {
      const result = await runtime.submitExecutorJob(request.body);
      return response.status(result.created ? 202 : 200).json(result);
    } catch (error) {
      return next(error);
    }
  });

  app.get('/v1/runs/:runId', async (request, response, next) => {
    try {
      const state = await runtime.getState(request.params.runId);
      if (!state) return response.status(404).json({ error: { code: 'RUN_NOT_FOUND' } });
      return response.json(state);
    } catch (error) {
      return next(error);
    }
  });

  app.get('/v1/runs/:runId/events', async (request, response, next) => {
    try {
      const result = await runtime.listEvents(request.params.runId, {
        after: request.query.after,
      });
      if (!result) return response.status(404).json({ error: { code: 'RUN_NOT_FOUND' } });
      return response.json(result);
    } catch (error) {
      return next(error);
    }
  });

  app.post('/v1/runs/:runId/resume', async (request, response, next) => {
    try {
      const state = await runtime.resume(request.params.runId, request.body);
      if (!state) return response.status(404).json({ error: { code: 'RUN_NOT_FOUND' } });
      return response.json(state);
    } catch (error) {
      return next(error);
    }
  });

  app.post('/v1/runs/:runId/cancel', async (request, response, next) => {
    try {
      const state = await runtime.cancel(request.params.runId, request.body);
      if (!state) return response.status(404).json({ error: { code: 'RUN_NOT_FOUND' } });
      return response.json(state);
    } catch (error) {
      return next(error);
    }
  });

  app.delete('/v1/runs/:runId', async (request, response, next) => {
    try {
      const result = await runtime.deleteRun(request.params.runId);
      if (!result) return response.status(404).json({ error: { code: 'RUN_NOT_FOUND' } });
      return response.json(result);
    } catch (error) {
      return next(error);
    }
  });

  app.post('/v1/maintenance/purge-terminal-runs', async (request, response, next) => {
    try {
      return response.json(await runtime.purgeTerminalRuns({
        before: request.body?.before,
        limit: request.body?.limit,
        dryRun: request.body?.apply !== true,
      }));
    } catch (error) {
      return next(error);
    }
  });

  app.get('/v1/executor-jobs/:jobId', async (request, response, next) => {
    try {
      const job = await runtime.getExecutorJob(request.params.jobId);
      if (!job) {
        return response.status(404).json({ error: { code: 'EXECUTOR_JOB_NOT_FOUND' } });
      }
      return response.json(job);
    } catch (error) {
      return next(error);
    }
  });

  app.get('/v1/executor-jobs/:jobId/events', async (request, response, next) => {
    try {
      const result = await runtime.listExecutorJobEvents(request.params.jobId, {
        after: request.query.after,
        limit: request.query.limit,
      });
      if (!result) {
        return response.status(404).json({ error: { code: 'EXECUTOR_JOB_NOT_FOUND' } });
      }
      return response.json(result);
    } catch (error) {
      return next(error);
    }
  });

  const requireWorker = authorizeWorker(workerToken);

  app.post('/v1/workers/claim', requireWorker, async (request, response, next) => {
    try {
      const claim = await runtime.claimExecutorJob(request.body);
      return claim ? response.json(claim) : response.status(204).end();
    } catch (error) {
      return next(error);
    }
  });

  app.post(
    '/v1/workers/jobs/:jobId/start',
    requireWorker,
    async (request, response, next) => {
      try {
        return response.json(await runtime.startExecutorJob(request.params.jobId, request.body));
      } catch (error) {
        return next(error);
      }
    },
  );

  app.post(
    '/v1/workers/jobs/:jobId/heartbeat',
    requireWorker,
    async (request, response, next) => {
      try {
        return response.json(await runtime.heartbeatExecutorJob(request.params.jobId, request.body));
      } catch (error) {
        return next(error);
      }
    },
  );

  app.post(
    '/v1/workers/jobs/:jobId/progress',
    requireWorker,
    async (request, response, next) => {
      try {
        return response.json(await runtime.progressExecutorJob(request.params.jobId, request.body));
      } catch (error) {
        return next(error);
      }
    },
  );

  app.post(
    '/v1/workers/jobs/:jobId/complete',
    requireWorker,
    async (request, response, next) => {
      try {
        return response.json(await runtime.completeExecutorJob(request.params.jobId, request.body));
      } catch (error) {
        return next(error);
      }
    },
  );

  app.post(
    '/v1/workers/jobs/:jobId/fail',
    requireWorker,
    async (request, response, next) => {
      try {
        return response.json(await runtime.failExecutorJob(request.params.jobId, request.body));
      } catch (error) {
        return next(error);
      }
    },
  );

  app.post('/v1/workers/recover-expired', requireWorker, async (request, response, next) => {
    try {
      return response.json(await runtime.recoverExpiredExecutorJobs(request.body));
    } catch (error) {
      return next(error);
    }
  });

  app.get('/v1/workers/stats', requireWorker, async (_request, response, next) => {
    try {
      return response.json(await runtime.getExecutorQueueStats());
    } catch (error) {
      return next(error);
    }
  });

  app.post('/v1/executor-jobs/:jobId/cancel', async (request, response, next) => {
    try {
      const job = await runtime.cancelExecutorJob(request.params.jobId, request.body);
      if (!job) {
        return response.status(404).json({ error: { code: 'EXECUTOR_JOB_NOT_FOUND' } });
      }
      return response.json(job);
    } catch (error) {
      return next(error);
    }
  });

  app.use((error, _request, response, _next) => {
    response.status(errorStatus(error)).json({
      error: {
        code: error?.code ?? 'ORCHESTRATOR_ERROR',
        message: error instanceof Error ? error.message : 'Orchestrator request failed',
      },
    });
  });

  return app;
}

export const orchestratorAppInternals = {
  authorize,
  authorizeWorker,
  errorStatus,
  prometheusMetrics,
};