refactor: Update event and order processing logic to improve subscription handling and logging

Author: dpiyumal2319

internal/kafka/event_consumer.go

@@ -79,19 +79,41 @@ func (c *EventConsumer) processEventEvent(value []byte) error {
 
 	// Handle different operations
 	switch rawEvent.Payload.Op {
-	case "c": // Event creation - notify organization subscribers
-		if err := c.SubscriberService.ProcessEventCreation(&eventEvent); err != nil {
-			log.Printf("Error processing event creation notification from Debezium: %v", err)
-			return err
+	case "c": // Skip event creation notification
+		log.Printf("Skipping notification for event creation: %s", eventID)
+
+	case "u": // Event update - notify only if before=PENDING and after=APPROVED
+		// Check status transition for updates
+		if rawEvent.Payload.Before != nil && rawEvent.Payload.After != nil {
+			beforeStatus := rawEvent.Payload.Before.Status
+			afterStatus := rawEvent.Payload.After.Status
+
+			if beforeStatus == "PENDING" && afterStatus == "APPROVED" {
+				// This is a status change from PENDING to APPROVED - treat as creation
+				if err := c.SubscriberService.ProcessEventCreation(&eventEvent); err != nil {
+					log.Printf("Error processing event approval notification from Debezium: %v", err)
+					return err
+				}
+				log.Printf("Successfully processed event approval (PENDING->APPROVED) notification for event %s", eventID)
+			} else if afterStatus == "APPROVED" {
+				// Other changes but final status is still APPROVED - process as update
+				if err := c.SubscriberService.ProcessEventUpdate(&eventEvent); err != nil {
+					log.Printf("Error processing event update notification from Debezium: %v", err)
+					return err
+				}
+				log.Printf("Successfully processed event update notification for approved event %s", eventID)
+			} else {
+				// Status is not APPROVED - skip notification
+				log.Printf("Skipping notification for event %s - status is %s (not APPROVED)", eventID, afterStatus)
+			}
 		}
-		log.Printf("Successfully processed event creation notification for event %s", eventID)
 
-	case "u", "d": // Event update/delete - notify event subscribers
+	case "d": // Event deletion - process normally for subscribers
 		if err := c.SubscriberService.ProcessEventUpdate(&eventEvent); err != nil {
-			log.Printf("Error processing event update notification from Debezium: %v", err)
+			log.Printf("Error processing event deletion notification from Debezium: %v", err)
 			return err
 		}
-		log.Printf("Successfully processed event update notification for event %s", eventID)
+		log.Printf("Successfully processed event deletion notification for event %s", eventID)
 
 	default:
 		log.Printf("Unhandled operation '%s' for event %s", rawEvent.Payload.Op, eventID)

internal/kafka/order_consumer.go

@@ -150,22 +150,24 @@ func (c *OrderConsumer) processOrderUpdated(value []byte) error {
 	// For orders changing to 'completed' status, add subscriptions
 	if order.Status == "completed" {
 		// Add subscription to the event and session
+		log.Printf("Order %s is now completed - adding subscriptions", order.OrderID)
+
+		log.Printf("Adding event subscription for event %s", order.EventID)
 		if err := c.SubscriberService.AddSubscription(subscriber.SubscriberID, models.SubscriptionCategoryEvent, order.EventID); err != nil {
 			log.Printf("Error adding event subscription: %v", err)
 		}
 
+		log.Printf("Adding session subscription for session %s", order.SessionID)
 		if err := c.SubscriberService.AddSubscription(subscriber.SubscriberID, models.SubscriptionCategorySession, order.SessionID); err != nil {
 			log.Printf("Error adding session subscription: %v", err)
 		}
 
-		if order.OrganizationID != "" {
-			if err := c.SubscriberService.AddSubscription(subscriber.SubscriberID, models.SubscriptionCategoryOrganization, order.OrganizationID); err != nil {
-				log.Printf("Error adding organization subscription: %v", err)
-			}
+		log.Printf("Adding organization subscription for organization %s", order.OrganizationID)
+		if err := c.SubscriberService.AddSubscription(subscriber.SubscriberID, models.SubscriptionCategoryOrganization, order.OrganizationID); err != nil {
+			log.Printf("Error adding organization subscription: %v", err)
 		}
-
-		log.Printf("Added subscriptions for completed order %s", order.OrderID)
 	}
+	log.Printf("Added subscriptions for completed order %s", order.OrderID)
 
 	// Send appropriate order email based on status
 	if err := c.SubscriberService.SendOrderConfirmationEmail(subscriber, &order); err != nil {