Skip to content

chore: move dispatch jobs to events package #588

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Jun 13, 2025
Merged

chore: move dispatch jobs to events package #588

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
06c7950
chore: move dispatch jobs to events package
adityachoudhari26 Jun 13, 2025
a552f78
fix
adityachoudhari26 Jun 13, 2025
56567f2
rabbit comments
adityachoudhari26 Jun 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 0 additions & 14 deletions apps/event-worker/src/utils/dispatch-compute-deployment-jobs.ts
View file
Open in desktop

This file was deleted.

14 changes: 0 additions & 14 deletions apps/event-worker/src/utils/dispatch-compute-env-jobs.ts
View file
Open in desktop

This file was deleted.

14 changes: 0 additions & 14 deletions apps/event-worker/src/utils/dispatch-compute-policy-target-selector-jobs.ts
View file
Open in desktop

This file was deleted.

18 changes: 0 additions & 18 deletions apps/event-worker/src/utils/dispatch-compute-system-jobs.ts
View file
Open in desktop

This file was deleted.

20 changes: 0 additions & 20 deletions apps/event-worker/src/utils/dispatch-evaluate-jobs.ts
View file
Open in desktop

This file was deleted.

12 changes: 6 additions & 6 deletions apps/event-worker/src/workers/compute-deployment-resource-selector.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,7 @@
import { and, eq, isNull, selector, sql } from "@ctrlplane/db";
import { db } from "@ctrlplane/db/client";
import * as schema from "@ctrlplane/db/schema";
import { Channel, createWorker } from "@ctrlplane/events";

import { dispatchComputeDeploymentResourceSelectorJobs } from "../utils/dispatch-compute-deployment-jobs.js";
import { dispatchComputeSystemReleaseTargetsJobs } from "../utils/dispatch-compute-system-jobs.js";
import { Channel, createWorker, dispatchQueueJob } from "@ctrlplane/events";

export const computeDeploymentResourceSelectorWorkerEvent = createWorker(
Channel.ComputeDeploymentResourceSelector,
Expand Down Expand Up @@ -61,11 +58,14 @@ export const computeDeploymentResourceSelectorWorkerEvent = createWorker(
.onConflictDoNothing();
});

dispatchComputeSystemReleaseTargetsJobs(deployment.system);
dispatchQueueJob().toCompute().system(deployment.system).releaseTargets();
Copy link
Contributor

@coderabbitai coderabbitai bot Jun 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Forgot to await the compute-releaseTargets dispatch

Same reasoning as other workers: lack of await hides failures and may exit the worker before the dispatch is persisted.

-      dispatchQueueJob().toCompute().system(deployment.system).releaseTargets();
+      await dispatchQueueJob()
+        .toCompute()
+        .system(deployment.system)
+        .releaseTargets();
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
dispatchQueueJob().toCompute().system(deployment.system).releaseTargets();
await dispatchQueueJob()
.toCompute()
.system(deployment.system)
.releaseTargets();
🤖 Prompt for AI Agents 
In apps/event-worker/src/workers/compute-deployment-resource-selector.ts at line
61, the call to
dispatchQueueJob().toCompute().system(deployment.system).releaseTargets() is
missing an await. Add the await keyword before this call to ensure the dispatch
completes before proceeding, which will properly handle any errors and prevent
the worker from exiting prematurely.

} catch (e: any) {
const isRowLocked = e.code === "55P03";
if (isRowLocked) {
dispatchComputeDeploymentResourceSelectorJobs(deployment);
dispatchQueueJob()
.toCompute()
.deployment(deployment)
.resourceSelector();
Comment on lines +65 to +68
Copy link
Contributor

@coderabbitai coderabbitai bot Jun 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Ditto in retry branch

Add await here as well to maintain consistency and error visibility.

🤖 Prompt for AI Agents 
In apps/event-worker/src/workers/compute-deployment-resource-selector.ts at
lines 65 to 68, the call to
dispatchQueueJob().toCompute().deployment(deployment).resourceSelector() is
missing an await keyword. Add await before this call to ensure the asynchronous
operation completes before proceeding, maintaining consistency with other
branches and improving error visibility.

return;
}

Expand Down
16 changes: 9 additions & 7 deletions apps/event-worker/src/workers/compute-environment-resource-selector.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,7 @@
import { and, eq, isNull, selector, sql } from "@ctrlplane/db";
import { db } from "@ctrlplane/db/client";
import * as schema from "@ctrlplane/db/schema";
import { Channel, createWorker } from "@ctrlplane/events";

import { dispatchComputeEnvironmentResourceSelectorJobs } from "../utils/dispatch-compute-env-jobs.js";
import { dispatchComputeSystemReleaseTargetsJobs } from "../utils/dispatch-compute-system-jobs.js";
import { Channel, createWorker, dispatchQueueJob } from "@ctrlplane/events";

/**
* Worker that computes and updates the resources associated with an environment
Expand Down Expand Up @@ -81,12 +78,17 @@ export const computeEnvironmentResourceSelectorWorkerEvent = createWorker(
.values(computedEnvironmentResources)
.onConflictDoNothing();
});

dispatchComputeSystemReleaseTargetsJobs(environment.system);
dispatchQueueJob()
.toCompute()
.system(environment.system)
.releaseTargets();
Comment on lines +81 to +84
Copy link
Contributor

@coderabbitai coderabbitai bot Jun 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Missing await may drop job if queue add fails

Same pattern as in the system worker: the promise from dispatchQueueJob() is ignored.
Either await it or add an error handler to prevent unobserved rejections.

-      dispatchQueueJob()
+      await dispatchQueueJob()
         .toCompute()
         .system(environment.system)
         .releaseTargets();
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
dispatchQueueJob()
.toCompute()
.system(environment.system)
.releaseTargets();
await dispatchQueueJob()
.toCompute()
.system(environment.system)
.releaseTargets();
🤖 Prompt for AI Agents 
In apps/event-worker/src/workers/compute-environment-resource-selector.ts around
lines 81 to 84, the call to dispatchQueueJob() returns a promise that is not
awaited or handled, risking dropped jobs if the queue addition fails. Fix this
by adding an await before dispatchQueueJob() to ensure the promise is resolved
or rejected properly, or attach a catch handler to handle any errors and prevent
unobserved promise rejections.

} catch (e: any) {
const isRowLocked = e.code === "55P03";
if (isRowLocked) {
dispatchComputeEnvironmentResourceSelectorJobs(environment);
dispatchQueueJob()
.toCompute()
.environment(environment)
.resourceSelector();
Comment on lines +88 to +91
Copy link
Contributor

@coderabbitai coderabbitai bot Jun 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Add awaiting / error handling for retry dispatch

For the retry path you probably want to make sure the retry job is actually enqueued before returning.

-        dispatchQueueJob()
+        await dispatchQueueJob()
           .toCompute()
           .environment(environment)
           .resourceSelector();
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
dispatchQueueJob()
.toCompute()
.environment(environment)
.resourceSelector();
await dispatchQueueJob()
.toCompute()
.environment(environment)
.resourceSelector();
🤖 Prompt for AI Agents 
In apps/event-worker/src/workers/compute-environment-resource-selector.ts around
lines 88 to 91, the dispatchQueueJob call for the retry path is missing awaiting
and error handling. Modify the code to await the dispatchQueueJob chain to
ensure the retry job is enqueued before returning, and add try-catch error
handling to properly handle any errors during dispatch.

return;
}

Expand Down
12 changes: 6 additions & 6 deletions apps/event-worker/src/workers/compute-policy-target-release-target-selector.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,7 @@ import { and, eq, inArray, isNull } from "@ctrlplane/db";
import { db } from "@ctrlplane/db/client";
import { computePolicyTargets } from "@ctrlplane/db/queries";
import * as schema from "@ctrlplane/db/schema";
import { Channel, createWorker } from "@ctrlplane/events";

import { dispatchComputePolicyTargetReleaseTargetSelectorJobs } from "../utils/dispatch-compute-policy-target-selector-jobs.js";
import { dispatchEvaluateJobs } from "../utils/dispatch-evaluate-jobs.js";
import { Channel, createWorker, dispatchQueueJob } from "@ctrlplane/events";

export const computePolicyTargetReleaseTargetSelectorWorkerEvent = createWorker(
Channel.ComputePolicyTargetReleaseTargetSelector,
Expand Down Expand Up @@ -39,11 +36,14 @@ export const computePolicyTargetReleaseTargetSelectorWorkerEvent = createWorker(
)
.then((rows) => rows.map((row) => row.release_target));

dispatchEvaluateJobs(releaseTargets);
dispatchQueueJob().toEvaluate().releaseTargets(releaseTargets);
Copy link
Contributor

@coderabbitai coderabbitai bot Jun 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Missing await on evaluate-releaseTargets dispatch

-      dispatchQueueJob().toEvaluate().releaseTargets(releaseTargets);
+      await dispatchQueueJob().toEvaluate().releaseTargets(releaseTargets);

Ensures the enqueue finishes or throws inside the current try/​catch.

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
dispatchQueueJob().toEvaluate().releaseTargets(releaseTargets);
await dispatchQueueJob().toEvaluate().releaseTargets(releaseTargets);
🤖 Prompt for AI Agents 
In
apps/event-worker/src/workers/compute-policy-target-release-target-selector.ts
at line 39, the call to
dispatchQueueJob().toEvaluate().releaseTargets(releaseTargets) is missing an
await, which means the enqueue operation may not complete or throw within the
current try/catch block. Add the await keyword before this call to ensure the
promise resolves or rejects inside the try/catch for proper error handling.

} catch (e: any) {
const isRowLocked = e.code === "55P03";
if (isRowLocked) {
dispatchComputePolicyTargetReleaseTargetSelectorJobs(policyTarget);
dispatchQueueJob()
.toCompute()
.policyTarget(policyTarget)
.releaseTargetSelector();
Comment on lines +43 to +46
Copy link
Contributor

@coderabbitai coderabbitai bot Jun 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Retry dispatch should also be awaited

-        dispatchQueueJob()
+        await dispatchQueueJob()
           .toCompute()
           .policyTarget(policyTarget)
           .releaseTargetSelector();
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
dispatchQueueJob()
.toCompute()
.policyTarget(policyTarget)
.releaseTargetSelector();
await dispatchQueueJob()
.toCompute()
.policyTarget(policyTarget)
.releaseTargetSelector();
🤖 Prompt for AI Agents 
In
apps/event-worker/src/workers/compute-policy-target-release-target-selector.ts
around lines 43 to 46, the dispatchQueueJob call chain that includes toCompute,
policyTarget, and releaseTargetSelector is missing an await for the retry
dispatch. Modify the code to await the dispatchQueueJob call to ensure the retry
dispatch completes before proceeding.

return;
}

Expand Down
25 changes: 15 additions & 10 deletions apps/event-worker/src/workers/compute-systems-release-targets.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,14 @@ import { and, eq, inArray, isNull, notInArray, or, sql } from "@ctrlplane/db";
import { db } from "@ctrlplane/db/client";
import { computePolicyTargets } from "@ctrlplane/db/queries";
import * as schema from "@ctrlplane/db/schema";
import { Channel, createWorker, getQueue } from "@ctrlplane/events";
import {
Channel,
createWorker,
dispatchQueueJob,
getQueue,
} from "@ctrlplane/events";
import { logger } from "@ctrlplane/logger";

import { dispatchComputeSystemReleaseTargetsJobs } from "../utils/dispatch-compute-system-jobs.js";
import { dispatchEvaluateJobs } from "../utils/dispatch-evaluate-jobs.js";

const log = logger.child({
component: "computeSystemsReleaseTargetsWorker",
});
Expand Down Expand Up @@ -206,10 +208,13 @@ export const computeSystemsReleaseTargetsWorker = createWorker(
additionalProcessedPolicyTargetIds.push(policyTarget.id);
} catch (e: any) {
if (e.code === "55P03") {
dispatchComputeSystemReleaseTargetsJobs(system, true, [
...(processedPolicyTargetIds ?? []),
...additionalProcessedPolicyTargetIds,
]);
dispatchQueueJob()
.toCompute()
.system(system)
.releaseTargets(true, [
...(processedPolicyTargetIds ?? []),
...additionalProcessedPolicyTargetIds,
]);
Comment on lines +211 to +217
Copy link
Contributor

@coderabbitai coderabbitai bot Jun 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Fire-and-forget dispatch may swallow Redis / BullMQ errors

dispatchQueueJob().toCompute().system(system).releaseTargets(...) returns a promise from the underlying queue API, but it is not await-ed (nor chained with .catch()).
If Redis is unreachable or BullMQ rejects the add-job call, the exception becomes an unhandled-rejection and the worker will exit silently in many Node runtimes.

-            dispatchQueueJob()
-              .toCompute()
-              .system(system)
-              .releaseTargets(true, [
+            await dispatchQueueJob()
+              .toCompute()
+              .system(system)
+              .releaseTargets(true, [

Apply the same treatment to the analogous call two blocks below (lines 230-231).
If the intention is truly “fire-and-forget”, add an explicit .catch((e) => log.error(e)) to avoid process crashes.

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
dispatchQueueJob()
.toCompute()
.system(system)
.releaseTargets(true, [
...(processedPolicyTargetIds ?? []),
...additionalProcessedPolicyTargetIds,
]);
await dispatchQueueJob()
.toCompute()
.system(system)
.releaseTargets(true, [
...(processedPolicyTargetIds ?? []),
...additionalProcessedPolicyTargetIds,
]);
🤖 Prompt for AI Agents 
In apps/event-worker/src/workers/compute-systems-release-targets.ts around lines
211 to 217, the dispatchQueueJob call returns a promise that is not awaited or
handled for errors, risking unhandled promise rejections if Redis or BullMQ
fail. Fix this by adding a .catch handler to the promise to log any errors. Also
apply the same fix to the similar dispatchQueueJob call around lines 230 to 231
to ensure all fire-and-forget calls handle errors properly.

return;
}
throw e;
Expand All @@ -218,11 +223,11 @@ export const computeSystemsReleaseTargetsWorker = createWorker(

const toEvaluate = [...created, ...(redeployAll ? unchanged : [])];

await dispatchEvaluateJobs(toEvaluate);
await dispatchQueueJob().toEvaluate().releaseTargets(toEvaluate);
} catch (e: any) {
const isRowLocked = e.code === "55P03";
if (isRowLocked) {
dispatchComputeSystemReleaseTargetsJobs(system);
dispatchQueueJob().toCompute().system(system).releaseTargets();
return;
}
log.error("Failed to compute release targets", { error: e });
Expand Down
11 changes: 7 additions & 4 deletions apps/event-worker/src/workers/delete-resource.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,12 @@ import { eq, inArray } from "@ctrlplane/db";
import { db } from "@ctrlplane/db/client";
import { getResourceChildren } from "@ctrlplane/db/queries";
import * as SCHEMA from "@ctrlplane/db/schema";
import { Channel, createWorker, getQueue } from "@ctrlplane/events";

import { dispatchEvaluateJobs } from "../utils/dispatch-evaluate-jobs.js";
import {
Channel,
createWorker,
dispatchQueueJob,
getQueue,
} from "@ctrlplane/events";

const softDeleteResource = async (tx: Tx, resource: SCHEMA.Resource) =>
tx
Expand Down Expand Up @@ -47,7 +50,7 @@ const dispatchAffectedTargetJobs = async (
),
);

await dispatchEvaluateJobs(affectedReleaseTargets);
await dispatchQueueJob().toEvaluate().releaseTargets(affectedReleaseTargets);
};

export const deleteResourceWorker = createWorker(
Expand Down
11 changes: 7 additions & 4 deletions apps/event-worker/src/workers/evaluate-release-target.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,18 @@ import { and, desc, eq, sql, takeFirst } from "@ctrlplane/db";
import { db } from "@ctrlplane/db/client";
import { createReleaseJob } from "@ctrlplane/db/queries";
import * as schema from "@ctrlplane/db/schema";
import { Channel, createWorker, getQueue } from "@ctrlplane/events";
import {
Channel,
createWorker,
dispatchQueueJob,
getQueue,
} from "@ctrlplane/events";
import { logger, makeWithSpan, trace } from "@ctrlplane/logger";
import {
VariableReleaseManager,
VersionReleaseManager,
} from "@ctrlplane/rule-engine";

import { dispatchEvaluateJobs } from "../utils/dispatch-evaluate-jobs.js";

const tracer = trace.getTracer("evaluate-release-target");
const withSpan = makeWithSpan(tracer);
const log = logger.child({ module: "evaluate-release-target" });
Expand Down Expand Up @@ -187,7 +190,7 @@ export const evaluateReleaseTargetWorker = createWorker(
const isRowLocked = e.code === "55P03";
const isReleaseTargetNotCommittedYet = e.code === "23503";
if (isRowLocked || isReleaseTargetNotCommittedYet) {
dispatchEvaluateJobs([job.data]);
dispatchQueueJob().toEvaluate().releaseTargets([job.data]);
return;
}
Comment on lines +193 to 195
Copy link
Contributor

@coderabbitai coderabbitai bot Jun 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Missing await may swallow dispatch errors

dispatchQueueJob().toEvaluate().releaseTargets(...) is very likely asynchronous (every other call site awaits it).
Not awaiting here means any rejection is detached from the surrounding try/​catch, causing an unhandled-rejection and silent job-loss.

-        dispatchQueueJob().toEvaluate().releaseTargets([job.data]);
+        await dispatchQueueJob().toEvaluate().releaseTargets([job.data]);
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
dispatchQueueJob().toEvaluate().releaseTargets([job.data]);
return;
}
await dispatchQueueJob().toEvaluate().releaseTargets([job.data]);
return;
}
🤖 Prompt for AI Agents 
In apps/event-worker/src/workers/evaluate-release-target.ts around lines 193 to
195, the call to dispatchQueueJob().toEvaluate().releaseTargets([job.data]) is
asynchronous but lacks an await, which can cause errors to be unhandled and
swallowed. Add the await keyword before this call to ensure any rejections are
caught by the surrounding try/catch and prevent silent job failures.

const isJobAgentError =
Expand Down
6 changes: 2 additions & 4 deletions apps/event-worker/src/workers/new-deployment-version.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,7 @@
import { eq } from "@ctrlplane/db";
import { db } from "@ctrlplane/db/client";
import * as schema from "@ctrlplane/db/schema";
import { Channel, createWorker } from "@ctrlplane/events";

import { dispatchEvaluateJobs } from "../utils/dispatch-evaluate-jobs.js";
import { Channel, createWorker, dispatchQueueJob } from "@ctrlplane/events";

/**
* Worker that processes new deployment version events.
Expand All @@ -20,6 +18,6 @@ export const newDeploymentVersionWorker = createWorker(
const releaseTargets = await db.query.releaseTarget.findMany({
where: eq(schema.releaseTarget.deploymentId, version.deploymentId),
});
await dispatchEvaluateJobs(releaseTargets);
await dispatchQueueJob().toEvaluate().releaseTargets(releaseTargets);
},
);
9 changes: 5 additions & 4 deletions apps/event-worker/src/workers/new-deployment.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
import { Channel, createWorker } from "@ctrlplane/events";

import { dispatchComputeDeploymentResourceSelectorJobs } from "../utils/dispatch-compute-deployment-jobs.js";
import { Channel, createWorker, dispatchQueueJob } from "@ctrlplane/events";

export const newDeploymentWorker = createWorker(
Channel.NewDeployment,
async (job) => {
const { data: deployment } = job;
await dispatchComputeDeploymentResourceSelectorJobs(deployment);
await dispatchQueueJob()
.toCompute()
.deployment(deployment)
.resourceSelector();
},
);
9 changes: 5 additions & 4 deletions apps/event-worker/src/workers/new-environment.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
import { Channel, createWorker } from "@ctrlplane/events";

import { dispatchComputeEnvironmentResourceSelectorJobs } from "../utils/dispatch-compute-env-jobs.js";
import { Channel, createWorker, dispatchQueueJob } from "@ctrlplane/events";

export const newEnvironmentWorker = createWorker(
Channel.NewEnvironment,
async (job) => {
const { data: environment } = job;
await dispatchComputeEnvironmentResourceSelectorJobs(environment);
await dispatchQueueJob()
.toCompute()
.environment(environment)
.resourceSelector();
},
);
9 changes: 5 additions & 4 deletions apps/event-worker/src/workers/new-policy.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,16 @@
import { eq } from "@ctrlplane/db";
import { db } from "@ctrlplane/db/client";
import * as schema from "@ctrlplane/db/schema";
import { Channel, createWorker } from "@ctrlplane/events";

import { dispatchComputePolicyTargetReleaseTargetSelectorJobs } from "../utils/dispatch-compute-policy-target-selector-jobs.js";
import { Channel, createWorker, dispatchQueueJob } from "@ctrlplane/events";

export const newPolicyWorker = createWorker(Channel.NewPolicy, async (job) => {
const policyTargets = await db.query.policyTarget.findMany({
where: eq(schema.policyTarget.policyId, job.data.id),
});

for (const policyTarget of policyTargets)
dispatchComputePolicyTargetReleaseTargetSelectorJobs(policyTarget);
dispatchQueueJob()
.toCompute()
.policyTarget(policyTarget)
.releaseTargetSelector();
Comment on lines 11 to +15
Copy link
Contributor

@coderabbitai coderabbitai bot Jun 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Mirror the awaiting strategy used elsewhere for consistency

Same pattern as in update-policy.ts: the dispatch is fire-and-forget, whereas new-deployment awaits the promise. Aligning behaviour avoids hidden race conditions and makes error-handling uniform across workers.

🤖 Prompt for AI Agents 
In apps/event-worker/src/workers/new-policy.ts around lines 11 to 15, the
dispatchQueueJob calls are currently fire-and-forget without awaiting their
promises. To align with the awaiting strategy used in new-deployment and ensure
consistent error handling and avoid race conditions, modify the code to await
each dispatchQueueJob call inside the loop or collect the promises and await
them properly.

});
15 changes: 9 additions & 6 deletions apps/event-worker/src/workers/new-resource.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,7 @@
import { eq } from "@ctrlplane/db";
import { db } from "@ctrlplane/db/client";
import * as schema from "@ctrlplane/db/schema";
import { Channel, createWorker } from "@ctrlplane/events";

import { dispatchComputeDeploymentResourceSelectorJobs } from "../utils/dispatch-compute-deployment-jobs.js";
import { dispatchComputeEnvironmentResourceSelectorJobs } from "../utils/dispatch-compute-env-jobs.js";
import { Channel, createWorker, dispatchQueueJob } from "@ctrlplane/events";

/**
* Worker that processes new resource events.
Expand All @@ -30,12 +27,18 @@ export const newResourceWorker = createWorker(
const environments = systems.flatMap((s) => s.environments);

for (const environment of environments) {
await dispatchComputeEnvironmentResourceSelectorJobs(environment);
await dispatchQueueJob()
.toCompute()
.environment(environment)
.resourceSelector();
}

const deployments = systems.flatMap((s) => s.deployments);
for (const deployment of deployments) {
await dispatchComputeDeploymentResourceSelectorJobs(deployment);
await dispatchQueueJob()
.toCompute()
.deployment(deployment)
.resourceSelector();
}
},
);
Loading
Loading