feat: Mixpanel cohort sync webhook - #8338
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub. 3 Skipped Deployments
|
📝 WalkthroughWalkthroughThe change adds Mixpanel cohort synchronisation through a webhook endpoint. It supports Bearer and Basic authentication. Cohorts store optional external identifiers and support the Mixpanel source type. The webhook handles cohort creation, reuse, member additions, removals, validation failures, and rejection logging. API documentation, observability records, fixtures, and tests are updated. Estimated code review effort: 4 (Complex) | ~60 minutes Merge Risk: 🟠 High · up to The webhook introduces Mixpanel-driven cohort creation and membership updates. At the current head, unresolved authorization, concurrency/data-consistency, task-ordering, and schema-generation issues could permit unauthorized synchronization or leave cohort state divergent, so merge should be blocked until these are fixed or explicitly accepted. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
github-actions
Bot
added
api
8ccf977 to
753f5db
Compare
github-actions
Bot
added
feature
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #8338 +/- ##
========================================
Coverage 98.79% 98.80%
========================================
Files 1616 1617 +1
Lines 65283 65509 +226
========================================
+ Hits 64499 64725 +226
Misses 784 784 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
github-actions
Bot
added
feature
github-actions
Bot
added
feature
Docker builds report
|
✅ oss · depot-ubuntu-latest-16 — run #19727 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-16)
Details
🗂️ Previous results✅ oss · depot-ubuntu-latest-arm-16 — run #19727 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-arm-16)
Details
✅ oss · depot-ubuntu-latest-arm-16 — run #19726 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-arm-16)
Details
✅ oss · depot-ubuntu-latest-16 — run #19726 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-16)
Details
✅ private-cloud · depot-ubuntu-latest-arm-16 — run #19721 (attempt 1)Playwright Test Results (private-cloud - depot-ubuntu-latest-arm-16)
Details
✅ private-cloud · depot-ubuntu-latest-16 — run #19721 (attempt 1)Playwright Test Results (private-cloud - depot-ubuntu-latest-16)
Details
✅ oss · depot-ubuntu-latest-arm-16 — run #19721 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-arm-16)
Details
✅ oss · depot-ubuntu-latest-16 — run #19721 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-16)
Details
✅ oss · depot-ubuntu-latest-arm-16 — run #19722 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-arm-16)
Details
✅ oss · depot-ubuntu-latest-16 — run #19722 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-16)
Details
✅ private-cloud · depot-ubuntu-latest-16 — run #19648 (attempt 1)Playwright Test Results (private-cloud - depot-ubuntu-latest-16)
Details
✅ oss · depot-ubuntu-latest-arm-16 — run #19649 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-arm-16)
Details
|
Visual RegressionNo baseline found — first run. Baselines will be generated after merge to main. |
|
@coderabbitai review |
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: f6c11a14-bb0e-47d0-b8b7-a2a4f9a2781e
📒 Files selected for processing (15)
api/api/openapi.pyapi/cohorts/authentication.pyapi/cohorts/migrations/0004_mixpanel_source.pyapi/cohorts/models.pyapi/cohorts/serializers.pyapi/cohorts/services.pyapi/cohorts/sync_urls.pyapi/cohorts/sync_views.pyapi/tests/unit/cohorts/conftest.pyapi/tests/unit/cohorts/test_services.pyapi/tests/unit/cohorts/test_sync_views.pydocs/docs/deployment-self-hosting/observability/_events-catalogue.mdmcp/src/flagsmith_mcp/openapi.jsonopenapi.yamlsdk/openapi.yaml
Included review availability: Your plan provides up to 8 included reviews per hour; 7 remain after this review.
| # 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) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Make active cohort lookup and creation atomic.
Concurrent initial members requests can both observe no cohort and create separate active cohorts for the same environment, source type, and external ID. Later incremental requests may select only one duplicate, splitting memberships between cohorts.
Add a partial unique constraint for active cohorts with non-null external IDs, and use an atomic get-or-create flow that re-fetches after an integrity conflict.
📍 Affects 2 files
api/cohorts/models.py#L32-L36(this comment)api/cohorts/sync_views.py#L140-L153
There was a problem hiding this comment.
Deliberately dropped to keep v1 lean — we'll add the constraint and conflict recovery if duplicate creation is observed in practice.
There was a problem hiding this comment.
@gagantrivedi, understood. The concurrency risk remains deferred from this PR.
Would you like me to create a follow-up GitHub issue for the active-cohort uniqueness constraint and atomic conflict recovery?
🐇
✏️ Learnings added
Learnt from: gagantrivedi
Repo: Flagsmith/flagsmith PR: 8338
File: api/cohorts/models.py:32-36
Timestamp: 2026-08-24T04:25:26.515Z
Learning: For the Mixpanel cohort synchronisation v1, the maintainers deliberately defer the active-cohort uniqueness constraint and atomic conflict recovery for concurrent initial `members` requests. They will add this protection if duplicate cohort creation is observed in practice.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
You are interacting with an AI system.
github-actions
Bot
added
feature
aab49fc to
a34a1ce
Compare
github-actions
Bot
added
feature
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
api/cohorts/services.py (1)
217-229: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winDispatch membership tasks after the transaction commits.
add_cohort_members()andremove_cohort_members()callapply_cohort_membership_deltas.delay()insidetransaction.atomic(). With a separate task database, the task can start before the membership transaction commits, find no pending memberships, and exit without retrying. The caller can succeed while memberships remain pending.Wrap both dispatches in
transaction.on_commit().Source: Learnings
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 94132278-03ff-4a4a-8107-9df5889f5e10
📒 Files selected for processing (11)
api/api/openapi.pyapi/cohorts/serializers.pyapi/cohorts/services.pyapi/cohorts/sync_views.pyapi/tests/unit/cohorts/conftest.pyapi/tests/unit/cohorts/test_services.pyapi/tests/unit/cohorts/test_sync_views.pydocs/docs/deployment-self-hosting/observability/_events-catalogue.mdmcp/src/flagsmith_mcp/openapi.jsonopenapi.yamlsdk/openapi.yaml
Included review availability: Your plan provides up to 8 included reviews per hour; 7 remain after this review.
| 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] |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🔎 Supported by static analysis
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
files=(
api/api/openapi.py
openapi.yaml
mcp/src/flagsmith_mcp/openapi.json
sdk/openapi.yaml
)
for file in "${files[@]}"; do
test -f "$file"
done
if rg -n -P 'name\s*=\s*\[.*Cohort Sync Key|Cohort Sync Key(?: \(Basic\))?\s*:' "${files[@]}"; then
echo "Found non-conforming cohort security-scheme names." >&2
exit 1
fiRepository: Flagsmith/flagsmith
Length of output: 885
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
printf '%s\n' '--- api/api/openapi.py ---'
sed -n '145,205p' api/api/openapi.py
printf '%s\n' '--- generated security schemes ---'
sed -n '555,575p' sdk/openapi.yaml
sed -n '29645,29672p' openapi.yaml
printf '%s\n' '--- generated security requirements ---'
sed -n '1675,1692p' openapi.yamlRepository: Flagsmith/flagsmith
Length of output: 4676
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
printf '%s\n' '--- drf-spectacular version pins ---'
rg -n -m 20 'drf-spectacular|drf_spectacular' \
pyproject.toml poetry.lock requirements*.txt setup.cfg setup.py 2>/dev/null || true
printf '%s\n' '--- extension references ---'
rg -n -m 20 'CohortSyncKeyAuthenticationExtension|get_security_requirement|get_security_definition' api/api/openapi.pyRepository: Flagsmith/flagsmith
Length of output: 485
🌐 Web query:
drf-spectacular 0.28.0 OpenApiAuthenticationExtension name list get_security_definition implementation OpenAPI component key pattern 3.0.3
💡 Result:
In drf-spectacular 0.28.0, the OpenApiAuthenticationExtension allows you to define authentication schemes that are registered in the OpenAPI components section [1][2]. Implementation for Single vs. Multiple Security Definitions: To implement a single security definition, set the name attribute to a string and return a dictionary from get_security_definition [1][3]. For scenarios requiring multiple parameters or a group of headers (e.g., to model security requirements that include multiple headers), you should define name as a list of strings and return a corresponding list of dictionary definitions from get_security_definition [1][2][4]. Example Implementation: class MyCustomAuthExtension(OpenApiAuthenticationExtension): target_class = 'path.to.YourAuthenticationClass' name = ['HeaderOne', 'HeaderTwo'] def get_security_definition(self, auto_schema): return [ {'type': 'apiKey', 'in': 'header', 'name': 'X-Header-One'}, {'type': 'apiKey', 'in': 'header', 'name': 'X-Header-Two'} ] OpenAPI Component Key Pattern: When drf-spectacular processes these extensions, it registers them under the securitySchemes component in the OpenAPI specification [5][6]. Under OpenAPI 3.0.3, these components are structured under the components/securitySchemes key [6]. The library ensures that each name provided in your extension is registered as a unique component key within the resulting OpenAPI schema [5]. If you provide a list of names, the library iterates through them and maps each to the corresponding definition returned by get_security_definition, ensuring they are correctly placed in the schema components [5]. Key Technical Details: - The get_security_definition method is expected to return a dictionary (or a list of dictionaries) compliant with the OpenAPI 3.0.3 Security Scheme Object specification [1][3]. - Internally, drf-spectacular uses the ResolvedComponent class to handle these, where the security component type is defined as securitySchemes [6]. - The library automatically warns if it detects component naming collisions [6]. For most standard implementations, refer to the official documentation on workflow and customization [3][7].
Citations:
- 1: https://drf-spectacular.readthedocs.io/en/stable/drf_spectacular.html
- 2: https://drf-spectacular.readthedocs.io/en/latest/drf_spectacular.html
- 3: https://drf-spectacular.readthedocs.io/en/latest/customization.html
- 4: Example on how to add multiple security headers using the following method tfranzel/drf-spectacular#1173
- 5: https://github.com/tfranzel/drf-spectacular/blob/master/drf_spectacular/openapi.py
- 6: https://github.com/tfranzel/drf-spectacular/blob/master/drf_spectacular/plumbing.py
- 7: https://github.com/tfranzel/drf-spectacular/blob/master/docs/customization.rst
🌐 Web query:
OpenAPI 3.0.3 specification components securitySchemes map keys pattern ^[a-zA-Z0-9.-_]+$
💡 Result:
In the OpenAPI 3.0.3 specification, the keys used within the components/securitySchemes object must conform to the regular expression ^[a-zA-Z0-9.-_]+$ [1][2][3]. This requirement applies generally to all fixed fields declared within the components object (such as schemas, responses, parameters, examples, requestBodies, headers, links, and callbacks) [2][3]. The specification explicitly mandates that the names used as keys in these maps must match this pattern [1][4]. While these names are arbitrary and used to refer to the security schemes from other parts of the API definition (such as the global or operation-level security requirements), they must adhere to the specified character constraints to be compliant with the OpenAPI 3.0.3 standard [5]. Note that while the textual specification imposes this constraint, it has been noted in community discussions that the corresponding JSON schema files for version 3.0 historically lacked the automated validation (such as additionalProperties: false) to strictly enforce this pattern, which was considered an oversight in the 3.0 schema design [6].
Citations:
- 1: https://github.com/oai/openapi-specification/blob/master/versions/3.0.3.md
- 2: https://spec.openapis.org/oas/v3.0.3
- 3: https://spec.openapis.org/oas/v3.0.3.html
- 4: https://github.com/OAI/OpenAPI-Specification/blob/aa91a19c43f8a12c02efa42d64794e396473f3b1/versions/3.0.3.md
- 5: https://swagger.io/docs/specification/v3_0/authentication/
- 6: Why does the JSON schema for 3.0 specify patternProperties for component names but allows additional properties? OAI/OpenAPI-Specification#3720
Use OpenAPI-safe names for the security schemes.
CohortSyncKeyAuthenticationExtension.name generates the components.securitySchemes keys. The current names contain spaces and parentheses, which violate the OpenAPI 3.0.3 component-key pattern. The invalid names are also present in openapi.yaml and sdk/openapi.yaml. Rename both names and regenerate these documents.
Source: MCP tools
| authentication_classes = [CohortSyncKeyAuthentication] | ||
| permission_classes = [HasCohortSyncKey] |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Enforce the cohort-sync plan gate for Mixpanel.
MixpanelCohortSyncView currently accepts any valid sync key, while the Amplitude endpoint also requires CohortSyncPlanPermission. Add the same permission to the Mixpanel view, cover the rejected-plan path, and regenerate the API descriptions so the endpoint advertises the START_UP minimum-plan requirement in all generated specifications.
📍 Affects 2 files
api/cohorts/sync_views.py#L124-L125(this comment)openapi.yaml#L1750-L1789
There was a problem hiding this comment.
Agreed. Also as in amplitude we should keep them in sync with the max_segments_allowed limit
| cohort = services.get_cohort_for_source( | ||
| environment=environment, | ||
| source_type=CohortSourceType.MIXPANEL, | ||
| external_id=parameters["mixpanel_cohort_id"], | ||
| ) or services.create_cohort_for_source( | ||
| environment=environment, | ||
| name=parameters["mixpanel_cohort_name"], | ||
| source_type=CohortSourceType.MIXPANEL, | ||
| external_id=parameters["mixpanel_cohort_id"], | ||
| ) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🔎 Supported by static analysis
🏁 Script executed:
#!/bin/bash
set -e
printf '%s\n' '--- api/cohorts/sync_views.py ---'
sed -n '120,190p' api/cohorts/sync_views.py
printf '%s\n' '--- api/cohorts/models.py ---'
rg -n -C 8 'class Cohort|source_type|external_id|Unique|unique' api/cohorts/models.py
printf '%s\n' '--- api/cohorts/migrations/0004_mixpanel_source.py ---'
cat -n api/cohorts/migrations/0004_mixpanel_source.py
printf '%s\n' '--- cohort creation and lookup definitions/usages ---'
rg -n -C 5 'get_cohort_for_source|create_cohort_for_source' api/cohortsRepository: Flagsmith/flagsmith
Length of output: 12594
🏁 Script executed:
#!/bin/bash
set -e
printf '%s\n' '--- api/cohorts/services.py ---'
sed -n '120,215p' api/cohorts/services.py
printf '%s\n' '--- SoftDeleteExportableModel definition ---'
rg -n -C 12 'class SoftDeleteExportableModel|deleted_at' core api | head -160
printf '%s\n' '--- cohort migrations and migration state ---'
rg -n -C 4 'unique_active_cohort_per_segment|UniqueConstraint|deleted_at|external_id' api/cohorts/migrationsRepository: Flagsmith/flagsmith
Length of output: 20758
🏁 Script executed:
#!/bin/bash
set -e
printf '%s\n' '--- cohort deletion lifecycle ---'
rg -n -C 6 'deletion_requested_at|deleted_at' api/cohorts api/core core 2>/dev/null | head -240Repository: Flagsmith/flagsmith
Length of output: 11489
Make initial cohort creation idempotent under concurrent requests
api/cohorts/migrations/0004_mixpanel_source.py adds external_id without a uniqueness constraint. Concurrent members requests can both miss get_cohort_for_source() and create separate cohorts. Later delta requests use .first(), so updates can target only one duplicate cohort.
Add a uniqueness constraint for active Mixpanel cohorts and reload the existing cohort after a creation conflict. Use deletion_requested_at__isnull=True for consistency with get_cohort_for_source(); deleted_at__isnull=True would still conflict with cohorts awaiting deletion.
github-actions
Bot
added
feature
There was a problem hiding this comment.
Actionable comments posted: 1
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: f3174a6d-ff87-4af4-8e84-088565d372fa
📒 Files selected for processing (3)
api/cohorts/serializers.pyapi/tests/unit/cohorts/test_sync_views.pyopenapi.yaml
💤 Files with no reviewable changes (1)
- openapi.yaml
Included review availability: Your plan provides up to 8 included reviews per hour; 6 remain after this review.
| body = response.json() | ||
| assert body["status"] == "failure" | ||
| assert "1024 bytes" in body["error"]["message"] | ||
| assert not CohortMembership.objects.exists() |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Scope the membership assertion to the target cohort.
CohortMembership.objects.exists() queries the entire table. If another fixture creates a membership, this test fails even when the request correctly creates no membership for mixpanel_cohort. Filter by cohort=mixpanel_cohort or compare that cohort’s membership count before and after the request.
Proposed fix
- assert not CohortMembership.objects.exists()
+ assert not CohortMembership.objects.filter(cohort=mixpanel_cohort).exists()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| assert not CohortMembership.objects.exists() | |
| assert not CohortMembership.objects.filter(cohort=mixpanel_cohort).exists() |
Zaimwa9
left a comment
Zaimwa9
left a comment
There was a problem hiding this comment.
Looks good, just needs to mirror the plan gating and max allowed segment.
Also shared my opinion on adding some robustness regarding the decision to drop the unique constraint
| authentication_classes = [CohortSyncKeyAuthentication] | ||
| permission_classes = [HasCohortSyncKey] |
There was a problem hiding this comment.
Agreed. Also as in amplitude we should keep them in sync with the max_segments_allowed limit
| return self._failure( | ||
| request, | ||
| message=( | ||
| f"Invalid payload: {json.dumps(serializer.errors, default=str)}" |
There was a problem hiding this comment.
Should we be worried about the size of the dumps in case of completely wrong payload ? It would be for 1k member at most right ?
| if webhook_action == "members": | ||
| # A large first sync arrives as several requests, each one page | ||
| # of members. Every page only adds; removals can't be detected | ||
| # without seeing all pages at once. |
There was a problem hiding this comment.
Hey, to improve the deletion lifecycle related to the dropped unique constraint discussion raised by coderabbit, I wonder if we could add a check here that raises a Cohort is being deleted and return a 404 if the cohort is processing the deletions.
My understanding is that it would pause the sync and send an email. So it would remind the user to completely shutdown his mixpanel sync or resume it intentionally.
Sounds like a cheap trade-off
2 participants
docs/if required so people know about the feature.Changes
Contributes to https://github.com/Flagsmith/flagsmith-private/issues/261
Adds the receiving end of Mixpanel's Custom Webhook cohort sync, on top of the Amplitude branch (#8290):
members,add_membersandremove_membersactions, answering in the{action, status}envelope Mixpanel expects.membersfirst sync creates the cohort on the fly, keyed by a newCohort.external_id; snapshot pages are treated as adds only.sync_webhook.rejectedevent.How did you test this code?
Unit tests for the webhook actions, cohort auto-creation (including the simultaneous-creation and deleted-cohort cases), Basic credential handling, and cross-environment isolation; 100% coverage on the cohorts app. The real integration test happens against Mixpanel once the endpoint is deployed.