- ID
- 3ccf7052-1a35-49ed-af73-66bca853aa20
BE-7152 Investigate incorrect data types in Segement events (signed in and reactivation)
- source :: https://hungryroot.atlassian.net/browse/BE-7152
- tags :: Hungryroot API Project
- epic :: EP-411 Eng Parking Lot
- sprint :: HR BE Sprint 104
- status :: Status - Complete
Tasks
DONE Investigate Segment inconsistencies in Datadog
- ID
- a9251158-46c8-03eb-51b0-3d13e00b40d1
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)
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,
)
@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