platform/src/servers/api/tasks/pipeline-jobs-routes.ts

import { stat, open } from 'node:fs/promises';
import { createRouter } from '../../create-router';
import * as errors from '../../custom-errors';
import { getPipelineJobsForUser, getPipelineJob } from 'officerdb';
import { getTaskByDirName } from './task-files';
import * as jobManager from './pipeline-job-manager';
import { jobLogPath } from './execute-script';

export const pipelineJobsRouter = createRouter();

// GET / — list the user's jobs (optionally only the live ones).
pipelineJobsRouter.get('/', async (c) => {
  const user = c.get('user');
  const liveOnly = c.req.query('live') === '1';
  const jobs = await jobManager.getJobsForUser(user.id);
  return c.json(
    jobs
      .filter((j) => (liveOnly ? j.isLive : true))
      .map((j) => ({
        id: j.id,
        mode: j.mode,
        taskDirName: j.taskDirName,
        taskName: j.taskName,
        status: j.status,
        isLive: j.isLive,
        exitCode: j.exitCode,
        // The file/folder the job is working on (for the list rows).
        target: (j.inputs as Record<string, unknown> | null)?.file_path ?? j.cwd ?? null,
        totalCost: j.totalCost,
        createdAt: j.createdAt,
        startedAt: j.startedAt,
        completedAt: j.completedAt,
        error: j.error,
      })),
  );
});

// POST / — create a job (script or pipeline). action 'start' runs now; 'queue' waits behind a running
// job. Returns { jobId, status }. This is the REST creation path the phone / unattended runs use.
pipelineJobsRouter.post('/', async (c) => {
  const user = c.get('user');
  const body = await c.req.json<{
    taskDirName: string;
    inputs?: Record<string, string>;
    cwd?: string;
    action?: 'start' | 'queue';
  }>();
  if (!body.taskDirName) throw errors.BAD_REQUEST('taskDirName is required');

  const task = await getTaskByDirName(body.taskDirName);
  if (!task) throw errors.NOT_FOUND(`Task not found: ${body.taskDirName}`);
  const mode = task.mode as 'pipeline' | 'script' | 'agentic';
  if (mode !== 'script' && mode !== 'pipeline') throw errors.BAD_REQUEST(`Task mode '${mode}' can't run as a job yet`);

  const { jobId, status } = await jobManager.enqueueJob(
    {
      userId: user.id,
      email: user.email,
      username: user.username ?? '',
      mode,
      taskDirName: body.taskDirName,
      taskName: task.name,
      inputs: body.inputs ?? {},
      cwd: body.cwd,
      config: task.config,
    },
    body.action === 'queue' ? 'queue' : 'start',
  );

  return c.json({ jobId, status });
});

// GET /counts — header-badge summary { running, runningJobId, queued }. Before /:id so it isn't
// captured as an id.
pipelineJobsRouter.get('/counts', async (c) => {
  const user = c.get('user');
  return c.json(await jobManager.getCounts(user.id));
});

// GET /:id/log?offset= — tail the persisted output log (script jobs). Returns text from `offset`.
pipelineJobsRouter.get('/:id/log', async (c) => {
  const user = c.get('user');
  const job = await getPipelineJob(c.req.param('id'));
  if (!job || job.userId !== user.id) return c.json({ error: 'Not found' }, 404);

  const offset = Math.max(0, parseInt(c.req.query('offset') ?? '0', 10) || 0);
  const path = jobLogPath(job.id);
  let size = 0;
  try {
    size = (await stat(path)).size;
  } catch {
    return c.json({ text: '', offset, size: 0 }); // no log yet
  }
  if (offset >= size) return c.json({ text: '', offset: size, size });

  const fh = await open(path, 'r');
  try {
    const buf = Buffer.alloc(size - offset);
    await fh.read(buf, 0, buf.length, offset);
    return c.json({ text: buf.toString('utf8'), offset: size, size });
  } finally {
    await fh.close();
  }
});

// POST /:id/stop — stop a running job or cancel a queued one.
pipelineJobsRouter.post('/:id/stop', async (c) => {
  const user = c.get('user');
  const job = await getPipelineJob(c.req.param('id'));
  if (!job || job.userId !== user.id) return c.json({ error: 'Not found' }, 404);
  const result = await jobManager.requestStop(job.id);
  return c.json({ ok: true, result });
});

// DELETE /history — clear all finished jobs (rows + logs). Before /:id so it isn't captured as an id.
pipelineJobsRouter.delete('/history', async (c) => {
  const user = c.get('user');
  const cleared = await jobManager.clearHistory(user.id);
  return c.json({ ok: true, cleared });
});

// DELETE /:id — delete a job (queued or finished). Running jobs must be stopped first (409).
pipelineJobsRouter.delete('/:id', async (c) => {
  const user = c.get('user');
  const job = await getPipelineJob(c.req.param('id'));
  if (!job || job.userId !== user.id) return c.json({ error: 'Not found' }, 404);
  const result = await jobManager.deleteJob(job.id);
  if (result === 'running') return c.json({ error: 'Stop the job before deleting' }, 409);
  return c.json({ ok: true, result });
});

// GET /:id — single job detail (full row).
pipelineJobsRouter.get('/:id', async (c) => {
  const user = c.get('user');
  const job = await getPipelineJob(c.req.param('id'));
  if (!job || job.userId !== user.id) return c.json({ error: 'Not found' }, 404);
  return c.json(job);
});