fix: Move events to job dispatch and use deleteResources utility func…

…tion everywhere (#222)

apps/jobs/Dockerfile

@@ -27,7 +27,6 @@ COPY packages/validators/package.json ./packages/validators/package.json
 COPY packages/logger/package.json ./packages/logger/package.json
 COPY packages/job-dispatch/package.json ./packages/job-dispatch/package.json
 COPY packages/secrets/package.json ./packages/secrets/package.json
-COPY packages/events/package.json ./packages/events/package.json
 
 COPY apps/jobs/package.json ./apps/jobs/package.json
 

apps/jobs/package.json

@@ -14,7 +14,6 @@
   },
   "dependencies": {
     "@ctrlplane/db": "workspace:*",
-    "@ctrlplane/events": "workspace:*",
     "@ctrlplane/job-dispatch": "workspace:*",
     "@ctrlplane/logger": "workspace:*",
     "@ctrlplane/validators": "workspace:*",

apps/jobs/src/expired-env-checker/index.ts

@@ -1,7 +1,10 @@
 import { inArray, lte } from "@ctrlplane/db";
 import { db } from "@ctrlplane/db/client";
 import * as SCHEMA from "@ctrlplane/db/schema";
-import { getEventsForEnvironmentDeleted, handleEvent } from "@ctrlplane/events";
+import {
+  getEventsForEnvironmentDeleted,
+  handleEvent,
+} from "@ctrlplane/job-dispatch";
 
 export const run = async () => {
   const expiredEnvironments = await db

apps/pty-proxy/src/controller/sockets.ts

@@ -1,6 +1,8 @@
 import ms from "ms";
 
+import { eq } from "@ctrlplane/db";
 import { db } from "@ctrlplane/db/client";
+import * as SCHEMA from "@ctrlplane/db/schema";
 import { deleteResources } from "@ctrlplane/job-dispatch";
 import { logger } from "@ctrlplane/logger";
 
@@ -26,9 +28,10 @@ setInterval(() => {
 
       agent.socket.close(1000, "Agent connection timed out");
       agents.delete(agentId);
-      deleteResources(db, [agentId]).then(() => {
-        logger.info("Deleted stale agent resource", { agentId });
-      });
+      db.query.resource
+        .findFirst({ where: eq(SCHEMA.resource.id, agentId) })
+        .then((resource) => resource && deleteResources(db, [resource]))
+        .then(() => logger.info("Deleted stale agent resource", { agentId }));
     }
   }
 }, ms("1m"));

apps/webservice/Dockerfile

@@ -36,7 +36,6 @@ COPY packages/job-dispatch/package.json ./packages/job-dispatch/package.json
 COPY packages/ui/package.json ./packages/ui/package.json
 COPY packages/logger/package.json ./packages/logger/package.json
 COPY packages/secrets/package.json ./packages/secrets/package.json
-COPY packages/events/package.json ./packages/events/package.json
 
 COPY apps/webservice/package.json ./apps/webservice/package.json
 

apps/webservice/src/app/[workspaceSlug]/(app)/systems/[systemSlug]/deployments/[deploymentSlug]/hooks/EditHookDialog.tsx

@@ -76,6 +76,7 @@ export const EditHookDialog: React.FC<EditHookDialogProps> = ({
 
   const defaultValues = {
     ...hook,
+    action: hook.action as HookAction,
     jobAgentId: hook.runhook?.runbook.jobAgentId ?? null,
     jobAgentConfig: hook.runhook?.runbook.jobAgentConfig ?? null,
   };
@@ -144,7 +145,7 @@ export const EditHookDialog: React.FC<EditHookDialogProps> = ({
                           className="items-center justify-start gap-2 px-2"
                         >
                           <IconSelector className="h-4 w-4" />
-                          {value === "" ? "Select action..." : value}
+                          {value}
                         </Button>
                       </PopoverTrigger>
                       <PopoverContent align="start" className="w-[462px] p-0">

packages/api/package.json

@@ -21,7 +21,6 @@
   "dependencies": {
     "@ctrlplane/auth": "workspace:*",
     "@ctrlplane/db": "workspace:*",
-    "@ctrlplane/events": "workspace:*",
     "@ctrlplane/job-dispatch": "workspace:*",
     "@ctrlplane/logger": "workspace:*",
     "@ctrlplane/secrets": "workspace:*",

packages/api/src/router/deployment.ts

@@ -37,7 +37,10 @@ import {
   updateReleaseChannel,
   workspace,
 } from "@ctrlplane/db/schema";
-import { getEventsForDeploymentDeleted, handleEvent } from "@ctrlplane/events";
+import {
+  getEventsForDeploymentDeleted,
+  handleEvent,
+} from "@ctrlplane/job-dispatch";
 import { Permission } from "@ctrlplane/validators/auth";
 import { JobStatus } from "@ctrlplane/validators/jobs";
 

packages/api/src/router/environment.ts

@@ -26,8 +26,11 @@ import {
   system,
   updateEnvironment,
 } from "@ctrlplane/db/schema";
-import { getEventsForEnvironmentDeleted, handleEvent } from "@ctrlplane/events";
-import { dispatchJobsForNewResources } from "@ctrlplane/job-dispatch";
+import {
+  dispatchJobsForNewResources,
+  getEventsForEnvironmentDeleted,
+  handleEvent,
+} from "@ctrlplane/job-dispatch";
 import { Permission } from "@ctrlplane/validators/auth";
 import {
   ComparisonOperator,

packages/api/src/router/resources.ts

@@ -17,11 +17,11 @@ import {
   takeFirstOrNull,
 } from "@ctrlplane/db";
 import * as schema from "@ctrlplane/db/schema";
-import { getEventsForResourceDeleted, handleEvent } from "@ctrlplane/events";
 import {
   cancelOldReleaseJobTriggersOnJobDispatch,
   createJobApprovals,
   createReleaseJobTriggers,
+  deleteResources,
   dispatchReleaseJobTriggers,
   isPassingAllPoliciesExceptNewerThanLastActive,
   isPassingNoPendingJobsPolicy,
@@ -577,20 +577,11 @@ export const resourceRouter = createTRPCRouter({
         ),
     })
     .input(z.array(z.string().uuid()))
-    .mutation(async ({ ctx, input }) => {
-      const resources = await ctx.db.query.resource.findMany({
-        where: inArray(schema.resource.id, input),
-      });
-      const events = (
-        await Promise.allSettled(resources.map(getEventsForResourceDeleted))
-      ).flatMap((r) => (r.status === "fulfilled" ? r.value : []));
-      await Promise.allSettled(events.map(handleEvent));
-
-      return ctx.db
-        .delete(schema.resource)
-        .where(inArray(schema.resource.id, input))
-        .returning();
-    }),
+    .mutation(async ({ ctx, input }) =>
+      ctx.db.query.resource
+        .findMany({ where: inArray(schema.resource.id, input) })
+        .then((resources) => deleteResources(ctx.db, resources)),
+    ),
 
   metadataKeys: protectedProcedure
     .meta({

packages/api/src/router/runbook.ts

@@ -35,43 +35,16 @@ export const runbookRouter = createTRPCRouter({
           .on({ type: "system", id: input }),
     })
     .input(z.string().uuid())
-    .query(({ ctx, input }) => {
-      return ctx.db.query.runbook.findMany({
+    .query(({ ctx, input }) =>
+      ctx.db.query.runbook.findMany({
         where: eq(SCHEMA.runbook.systemId, input),
         with: {
           runhooks: { with: { hook: true } },
           jobAgent: true,
           variables: true,
         },
-      });
-      // return ctx.db
-      //   .select()
-      //   .from(SCHEMA.runbook)
-      //   .leftJoin(
-      //     SCHEMA.runbookVariable,
-      //     eq(SCHEMA.runbookVariable.runbookId, SCHEMA.runbook.id),
-      //   )
-      //   .leftJoin(
-      //     SCHEMA.jobAgent,
-      //     eq(SCHEMA.runbook.jobAgentId, SCHEMA.jobAgent.id),
-      //   )
-      //   .leftJoin(
-      //     SCHEMA.runbookJobTrigger,
-      //     eq(SCHEMA.runbook.id, SCHEMA.runbookJobTrigger.runbookId),
-      //   )
-      //   .leftJoin(SCHEMA.job, eq(SCHEMA.runbookJobTrigger.jobId, SCHEMA.job.id))
-      //   .where(eq(SCHEMA.runbook.systemId, input))
-      //   .then((rbs) =>
-      //     _.chain(rbs)
-      //       .groupBy((rb) => rb.runbook.id)
-      //       .map((rb) => ({
-      //         ...rb[0]!.runbook,
-      //         variables: rb.map((v) => v.runbook_variable).filter(isPresent),
-      //         jobAgent: rb[0]!.job_agent,
-      //       }))
-      //       .value(),
-      //   );
-    }),
+      }),
+    ),
 
   create: protectedProcedure
     .meta({

packages/db/src/schema/event.ts

@@ -56,9 +56,7 @@ export const runhook = pgTable(
       .notNull()
       .references(() => runbook.id, { onDelete: "cascade" }),
   },
-  (t) => ({
-    uniq: uniqueIndex().on(t.hookId, t.runbookId),
-  }),
+  (t) => ({ uniq: uniqueIndex().on(t.hookId, t.runbookId) }),
 );
 
 export const runhookRelations = relations(runhook, ({ one }) => ({

packages/job-dispatch/src/events/handlers/index.ts

@@ -0,0 +1 @@
+export * from "./resource-removed.js";

packages/events/src/handlers/target-removed.ts → packages/job-dispatch/src/events/handlers/resource-removed.ts

@@ -3,7 +3,8 @@ import type { ResourceRemoved } from "@ctrlplane/validators/events";
 import { and, eq } from "@ctrlplane/db";
 import { db } from "@ctrlplane/db/client";
 import * as SCHEMA from "@ctrlplane/db/schema";
-import { dispatchRunbook } from "@ctrlplane/job-dispatch";
+
+import { dispatchRunbook } from "../../job-dispatch.js";
 
 export const handleResourceRemoved = async (event: ResourceRemoved) => {
   const { resource, deployment } = event.payload;

packages/events/src/triggers/index.ts → packages/job-dispatch/src/events/triggers/index.ts

@@ -1,3 +1,3 @@
 export * from "./environment-deleted.js";
 export * from "./deployment-deleted.js";
-export * from "./target-deleted.js";
+export * from "./resource-deleted.js";

packages/job-dispatch/src/index.ts

@@ -23,3 +23,4 @@ export * from "./policies/concurrency-policy.js";
 export * from "./policies/release-window.js";
 export * from "./environment-creation.js";
 export * from "./pending-job-checker.js";
+export * from "./events/index.js";

packages/job-dispatch/src/resource.ts

@@ -22,6 +22,7 @@ import {
 import { logger } from "@ctrlplane/logger";
 import { variablesAES256 } from "@ctrlplane/secrets";
 
+import { getEventsForResourceDeleted, handleEvent } from "./events/index.js";
 import { dispatchJobsForNewResources } from "./new-resource.js";
 
 const log = logger.child({ label: "upsert-resources" });
@@ -320,10 +321,7 @@ export const upsertResources = async (
     );
 
     if (resourcesToDelete.length > 0) {
-      await deleteResources(
-        tx,
-        resourcesToDelete.map((r) => r.id),
-      ).catch((err) => {
+      await deleteResources(tx, resourcesToDelete).catch((err) => {
         log.error("Error deleting resources", { error: err });
         throw err;
       });
@@ -346,6 +344,12 @@ export const upsertResources = async (
  * @param tx - The transaction to use.
  * @param resourceIds - The ids of the resources to delete.
  */
-export const deleteResources = async (tx: Tx, resourceIds: string[]) => {
+export const deleteResources = async (tx: Tx, resources: Resource[]) => {
+  const eventsPromises = Promise.all(
+    resources.map(getEventsForResourceDeleted),
+  );
+  const events = await eventsPromises.then((res) => res.flat());
+  await Promise.all(events.map(handleEvent));
+  const resourceIds = resources.map((r) => r.id);
   await tx.delete(resource).where(inArray(resource.id, resourceIds));
 };