feat: Add Segment events for CheckoutIntent state transitions

Track state changes in CheckoutIntent lifecycle by emitting
appropriate Segment analytics events on transitions

Author: zwidekalanga

enterprise_access/apps/customer_billing/constants.py

@@ -73,6 +73,16 @@ class CheckoutIntentState(StrEnum):
     ERRORED_PROVISIONING = 'errored_provisioning'
     EXPIRED = 'expired'
 
+class CheckoutIntentSegmentEvents:
+    """
+    Segment events for CheckoutIntent lifecycle tracking.
+    """
+    CREATED = 'edx.server.enterprise-access.checkout-intent-lifecycle.created'
+    TRANSITION_TO_PAID = 'edx.server.enterprise-access.checkout-intent-lifecycle.transition-to-paid'
+    TRANSITION_TO_FULFILLED = 'edx.server.enterprise-access.checkout-intent-lifecycle.transition-to-fulfilled'
+    TRANSITION_TO_ERRORED_STRIPE_CHECKOUT = 'edx.server.enterprise-access.checkout-intent-lifecycle.transition-to-errored-stripe-checkout'
+    TRANSITION_TO_ERRORED_PROVISIONING = 'edx.server.enterprise-access.checkout-intent-lifecycle.transition-to-errored-provisioning'
+
 
 ALLOWED_CHECKOUT_INTENT_STATE_TRANSITIONS = {
     CheckoutIntentState.CREATED: [

enterprise_access/apps/customer_billing/models.py

@@ -185,23 +185,35 @@ def mark_as_paid(self, stripe_session_id=None):
             if self.state == CheckoutIntentState.PAID and stripe_session_id != self.stripe_checkout_session_id:
                 raise ValueError("Cannot transition from PAID to PAID with a different stripe_checkout_session_id")
 
+        previous_state = self.state
         self.state = CheckoutIntentState.PAID
         if stripe_session_id:
             self.stripe_checkout_session_id = stripe_session_id
         self.save(update_fields=['state', 'stripe_checkout_session_id', 'modified'])
         logger.info(f'CheckoutIntent {self} marked as {CheckoutIntentState.PAID}.')
+
+        # Track state transition
+        from .segment_events import track_checkout_intent_event
+        track_checkout_intent_event(self, previous_state, self.state)
+
         return self
 
     def mark_as_fulfilled(self, workflow=None):
         """Mark the intent as fulfilled after successful provisioning."""
         if not self.is_valid_state_transition(CheckoutIntentState(self.state), CheckoutIntentState.FULFILLED):
             raise ValueError(f"Cannot transition from {self.state} to {CheckoutIntentState.FULFILLED}.")
 
+        previous_state = self.state
         self.state = CheckoutIntentState.FULFILLED
         if workflow:
             self.workflow = workflow
         self.save(update_fields=['state', 'workflow', 'modified'])
         logger.info(f'CheckoutIntent {self} marked as {CheckoutIntentState.FULFILLED}.')
+
+        # Track state transition
+        from .segment_events import track_checkout_intent_event
+        track_checkout_intent_event(self, previous_state, self.state)
+
         return self
 
     def mark_checkout_error(self, error_message):
@@ -212,10 +224,16 @@ def mark_checkout_error(self, error_message):
         ):
             raise ValueError(f"Cannot transition from {self.state} to {CheckoutIntentState.ERRORED_STRIPE_CHECKOUT}.")
 
+        previous_state = self.state
         self.state = CheckoutIntentState.ERRORED_STRIPE_CHECKOUT
         self.last_checkout_error = error_message
         self.save(update_fields=['state', 'last_checkout_error', 'modified'])
         logger.info(f'CheckoutIntent {self} marked as {CheckoutIntentState.ERRORED_STRIPE_CHECKOUT}.')
+
+        # Track state transition
+        from .segment_events import track_checkout_intent_event
+        track_checkout_intent_event(self, previous_state, self.state)
+
         return self
 
     def mark_provisioning_error(self, error_message, workflow=None):
@@ -226,12 +244,18 @@ def mark_provisioning_error(self, error_message, workflow=None):
         ):
             raise ValueError(f"Cannot transition from {self.state} to {CheckoutIntentState.ERRORED_PROVISIONING}.")
 
+        previous_state = self.state
         self.state = CheckoutIntentState.ERRORED_PROVISIONING
         self.last_provisioning_error = error_message
         if workflow:
             self.workflow = workflow
         self.save(update_fields=['state', 'last_provisioning_error', 'workflow', 'modified'])
         logger.info(f'CheckoutIntent {self} marked as {CheckoutIntentState.ERRORED_PROVISIONING}.')
+
+        # Track state transition
+        from .segment_events import track_checkout_intent_event
+        track_checkout_intent_event(self, previous_state, self.state)
+
         return self
 
     @property
@@ -429,9 +453,14 @@ def create_intent(
                 existing_intent.country = country
                 existing_intent.terms_metadata = terms_metadata
                 existing_intent.save()
+
+                # Track creation event for updated intent
+                from .segment_events import track_checkout_intent_event
+                track_checkout_intent_event(existing_intent)
+
                 return existing_intent
 
-            return cls.objects.create(
+            intent = cls.objects.create(
                 user=user,
                 state=CheckoutIntentState.CREATED,
                 enterprise_slug=slug,
@@ -442,6 +471,12 @@ def create_intent(
                 terms_metadata=terms_metadata,
             )
 
+            # Track creation event
+            from .segment_events import track_checkout_intent_event
+            track_checkout_intent_event(intent)
+
+            return intent
+
     @classmethod
     def for_user(cls, user):
         """

enterprise_access/apps/customer_billing/segment_events.py

@@ -0,0 +1,37 @@
+"""
+Segment event tracking for CheckoutIntent lifecycle events.
+"""
+
+from enterprise_access.apps.api.serializers import CheckoutIntentReadOnlySerializer
+from enterprise_access.apps.customer_billing.constants import (
+    CheckoutIntentSegmentEvents,
+)
+from enterprise_access.apps.track.segment import track_event
+
+
+def track_checkout_intent_event(checkout_intent, previous_state=None, new_state=None):
+    """Track CheckoutIntent lifecycle events (creation and state transitions)."""
+    properties = CheckoutIntentReadOnlySerializer(checkout_intent).data
+    properties["previous_state"] = previous_state
+    properties["new_state"] = new_state or checkout_intent.state
+
+    # Get event name based on new state
+    event_name = _get_event_name_for_state(new_state or checkout_intent.state)
+
+    track_event(
+        lms_user_id=str(checkout_intent.user.id),
+        event_name=event_name,
+        properties=properties,
+    )
+
+
+def _get_event_name_for_state(new_state):
+    """Map state to event name."""
+    mapping = {
+        "created": CheckoutIntentSegmentEvents.CREATED,
+        "paid": CheckoutIntentSegmentEvents.TRANSITION_TO_PAID,
+        "fulfilled": CheckoutIntentSegmentEvents.TRANSITION_TO_FULFILLED,
+        "errored_stripe_checkout": CheckoutIntentSegmentEvents.TRANSITION_TO_ERRORED_STRIPE_CHECKOUT,
+        "errored_provisioning": CheckoutIntentSegmentEvents.TRANSITION_TO_ERRORED_PROVISIONING,
+    }
+    return mapping.get(new_state)