chore: send system events

Author: adityachoudhari26

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

@@ -70,6 +70,12 @@ const workspaceHandlers: Record<Event, Handler<any>> = {
   [Event.PolicyDeleted]: deletedPolicy,
   [Event.JobUpdated]: updateJob,
   [Event.EvaluateReleaseTarget]: evaluateReleaseTarget,
+  [Event.SystemCreated]: () => Promise.resolve(),
+  [Event.SystemUpdated]: () => Promise.resolve(),
+  [Event.SystemDeleted]: () => Promise.resolve(),
+  [Event.UserApprovalRecordCreated]: () => Promise.resolve(),
+  [Event.UserApprovalRecordUpdated]: () => Promise.resolve(),
+  [Event.UserApprovalRecordDeleted]: () => Promise.resolve(),
 };
 
 export type Handler<T extends keyof EventPayload> = (

apps/webservice/src/app/api/v1/systems/[systemId]/route.ts

@@ -4,6 +4,7 @@ import httpStatus from "http-status";
 
 import { eq, takeFirst } from "@ctrlplane/db";
 import * as schema from "@ctrlplane/db/schema";
+import { eventDispatcher } from "@ctrlplane/events";
 import { Permission } from "@ctrlplane/validators/auth";
 
 import { authn, authz } from "../../auth";
@@ -58,6 +59,10 @@ export const PATCH = request()
       .where(eq(schema.system.id, systemId))
       .returning()
       .then(takeFirst)
+      .then(async (system) => {
+        await eventDispatcher.dispatchSystemUpdated(system);
+        return system;
+      })
       .then((system) => NextResponse.json(system, { status: httpStatus.OK }))
       .catch((error) =>
         NextResponse.json(
@@ -93,6 +98,10 @@ export const DELETE = request()
         .where(eq(schema.system.id, systemId))
         .returning()
         .then(takeFirst)
+        .then(async (system) => {
+          await eventDispatcher.dispatchSystemDeleted(system);
+          return system;
+        })
         .then((system) =>
           NextResponse.json(
             { message: "System deleted", system },

apps/webservice/src/app/api/v1/systems/route.ts

@@ -5,6 +5,7 @@ import { z } from "zod";
 
 import { and, eq, takeFirst, takeFirstOrNull } from "@ctrlplane/db";
 import * as schema from "@ctrlplane/db/schema";
+import { eventDispatcher } from "@ctrlplane/events";
 import { logger } from "@ctrlplane/logger";
 import { Permission } from "@ctrlplane/validators/auth";
 
@@ -38,24 +39,26 @@ export const POST = request()
         .then(takeFirstOrNull);
 
       if (existingSystem != null) {
-        // Update existing system
         const updatedSystem = await ctx.db
           .update(schema.system)
           .set(ctx.body)
           .where(eq(schema.system.id, existingSystem.id))
           .returning()
           .then(takeFirst);
 
+        await eventDispatcher.dispatchSystemUpdated(updatedSystem);
+
         return NextResponse.json(updatedSystem, { status: httpStatus.OK });
       }
 
-      // Create new system
       const newSystem = await ctx.db
         .insert(schema.system)
         .values(ctx.body)
         .returning()
         .then(takeFirst);
 
+      await eventDispatcher.dispatchSystemCreated(newSystem);
+
       return NextResponse.json(newSystem, { status: httpStatus.CREATED });
     } catch (error) {
       if (error instanceof z.ZodError)

packages/api/src/router/system.ts

@@ -28,6 +28,7 @@ import {
   updateSystem,
   workspace,
 } from "@ctrlplane/db/schema";
+import { eventDispatcher } from "@ctrlplane/events";
 import { Permission } from "@ctrlplane/validators/auth";
 import {
   ComparisonOperator,
@@ -199,11 +200,20 @@ export const systemRouter = createTRPCRouter({
           .returning()
           .then(takeFirst);
 
-        await Promise.all([
+        await eventDispatcher.dispatchSystemCreated(sys);
+
+        const defaultEnvs = await Promise.all([
           upsertEnv(db, { systemId: sys.id, name: "Production" }),
           upsertEnv(db, { systemId: sys.id, name: "QA" }),
           upsertEnv(db, { systemId: sys.id, name: "Staging" }),
         ]);
+
+        await Promise.all(
+          defaultEnvs.map((env) =>
+            eventDispatcher.dispatchEnvironmentCreated(env),
+          ),
+        );
+
         return sys;
       }),
     ),
@@ -222,7 +232,11 @@ export const systemRouter = createTRPCRouter({
         .set(input.data)
         .where(eq(system.id, input.id))
         .returning()
-        .then(takeFirst),
+        .then(takeFirst)
+        .then((sys) => {
+          eventDispatcher.dispatchSystemUpdated(sys);
+          return sys;
+        }),
     ),
 
   delete: protectedProcedure
@@ -238,7 +252,11 @@ export const systemRouter = createTRPCRouter({
         .delete(system)
         .where(eq(system.id, input))
         .returning()
-        .then(takeFirst),
+        .then(takeFirst)
+        .then((sys) => {
+          eventDispatcher.dispatchSystemDeleted(sys);
+          return sys;
+        }),
     ),
 
   resources: protectedProcedure

packages/events/src/dispatch-jobs.ts

@@ -163,6 +163,18 @@ export const dispatchQueueJob = () => ({
 });
 
 export class BullMQEventDispatcher implements EventDispatcher {
+  async dispatchSystemCreated(_: schema.System): Promise<void> {
+    return Promise.resolve();
+  }
+
+  async dispatchSystemUpdated(_: schema.System): Promise<void> {
+    return Promise.resolve();
+  }
+
+  async dispatchSystemDeleted(_: schema.System): Promise<void> {
+    return Promise.resolve();
+  }
+
   async dispatchResourceCreated(resource: schema.Resource): Promise<void> {
     await getQueue(Channel.NewResource).add(resource.id, resource);
   }

packages/events/src/event-dispatcher.ts

@@ -13,6 +13,10 @@ export type FullPolicy = schema.Policy & {
 };
 
 export interface EventDispatcher {
+  dispatchSystemCreated(system: schema.System): Promise<void>;
+  dispatchSystemUpdated(system: schema.System): Promise<void>;
+  dispatchSystemDeleted(system: schema.System): Promise<void>;
+
   dispatchResourceCreated(resource: schema.Resource): Promise<void>;
   dispatchResourceUpdated(
     previous: schema.Resource,

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

@@ -9,9 +9,22 @@ import * as jobDispatch from "./job.js";
 import * as policyDispatch from "./policy.js";
 import * as releaseTargetDispatch from "./release-target.js";
 import * as resourceDispatch from "./resource.js";
+import * as systemDispatch from "./system.js";
 import * as userApprovalRecordDispatch from "./user-approval-record.js";
 
 export class KafkaEventDispatcher implements EventDispatcher {
+  async dispatchSystemCreated(system: schema.System): Promise<void> {
+    await systemDispatch.dispatchSystemCreated(system);
+  }
+
+  async dispatchSystemUpdated(system: schema.System): Promise<void> {
+    await systemDispatch.dispatchSystemUpdated(system);
+  }
+
+  async dispatchSystemDeleted(system: schema.System): Promise<void> {
+    await systemDispatch.dispatchSystemDeleted(system);
+  }
+
   async dispatchResourceCreated(resource: schema.Resource): Promise<void> {
     await resourceDispatch.dispatchResourceCreated(resource);
   }

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

@@ -0,0 +1,63 @@
+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 getOapiSystem = (
+  system: schema.System,
+): WorkspaceEngine["schemas"]["System"] => ({
+  id: system.id,
+  workspaceId: system.workspaceId,
+  name: system.name,
+  description: system.description,
+});
+
+const convertSystemToGoEvent = (
+  system: schema.System,
+  eventType: keyof GoEventPayload,
+): GoMessage<keyof GoEventPayload> => ({
+  workspaceId: system.workspaceId,
+  eventType,
+  data: getOapiSystem(system),
+  timestamp: Date.now(),
+});
+
+export const dispatchSystemCreated = createSpanWrapper(
+  "dispatchSystemCreated",
+  async (span: Span, system: schema.System) => {
+    span.setAttribute("system.id", system.id);
+    span.setAttribute("system.workspaceId", system.workspaceId);
+    span.setAttribute("system.name", system.name);
+    span.setAttribute("system.description", system.description);
+
+    await sendGoEvent(convertSystemToGoEvent(system, Event.SystemCreated));
+  },
+);
+
+export const dispatchSystemUpdated = createSpanWrapper(
+  "dispatchSystemUpdated",
+  async (span: Span, system: schema.System) => {
+    span.setAttribute("system.id", system.id);
+    span.setAttribute("system.workspaceId", system.workspaceId);
+    span.setAttribute("system.name", system.name);
+    span.setAttribute("system.description", system.description);
+
+    await sendGoEvent(convertSystemToGoEvent(system, Event.SystemUpdated));
+  },
+);
+
+export const dispatchSystemDeleted = createSpanWrapper(
+  "dispatchSystemDeleted",
+  async (span: Span, system: schema.System) => {
+    span.setAttribute("system.id", system.id);
+    span.setAttribute("system.workspaceId", system.workspaceId);
+    span.setAttribute("system.name", system.name);
+    span.setAttribute("system.description", system.description);
+
+    await sendGoEvent(convertSystemToGoEvent(system, Event.SystemDeleted));
+  },
+);

packages/events/src/kafka/events.ts

@@ -4,6 +4,10 @@ import type { WorkspaceEngine } from "@ctrlplane/workspace-engine-sdk";
 import type * as PB from "../workspace-engine/types/index.js";
 
 export enum Event {
+  SystemCreated = "system.created",
+  SystemUpdated = "system.updated",
+  SystemDeleted = "system.deleted",
+
   ResourceCreated = "resource.created",
   ResourceUpdated = "resource.updated",
   ResourceDeleted = "resource.deleted",
@@ -130,6 +134,9 @@ export type EventPayload = {
   [Event.UserApprovalRecordCreated]: schema.PolicyRuleAnyApprovalRecord;
   [Event.UserApprovalRecordUpdated]: schema.PolicyRuleAnyApprovalRecord;
   [Event.UserApprovalRecordDeleted]: schema.PolicyRuleAnyApprovalRecord;
+  [Event.SystemCreated]: schema.System;
+  [Event.SystemUpdated]: schema.System;
+  [Event.SystemDeleted]: schema.System;
   // [Event.JobCreated]: schema.Job;
   // [Event.JobDeleted]: schema.Job;
   // [Event.SystemCreated]: schema.System;
@@ -138,6 +145,9 @@ export type EventPayload = {
 };
 
 export type GoEventPayload = {
+  [Event.SystemCreated]: WorkspaceEngine["schemas"]["System"];
+  [Event.SystemUpdated]: WorkspaceEngine["schemas"]["System"];
+  [Event.SystemDeleted]: WorkspaceEngine["schemas"]["System"];
   [Event.ResourceCreated]: WorkspaceEngine["schemas"]["Resource"];
   [Event.ResourceUpdated]: WorkspaceEngine["schemas"]["Resource"];
   [Event.ResourceDeleted]: WorkspaceEngine["schemas"]["Resource"];