ScheduleTaskManager/server/src/services/taskSync.ts

import { and, eq, notInArray } from "drizzle-orm";
import { db } from "../db/client.js";
import { scheduledTasks, servers } from "../db/schema.js";

export interface IncomingTask {
  scheduleType: "cron" | "systemd_timer";
  name: string;
  command?: string;
  scheduleExpression?: string;
  source?: string;
  enabled?: boolean;
  nextRunAt?: string;
  metadata?: unknown;
}

export interface AgentReport {
  hostname?: string;
  tasks: IncomingTask[];
}

export async function syncServerTasks(serverId: number, report: AgentReport) {
  const now = new Date().toISOString();

  const existing = await db.query.scheduledTasks.findMany({
    where: and(eq(scheduledTasks.serverId, serverId), eq(scheduledTasks.origin, "agent")),
  });

  const seenIds: number[] = [];

  for (const task of report.tasks) {
    const match = existing.find(
      (t) =>
        t.scheduleType === task.scheduleType &&
        t.name === task.name &&
        (t.source ?? "") === (task.source ?? ""),
    );

    if (match) {
      const [updated] = await db
        .update(scheduledTasks)
        .set({
          command: task.command,
          scheduleExpression: task.scheduleExpression,
          enabled: task.enabled ?? true,
          nextRunAt: task.nextRunAt,
          rawMetadata: task.metadata ? JSON.stringify(task.metadata) : null,
          isStale: false,
          lastSeenAt: now,
        })
        .where(eq(scheduledTasks.id, match.id))
        .returning({ id: scheduledTasks.id });
      seenIds.push(updated.id);
    } else {
      const [created] = await db
        .insert(scheduledTasks)
        .values({
          serverId,
          scheduleType: task.scheduleType,
          origin: "agent",
          name: task.name,
          command: task.command,
          scheduleExpression: task.scheduleExpression,
          source: task.source,
          enabled: task.enabled ?? true,
          nextRunAt: task.nextRunAt,
          rawMetadata: task.metadata ? JSON.stringify(task.metadata) : null,
          firstSeenAt: now,
          lastSeenAt: now,
        })
        .returning({ id: scheduledTasks.id });
      seenIds.push(created.id);
    }
  }

  // Anything agent-sourced for this server that wasn't in this report is now stale,
  // rather than deleted, so a bad/partial agent run doesn't wipe history. Manually
  // entered tasks (origin = 'manual') are never touched by agent sync.
  const agentScope = and(eq(scheduledTasks.serverId, serverId), eq(scheduledTasks.origin, "agent"));
  if (seenIds.length > 0) {
    await db
      .update(scheduledTasks)
      .set({ isStale: true })
      .where(and(agentScope, notInArray(scheduledTasks.id, seenIds)));
  } else {
    await db.update(scheduledTasks).set({ isStale: true }).where(agentScope);
  }

  await db
    .update(servers)
    .set({ lastSeenAt: now, ...(report.hostname ? { hostname: report.hostname } : {}) })
    .where(eq(servers.id, serverId));
}