feat: Mixpanel cohort sync webhook (#8338)

Co-authored-by: flagsmith-engineering[bot] <flagsmith-engineering[bot]@users.noreply.github.com>

Author: gagantrivedi

api/api/openapi.py

@@ -165,19 +165,38 @@ def get_security_definition(
 
 class CohortSyncKeyAuthenticationExtension(OpenApiAuthenticationExtension):  # type: ignore[no-untyped-call]
     target_class = "cohorts.authentication.CohortSyncKeyAuthentication"
-    name = "Cohort Sync Key"
+    name = ["Cohort Sync Key", "Cohort Sync Key (Basic)"]
+
+    def get_security_requirement(
+        self, auto_schema: openapi.AutoSchema
+    ) -> list[dict[str, list[Any]]]:
+        # Separate entries: the caller sends the key with either scheme,
+        # not both at once.
+        return [{name: []} for name in self.name]
 
     def get_security_definition(
         self, auto_schema: openapi.AutoSchema | None = None
-    ) -> dict[str, Any]:
-        return {
-            "type": "http",
-            "scheme": "bearer",
-            "description": (
-                "For cohort sync endpoints called by an external cohort "
-                "source, such as Amplitude."
-            ),
-        }
+    ) -> list[dict[str, Any]]:
+        return [
+            {
+                "type": "http",
+                "scheme": "bearer",
+                "description": (
+                    "For cohort sync endpoints called by an external cohort "
+                    "source, such as Amplitude."
+                ),
+            },
+            {
+                "type": "http",
+                "scheme": "basic",
+                "description": (
+                    "For cohort sync endpoints called by an external cohort "
+                    "source that can only send Basic credentials, such as "
+                    "Mixpanel. The key is the password; the username is "
+                    "ignored."
+                ),
+            },
+        ]
 
 
 # Tag definitions controlling the order and display of sections in the Swagger UI.

api/cohorts/authentication.py

@@ -1,4 +1,4 @@
-import typing
+import base64
 from contextlib import suppress
 
 from django.contrib.auth.models import AnonymousUser
@@ -9,18 +9,36 @@
 
 
 class CohortSyncKeyAuthentication(authentication.BaseAuthentication):
+    """
+    Accepts a cohort sync key sent either as a Bearer token or as the
+    password of Basic credentials. Amplitude sends Bearer; Mixpanel's
+    webhook setup only offers a username/password form, so its customers
+    enter any username and the key as the password. The username is
+    ignored.
+    """
+
     def authenticate(
         self, request: Request
     ) -> tuple[AnonymousUser, CohortSyncKey] | None:
         header = request.headers.get("Authorization", "")
-        if not header.startswith("Bearer "):
+        if header.startswith("Bearer "):
+            raw_key = header.removeprefix("Bearer ")
+        elif header.startswith("Basic "):
+            try:
+                decoded = base64.b64decode(
+                    header.removeprefix("Basic "), validate=True
+                ).decode()
+            except ValueError:
+                # Covers malformed base64, header bytes outside ASCII, and
+                # decoded credentials that are not valid UTF-8.
+                raise exceptions.AuthenticationFailed("Invalid Basic credentials.")
+            # Split at the first colon, so a key containing colons survives.
+            _, _, raw_key = decoded.partition(":")
+        else:
             return None
 
         with suppress(CohortSyncKey.DoesNotExist):
-            key = typing.cast(
-                CohortSyncKey,
-                CohortSyncKey.objects.get_from_key(header.removeprefix("Bearer ")),
-            )
+            key = CohortSyncKey.objects.get_from_key(raw_key)
             if not key.has_expired:
                 # No person is acting here, so no user is returned: the key
                 # alone carries authority, and audit trails record the source

api/cohorts/migrations/0004_mixpanel_source.py

@@ -0,0 +1,33 @@
+# Generated by Django 5.2.16 on 2026-08-20 10:06
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cohorts", "0003_cohort_sync_key"),
+        ("environments", "0039_use_no_ssrf_url_field"),
+        ("segments", "0032_add_segment_rules_data"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="cohort",
+            name="external_id",
+            field=models.CharField(blank=True, max_length=255, null=True),
+        ),
+        migrations.AlterField(
+            model_name="cohort",
+            name="source_type",
+            field=models.CharField(
+                choices=[
+                    ("csv", "CSV"),
+                    ("amplitude", "Amplitude"),
+                    ("mixpanel", "Mixpanel"),
+                ],
+                default="csv",
+                max_length=50,
+            ),
+        ),
+    ]

api/cohorts/models.py

@@ -1,5 +1,7 @@
+import typing
+
 from django.db import models
-from rest_framework_api_key.models import AbstractAPIKey
+from rest_framework_api_key.models import AbstractAPIKey, APIKeyManager
 
 from cohorts.constants import COHORT_SYSTEM_TRAIT_KEY_PREFIX
 from core.models import SoftDeleteExportableModel
@@ -8,6 +10,7 @@
 class CohortSourceType(models.TextChoices):
     CSV = "csv", "CSV"
     AMPLITUDE = "amplitude", "Amplitude"
+    MIXPANEL = "mixpanel", "Mixpanel"
 
 
 class Cohort(SoftDeleteExportableModel):
@@ -26,6 +29,11 @@ class Cohort(SoftDeleteExportableModel):
         choices=CohortSourceType.choices,
         default=CohortSourceType.CSV,
     )
+    # The cohort's identifier in the external source. Mixpanel pushes under
+    # its own cohort ID, so we store it to route later requests; Amplitude
+    # uses the ID we hand back at list creation, and CSV cohorts have no
+    # external system, so both leave this null.
+    external_id = models.CharField(max_length=255, null=True, blank=True)
     version = models.PositiveIntegerField(default=0)
     created_at = models.DateTimeField(auto_now_add=True)
     # Deletion drains memberships from the identity store first; the cohort is
@@ -48,7 +56,19 @@ class Meta:
         ]
 
 
+class CohortSyncKeyManager(APIKeyManager):
+    def get_from_key(self, key: str) -> "CohortSyncKey":
+        if "\x00" in key:
+            # A NUL can't travel in a raw header, but base64 credentials can
+            # decode to one, and the database driver refuses to build a query
+            # containing it. No real key holds one, so treat it as absent.
+            raise self.model.DoesNotExist("Key contains a NUL character.")
+        return typing.cast("CohortSyncKey", super().get_from_key(key))
+
+
 class CohortSyncKey(AbstractAPIKey):
+    objects: typing.ClassVar[CohortSyncKeyManager] = CohortSyncKeyManager()
+
     environment = models.ForeignKey(
         "environments.Environment",
         on_delete=models.CASCADE,

api/cohorts/serializers.py

@@ -98,6 +98,33 @@ def _validate_identifier_byte_length(value: str) -> None:
         )
 
 
+class MixpanelMemberSerializer(serializers.Serializer[None]):
+    mixpanel_distinct_id = serializers.CharField(
+        validators=[_validate_identifier_byte_length]
+    )
+
+
+class MixpanelParametersSerializer(serializers.Serializer[None]):
+    mixpanel_cohort_id = serializers.CharField(max_length=255)
+    mixpanel_cohort_name = serializers.CharField(max_length=2000)
+    # An empty page is valid: a first sync of an empty cohort has no members.
+    # Mixpanel sends at most 1000 members per message; the cap stops anything
+    # else from posting an arbitrarily large page.
+    # The stubs don't know many=True forwards max_length to the list serialiser.
+    members = MixpanelMemberSerializer(  # type: ignore[call-arg]
+        many=True, allow_empty=True, max_length=1000
+    )
+
+
+class MixpanelWebhookSerializer(serializers.Serializer[None]):
+    # "members" carries the full membership on the first sync;
+    # "add_members"/"remove_members" carry changes since the last sync.
+    action = serializers.ChoiceField(
+        choices=["members", "add_members", "remove_members"]
+    )
+    parameters = MixpanelParametersSerializer()
+
+
 class CohortSyncMembersSerializer(serializers.Serializer[None]):
     user_ids = serializers.ListField(
         child=serializers.CharField(validators=[_validate_identifier_byte_length]),

api/cohorts/services.py

@@ -116,6 +116,7 @@ def create_cohort(
     name: str,
     description: str | None = None,
     source_type: CohortSourceType = CohortSourceType.CSV,
+    external_id: str | None = None,
 ) -> Cohort:
     project = environment.project
     # Mirrors the segment limit enforced by SegmentSerializer, which cohort
@@ -137,7 +138,10 @@ def create_cohort(
         )
         rule = SegmentRule.objects.create(segment=segment, type=SegmentRule.ALL_RULE)
         cohort: Cohort = Cohort.objects.create(
-            environment=environment, segment=segment, source_type=source_type
+            environment=environment,
+            segment=segment,
+            source_type=source_type,
+            external_id=external_id,
         )
         Condition.objects.create(
             rule=rule,
@@ -161,10 +165,16 @@ def create_cohort_for_source(
     environment: "Environment",
     name: str,
     source_type: CohortSourceType,
+    external_id: str | None = None,
 ) -> Cohort:
     """Create a cohort on behalf of an external source, where no Flagsmith
     user is acting."""
-    cohort = create_cohort(environment=environment, name=name, source_type=source_type)
+    cohort = create_cohort(
+        environment=environment,
+        name=name,
+        source_type=source_type,
+        external_id=external_id,
+    )
     # Nothing records a user for these calls, so the audit log that Flagsmith
     # derives from historical records is skipped — and with it the environment
     # document rebuild that makes the new segment visible to SDKs. Write the
@@ -182,6 +192,37 @@ def create_cohort_for_source(
     return cohort
 
 
+def get_cohort_for_source(
+    *,
+    environment: "Environment",
+    source_type: CohortSourceType,
+    external_id: str,
+) -> Cohort | None:
+    cohort: Cohort | None = Cohort.objects.filter(
+        environment=environment,
+        source_type=source_type,
+        external_id=external_id,
+        deletion_requested_at__isnull=True,
+    ).first()
+    return cohort
+
+
+def cohort_deletion_in_progress(
+    *,
+    environment: "Environment",
+    source_type: CohortSourceType,
+    external_id: str,
+) -> bool:
+    return bool(
+        Cohort.objects.filter(
+            environment=environment,
+            source_type=source_type,
+            external_id=external_id,
+            deletion_requested_at__isnull=False,
+        ).exists()
+    )
+
+
 def add_cohort_members(cohort: Cohort, identifiers: "typing.Iterable[str]") -> None:
     from cohorts.tasks import apply_cohort_membership_deltas
 

api/cohorts/sync_urls.py

@@ -1,11 +1,15 @@
+from django.urls import path
 from rest_framework.routers import SimpleRouter
 
-from cohorts.sync_views import AmplitudeCohortSyncViewSet
+from cohorts.sync_views import AmplitudeCohortSyncViewSet, MixpanelCohortSyncView
 
 app_name = "cohort-sync"
 
 # SimpleRouter: nothing here is browsed by a person.
 router = SimpleRouter()
 router.register(r"amplitude/lists", AmplitudeCohortSyncViewSet, basename="amplitude")
 
-urlpatterns = router.urls
+urlpatterns = [
+    path("mixpanel/webhook/", MixpanelCohortSyncView.as_view(), name="mixpanel"),
+    *router.urls,
+]