pr-preview/backend/dist/workers/jobWorker.js

"use strict";
var __defProp = Object.defineProperty;
var __getOwnPropDesc = Object.getOwnPropertyDescriptor;
var __getOwnPropNames = Object.getOwnPropertyNames;
var __hasOwnProp = Object.prototype.hasOwnProperty;
var __export = (target, all) => {
  for (var name in all)
    __defProp(target, name, { get: all[name], enumerable: true });
};
var __copyProps = (to, from, except, desc) => {
  if (from && typeof from === "object" || typeof from === "function") {
    for (let key of __getOwnPropNames(from))
      if (!__hasOwnProp.call(to, key) && key !== except)
        __defProp(to, key, { get: () => from[key], enumerable: !(desc = __getOwnPropDesc(from, key)) || desc.enumerable });
  }
  return to;
};
var __toCommonJS = (mod) => __copyProps(__defProp({}, "__esModule", { value: true }), mod);
var jobWorker_exports = {};
__export(jobWorker_exports, {
  startJobWorker: () => startJobWorker,
  stopJobWorker: () => stopJobWorker
});
module.exports = __toCommonJS(jobWorker_exports);
var import_db = require("../lib/db");
var import_logger = require("../lib/logger");
var import_deploy = require("../services/deploy");
const log = (0, import_logger.createLogger)("JOB_WORKER");
let running = false;
async function startJobWorker() {
  if (running) return;
  running = true;
  log.info("Job worker started");
  await resetStuckJobs();
  pollLoop();
}
async function resetStuckJobs() {
  const count = await import_db.prisma.job.updateMany({
    where: { status: "RUNNING" },
    data: { status: "PENDING", startedAt: null }
  });
  if (count.count > 0) log.info({ count: count.count }, "Reset stuck running jobs to PENDING");
}
async function pollLoop() {
  while (running) {
    try {
      await processPendingJobs();
    } catch (e) {
      log.error({ e }, "Job worker poll error");
    }
    await sleep(1e3);
  }
}
const activePreviewJobs = /* @__PURE__ */ new Map();
async function processPendingJobs() {
  const pending = await import_db.prisma.job.findMany({
    where: { status: "PENDING" },
    orderBy: { createdAt: "asc" },
    take: 20
  });
  for (const job of pending) {
    const previewId = job.previewId;
    if (!previewId) continue;
    if (activePreviewJobs.has(previewId)) {
      const existingJobId = activePreviewJobs.get(previewId);
      if (job.type === "DEPLOY") {
        log.info({ previewId, newJobId: job.id, abortingJobId: existingJobId }, "New deploy cancels existing");
        (0, import_deploy.signalAbort)(previewId);
        await sleep(500);
      } else {
        continue;
      }
    }
    activePreviewJobs.set(previewId, job.id);
    processJob(job).finally(() => {
      if (activePreviewJobs.get(previewId) === job.id) {
        activePreviewJobs.delete(previewId);
      }
    });
  }
}
async function processJob(job) {
  log.info({ jobId: job.id, type: job.type, previewId: job.previewId }, "Processing job");
  try {
    if (job.type === "DEPLOY") {
      await (0, import_deploy.runDeploy)(job.id);
    } else if (job.type === "STOP" || job.type === "INACTIVITY_STOP") {
      if (job.previewId) {
        await import_db.prisma.job.update({ where: { id: job.id }, data: { status: "RUNNING", startedAt: /* @__PURE__ */ new Date() } });
        await (0, import_deploy.stopPreview)(job.previewId);
        await import_db.prisma.job.update({ where: { id: job.id }, data: { status: "DONE", finishedAt: /* @__PURE__ */ new Date() } });
      }
    }
  } catch (e) {
    log.error({ e, jobId: job.id }, "Job processing error");
    await import_db.prisma.job.update({
      where: { id: job.id },
      data: { status: "FAILED", error: e.message, finishedAt: /* @__PURE__ */ new Date() }
    }).catch(() => {
    });
  }
}
function stopJobWorker() {
  running = false;
}
function sleep(ms) {
  return new Promise((r) => setTimeout(r, ms));
}
// Annotate the CommonJS export names for ESM import in node:
0 && (module.exports = {
  startJobWorker,
  stopJobWorker
});
//# sourceMappingURL=jobWorker.js.map