chore: remove event worker #691
Conversation
|
Caution Review failedThe pull request is closed. WalkthroughThis PR removes the apps/event-worker application and its workers, deletes Redis/BullMQ wiring and redis-related clients/config, removes many resource-scan modules and job/queue dispatching code, and disables ioredis OpenTelemetry instrumentation; event dispatching is consolidated to Kafka (eventDispatcher). Changes
Sequence Diagram(s)mermaid mermaid Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Suggested reviewers
Poem
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
📜 Recent review detailsConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Disabled knowledge base sources:
⛔ Files ignored due to path filters (1)
📒 Files selected for processing (1)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
📊 Code Coverage Reportworkspace-engine coverage: |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (5)
packages/job-dispatch/src/job-dispatch.ts (1)
13-18: Add null check for runbook before passing tocreateTriggeredRunbookJob.The
takeFirsthelper can returnundefinedif no runbook matches the query. Passing an undefined value tocreateTriggeredRunbookJob(which expects a non-optionalschema.Runbookparameter) will cause a runtime error when the function attempts to access runbook properties.Apply this diff to add a null check:
const runbook = await db .select() .from(schema.runbook) .where(eq(schema.runbook.id, runbookId)) .then(takeFirst); + + if (!runbook) { + throw new Error(`Runbook not found: ${runbookId}`); + } + return createTriggeredRunbookJob(db, runbook, values);packages/api/src/router/policy/approval/add-record.ts (1)
43-47: Add release-target evaluation trigger for UserApprovalRecordCreated events.The
UserApprovalRecordCreatedevent is dispatched but the handler atapps/event-queue/src/events/index.ts:76is a no-op (() => Promise.resolve()). The removed logic previously queried affected release targets and dispatched evaluation jobs. This functionality is now lost—release targets are never evaluated when approval records are created.You must either:
- Implement the
UserApprovalRecordCreatedhandler to triggerdispatchEvaluateReleaseTargetfor affected release targets, or- Dispatch
dispatchEvaluateReleaseTargetdirectly in the approval record creation endpoints (alongsidedispatchUserApprovalRecordCreated)Ensure the release targets associated with the approval records are re-evaluated after creation.
packages/api/src/router/resource/resource-provider.ts (3)
214-249: Remove unusedrepeatSecondsparameter from input schema.The
repeatSecondsparameter (line 222) is no longer used after removing the queue-based scan scheduling. This dead parameter should be removed from the input schema to avoid confusion.Apply this diff:
update: protectedProcedure .input( z.object({ resourceProviderId: z.string().uuid(), name: z.string().optional(), config: updateResourceProviderGoogle.omit({ resourceProviderId: true, }), - repeatSeconds: z.number().min(1).nullable(), }), )
289-328: Remove unusedrepeatSecondsparameter from input schema.The
repeatSecondsparameter (line 303) is no longer used after removing the queue-based scan scheduling. This dead parameter should be removed from the input schema to avoid confusion.Apply this diff:
update: protectedProcedure .meta({ authorizationCheck: ({ canUser, input }) => canUser .perform(Permission.ResourceProviderUpdate) .on({ type: "resourceProvider", id: input.resourceProviderId }), }) .input( z.object({ resourceProviderId: z.string().uuid(), name: z.string().optional(), config: updateResourceProviderAws.omit({ resourceProviderId: true, }), - repeatSeconds: z.number().min(1).nullable(), }), )
167-175: Replace no-op mutation with error response or restore sync functionality.The
managed.syncmutation has been replaced with a silent no-op while remaining publicly exposed to active UI consumers:
- ProviderActionsDropdown.tsx:45
- PermissionsButton.tsx:18
When these components call this endpoint, they receive
Promise.resolve()instead of triggering a resource scan, creating a breaking change with no deprecation notice or error response.Recommended actions:
- Option 1: Return an error response:
throw new Error("Resource provider sync has been deprecated")- Option 2: Wire up replacement functionality using
handleResourceProviderScanandeventDispatcher(already implemented in the codebase)- Option 3: Remove the endpoint entirely and update all UI consumers
The current state silently breaks production functionality.
🧹 Nitpick comments (1)
packages/api/src/router/policy/approval/add-record.ts (1)
43-47: LGTM: Event dispatching is correct.The use of
Promise.all()ensures atomic dispatch behavior—if any event fails, the entire operation fails, maintaining consistency. This is appropriate for this approval flow.Optional: If you prefer more resilience (e.g., logging failed dispatches without blocking record creation), consider using
Promise.allSettled():- await Promise.all( + await Promise.allSettled( record.map((record) => eventDispatcher.dispatchUserApprovalRecordCreated(record), ), );This would allow partial failures to be handled gracefully, though the current fail-fast behavior may be intentional.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Disabled knowledge base sources:
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (87)
apps/event-queue/src/instrumentation-node.ts(0 hunks)apps/event-worker/Dockerfile(0 hunks)apps/event-worker/README.md(0 hunks)apps/event-worker/eslint.config.js(0 hunks)apps/event-worker/package.json(0 hunks)apps/event-worker/src/config.ts(0 hunks)apps/event-worker/src/github-utils.ts(0 hunks)apps/event-worker/src/index.ts(0 hunks)apps/event-worker/src/instrumentation-node.ts(0 hunks)apps/event-worker/src/instrumentation.ts(0 hunks)apps/event-worker/src/redis.ts(0 hunks)apps/event-worker/src/utils/omit-null-undefined.ts(0 hunks)apps/event-worker/src/workers/compute-deployment-resource-selector.ts(0 hunks)apps/event-worker/src/workers/compute-environment-resource-selector.ts(0 hunks)apps/event-worker/src/workers/compute-policy-target-release-target-selector.ts(0 hunks)apps/event-worker/src/workers/compute-systems-release-targets.ts(0 hunks)apps/event-worker/src/workers/compute-workspace-policy-targets.ts(0 hunks)apps/event-worker/src/workers/delete-deployment.ts(0 hunks)apps/event-worker/src/workers/delete-environment.ts(0 hunks)apps/event-worker/src/workers/delete-resource.ts(0 hunks)apps/event-worker/src/workers/deleted-release-target.ts(0 hunks)apps/event-worker/src/workers/evaluate-release-target.ts(0 hunks)apps/event-worker/src/workers/index.ts(0 hunks)apps/event-worker/src/workers/job-dispatch/github.ts(0 hunks)apps/event-worker/src/workers/job-dispatch/index.ts(0 hunks)apps/event-worker/src/workers/job-update/concurrency.ts(0 hunks)apps/event-worker/src/workers/job-update/db-update-job.ts(0 hunks)apps/event-worker/src/workers/job-update/index.ts(0 hunks)apps/event-worker/src/workers/job-update/job-metadata.ts(0 hunks)apps/event-worker/src/workers/job-update/job-retry.ts(0 hunks)apps/event-worker/src/workers/job-update/matched-policies.ts(0 hunks)apps/event-worker/src/workers/job-update/trigger-dependent-targets.ts(0 hunks)apps/event-worker/src/workers/job-update/utils.ts(0 hunks)apps/event-worker/src/workers/new-deployment-version.ts(0 hunks)apps/event-worker/src/workers/new-deployment.ts(0 hunks)apps/event-worker/src/workers/new-environment.ts(0 hunks)apps/event-worker/src/workers/new-policy.ts(0 hunks)apps/event-worker/src/workers/new-resource.ts(0 hunks)apps/event-worker/src/workers/resource-scan/aws/aws.ts(0 hunks)apps/event-worker/src/workers/resource-scan/aws/eks.ts(0 hunks)apps/event-worker/src/workers/resource-scan/aws/vpc.ts(0 hunks)apps/event-worker/src/workers/resource-scan/azure/aks.ts(0 hunks)apps/event-worker/src/workers/resource-scan/azure/cluster-to-resource.ts(0 hunks)apps/event-worker/src/workers/resource-scan/google/client.ts(0 hunks)apps/event-worker/src/workers/resource-scan/google/cluster-to-resource.ts(0 hunks)apps/event-worker/src/workers/resource-scan/google/gke.ts(0 hunks)apps/event-worker/src/workers/resource-scan/google/kube.ts(0 hunks)apps/event-worker/src/workers/resource-scan/google/vcluster.ts(0 hunks)apps/event-worker/src/workers/resource-scan/google/vm.ts(0 hunks)apps/event-worker/src/workers/resource-scan/google/vpc.ts(0 hunks)apps/event-worker/src/workers/resource-scan/index.ts(0 hunks)apps/event-worker/src/workers/resource-scan/utils/extract-variables.ts(0 hunks)apps/event-worker/src/workers/update-deployment-variable.ts(0 hunks)apps/event-worker/src/workers/update-deployment.ts(0 hunks)apps/event-worker/src/workers/update-environment.ts(0 hunks)apps/event-worker/src/workers/update-policy.ts(0 hunks)apps/event-worker/src/workers/update-resource-variable.ts(0 hunks)apps/event-worker/src/workers/updated-resources/index.ts(0 hunks)apps/event-worker/src/workers/updated-resources/span.ts(0 hunks)apps/event-worker/tsconfig.json(0 hunks)apps/webservice/src/app/api/azure/[workspaceId]/[tenantId]/[subscriptionId]/[name]/route.ts(0 hunks)apps/webservice/src/app/api/azure/consent/route.ts(0 hunks)apps/webservice/src/app/api/v1/release-targets/[releaseTargetId]/evaluate/route.ts(0 hunks)apps/webservice/src/env.ts(0 hunks)apps/webservice/src/instrumentation-node.ts(0 hunks)docker-compose.dev.yaml(0 hunks)docker-compose.yaml(0 hunks)packages/api/package.json(1 hunks)packages/api/src/config.ts(0 hunks)packages/api/src/index.ts(0 hunks)packages/api/src/redis.ts(0 hunks)packages/api/src/router/policy/approval/add-record.ts(1 hunks)packages/api/src/router/policy/version-selector.ts(3 hunks)packages/api/src/router/redeploy.ts(2 hunks)packages/api/src/router/resource/resource-provider.ts(2 hunks)packages/db/package.json(0 hunks)packages/db/src/config.ts(0 hunks)packages/db/src/redis.ts(0 hunks)packages/events/package.json(0 hunks)packages/events/src/config.ts(0 hunks)packages/events/src/dispatch-jobs.ts(0 hunks)packages/events/src/index.ts(0 hunks)packages/events/src/redis.ts(0 hunks)packages/events/src/types.ts(0 hunks)packages/job-dispatch/src/config.ts(1 hunks)packages/job-dispatch/src/job-dispatch.ts(1 hunks)pnpm-workspace.yaml(0 hunks)
💤 Files with no reviewable changes (80)
- apps/webservice/src/env.ts
- docker-compose.dev.yaml
- apps/event-worker/package.json
- apps/webservice/src/instrumentation-node.ts
- apps/event-worker/README.md
- packages/events/src/config.ts
- apps/event-worker/src/workers/new-deployment-version.ts
- apps/event-worker/src/instrumentation.ts
- apps/event-worker/src/workers/update-policy.ts
- apps/event-worker/src/utils/omit-null-undefined.ts
- packages/api/src/index.ts
- apps/event-worker/src/workers/job-update/utils.ts
- packages/db/package.json
- apps/event-worker/src/workers/resource-scan/index.ts
- apps/event-worker/src/workers/job-dispatch/github.ts
- apps/event-worker/src/workers/deleted-release-target.ts
- apps/event-worker/eslint.config.js
- apps/event-worker/src/workers/update-deployment-variable.ts
- apps/event-worker/src/redis.ts
- apps/event-worker/src/workers/job-update/matched-policies.ts
- apps/event-worker/src/instrumentation-node.ts
- apps/event-worker/src/workers/compute-systems-release-targets.ts
- apps/event-worker/src/workers/resource-scan/google/vpc.ts
- apps/event-worker/tsconfig.json
- apps/event-worker/src/workers/compute-deployment-resource-selector.ts
- apps/event-worker/src/workers/update-environment.ts
- apps/event-worker/src/workers/new-environment.ts
- apps/event-worker/src/workers/compute-workspace-policy-targets.ts
- apps/event-worker/src/workers/delete-deployment.ts
- apps/event-worker/src/workers/evaluate-release-target.ts
- apps/event-worker/src/workers/delete-resource.ts
- apps/event-worker/src/workers/resource-scan/google/kube.ts
- apps/event-worker/src/workers/compute-policy-target-release-target-selector.ts
- docker-compose.yaml
- apps/event-worker/src/workers/new-policy.ts
- packages/events/package.json
- apps/event-worker/src/index.ts
- apps/event-worker/src/workers/resource-scan/google/client.ts
- apps/event-queue/src/instrumentation-node.ts
- apps/event-worker/src/workers/index.ts
- packages/db/src/config.ts
- apps/event-worker/src/workers/job-update/trigger-dependent-targets.ts
- packages/events/src/redis.ts
- packages/api/src/config.ts
- apps/event-worker/src/workers/resource-scan/google/gke.ts
- apps/event-worker/src/workers/updated-resources/index.ts
- apps/event-worker/src/workers/resource-scan/aws/eks.ts
- apps/event-worker/src/workers/new-resource.ts
- packages/events/src/types.ts
- apps/event-worker/src/workers/resource-scan/azure/aks.ts
- apps/event-worker/src/workers/resource-scan/google/vcluster.ts
- apps/event-worker/src/workers/job-update/job-metadata.ts
- apps/event-worker/src/workers/delete-environment.ts
- apps/event-worker/src/workers/resource-scan/utils/extract-variables.ts
- apps/event-worker/src/workers/resource-scan/aws/vpc.ts
- packages/db/src/redis.ts
- packages/events/src/index.ts
- apps/webservice/src/app/api/v1/release-targets/[releaseTargetId]/evaluate/route.ts
- pnpm-workspace.yaml
- apps/event-worker/src/config.ts
- apps/event-worker/src/workers/job-update/index.ts
- apps/event-worker/src/workers/job-dispatch/index.ts
- apps/event-worker/src/workers/job-update/db-update-job.ts
- packages/api/src/redis.ts
- packages/events/src/dispatch-jobs.ts
- apps/event-worker/src/workers/resource-scan/google/cluster-to-resource.ts
- apps/webservice/src/app/api/azure/consent/route.ts
- apps/event-worker/src/workers/resource-scan/google/vm.ts
- apps/event-worker/src/workers/compute-environment-resource-selector.ts
- apps/event-worker/src/workers/job-update/concurrency.ts
- apps/event-worker/src/workers/resource-scan/azure/cluster-to-resource.ts
- apps/webservice/src/app/api/azure/[workspaceId]/[tenantId]/[subscriptionId]/[name]/route.ts
- apps/event-worker/Dockerfile
- apps/event-worker/src/workers/job-update/job-retry.ts
- apps/event-worker/src/workers/update-deployment.ts
- apps/event-worker/src/github-utils.ts
- apps/event-worker/src/workers/update-resource-variable.ts
- apps/event-worker/src/workers/new-deployment.ts
- apps/event-worker/src/workers/updated-resources/span.ts
- apps/event-worker/src/workers/resource-scan/aws/aws.ts
🧰 Additional context used
📓 Path-based instructions (2)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.{ts,tsx}: Use TypeScript with explicit types (prefer interfaces for public APIs)
Import styles: Use named imports, group imports by source (std lib > external > internal)
Consistent type imports:import type { Type } from "module"
Prefer async/await over raw promises
Handle errors explicitly (use try/catch and typed error responses)
Files:
packages/api/src/router/policy/version-selector.tspackages/api/src/router/policy/approval/add-record.tspackages/api/src/router/redeploy.tspackages/api/src/router/resource/resource-provider.tspackages/job-dispatch/src/config.tspackages/job-dispatch/src/job-dispatch.ts
⚙️ CodeRabbit configuration file
**/*.{ts,tsx}: Note on Error Handling:
Avoid strict enforcement of try/catch blocks. Code may use early returns, Promise chains (.then().catch()), or other patterns for error handling. These are acceptable as long as they maintain clarity and predictability.
Files:
packages/api/src/router/policy/version-selector.tspackages/api/src/router/policy/approval/add-record.tspackages/api/src/router/redeploy.tspackages/api/src/router/resource/resource-provider.tspackages/job-dispatch/src/config.tspackages/job-dispatch/src/job-dispatch.ts
**/*.{js,jsx,ts,tsx,json,md,yml,yaml}
📄 CodeRabbit inference engine (CLAUDE.md)
Formatting: Prettier is used with
@ctrlplane/prettier-config
Files:
packages/api/src/router/policy/version-selector.tspackages/api/src/router/policy/approval/add-record.tspackages/api/package.jsonpackages/api/src/router/redeploy.tspackages/api/src/router/resource/resource-provider.tspackages/job-dispatch/src/config.tspackages/job-dispatch/src/job-dispatch.ts
🧬 Code graph analysis (3)
packages/api/src/router/policy/version-selector.ts (3)
packages/db/src/common.ts (1)
Tx(22-22)packages/db/src/utils/policies/index.ts (1)
rulesAndTargets(13-13)packages/events/src/index.ts (1)
eventDispatcher(7-7)
packages/api/src/router/redeploy.ts (2)
apps/event-queue/src/job-manager/job-manager.ts (1)
createReleaseJob(231-260)packages/db/src/queries/create-release-job.ts (1)
createReleaseJob(19-95)
packages/job-dispatch/src/job-dispatch.ts (2)
packages/job-dispatch/src/job-creation.ts (1)
createTriggeredRunbookJob(9-54)packages/db/src/schema/runbook.ts (1)
runbook(20-38)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (4)
- GitHub Check: build (linux/amd64)
- GitHub Check: build (linux/amd64)
- GitHub Check: Typecheck
- GitHub Check: Lint
🔇 Additional comments (8)
packages/api/package.json (1)
27-27: Verify AWS SDK version alignment and usage context.The STS client version (
^3.913.0) is significantly newer than the IAM client (^3.696.0). AWS SDK packages are typically kept in sync to avoid incompatibilities and maintenance burden. Additionally, the relationship between adding STS and removing event-worker-v2 is unclear.Please verify:
- Whether the STS client version should be aligned with the IAM client version (
^3.696.0).- The code changes that introduce the STS client usage in this package to confirm the necessity of this dependency addition and its relationship to the event-worker removal.
- Whether there are any security advisories or compatibility issues with the chosen version.
packages/job-dispatch/src/config.ts (1)
4-9: The config file is part of the package's public API and should not be removed.The original review comment incorrectly assumes this file is unused. However,
packages/job-dispatch/src/index.tsexplicitly exports fromconfig.js:export * from "./config.js";. Since this package has a public export field inpackage.json, theenvexport fromconfig.tsis part of the package's public API. Removing it would be a breaking change for consumers.While the empty
server: {}object correctly reflects the removal ofREDIS_URLand Redis infrastructure, the file itself serves a legitimate purpose as the centralized environment configuration export point. The structure is appropriate for potential future environment variables.Likely an incorrect or invalid review comment.
packages/job-dispatch/src/job-dispatch.ts (1)
18-18: All callers correctly handle the direct job return transition.Verification confirms:
packages/api/src/router/runbook.ts:44 — Mutation directly returns
dispatchRunbook()result, which is the correct pattern for TRPC mutations.packages/job-dispatch/src/events/handlers/resource-removed.ts:31 — Event handler properly collects promises via
map()and awaits them withPromise.all().Both call sites are appropriately updated and no queue-based dependencies remain in the callers.
packages/api/src/router/policy/version-selector.ts (2)
7-7: LGTM: Import additions are appropriate.The new imports for
rulesAndTargetsandeventDispatchercorrectly support the transition to Kafka-based event dispatching.Also applies to: 12-12
107-107: Verify: Fire-and-forget event dispatch pattern.The
dispatchPolicyUpdatedcalls are not awaited, meaning mutations complete before confirming event dispatch succeeded. This introduces potential for event loss if the dispatch fails.Is this fire-and-forget pattern intentional? Consider:
- If eventual consistency is acceptable: Add error handling (see comment on lines 19-28) to log failures and implement retry logic in the Kafka producer.
- If strong consistency is required: Add
awaitto ensure events are dispatched before mutations complete.Also applies to: 140-140
packages/api/src/router/policy/approval/add-record.ts (2)
4-4: LGTM: Import simplified correctly.The import now includes only
eventDispatcher, which aligns with the transition from queue-based (dispatchQueueJob) to event-driven processing.
28-41: LGTM: Record creation and insertion logic is sound.The mapping of environment IDs to approval records is clear, and
onConflictDoNothing()appropriately handles duplicate entries.packages/api/src/router/redeploy.ts (1)
9-9: LGTM! Transition to event dispatcher.The import of
eventDispatchercorrectly replaces the removed queue-based infrastructure and is properly utilized in the non-force deployment path.
| const dispatchPolicyUpdated = (db: Tx, policyId: string) => | ||
| db.query.policy | ||
| .findFirst({ | ||
| where: eq(schema.policy.id, policyId), | ||
| with: rulesAndTargets, | ||
| }) | ||
| .then((fullPolicy) => { | ||
| if (fullPolicy == null) return; | ||
| eventDispatcher.dispatchPolicyUpdated(fullPolicy, fullPolicy); | ||
| }); |
There was a problem hiding this comment.
Critical: Missing error handling for event dispatch.
The dispatchPolicyUpdated helper has no error handling. Since it's called without await (lines 107, 140), any failures in the query or event dispatch will result in unhandled promise rejections, causing silent event loss.
Apply this diff to add error handling:
const dispatchPolicyUpdated = (db: Tx, policyId: string) =>
db.query.policy
.findFirst({
where: eq(schema.policy.id, policyId),
with: rulesAndTargets,
})
.then((fullPolicy) => {
if (fullPolicy == null) return;
eventDispatcher.dispatchPolicyUpdated(fullPolicy, fullPolicy);
- });
+ })
+ .catch((error) => {
+ console.error(
+ `Failed to dispatch policy updated event for policy ${policyId}:`,
+ error,
+ );
+ });Alternatively, if event dispatch should be synchronous and failures should block mutations, add await at the call sites (lines 107, 140).
📝 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.
| const dispatchPolicyUpdated = (db: Tx, policyId: string) => | |
| db.query.policy | |
| .findFirst({ | |
| where: eq(schema.policy.id, policyId), | |
| with: rulesAndTargets, | |
| }) | |
| .then((fullPolicy) => { | |
| if (fullPolicy == null) return; | |
| eventDispatcher.dispatchPolicyUpdated(fullPolicy, fullPolicy); | |
| }); | |
| const dispatchPolicyUpdated = (db: Tx, policyId: string) => | |
| db.query.policy | |
| .findFirst({ | |
| where: eq(schema.policy.id, policyId), | |
| with: rulesAndTargets, | |
| }) | |
| .then((fullPolicy) => { | |
| if (fullPolicy == null) return; | |
| eventDispatcher.dispatchPolicyUpdated(fullPolicy, fullPolicy); | |
| }) | |
| .catch((error) => { | |
| console.error( | |
| `Failed to dispatch policy updated event for policy ${policyId}:`, | |
| error, | |
| ); | |
| }); |
🤖 Prompt for AI Agents
In packages/api/src/router/policy/version-selector.ts around lines 19 to 28,
dispatchPolicyUpdated currently performs async work without any error handling
which leads to unhandled promise rejections when called without await; wrap the
function body in a try/catch that catches errors from the DB query and
eventDispatcher, log the error with context (including policyId and the error
message/stack) and prevent an unhandled rejection (either swallow after logging
or rethrow based on desired behavior), or alternatively update the two call
sites at lines 107 and 140 to await the helper so failures propagate and block
the mutation.
| }) | ||
| .then((fullPolicy) => { | ||
| if (fullPolicy == null) return; | ||
| eventDispatcher.dispatchPolicyUpdated(fullPolicy, fullPolicy); |
There was a problem hiding this comment.
🧩 Analysis chain
Verify: Passing the same policy twice to dispatchPolicyUpdated.
The method signature appears to dispatch an update event but receives the same fullPolicy object for both arguments. Typically, update events distinguish between old and new state. If the second parameter should represent the previous state, this is a logic error.
Run this script to verify the method signature:
🏁 Script executed:
#!/bin/bash
# Verify the signature of eventDispatcher.dispatchPolicyUpdated
# Find the EventDispatcher interface and KafkaEventDispatcher implementation
ast-grep --pattern $'class KafkaEventDispatcher {
$$$
dispatchPolicyUpdated($$$) {
$$$
}
$$$
}'
# Also search for the interface definition
ast-grep --pattern $'interface EventDispatcher {
$$$
dispatchPolicyUpdated($$$)
$$$
}'
# Search for method signature with rg as backup
rg -n -C5 'dispatchPolicyUpdated.*\(' --type=tsLength of output: 17637
Fix: Line 27 passes the same policy twice instead of previous and current states.
The interface signature requires dispatchPolicyUpdated(previous: FullPolicy, current: FullPolicy), but line 27 calls it with the same object twice: eventDispatcher.dispatchPolicyUpdated(fullPolicy, fullPolicy). This breaks the event semantics—all other callers in the codebase (router.ts, approval/router.ts, webservice) correctly pass the previous and current states. The local helper function needs refactoring to track the previous policy state.
🤖 Prompt for AI Agents
In packages/api/src/router/policy/version-selector.ts around line 27, the call
eventDispatcher.dispatchPolicyUpdated(fullPolicy, fullPolicy) incorrectly passes
the same FullPolicy as both previous and current; change the helper so it
captures or retrieves the previous FullPolicy state and pass (previousPolicy,
fullPolicy) to dispatchPolicyUpdated. Concretely: obtain the prior policy before
applying updates (e.g., load from DB or keep a prev variable outside the update
logic), ensure the helper returns or has access to that previousPolicy, and
replace the current call with
eventDispatcher.dispatchPolicyUpdated(previousPolicy, fullPolicy) so event
semantics match other callers.
|
|
||
| const job = await createReleaseJob(tx, existingRelease.release); | ||
| getQueue(Channel.DispatchJob).add(job.id, { jobId: job.id }); | ||
| await createReleaseJob(tx, existingRelease.release); |
There was a problem hiding this comment.
🧩 Analysis chain
Verify that pending jobs are processed after creation.
After creating the release job with createReleaseJob, no dispatch or event is emitted to trigger job processing. Previously, the job was enqueued to a worker queue. In the current implementation, the job is created with status "pending" but it's unclear what mechanism picks it up for execution.
Run the following script to verify the job processing mechanism:
🏁 Script executed:
#!/bin/bash
# Search for job polling, Kafka consumers, or other mechanisms that process pending jobs
echo "=== Searching for job status transitions from 'pending' ==="
rg -nP --type=ts -C3 'status.*=.*["\']pending["\']|job\.status.*pending'
echo -e "\n=== Searching for Kafka consumers or event handlers for job processing ==="
rg -nP --type=ts -C3 'consumer.*job|handleJob|processJob|JOB_CREATED|job.*event'
echo -e "\n=== Searching for job polling or scheduler mechanisms ==="
rg -nP --type=ts -C3 'pollJobs|scheduleJob|job.*scheduler|pending.*jobs'
echo -e "\n=== Searching for eventDispatcher.dispatch calls related to jobs ==="
rg -nP --type=ts -C3 'eventDispatcher\.dispatch.*[Jj]ob'Length of output: 129
Let me fix the script syntax errors and re-run with simpler patterns:The web search results don't provide specific information about the codebase. Let me execute a corrected shell script to search the codebase directly:
🏁 Script executed:
#!/bin/bash
# Fixed shell script with simpler patterns
echo "=== Searching for eventDispatcher.dispatch calls ==="
rg -nP --type=ts 'eventDispatcher\.dispatch' -C2
echo -e "\n=== Searching for createReleaseJob usage ==="
rg -nP --type=ts 'createReleaseJob' -C3
echo -e "\n=== Searching for job processing functions ==="
rg -nP --type=ts 'processJob|handleJob|worker|consumer' -C2
echo -e "\n=== Searching for release job status ==="
rg -nP --type=ts 'releaseJob.*status|pending' -C2Length of output: 77331
🏁 Script executed:
#!/bin/bash
echo "=== Checking createReleaseJob implementation ==="
cat packages/db/src/queries/create-release-job.ts
echo -e "\n=== Searching for Kafka event names/topics ==="
rg -nP --type=ts 'topic|TOPIC|event.*name' packages/events/src -C2
echo -e "\n=== Checking event-queue message handler ==="
head -120 apps/event-queue/src/index.tsLength of output: 7230
🏁 Script executed:
#!/bin/bash
echo "=== Searching for job-related event handlers in event-queue ==="
find apps/event-queue/src/events -name "*.ts" -type f | head -20
echo -e "\n=== Checking events index/handler registration ==="
cat apps/event-queue/src/events/index.ts
echo -e "\n=== Searching for job or release event types ==="
rg -nP --type=ts 'JOB|RELEASE' packages/events/src -i | grep -i 'event\|type' | head -30
echo -e "\n=== Searching for polling or scheduler mechanisms for jobs ==="
rg -nP --type=ts 'setInterval|setTimeout|cron|schedule|poll' apps/event-queue/src -C2Length of output: 8576
Dispatch event missing after job creation in force deployment path.
After createReleaseJob at line 41, there's no event dispatch to trigger job processing. While other create operations throughout the codebase follow the pattern: create → eventDispatcher.dispatch* → handler, the job creation in redeploy.ts breaks this pattern. The job is created with status "pending" but nothing consumes or processes it.
The event-queue has handlers for resource/deployment/version creation but no handler for job creation (only Event.JobUpdated exists). Without an event dispatch, the pending job will never be picked up for execution.
await createReleaseJob(tx, existingRelease.release);Add event dispatch after job creation:
const job = await createReleaseJob(tx, existingRelease.release);
await eventDispatcher.dispatchJobUpdated(null, job); // or create a dispatchJobCreated eventAlternatively, verify if there's a background polling mechanism that processes pending jobs and ensure it's active.
🤖 Prompt for AI Agents
In packages/api/src/router/redeploy.ts around line 41 the code calls
createReleaseJob(tx, existingRelease.release) but does not dispatch an event to
trigger job processing; change it to capture the returned job and immediately
call the event dispatcher (e.g., await eventDispatcher.dispatchJobUpdated(null,
job) or the appropriate dispatchJobCreated variant) so the newly-created pending
job is published to the event system and picked up by handlers; ensure the
dispatch call is awaited and uses the correct eventDispatcher method available
in this module.
📊 Code Coverage Reportworkspace-engine coverage: |
Summary by CodeRabbit
Removals
Infrastructure Changes