refactor: move upsert release targets to selector utility

apps/event-worker/src/utils/dispatch-evaluate-jobs.ts

@@ -0,0 +1,11 @@
+import type * as schema from "@ctrlplane/db/schema";
+
+import { Channel, getQueue } from "@ctrlplane/events";
+
+export const dispatchEvaluateJobs = async (rts: schema.ReleaseTarget[]) => {
+  const jobs = rts.map((rt) => ({
+    name: `${rt.resourceId}-${rt.environmentId}-${rt.deploymentId}`,
+    data: rt,
+  }));
+  await getQueue(Channel.EvaluateReleaseTarget).addBulk(jobs);
+};

apps/event-worker/src/workers/new-deployment-version.ts

@@ -1,7 +1,9 @@
 import { eq } from "@ctrlplane/db";
 import { db } from "@ctrlplane/db/client";
 import * as schema from "@ctrlplane/db/schema";
-import { Channel, createWorker, getQueue } from "@ctrlplane/events";
+import { Channel, createWorker } from "@ctrlplane/events";
+
+import { dispatchEvaluateJobs } from "../utils/dispatch-evaluate-jobs.js";
 
 /**
  * Worker that processes new deployment version events.
@@ -18,12 +20,6 @@ export const newDeploymentVersionWorker = createWorker(
     const releaseTargets = await db.query.releaseTarget.findMany({
       where: eq(schema.releaseTarget.deploymentId, version.deploymentId),
     });
-
-    await getQueue(Channel.EvaluateReleaseTarget).addBulk(
-      releaseTargets.map((rt) => ({
-        name: `${rt.resourceId}-${rt.environmentId}-${rt.deploymentId}`,
-        data: rt,
-      })),
-    );
+    await dispatchEvaluateJobs(releaseTargets);
   },
 );

apps/event-worker/src/workers/new-deployment.ts

@@ -1,13 +1,22 @@
-import { eq, selector, takeFirst } from "@ctrlplane/db";
+import { eq, selector } from "@ctrlplane/db";
 import { db } from "@ctrlplane/db/client";
 import * as schema from "@ctrlplane/db/schema";
-import { Channel, createWorker, getQueue } from "@ctrlplane/events";
+import { Channel, createWorker } from "@ctrlplane/events";
 import { logger } from "@ctrlplane/logger";
 
-import { replaceReleaseTargets } from "../utils/replace-release-targets.js";
+import { dispatchEvaluateJobs } from "../utils/dispatch-evaluate-jobs.js";
 
 const log = logger.child({ module: "new-deployment" });
 
+const recomputeReleaseTargets = async (
+  deployment: schema.Deployment,
+  resources: schema.Resource[],
+) => {
+  const computeBuilder = selector().compute();
+  await computeBuilder.deployments([deployment]).resourceSelectors();
+  return computeBuilder.resources(resources).releaseTargets();
+};
+
 /**
  * Worker that processes new deployment events.
  *
@@ -24,43 +33,17 @@ export const newDeploymentWorker = createWorker(
   Channel.NewDeployment,
   async (job) => {
     try {
-      await selector()
-        .compute()
-        .deployments([job.data.id])
-        .resourceSelectors()
-        .replace();
-
-      const system = await db
-        .select()
-        .from(schema.system)
-        .where(eq(schema.system.id, job.data.systemId))
-        .then(takeFirst);
-      const { workspaceId } = system;
+      const environments = await db.query.environment.findMany({
+        where: eq(schema.environment.systemId, job.data.systemId),
+        with: { computedResources: { with: { resource: true } } },
+      });
 
-      const computedDeploymentResources = await db
-        .select()
-        .from(schema.computedDeploymentResource)
-        .innerJoin(
-          schema.resource,
-          eq(schema.computedDeploymentResource.resourceId, schema.resource.id),
-        )
-        .where(eq(schema.computedDeploymentResource.deploymentId, job.data.id));
-      const resources = computedDeploymentResources.map((r) => r.resource);
-
-      const releaseTargetPromises = resources.map(async (r) =>
-        replaceReleaseTargets(db, r),
+      const resources = environments.flatMap((e) =>
+        e.computedResources.map((r) => r.resource),
       );
-      const fulfilled = await Promise.all(releaseTargetPromises);
-      const rts = fulfilled.flat();
-
-      await selector()
-        .compute()
-        .allPolicies(workspaceId)
-        .releaseTargetSelectors()
-        .replace();
 
-      const evaluateJobs = rts.map((rt) => ({ name: rt.id, data: rt }));
-      await getQueue(Channel.EvaluateReleaseTarget).addBulk(evaluateJobs);
+      const releaseTargets = await recomputeReleaseTargets(job.data, resources);
+      await dispatchEvaluateJobs(releaseTargets);
     } catch (error) {
       log.error("Error upserting release targets", { error });
       throw error;

apps/event-worker/src/workers/new-resource.ts

@@ -1,10 +1,16 @@
+import type * as schema from "@ctrlplane/db/schema";
+
 import { selector } from "@ctrlplane/db";
-import { db } from "@ctrlplane/db/client";
-import { Channel, createWorker, getQueue } from "@ctrlplane/events";
+import { Channel, createWorker } from "@ctrlplane/events";
 
-import { replaceReleaseTargets } from "../utils/replace-release-targets.js";
+import { dispatchEvaluateJobs } from "../utils/dispatch-evaluate-jobs.js";
 
-const queue = getQueue(Channel.EvaluateReleaseTarget);
+const recomputeReleaseTargets = async (resource: schema.Resource) => {
+  const computeBuilder = selector().compute();
+  const { workspaceId } = resource;
+  await computeBuilder.allResourceSelectors(workspaceId);
+  return computeBuilder.resources([resource]).releaseTargets();
+};
 
 /**
  * Worker that processes new resource events.
@@ -21,23 +27,7 @@ const queue = getQueue(Channel.EvaluateReleaseTarget);
 export const newResourceWorker = createWorker(
   Channel.NewResource,
   async ({ data: resource }) => {
-    const cb = selector().compute();
-
-    await Promise.all([
-      cb.allEnvironments(resource.workspaceId).resourceSelectors().replace(),
-      cb.allDeployments(resource.workspaceId).resourceSelectors().replace(),
-    ]);
-
-    const rts = await replaceReleaseTargets(db, resource);
-    await cb
-      .allPolicies(resource.workspaceId)
-      .releaseTargetSelectors()
-      .replace();
-
-    const jobs = rts.map((rt) => ({
-      name: `${rt.resourceId}-${rt.environmentId}-${rt.deploymentId}`,
-      data: rt,
-    }));
-    await queue.addBulk(jobs);
+    const releaseTargets = await recomputeReleaseTargets(resource);
+    await dispatchEvaluateJobs(releaseTargets);
   },
 );

apps/event-worker/src/workers/update-deployment-variable.ts

@@ -1,7 +1,9 @@
 import { eq } from "@ctrlplane/db";
 import { db } from "@ctrlplane/db/client";
 import * as schema from "@ctrlplane/db/schema";
-import { Channel, createWorker, getQueue } from "@ctrlplane/events";
+import { Channel, createWorker } from "@ctrlplane/events";
+
+import { dispatchEvaluateJobs } from "../utils/dispatch-evaluate-jobs.js";
 
 /**
  * Worker that handles deployment variable changes
@@ -27,11 +29,6 @@ export const updateDeploymentVariableWorker = createWorker(
       where: eq(schema.releaseTarget.deploymentId, variable.deploymentId),
     });
 
-    await getQueue(Channel.EvaluateReleaseTarget).addBulk(
-      releaseTargets.map((rt) => ({
-        name: `${rt.resourceId}-${rt.environmentId}-${rt.deploymentId}`,
-        data: rt,
-      })),
-    );
+    await dispatchEvaluateJobs(releaseTargets);
   },
 );