fix: shared processor event extractors

internal/processor/activity.go

@@ -142,27 +142,13 @@ func (b *ActivityBuilder) BuildFromEvent(
 	links []cel.Link,
 	resolveKind KindResolver,
 ) (*v1alpha1.Activity, error) {
-	regarding, _ := eventMap["regarding"].(map[string]interface{})
+	// Extract the involved/regarding object using the shared fallback chain:
+	// "regarding" (events.k8s.io/v1) -> "involvedObject" (core/v1).
+	regarding := ResolveInvolvedObject(eventMap)
 
-	// Extract timestamps
-	var timestamp time.Time
-	if ts, ok := eventMap["eventTime"].(string); ok {
-		if t, err := time.Parse(time.RFC3339Nano, ts); err == nil {
-			timestamp = t
-		}
-	}
-	if timestamp.IsZero() {
-		if metadata, ok := eventMap["metadata"].(map[string]interface{}); ok {
-			if ts, ok := metadata["creationTimestamp"].(string); ok {
-				if t, err := time.Parse(time.RFC3339, ts); err == nil {
-					timestamp = t
-				}
-			}
-		}
-	}
-	if timestamp.IsZero() {
-		timestamp = time.Now()
-	}
+	// Extract timestamp using the shared fallback chain: eventTime -> lastTimestamp
+	// -> firstTimestamp -> metadata.creationTimestamp -> now().
+	timestamp := resolveEventTimestamp(eventMap)
 
 	// Extract resource info from regarding
 	namespace := GetNestedString(regarding, "namespace")
@@ -173,15 +159,9 @@ func (b *ActivityBuilder) BuildFromEvent(
 	// Events are typically system-generated
 	changeSource := ChangeSourceSystem
 
-	// For events, actor is usually the reporting component
-	reportingController := GetNestedString(eventMap, "reportingController")
-	actor := v1alpha1.ActivityActor{
-		Type: ActorTypeSystem,
-		Name: reportingController,
-	}
-	if actor.Name == "" {
-		actor.Name = "unknown"
-	}
+	// Resolve actor from reporting controller or source component, using the
+	// fallback chain shared with the live event processor.
+	actor := resolveActorFromEvent(eventMap)
 
 	// Extract tenant from scope annotations; fall back to platform scope when absent.
 	tenant := ExtractTenantFromAnnotations(eventMap)
@@ -211,12 +191,7 @@ func (b *ActivityBuilder) BuildFromEvent(
 			Name:              name,
 			Namespace:         namespace,
 			CreationTimestamp: metav1.NewTime(timestamp),
-			Labels: map[string]string{
-				"activity.miloapis.com/origin-type":   "event",
-				"activity.miloapis.com/change-source": changeSource,
-				"activity.miloapis.com/api-group":     b.APIGroup,
-				"activity.miloapis.com/resource-kind": b.Kind,
-			},
+			Labels:            eventActivityLabels(changeSource, b.APIGroup, b.Kind, getStringFromMap(eventMap, "reason")),
 		},
 		Spec: v1alpha1.ActivitySpec{
 			Summary:      summary,

internal/processor/event.go

@@ -152,7 +152,7 @@ func (p *EventProcessor) processMessage(ctx context.Context, msg *nats.Msg) erro
 	// Extract involved object info to find matching policy.
 	// Kubernetes events have either "regarding" (events.k8s.io/v1) or
 	// "involvedObject" (core/v1) to identify the subject resource.
-	involvedObject := p.getInvolvedObject(event)
+	involvedObject := ResolveInvolvedObject(event)
 	if involvedObject == nil {
 		klog.V(4).Info("Event has no involved object, skipping")
 		return nil
@@ -234,20 +234,6 @@ func (p *EventProcessor) processMessage(ctx context.Context, msg *nats.Msg) erro
 	return nil
 }
 
-// getInvolvedObject extracts the involved object from a Kubernetes event.
-// Handles both v1.Event (regarding) and corev1.Event (involvedObject) formats.
-func (p *EventProcessor) getInvolvedObject(event map[string]interface{}) map[string]interface{} {
-	// Try "regarding" first (events.k8s.io/v1).
-	if regarding, ok := event["regarding"].(map[string]interface{}); ok {
-		return regarding
-	}
-	// Fall back to "involvedObject" (v1).
-	if involvedObject, ok := event["involvedObject"].(map[string]interface{}); ok {
-		return involvedObject
-	}
-	return nil
-}
-
 // normalizeEvent creates a copy of the event with a "regarding" field.
 // This ensures CEL expressions can consistently use event.regarding regardless
 // of whether the original event used "regarding" or "involvedObject".
@@ -275,41 +261,9 @@ func (p *EventProcessor) buildActivity(
 	summary string,
 	links []cel.Link,
 ) *v1alpha1.Activity {
-	// Extract timestamps - try eventTime first (events.k8s.io/v1).
-	var timestamp time.Time
-	if ts := getStringFromMap(event, "eventTime"); ts != "" {
-		if t, err := time.Parse(time.RFC3339Nano, ts); err == nil {
-			timestamp = t
-		}
-	}
-	// Fall back to lastTimestamp or firstTimestamp.
-	if timestamp.IsZero() {
-		if ts := getStringFromMap(event, "lastTimestamp"); ts != "" {
-			if t, err := time.Parse(time.RFC3339Nano, ts); err == nil {
-				timestamp = t
-			}
-		}
-	}
-	if timestamp.IsZero() {
-		if ts := getStringFromMap(event, "firstTimestamp"); ts != "" {
-			if t, err := time.Parse(time.RFC3339Nano, ts); err == nil {
-				timestamp = t
-			}
-		}
-	}
-	// Fall back to metadata.creationTimestamp.
-	if timestamp.IsZero() {
-		if metadata, ok := event["metadata"].(map[string]interface{}); ok {
-			if ts := getStringFromMap(metadata, "creationTimestamp"); ts != "" {
-				if t, err := time.Parse(time.RFC3339Nano, ts); err == nil {
-					timestamp = t
-				}
-			}
-		}
-	}
-	if timestamp.IsZero() {
-		timestamp = time.Now()
-	}
+	// Extract timestamp using the shared fallback chain: eventTime -> lastTimestamp
+	// -> firstTimestamp -> metadata.creationTimestamp -> now().
+	timestamp := resolveEventTimestamp(event)
 
 	// Extract resource info from involved object.
 	namespace := getStringFromMap(involvedObject, "namespace")
@@ -318,7 +272,7 @@ func (p *EventProcessor) buildActivity(
 	apiVersion := getStringFromMap(involvedObject, "apiVersion")
 
 	// Resolve actor from reporting controller or source component.
-	actor := p.resolveEventActor(event)
+	actor := resolveActorFromEvent(event)
 
 	// Events from controllers are always system-initiated.
 	changeSource := ChangeSourceSystem
@@ -356,13 +310,7 @@ func (p *EventProcessor) buildActivity(
 			Name:              name,
 			Namespace:         namespace,
 			CreationTimestamp: metav1.NewTime(timestamp),
-			Labels: map[string]string{
-				"activity.miloapis.com/origin-type":   "event",
-				"activity.miloapis.com/change-source": changeSource,
-				"activity.miloapis.com/api-group":     matched.APIGroup,
-				"activity.miloapis.com/resource-kind": matched.Kind,
-				"activity.miloapis.com/event-reason":  getStringFromMap(event, "reason"),
-			},
+			Labels:            eventActivityLabels(changeSource, matched.APIGroup, matched.Kind, getStringFromMap(event, "reason")),
 		},
 		Spec: v1alpha1.ActivitySpec{
 			Summary:      summary,
@@ -387,30 +335,6 @@ func (p *EventProcessor) buildActivity(
 	}
 }
 
-// resolveEventActor extracts actor information from a Kubernetes event.
-// Events are generated by controllers, so we extract the reporting controller or source component.
-func (p *EventProcessor) resolveEventActor(event map[string]interface{}) v1alpha1.ActivityActor {
-	// Try reportingController first (events.k8s.io/v1).
-	reportingController := getStringFromMap(event, "reportingController")
-
-	// Fall back to source.component (v1).
-	if reportingController == "" {
-		if source, ok := event["source"].(map[string]interface{}); ok {
-			reportingController = getStringFromMap(source, "component")
-		}
-	}
-
-	// Default to unknown if we can't find the controller.
-	if reportingController == "" {
-		reportingController = "unknown"
-	}
-
-	return v1alpha1.ActivityActor{
-		Type: ActorTypeController,
-		Name: reportingController,
-	}
-}
-
 // publishActivity serializes and publishes an Activity to the NATS ACTIVITIES stream.
 func (p *EventProcessor) publishActivity(ctx context.Context, activity *v1alpha1.Activity) error {
 	data, err := json.Marshal(activity)

internal/processor/event_activity_parity_test.go

@@ -0,0 +1,163 @@
+package processor
+
+import (
+	"reflect"
+	"testing"
+)
+
+// TestEventBuildersParity asserts EventProcessor.buildActivity (the live path)
+// and ActivityBuilder.BuildFromEvent (PolicyPreview/reindex) produce identical
+// output for the same legacy-format event.
+func TestEventBuildersParity(t *testing.T) {
+	event := map[string]any{
+		"metadata": map[string]any{
+			"uid": "event-legacy-123",
+			// Differs from lastTimestamp so a wrong fallback choice is visible.
+			"creationTimestamp": "2024-01-15T10:30:00Z",
+		},
+		// No eventTime; the fallback chain must pick this over creationTimestamp.
+		"lastTimestamp": "2024-01-14T09:00:00Z",
+		"reason":        "Scheduled",
+		"message":       "Successfully assigned default/my-pod to node-1",
+		// Legacy actor field; no reportingController.
+		"source": map[string]any{
+			"component": "kubelet",
+		},
+		// Legacy subject field; no regarding.
+		"involvedObject": map[string]any{
+			"kind":       "Pod",
+			"name":       "my-pod",
+			"namespace":  "default",
+			"uid":        "pod-456",
+			"apiVersion": "v1",
+		},
+	}
+
+	involvedObject := ResolveInvolvedObject(event)
+	if involvedObject == nil {
+		t.Fatal("ResolveInvolvedObject() = nil, want the legacy involvedObject map")
+	}
+
+	matched := &MatchedPolicy{
+		PolicyName: "core-pods",
+		Generation: 1,
+		APIGroup:   "",
+		Kind:       "Pod",
+		Summary:    "Pod my-pod was scheduled",
+	}
+
+	// Live path: EventProcessor.buildActivity.
+	p := &EventProcessor{}
+	liveActivity := p.buildActivity(event, matched, involvedObject, matched.Summary, nil)
+
+	// PolicyPreview/reindex path: ActivityBuilder.BuildFromEvent.
+	builder := &ActivityBuilder{APIGroup: matched.APIGroup, Kind: matched.Kind}
+	previewActivity, err := builder.BuildFromEvent(event, matched.Summary, nil, nil)
+	if err != nil {
+		t.Fatalf("BuildFromEvent() error = %v", err)
+	}
+
+	// Confirm the fallbacks actually fired, not just that both sides agree.
+
+	if liveActivity.Spec.Resource.Name != "my-pod" ||
+		liveActivity.Spec.Resource.Namespace != "default" ||
+		liveActivity.Spec.Resource.UID != "pod-456" ||
+		liveActivity.Spec.Resource.APIVersion != "v1" {
+		t.Fatalf("live Resource not populated from legacy involvedObject fallback: %+v", liveActivity.Spec.Resource)
+	}
+
+	if liveActivity.Spec.Actor.Type != ActorTypeController || liveActivity.Spec.Actor.Name != "kubelet" {
+		t.Fatalf("live Actor not resolved via source.component fallback: %+v", liveActivity.Spec.Actor)
+	}
+
+	wantTimestamp := "2024-01-14T09:00:00Z" // lastTimestamp, not metadata.creationTimestamp
+	if got := liveActivity.CreationTimestamp.UTC().Format("2006-01-02T15:04:05Z"); got != wantTimestamp {
+		t.Fatalf("live CreationTimestamp not resolved via lastTimestamp fallback: got %q, want %q", got, wantTimestamp)
+	}
+
+	// The two paths must now agree exactly.
+	if !previewActivity.CreationTimestamp.Time.Equal(liveActivity.CreationTimestamp.Time) {
+		t.Errorf("CreationTimestamp mismatch between buildActivity and BuildFromEvent: live=%v preview=%v",
+			liveActivity.CreationTimestamp.Time, previewActivity.CreationTimestamp.Time)
+	}
+
+	if previewActivity.Name != liveActivity.Name {
+		t.Errorf("Name mismatch between buildActivity and BuildFromEvent: live=%q preview=%q",
+			liveActivity.Name, previewActivity.Name)
+	}
+
+	if previewActivity.Spec.Resource != liveActivity.Spec.Resource {
+		t.Errorf("Resource mismatch between buildActivity and BuildFromEvent:\n live:    %+v\n preview: %+v",
+			liveActivity.Spec.Resource, previewActivity.Spec.Resource)
+	}
+
+	if previewActivity.Spec.Actor != liveActivity.Spec.Actor {
+		t.Errorf("Actor mismatch between buildActivity and BuildFromEvent:\n live:    %+v\n preview: %+v",
+			liveActivity.Spec.Actor, previewActivity.Spec.Actor)
+	}
+
+	if !reflect.DeepEqual(previewActivity.Labels, liveActivity.Labels) {
+		t.Errorf("Labels mismatch between buildActivity and BuildFromEvent:\n live:    %+v\n preview: %+v",
+			liveActivity.Labels, previewActivity.Labels)
+	}
+
+	// BuildFromEvent previously omitted this label entirely.
+	if got := previewActivity.Labels["activity.miloapis.com/event-reason"]; got != "Scheduled" {
+		t.Errorf("preview activity missing event-reason label: got %q, want %q", got, "Scheduled")
+	}
+
+	if liveActivity.Spec.ChangeSource != previewActivity.Spec.ChangeSource {
+		t.Errorf("ChangeSource mismatch: live=%q preview=%q", liveActivity.Spec.ChangeSource, previewActivity.Spec.ChangeSource)
+	}
+}
+
+// TestActivityBuilderBuildFromEventUsesControllerActorType asserts
+// BuildFromEvent uses ActorTypeController, matching the live path (it
+// previously used ActorTypeSystem).
+func TestActivityBuilderBuildFromEventUsesControllerActorType(t *testing.T) {
+	event := map[string]any{
+		"reportingController": "deployment-controller",
+		"regarding": map[string]any{
+			"kind": "Deployment",
+			"name": "my-deployment",
+		},
+	}
+
+	builder := &ActivityBuilder{APIGroup: "apps", Kind: "Deployment"}
+	activity, err := builder.BuildFromEvent(event, "summary", nil, nil)
+	if err != nil {
+		t.Fatalf("BuildFromEvent() error = %v", err)
+	}
+
+	if activity.Spec.Actor.Type != ActorTypeController {
+		t.Errorf("Actor.Type = %q, want %q", activity.Spec.Actor.Type, ActorTypeController)
+	}
+	if activity.Spec.Actor.Name != "deployment-controller" {
+		t.Errorf("Actor.Name = %q, want %q", activity.Spec.Actor.Name, "deployment-controller")
+	}
+}
+
+// TestActivityBuilderBuildFromEventTimestampFallbackChain asserts
+// BuildFromEvent falls back to lastTimestamp (it previously skipped straight
+// to creationTimestamp).
+func TestActivityBuilderBuildFromEventTimestampFallbackChain(t *testing.T) {
+	event := map[string]any{
+		"lastTimestamp": "2024-03-01T12:00:00Z",
+		"regarding": map[string]any{
+			"kind": "Pod",
+			"name": "my-pod",
+		},
+	}
+
+	builder := &ActivityBuilder{APIGroup: "", Kind: "Pod"}
+	activity, err := builder.BuildFromEvent(event, "summary", nil, nil)
+	if err != nil {
+		t.Fatalf("BuildFromEvent() error = %v", err)
+	}
+
+	want := "2024-03-01T12:00:00Z"
+	got := activity.CreationTimestamp.UTC().Format("2006-01-02T15:04:05Z")
+	if got != want {
+		t.Errorf("CreationTimestamp = %q, want %q (should fall back to lastTimestamp)", got, want)
+	}
+}