Org Web Adapter

Editing hungryroot/jira/be_7152_investigate_incorrect_data_types_in_segement_events_signed_in_and_reactivation.org

:PROPERTIES:
:ID:       3ccf7052-1a35-49ed-af73-66bca853aa20
:END:
#+title: BE-7152 Investigate incorrect data types in Segement events (signed in and reactivation)

- source :: https://hungryroot.atlassian.net/browse/BE-7152
- tags   :: [[id:3a477c02-e26d-40f7-a719-0aa1cd169dd8][Hungryroot API Project]] 
- epic   :: [[id:86327fb4-0f66-4bc6-9068-c414cea77042][EP-411 Eng Parking Lot]]
- sprint :: [[id:30370b03-0b10-4105-87da-641cdda1bcb7][HR BE Sprint 104]]
- status :: [[id:4d41d978-e56f-45c1-88d1-e2c76510972e][Status - Complete]]

* Tasks
** DONE Investigate Segment inconsistencies in Datadog :work:project:spike:sprint104:ep411:be7152:hungryroot:segment:
:PROPERTIES:
:ID:       a9251158-46c8-03eb-51b0-3d13e00b40d1
:END:
<2026-02-18 Wed>

* Description
https://hungryroot.slack.com/archives/C03MJA5K0QJ/p1771260417989769

We have identified inconsistencies in backend Segment event payloads where
certain fields are being sent with incorrect data types.

Issue Summary

“Signed In” Event

~35% of events are passing user_id as a string

Expected type: integer

This inconsistency may impact downstream analytics, schema enforcement, and
tracking plan validation.

“Reactivation” Event

~40% of events are passing campaign details as a string

Expected type: object

This likely breaks tracking plan expectations and could affect attribution or
reporting logic.

Scope of Investigation

Identify the source(s) of the incorrect payload formatting

Are multiple code paths emitting the same event?

Is there conditional logic that changes serialization?

Confirm expected schema definitions in Segment tracking plan

Determine whether:

The issue is caused by improper casting

Data is being serialized before being attached to the event

A legacy path is still in use

Assess downstream impact (analytics, dashboards, warehouse ingestion, etc.)

Propose a fix and outline safe rollout strategy

Acceptance Criteria

Root cause identified for both event inconsistencies

Clear explanation of why incorrect data types are being emitted

Recommended fix documented

Plan to normalize data types going forward

If needed: backfill or cleanup strategy proposed

Yes, I can send examples! The below screen in Segment, under "Protocols", has a
way to see the exact JSON of the violating events

* Test scripts
#+begin_src python
from types import SimpleNamespace
from unittest.mock import patch

from django.contrib.sessions.middleware import SessionMiddleware
from django.test import RequestFactory

from app.models import Customer
from app.services import async_helpers

def run_reactivation_segment_smoke_test(
    user_id: int,
    ,*,
    use_request_user: bool = False,
    utm_campaign: str = "ACQ_S946_7D1V-noEV",
    utm_source: str = "facebook-sub",
    utm_medium: str = "paid-social_Instagram_Stories_ig",
    utm_term: str = "6858413077315",
    utm_content: str = "FUN32871 - i-1x1-ClickFriendly_Biohack-v1",
):
    """
    Smoke-test enqueue_segment_precheckout_trk for Reactivation in shell.

Args:
        user_id: customer.user_id (NOT Customer.pk)
        use_request_user: if True, set request.user=customer.user (realistic path).
                          if False, use kwargs fallback path with SimpleNamespace().
        utm_*: session campaign fields used to build nested campaign object.
    """
    customer = Customer.objects.get(user_id=user_id)

factory = RequestFactory()
    request = factory.post(
        "/api/customer/reactivate",
        HTTP_USER_AGENT="qa-shell-test-agent",
        HTTP_X_FORWARDED_FOR="10.20.30.40",
    )

# attach session
    SessionMiddleware(lambda req: None).process_request(request)
    request.session.save()

# campaign/session fields
    request.session["utm_campaign"] = utm_campaign
    request.session["utm_source"] = utm_source
    request.session["utm_medium"] = utm_medium
    request.session["utm_term"] = utm_term
    request.session["utm_content"] = utm_content

# choose user path
    request.user = customer.user if use_request_user else SimpleNamespace()

with patch("app.services.async_helpers.async_segment_track.delay") as mock_delay:
        async_helpers.enqueue_segment_precheckout_trk(
            request,
            "Reactivation",
            user_id=customer.user_id,
        )

customer_id_arg, event_name_arg, properties_arg = mock_delay.call_args.args

result = {
        "customer_id_arg": customer_id_arg,
        "customer_id_type": type(customer_id_arg).__name__,
        "event_name_arg": event_name_arg,
        "campaign": properties_arg.get("campaign"),
        "has_flat_campaign_keys": any(
            key in properties_arg
            for key in ("campaign_source", "campaign_medium", "campaign_term", "campaign_content")
        ),
        "all_property_keys": sorted(properties_arg.keys()),
    }

async_helpers.enqueue_segment_precheckout_trk(
        request,
        "Reactivation",
        user_id=customer.user_id,
    )
    print(result)
    return result

def run_signed_in_apple_segment_smoke_test(
    user_id: int,
    ,*,
    use_request_user: bool = False,
    utm_campaign: str = "ACQ_S946_7D1V-noEV",
    utm_source: str = "facebook-sub",
    utm_medium: str = "paid-social_Instagram_Stories_ig",
    utm_term: str = "6858413077315",
    utm_content: str = "FUN32871 - i-1x1-ClickFriendly_Biohack-v1",
):
    """
    Smoke-test Signed In (apple) segment payload in Django shell.

Args:
        user_id: customer.user_id (NOT Customer.pk)
        use_request_user: if True, set request.user=customer.user; else use kwargs fallback path.
        utm_*: optional session values to inspect nested campaign payload.
    """
    customer = Customer.objects.get(user_id=user_id)

factory = RequestFactory()
    request = factory.post(
        "/api/apple_login",
        HTTP_USER_AGENT="qa-shell-test-agent",
        HTTP_X_FORWARDED_FOR="10.20.30.40",
    )

SessionMiddleware(lambda req: None).process_request(request)
    request.session.save()

request.session["utm_campaign"] = utm_campaign
    request.session["utm_source"] = utm_source
    request.session["utm_medium"] = utm_medium
    request.session["utm_term"] = utm_term
    request.session["utm_content"] = utm_content

request.user = customer.user if use_request_user else SimpleNamespace()

with patch("app.services.async_helpers.async_segment_track.delay") as mock_delay:
        async_helpers.enqueue_segment_precheckout_trk(
            request,
            "Signed In",
            method="apple",
            user_id=customer.user_id,
        )
        customer_id_arg, event_name_arg, properties_arg = mock_delay.call_args.args

result = {
        "customer_id_arg": customer_id_arg,
        "customer_id_type": type(customer_id_arg).__name__,
        "event_name_arg": event_name_arg,
        "method": properties_arg.get("method"),
        "campaign": properties_arg.get("campaign"),
        "has_flat_campaign_keys": any(
            key in properties_arg
            for key in ("campaign_source", "campaign_medium", "campaign_term", "campaign_content")
        ),
        "user_id_is_int": isinstance(customer_id_arg, int),
        "all_property_keys": sorted(properties_arg.keys()),
    }

async_helpers.enqueue_segment_precheckout_trk(
        request,
        "Signed In",
        method="apple",
        user_id=customer.user_id,
    )

print(result)
    return result
#+end_src

#+begin_src python
from argparse import ArgumentParser
import json
import random
from types import SimpleNamespace
from typing import Any

from django.contrib.sessions.middleware import SessionMiddleware
from django.core.management.base import BaseCommand
from django.core.management.base import CommandError
from django.test import RequestFactory

from app.models import Customer
from app.services import async_helpers
from app.services import segment_tracking

class Command(BaseCommand):
    help = "Smoke-test Segment precheckout payloads for Signed In and Reactivation events."

def add_arguments(self, parser: ArgumentParser) -> None:
        parser.add_argument(
            "user_id",
            nargs="?",
            type=int,
            default=None,
            help="Customer user_id (not customer pk). If omitted, command selects one at random.",
        )
        parser.add_argument(
            "--with-utm-campaign",
            action="store_true",
            default=False,
            help="Populate request.session utm_* values for campaign object testing",
        )
        parser.add_argument(
            "--dry-run",
            action="store_true",
            default=False,
            help="Do not enqueue Segment events; only print payloads",
        )

def handle(self, *args: Any, **options: Any) -> None:
        user_id = options["user_id"] or self._pick_random_user_id()
        with_utm_campaign = options["with_utm_campaign"]
        dry_run = options["dry_run"]

customer = Customer.objects.filter(user_id=user_id).first()
        if customer is None:
            raise CommandError(f"Could not find Customer with user_id={user_id}")

self.stdout.write(f"Using user_id={user_id} dry_run={dry_run}")
        signed_in_result = self._run_signed_in_apple_test(
            customer,
            with_utm_campaign=with_utm_campaign,
            dry_run=dry_run,
        )
        reactivation_result = self._run_reactivation_test(
            customer,
            with_utm_campaign=with_utm_campaign,
            dry_run=dry_run,
        )

self.stdout.write(self.style.SUCCESS("Segment smoke test completed"))
        self.stdout.write("Signed In payload:")
        self.stdout.write(json.dumps(signed_in_result, indent=2, sort_keys=True, default=str))
        self.stdout.write("Reactivation payload:")
        self.stdout.write(json.dumps(reactivation_result, indent=2, sort_keys=True, default=str))

def _make_request(self, path: str, with_utm_campaign: bool) -> Any:
        request = RequestFactory().post(
            path,
            HTTP_USER_AGENT="segment-smoke-test",
            HTTP_X_FORWARDED_FOR="10.20.30.40",
        )

SessionMiddleware(lambda req: None).process_request(request)
        request.session.save()

if with_utm_campaign:
            request.session["utm_campaign"] = "ACQ_S946_7D1V-noEV"
            request.session["utm_content"] = "FUN32871 - i-1x1-ClickFriendly_Biohack-v1"
            request.session["utm_medium"] = "paid-social_Instagram_Stories_ig"
            request.session["utm_source"] = "facebook-sub"
            request.session["utm_term"] = "6858413077315"

# Exercise the kwargs user_id fallback path in enqueue_segment_precheckout_trk.
        request.user = SimpleNamespace()
        return request

def _run_signed_in_apple_test(
        self, customer: Customer, *, with_utm_campaign: bool, dry_run: bool
    ) -> dict[str, Any]:
        request = self._make_request("/api/apple_login", with_utm_campaign=with_utm_campaign)
        event_kwargs = {
            "method": "apple",
            "user_id": customer.user_id,
        }
        customer_id_arg, event_name_arg, properties_arg = self._build_event_payload(
            request=request,
            event_name="Signed In",
            event_kwargs=event_kwargs,
        )
        if not dry_run:
            async_helpers.enqueue_segment_precheckout_trk(request, "Signed In", **event_kwargs)

return self._build_result(
            customer_id_arg=customer_id_arg,
            event_name_arg=event_name_arg,
            properties_arg=properties_arg,
            sent_to_segment=not dry_run,
        )

def _run_reactivation_test(self, customer: Customer, *, with_utm_campaign: bool, dry_run: bool) -> dict[str, Any]:
        request = self._make_request("/api/customer/reactivate", with_utm_campaign=with_utm_campaign)
        event_kwargs = {
            "user_id": customer.user_id,
        }
        customer_id_arg, event_name_arg, properties_arg = self._build_event_payload(
            request=request,
            event_name="Reactivation",
            event_kwargs=event_kwargs,
        )
        if not dry_run:
            async_helpers.enqueue_segment_precheckout_trk(request, "Reactivation", **event_kwargs)

@staticmethod
    def _build_result(
        customer_id_arg: Any,
        event_name_arg: Any,
        properties_arg: dict[str, Any],
        sent_to_segment: bool,
    ) -> dict[str, Any]:
        return {
            "event_name": event_name_arg,
            "customer_id": customer_id_arg,
            "customer_id_type": type(customer_id_arg).__name__,
            "user_id_is_int": isinstance(customer_id_arg, int),
            "sent_to_segment": sent_to_segment,
            "method": properties_arg.get("method"),
            "campaign": properties_arg.get("campaign"),
            "has_flat_campaign_keys": any(
                key in properties_arg
                for key in ("campaign_source", "campaign_medium", "campaign_term", "campaign_content")
            ),
            "all_property_keys": sorted(properties_arg.keys()),
        }

@staticmethod
    def _build_event_payload(
        request: Any,
        event_name: str,
        event_kwargs: dict[str, Any],
    ) -> tuple[int, str, dict[str, Any]]:
        kwargs = dict(event_kwargs)
        user_id = kwargs.get("user_id")
        if user_id is None:
            raise CommandError("user_id is required in event_kwargs for this smoke test command")

base_data = segment_tracking.make_segment_track_base_obj(request)
        data = {**base_data, **kwargs}
        return user_id, event_name, data

@staticmethod
    def _pick_random_user_id() -> int:
        qs = Customer.objects.exclude(user_id__isnull=True).values_list("user_id", flat=True)
        count = qs.count()
        if count == 0:
            raise CommandError("Could not find any Customer rows with a non-null user_id")
        return qs[random.randint(0, count - 1)]

#+end_src