update format

Author: jsbroks

apps/workspace-engine/pkg/events/handler/handler.go

@@ -57,16 +57,11 @@ const (
 	UserApprovalRecordDelete EventType = "user-approval-record.deleted"
 )
 
-// BaseEvent represents the common structure of all events
-type BaseEvent struct {
-	EventType   EventType `json:"eventType"`
-	WorkspaceID string    `json:"workspaceId"`
-}
-
 // RawEvent represents the raw event data received from Kafka messages
 type RawEvent struct {
-	BaseEvent
-	Data json.RawMessage `json:"payload,omitempty"`
+	EventType   EventType       `json:"eventType"`
+	WorkspaceID string          `json:"workspaceId"`
+	Data        json.RawMessage `json:"data,omitempty"`
 }
 
 // Handler defines the interface for processing events

apps/workspace-engine/pkg/events/handler/jobs/jobs.go

@@ -20,11 +20,11 @@ func HandleJobUpdated(
 	}
 
 	ws.Jobs().Upsert(ctx, job)
-	
+
 	rt := &pb.ReleaseTarget{
 		EnvironmentId: job.EnvironmentId,
-		DeploymentId: job.DeploymentId,
-		ResourceId: job.ResourceId,
+		DeploymentId:  job.DeploymentId,
+		ResourceId:    job.ResourceId,
 	}
 	ws.ReleaseManager().TaintReleaseTargets(rt)
 

apps/workspace-engine/pkg/events/handler/policies/policies.go

@@ -33,7 +33,7 @@ func HandlePolicyUpdated(
 	}
 
 	ws.Policies().Upsert(ctx, policy)
-	
+
 	return nil
 }
 

apps/workspace-engine/pkg/pb/selector.go

@@ -8,4 +8,4 @@ func NewJsonSelector(selector *structpb.Struct) *Selector {
 			Json: selector,
 		},
 	}
-}
+}

apps/workspace-engine/test/e2e/engine_deployment_test.go

@@ -257,11 +257,11 @@ func TestEngine_DeploymentJobAgentConfigMerging(t *testing.T) {
 	// Verify deployment has job agent config
 	d, _ := engine.Workspace().Deployments().Get(deploymentID)
 	config := d.GetJobAgentConfig().AsMap()
-	
+
 	if config["namespace"] != "custom-namespace" {
 		t.Fatalf("deployment job agent config namespace mismatch: got %v, want custom-namespace", config["namespace"])
 	}
-	
+
 	if timeout, ok := config["timeout"].(float64); !ok || timeout != 300 {
 		t.Fatalf("deployment job agent config timeout mismatch: got %v, want 300", config["timeout"])
 	}
@@ -278,12 +278,12 @@ func TestEngine_DeploymentJobAgentConfigMerging(t *testing.T) {
 		break
 	}
 	jobConfig := job.GetJobAgentConfig().AsMap()
-	
+
 	// Verify merged config includes deployment-specific settings
 	if jobConfig["namespace"] != "custom-namespace" {
 		t.Fatalf("job config namespace mismatch: got %v, want custom-namespace", jobConfig["namespace"])
 	}
-	
+
 	if timeout, ok := jobConfig["timeout"].(float64); !ok || timeout != 300 {
 		t.Fatalf("job config timeout mismatch: got %v, want 300", jobConfig["timeout"])
 	}
@@ -389,7 +389,7 @@ func TestEngine_DeploymentMultipleJobAgents(t *testing.T) {
 	// Verify each job has the correct job agent
 	k8sJobFound := false
 	dockerJobFound := false
-	
+
 	for _, job := range pendingJobs {
 		if job.DeploymentId == deploymentK8s {
 			k8sJobFound = true

apps/workspace-engine/test/e2e/engine_job_agent_test.go

@@ -187,11 +187,11 @@ func TestEngine_JobAgentWithConfig(t *testing.T) {
 	ja.Name = "Config Test Agent"
 	ja.WorkspaceId = engine.Workspace().ID
 	ja.Config = c.MustNewStructFromMap(map[string]any{
-		"namespace":      "default",
-		"timeout":        300,
-		"retries":        3,
-		"cleanupPolicy":  "always",
-		"imageRegistry":  "docker.io",
+		"namespace":     "default",
+		"timeout":       300,
+		"retries":       3,
+		"cleanupPolicy": "always",
+		"imageRegistry": "docker.io",
 		"resources": map[string]any{
 			"cpu":    "1000m",
 			"memory": "2Gi",
@@ -259,8 +259,8 @@ func TestEngine_JobAgentConfigUpdate(t *testing.T) {
 
 	// Update config
 	ja.Config = c.MustNewStructFromMap(map[string]any{
-		"timeout": 600,
-		"retries": 5,
+		"timeout":  600,
+		"retries":  5,
 		"newField": "newValue",
 	})
 
@@ -563,4 +563,3 @@ func TestEngine_JobAgentNameUniqueness(t *testing.T) {
 		t.Fatal("agents should have different IDs")
 	}
 }
-

apps/workspace-engine/test/e2e/engine_policy_test.go

@@ -257,9 +257,9 @@ func TestEngine_PolicyResourceSelector(t *testing.T) {
 	selector.ResourceSelector = pb.NewJsonSelector(c.MustNewStructFromMap(map[string]any{
 		"type":     "metadata",
 		"operator": "equals",
-			"key":      "priority",
-			"value":    "critical",
-		}),
+		"key":      "priority",
+		"value":    "critical",
+	}),
 	)
 	policy.Selectors = []*pb.PolicyTargetSelector{selector}
 	engine.PushEvent(ctx, handler.PolicyCreate, policy)
@@ -459,7 +459,7 @@ func TestEngine_PolicyMultipleSelectors(t *testing.T) {
 				}),
 			),
 			integration.WithPolicyTargetSelector(
-				integration.PolicyTargetJsonDeploymentSelector(map[string]any{	
+				integration.PolicyTargetJsonDeploymentSelector(map[string]any{
 					"type":     "name",
 					"operator": "contains",
 					"value":    "staging",

apps/workspace-engine/test/integration/workspace.go

@@ -64,25 +64,23 @@ func (tw *TestWorkspace) PushEvent(ctx context.Context, eventType handler.EventT
 	// Marshal the data payload
 	var dataBytes []byte
 	var err error
-	
+
 	// Check if data is a protobuf message
 	if protoMsg, ok := data.(proto.Message); ok {
 		dataBytes, err = protojson.Marshal(protoMsg)
 	} else {
 		dataBytes, err = json.Marshal(data)
 	}
-	
+
 	if err != nil {
 		tw.t.Fatalf("failed to marshal event data: %v", err)
 		return tw
 	}
 
 	// Create raw event
 	rawEvent := handler.RawEvent{
-		BaseEvent: handler.BaseEvent{
-			EventType:   eventType,
-			WorkspaceID: tw.workspace.ID,
-		},
+		EventType:   eventType,
+		WorkspaceID: tw.workspace.ID,
 		Data: dataBytes,
 	}
 

packages/events/src/kafka/client.ts

@@ -3,7 +3,12 @@ import { Kafka } from "kafkajs";
 
 import { logger } from "@ctrlplane/logger";
 
-import type { EventPayload, GoEventPayload, Message } from "./events.js";
+import type {
+  EventPayload,
+  GoEventPayload,
+  GoMessage,
+  Message,
+} from "./events.js";
 import { env } from "../config.js";
 
 const log = logger.child({ component: "kafka-client" });
@@ -26,7 +31,7 @@ const getProducer = async () => {
 };
 
 export const sendGoEvent = async <T extends keyof GoEventPayload>(
-  message: Message<T> | Message<T>[],
+  message: GoMessage<T> | GoMessage<T>[],
 ) => {
   const messages = Array.isArray(message) ? message : [message];
   const topic = "workspace-events";
@@ -36,7 +41,6 @@ export const sendGoEvent = async <T extends keyof GoEventPayload>(
     messages: messages.map((message) => ({
       key: message.workspaceId,
       value: JSON.stringify(message),
-      timestamp: message.timestamp.toString(),
     })),
   });
 };

packages/events/src/kafka/events.ts

@@ -163,24 +163,9 @@ export type Message<T extends keyof EventPayload> = {
   payload: EventPayload[T];
 };
 
-// Helper function to wrap a selector in the protobuf format
-function wrapSelector<T extends Record<string, any> | null | undefined>(
-  selector: T,
-): T extends null | undefined ? null : { json_selector: T } {
-  if (selector == null) return null as any;
-  return { json_selector: selector } as any;
-}
-
-// Helper to transform an object's selector fields to protobuf format
-export function wrapSelectorsInObject<T extends Record<string, any>>(
-  obj: T,
-  selectorFields: (keyof T)[],
-): T {
-  const result = { ...obj };
-  for (const field of selectorFields) {
-    if (field in result) {
-      (result as any)[field] = wrapSelector(result[field]);
-    }
-  }
-  return result;
-}
+export type GoMessage<T extends keyof GoEventPayload> = {
+  workspaceId: string;
+  eventType: T;
+  source: "api" | "scheduler" | "user-action";
+  data: GoEventPayload[T];
+};