Zacharyb/fix job dispatch status

apps/event-worker/src/index.ts

@@ -1,23 +1,13 @@
 import { logger } from "@ctrlplane/logger";
 
 import { register } from "./instrumentation.js";
-import { createDispatchExecutionJobWorker } from "./job-dispatch/index.js";
 import { redis } from "./redis.js";
-import { createReleaseNewVersionWorker } from "./releases/new-version/index.js";
-import { createReleaseVariableChangeWorker } from "./releases/variable-change/index.js";
-import { createResourceScanWorker } from "./resource-scan/index.js";
 import { workers } from "./workers/index.js";
 
 console.log("Registering instrumentation...");
 await register();
 
-const allWorkers = [
-  createResourceScanWorker(),
-  createDispatchExecutionJobWorker(),
-  createReleaseNewVersionWorker(),
-  createReleaseVariableChangeWorker(),
-  ...Object.values(workers),
-];
+const allWorkers = Object.values(workers);
 
 const shutdown = () => {
   logger.warn("Exiting...");

apps/event-worker/src/releases/create-job-from-release.ts

@@ -0,0 +1,72 @@
+import type { Tx } from "@ctrlplane/db";
+import _ from "lodash";
+
+import { eq, takeFirstOrNull } from "@ctrlplane/db";
+import { db } from "@ctrlplane/db/client";
+import * as schema from "@ctrlplane/db/schema";
+import { JobStatus } from "@ctrlplane/validators/jobs";
+
+const copyReleaseVariables = async (
+  tx: Tx,
+  jobId: string,
+  releaseId: string,
+): Promise<void> => {
+  const releaseVars = await tx.query.releaseVariable.findMany({
+    where: eq(schema.releaseVariable.releaseId, releaseId),
+  });
+
+  if (releaseVars.length === 0) return;
+
+  await tx.insert(schema.jobVariable).values(
+    releaseVars.map((v) => ({
+      jobId,
+      ..._.pick(v, ["key", "value", "sensitive"]),
+    })),
+  );
+};
+
+const getJobAgentConfig = async (tx: Tx, versionId: string) => {
+  const { job_agent: jobAgent } =
+    (await tx
+      .select()
+      .from(schema.deploymentVersion)
+      .innerJoin(
+        schema.deployment,
+        eq(schema.deploymentVersion.deploymentId, schema.deployment.id),
+      )
+      .innerJoin(
+        schema.jobAgent,
+        eq(schema.deployment.jobAgentId, schema.jobAgent.id),
+      )
+      .where(eq(schema.deploymentVersion.id, versionId))
+      .then(takeFirstOrNull)) ?? {};
+
+  if (jobAgent == null)
+    throw new Error(`Job agent not found for version ${versionId}`);
+  return {
+    jobAgentId: jobAgent.id,
+    jobAgentConfig: _.merge({}, jobAgent.config),
+  };
+};
+
+export const createJobFromRelease = async (
+  release: typeof schema.release.$inferSelect,
+) => {
+  return db.transaction(async (tx) => {
+    // Create the job
+    const job = await tx
+      .insert(schema.job)
+      .values({
+        status: JobStatus.Pending,
+        ...(await getJobAgentConfig(tx, release.versionId)),
+      })
+      .returning()
+      .then(takeFirstOrNull);
+
+    if (job == null) return null;
+
+    await copyReleaseVariables(tx, job.id, release.id);
+
+    return job;
+  });
+};

apps/event-worker/src/releases/create-release.ts

@@ -1,17 +1,15 @@
 import type { ReleaseRepository } from "@ctrlplane/rule-engine";
-import { Queue } from "bullmq";
 
+import { db } from "@ctrlplane/db/client";
+import * as schema from "@ctrlplane/db/schema";
+import { Channel, getQueue } from "@ctrlplane/events";
 import { ReleaseManager } from "@ctrlplane/release-manager";
-import { Channel } from "@ctrlplane/validators/events";
 
-import { redis } from "../redis.js";
 import { ReleaseRepositoryMutex } from "./mutex.js";
 
-const evaluate = new Queue(Channel.ReleaseEvaluate, {
-  connection: redis,
-});
+const policyEvaluateQueue = getQueue(Channel.PolicyEvaluate);
 
-export const createAndEvaluateRelease = async (
+const createReleaseWithLock = async (
   repo: ReleaseRepository,
   versionId?: string,
 ) => {
@@ -31,9 +29,34 @@ export const createAndEvaluateRelease = async (
     // it causes issues.
     await releaseManager.setDesiredRelease(release.id);
 
-    await evaluate.add(release.id, repo);
+    await policyEvaluateQueue.add(release.id, repo);
   } finally {
     // Always release the mutex lock
     await mutex.unlock();
   }
 };
+
+export const createReleases = async (
+  releaseTargets: (typeof schema.releaseTarget.$inferInsert & {
+    versionId?: string;
+  })[],
+) => {
+  // First upsert all release targets
+  await db
+    .insert(schema.releaseTarget)
+    .values(releaseTargets)
+    .onConflictDoNothing();
+
+  // Create releases and evaluate for each target
+  await Promise.all(
+    releaseTargets.map(async (target) => {
+      const repo = {
+        deploymentId: target.deploymentId,
+        environmentId: target.environmentId,
+        resourceId: target.resourceId,
+      };
+
+      await createReleaseWithLock(repo, target.versionId);
+    }),
+  );
+};