chore: send job agent events

Author: adityachoudhari26

apps/event-queue/src/events/index.ts

@@ -76,6 +76,9 @@ const workspaceHandlers: Record<Event, Handler<any>> = {
   [Event.UserApprovalRecordCreated]: () => Promise.resolve(),
   [Event.UserApprovalRecordUpdated]: () => Promise.resolve(),
   [Event.UserApprovalRecordDeleted]: () => Promise.resolve(),
+  [Event.JobAgentCreated]: () => Promise.resolve(),
+  [Event.JobAgentUpdated]: () => Promise.resolve(),
+  [Event.JobAgentDeleted]: () => Promise.resolve(),
 };
 
 export type Handler<T extends keyof EventPayload> = (

apps/webservice/src/app/api/github/installation/route.ts

@@ -11,6 +11,7 @@ import { auth } from "@ctrlplane/auth";
 import { eq, takeFirst, takeFirstOrNull } from "@ctrlplane/db";
 import { db } from "@ctrlplane/db/client";
 import { githubEntity, jobAgent, user, workspace } from "@ctrlplane/db/schema";
+import { eventDispatcher } from "@ctrlplane/events";
 
 import type { AuthedOctokitClient } from "../octokit";
 import { env } from "~/env";
@@ -162,15 +163,21 @@ export const GET = async (req: NextRequest) => {
     u.id,
   );
 
-  await db.insert(jobAgent).values({
-    workspaceId: activeWorkspace.id,
-    name: entity.slug,
-    type: "github-app",
-    config: {
-      installationId: entity.installationId,
-      owner: entity.slug,
-    },
-  });
+  const createdAgent = await db
+    .insert(jobAgent)
+    .values({
+      workspaceId: activeWorkspace.id,
+      name: entity.slug,
+      type: "github-app",
+      config: {
+        installationId: entity.installationId,
+        owner: entity.slug,
+      },
+    })
+    .returning()
+    .then(takeFirst);
+
+  await eventDispatcher.dispatchJobAgentCreated(createdAgent);
 
   return NextResponse.redirect(
     `${env.BASE_URL}/${activeWorkspace.slug}/settings/workspace/integrations/github`,

apps/webservice/src/app/api/v1/job-agents/name/route.ts

@@ -3,6 +3,7 @@ import { z } from "zod";
 
 import { eq, takeFirst, takeFirstOrNull } from "@ctrlplane/db";
 import { jobAgent, workspace } from "@ctrlplane/db/schema";
+import { eventDispatcher } from "@ctrlplane/events";
 import { Permission } from "@ctrlplane/validators/auth";
 
 import { authn, authz } from "~/app/api/v1/auth";
@@ -48,5 +49,7 @@ export const PATCH = request()
       .returning()
       .then(takeFirst);
 
+    await eventDispatcher.dispatchJobAgentCreated(tp);
+
     return NextResponse.json(tp);
   });

packages/api/src/router/job.ts

@@ -15,6 +15,7 @@ import {
   takeFirst,
 } from "@ctrlplane/db";
 import * as schema from "@ctrlplane/db/schema";
+import { eventDispatcher } from "@ctrlplane/events";
 import { updateJob } from "@ctrlplane/job-dispatch";
 import { Permission } from "@ctrlplane/validators/auth";
 import { JobAgentType } from "@ctrlplane/validators/jobs";
@@ -86,7 +87,15 @@ const jobAgentRouter = createTRPCRouter({
     })
     .input(schema.createJobAgent)
     .mutation(({ ctx, input }) =>
-      ctx.db.insert(schema.jobAgent).values(input).returning().then(takeFirst),
+      ctx.db
+        .insert(schema.jobAgent)
+        .values(input)
+        .returning()
+        .then(takeFirst)
+        .then(async (agent) => {
+          await eventDispatcher.dispatchJobAgentCreated(agent);
+          return agent;
+        }),
     ),
 
   update: protectedProcedure
@@ -103,7 +112,11 @@ const jobAgentRouter = createTRPCRouter({
         .set(input.data)
         .where(eq(schema.jobAgent.id, input.id))
         .returning()
-        .then(takeFirst),
+        .then(takeFirst)
+        .then(async (agent) => {
+          await eventDispatcher.dispatchJobAgentUpdated(agent);
+          return agent;
+        }),
     ),
 
   history: createTRPCRouter({

packages/events/src/dispatch-jobs.ts

@@ -324,6 +324,17 @@ export class BullMQEventDispatcher implements EventDispatcher {
     await Promise.resolve();
   }
 
+  async dispatchJobAgentCreated(_: schema.JobAgent): Promise<void> {
+    return Promise.resolve();
+  }
+  async dispatchJobAgentUpdated(_: schema.JobAgent): Promise<void> {
+    return Promise.resolve();
+  }
+
+  async dispatchJobAgentDeleted(_: schema.JobAgent): Promise<void> {
+    return Promise.resolve();
+  }
+
   async dispatchPolicyCreated(policy: FullPolicy): Promise<void> {
     await getQueue(Channel.NewPolicy).add(policy.id, policy);
   }

packages/events/src/event-dispatcher.ts

@@ -82,6 +82,10 @@ export interface EventDispatcher {
     deploymentVariableValue: schema.DeploymentVariableValue,
   ): Promise<void>;
 
+  dispatchJobAgentCreated(jobAgent: schema.JobAgent): Promise<void>;
+  dispatchJobAgentUpdated(jobAgent: schema.JobAgent): Promise<void>;
+  dispatchJobAgentDeleted(jobAgent: schema.JobAgent): Promise<void>;
+
   dispatchPolicyCreated(policy: FullPolicy): Promise<void>;
   dispatchPolicyUpdated(
     previous: FullPolicy,

packages/events/src/kafka/event-dispatch/index.ts

@@ -5,6 +5,7 @@ import * as deploymentVariableDispatch from "./deployment-variable.js";
 import * as deploymentVersionDispatch from "./deployment-version.js";
 import * as deploymentDispatch from "./deployment.js";
 import * as environmentDispatch from "./environment.js";
+import * as jobAgentDispatch from "./job-agent.js";
 import * as jobDispatch from "./job.js";
 import * as policyDispatch from "./policy.js";
 import * as releaseTargetDispatch from "./release-target.js";
@@ -175,6 +176,18 @@ export class KafkaEventDispatcher implements EventDispatcher {
     );
   }
 
+  async dispatchJobAgentCreated(jobAgent: schema.JobAgent): Promise<void> {
+    await jobAgentDispatch.dispatchJobAgentCreated(jobAgent);
+  }
+
+  async dispatchJobAgentUpdated(jobAgent: schema.JobAgent): Promise<void> {
+    await jobAgentDispatch.dispatchJobAgentUpdated(jobAgent);
+  }
+
+  async dispatchJobAgentDeleted(jobAgent: schema.JobAgent): Promise<void> {
+    await jobAgentDispatch.dispatchJobAgentDeleted(jobAgent);
+  }
+
   async dispatchPolicyCreated(policy: FullPolicy): Promise<void> {
     await policyDispatch.dispatchPolicyCreated(policy);
   }

packages/events/src/kafka/event-dispatch/job-agent.ts

@@ -0,0 +1,82 @@
+import type * as schema from "@ctrlplane/db/schema";
+import type { Span } from "@ctrlplane/logger";
+import type { WorkspaceEngine } from "@ctrlplane/workspace-engine-sdk";
+
+import type { GoEventPayload, GoMessage } from "../events.js";
+import { createSpanWrapper } from "../../span.js";
+import { sendGoEvent } from "../client.js";
+import { Event } from "../events.js";
+
+const getOapiJobAgent = (
+  jobAgent: schema.JobAgent,
+): WorkspaceEngine["schemas"]["JobAgent"] => ({
+  id: jobAgent.id,
+  workspaceId: jobAgent.workspaceId,
+  name: jobAgent.name,
+  type: jobAgent.type,
+  config: jobAgent.config,
+});
+
+const convertJobAgentToGoEvent = (
+  jobAgent: schema.JobAgent,
+  eventType: keyof GoEventPayload,
+): GoMessage<keyof GoEventPayload> => ({
+  workspaceId: jobAgent.workspaceId,
+  eventType,
+  data: getOapiJobAgent(jobAgent),
+  timestamp: Date.now(),
+});
+
+export const dispatchJobAgentCreated = createSpanWrapper(
+  "dispatchJobAgentCreated",
+  async (span: Span, jobAgent: schema.JobAgent) => {
+    span.setAttribute("job-agent.id", jobAgent.id);
+    span.setAttribute("job-agent.workspaceId", jobAgent.workspaceId);
+    span.setAttribute("job-agent.name", jobAgent.name);
+    span.setAttribute("job-agent.type", jobAgent.type);
+    span.setAttribute(
+      "job-agent.config",
+      JSON.stringify(jobAgent.config, null, 2),
+    );
+
+    await sendGoEvent(
+      convertJobAgentToGoEvent(jobAgent, Event.JobAgentCreated),
+    );
+  },
+);
+
+export const dispatchJobAgentUpdated = createSpanWrapper(
+  "dispatchJobAgentUpdated",
+  async (span: Span, jobAgent: schema.JobAgent) => {
+    span.setAttribute("job-agent.id", jobAgent.id);
+    span.setAttribute("job-agent.workspaceId", jobAgent.workspaceId);
+    span.setAttribute("job-agent.name", jobAgent.name);
+    span.setAttribute("job-agent.type", jobAgent.type);
+    span.setAttribute(
+      "job-agent.config",
+      JSON.stringify(jobAgent.config, null, 2),
+    );
+
+    await sendGoEvent(
+      convertJobAgentToGoEvent(jobAgent, Event.JobAgentUpdated),
+    );
+  },
+);
+
+export const dispatchJobAgentDeleted = createSpanWrapper(
+  "dispatchJobAgentDeleted",
+  async (span: Span, jobAgent: schema.JobAgent) => {
+    span.setAttribute("job-agent.id", jobAgent.id);
+    span.setAttribute("job-agent.workspaceId", jobAgent.workspaceId);
+    span.setAttribute("job-agent.name", jobAgent.name);
+    span.setAttribute("job-agent.type", jobAgent.type);
+    span.setAttribute(
+      "job-agent.config",
+      JSON.stringify(jobAgent.config, null, 2),
+    );
+
+    await sendGoEvent(
+      convertJobAgentToGoEvent(jobAgent, Event.JobAgentDeleted),
+    );
+  },
+);

packages/events/src/kafka/events.ts

@@ -32,6 +32,10 @@ export enum Event {
   DeploymentVersionUpdated = "deployment-version.updated",
   DeploymentVersionDeleted = "deployment-version.deleted",
 
+  JobAgentCreated = "job-agent.created",
+  JobAgentUpdated = "job-agent.updated",
+  JobAgentDeleted = "job-agent.deleted",
+
   EnvironmentCreated = "environment.created",
   EnvironmentUpdated = "environment.updated",
   EnvironmentDeleted = "environment.deleted",
@@ -117,6 +121,9 @@ export type EventPayload = {
     current: schema.DeploymentVersion;
   };
   [Event.DeploymentVersionDeleted]: schema.DeploymentVersion;
+  [Event.JobAgentCreated]: schema.JobAgent;
+  [Event.JobAgentUpdated]: schema.JobAgent;
+  [Event.JobAgentDeleted]: schema.JobAgent;
   [Event.EnvironmentCreated]: schema.Environment;
   [Event.EnvironmentUpdated]: {
     previous: schema.Environment;
@@ -163,6 +170,9 @@ export type GoEventPayload = {
   [Event.DeploymentVersionCreated]: WorkspaceEngine["schemas"]["DeploymentVersion"];
   [Event.DeploymentVersionUpdated]: WorkspaceEngine["schemas"]["DeploymentVersion"];
   [Event.DeploymentVersionDeleted]: WorkspaceEngine["schemas"]["DeploymentVersion"];
+  [Event.JobAgentCreated]: WorkspaceEngine["schemas"]["JobAgent"];
+  [Event.JobAgentUpdated]: WorkspaceEngine["schemas"]["JobAgent"];
+  [Event.JobAgentDeleted]: WorkspaceEngine["schemas"]["JobAgent"];
   [Event.EnvironmentCreated]: WorkspaceEngine["schemas"]["Environment"];
   [Event.EnvironmentUpdated]: WorkspaceEngine["schemas"]["Environment"];
   [Event.EnvironmentDeleted]: WorkspaceEngine["schemas"]["Environment"];