feat: Mixpanel cohort sync webhook - #8338
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
2 Skipped Deployments
|
|
Important Review skippedNo new commits to review since the last review. ⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughThe change adds Mixpanel cohort synchronisation through a webhook endpoint. It supports Bearer and Basic authentication, including keys supplied as Basic-auth passwords. Cohorts now store optional external identifiers and support the Mixpanel source type. The webhook handles cohort creation, reuse, member additions, removals, validation failures, and structured rejection logging. OpenAPI documents, observability records, fixtures, and unit tests are updated. Estimated code review effort: 4 (Complex) | ~60 minutes Merge Risk: 🟠 High · up to Concurrent initial syncs can create duplicate cohorts and split memberships, while oversized multibyte identifiers can pass validation and fail during processing after a success response. These current-head correctness issues should be fixed before merge; the remaining API documentation gaps are secondary. 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 @@
## feat/cohort-sync-amplitude #8338 +/- ##
============================================================
Coverage 98.79% 98.79%
============================================================
Files 1614 1615 +1
Lines 64934 65147 +213
============================================================
+ Hits 64150 64363 +213
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
|
✅ private-cloud · depot-ubuntu-latest-arm-16 — run #19609 (attempt 1)Playwright Test Results (private-cloud - depot-ubuntu-latest-arm-16)
Details
🗂️ Previous results✅ private-cloud · depot-ubuntu-latest-16 — run #19609 (attempt 1)Playwright Test Results (private-cloud - depot-ubuntu-latest-16)
Details
✅ oss · depot-ubuntu-latest-arm-16 — run #19609 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-arm-16)
Details
✅ oss · depot-ubuntu-latest-16 — run #19609 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-16)
Details
|
Visual Regression19 screenshots compared. See report for details. |
|
@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
| class MixpanelMemberSerializer(serializers.Serializer[None]): | ||
| # Length mirrors CohortMembership.identifier. | ||
| mixpanel_distinct_id = serializers.CharField(max_length=2000) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Reject identifiers that exceed the identity-store byte limit.
Line 65 validates 2000 characters. The downstream identity store permits only 1024 bytes. A multibyte identifier can pass validation, receive a success response, and then fail during identity processing.
Validate the UTF-8 byte length before calling the membership service.
Proposed validation
class MixpanelMemberSerializer(serializers.Serializer[None]):
# Length mirrors CohortMembership.identifier.
mixpanel_distinct_id = serializers.CharField(max_length=2000)
+
+ def validate_mixpanel_distinct_id(self, value: str) -> str:
+ if len(value.encode("utf-8")) > 1024:
+ raise serializers.ValidationError(
+ "Ensure this field is no longer than 1024 bytes."
+ )
+ return value📝 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.
| class MixpanelMemberSerializer(serializers.Serializer[None]): | |
| # Length mirrors CohortMembership.identifier. | |
| mixpanel_distinct_id = serializers.CharField(max_length=2000) | |
| class MixpanelMemberSerializer(serializers.Serializer[None]): | |
| # Length mirrors CohortMembership.identifier. | |
| mixpanel_distinct_id = serializers.CharField(max_length=2000) | |
| def validate_mixpanel_distinct_id(self, value: str) -> str: | |
| if len(value.encode("utf-8")) > 1024: | |
| raise serializers.ValidationError( | |
| "Ensure this field is no longer than 1024 bytes." | |
| ) | |
| return value |
| responses: | ||
| '200': | ||
| description: '' | ||
| content: | ||
| application/json: | ||
| schema: | ||
| $ref: '#/components/schemas/MixpanelWebhookResponse' |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Document the failure response contract.
The view returns structured 400 validation failures and 404 cohort failures. This operation documents only a 200 response. Add 400 and 404 responses with a schema that requires action, status: failure, and error.message and error.code. Regenerate the derived specifications afterwards.
| security: | ||
| - Cohort Sync Key: [] |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- openapi.yaml endpoint and scheme ranges ---'
sed -n '1745,1785p' openapi.yaml
sed -n '29425,29465p' openapi.yaml
printf '%s\n' '--- mcp openapi.json scheme range ---'
sed -n '7738,7775p' mcp/src/flagsmith_mcp/openapi.json
printf '%s\n' '--- sdk openapi.yaml scheme range ---'
sed -n '525,565p' sdk/openapi.yaml
printf '%s\n' '--- all Cohort Sync Key references ---'
rg -n -C 3 'Cohort Sync Key|cohort.?sync|cohort_sync' openapi.yaml mcp/src/flagsmith_mcp/openapi.json sdk/openapi.yaml
printf '%s\n' '--- repository references to the specification files and generation ---'
rg -n -i -C 2 'openapi(\.yaml|\.json)|generate.*(openapi|schema)|openapi.*generat|swagger' --glob '!openapi.yaml' --glob '!mcp/src/flagsmith_mcp/openapi.json' --glob '!sdk/openapi.yaml' .Repository: Flagsmith/flagsmith
Length of output: 47439
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- cohort-sync and Mixpanel source ---'
rg -n -i -C 4 'mixpanel|cohort.?sync|cohort sync key' api --glob '*.py' --glob '!**/migrations/**'
printf '%s\n' '--- authentication and OpenAPI security customisation ---'
rg -n -i -C 4 'authentication_classes|permission_classes|OpenApiAuthenticationExtension|basicAuth|http.*bearer|bearer.*basic|BasicAuthentication|TokenAuthentication' api --glob '*.py' --glob '!**/migrations/**'
printf '%s\n' '--- operation security entries in all specifications ---'
python3 - <<'PY'
import json
from pathlib import Path
def yaml_security(path):
# Extract only nearby operation blocks without parsing YAML dependencies.
text = Path(path).read_text()
lines = text.splitlines()
for i, line in enumerate(lines):
if 'cohort-sync' in line.lower() or 'mixpanel' in line.lower():
start = max(0, i - 2)
end = min(len(lines), i + 80)
block = '\n'.join(lines[start:end])
if 'security:' in block:
print(f'--- {path}:{i+1} ---')
for j in range(start, end):
if 'security:' in lines[j] or 'Cohort Sync Key' in lines[j] or 'basicAuth' in lines[j]:
print(f'{j+1}: {lines[j]}')
def json_security(path):
data = json.loads(Path(path).read_text())
print(f'--- {path} ---')
for route, methods in data.get('paths', {}).items():
if 'cohort-sync' in route.lower() or 'mixpanel' in route.lower():
for method, operation in methods.items():
if isinstance(operation, dict) and 'security' in operation:
print(route, method, operation['security'])
print('security schemes:', data.get('components', {}).get('securitySchemes', {}))
yaml_security('openapi.yaml')
yaml_security('sdk/openapi.yaml')
json_security('mcp/src/flagsmith_mcp/openapi.json')
PYRepository: Flagsmith/flagsmith
Length of output: 50376
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- targeted source files ---'
fd -t f -i 'cohort|sync|openapi|authentication|permission' api | head -200
printf '%s\n' '--- exact authentication and schema matches ---'
rg -n -C 8 'BasicAuthentication|TokenAuthentication|CohortSync|cohort_sync|Cohort Sync Key|OpenApiAuthenticationExtension|authentication_classes' \
api/api api/cohorts api/tests/unit/cohorts --glob '*.py' --glob '!**/migrations/**' \
| head -1000
printf '%s\n' '--- Mixpanel test helpers and authentication assertions ---'
sed -n '1,130p' api/tests/unit/cohorts/test_sync_views.py
sed -n '750,825p' api/tests/unit/cohorts/test_sync_views.py
printf '%s\n' '--- exact security arrays in the JSON specification ---'
python3 - <<'PY'
import json
from pathlib import Path
path = Path('mcp/src/flagsmith_mcp/openapi.json')
data = json.loads(path.read_text())
for route, methods in data.get('paths', {}).items():
if 'cohort-sync' in route.lower():
for method, operation in methods.items():
if isinstance(operation, dict):
print(route, method, 'security=', operation.get('security'))
print('schemes=', data.get('components', {}).get('securitySchemes'))
PY
printf '%s\n' '--- generator filters and related settings ---'
sed -n '100,205p' api/Makefile
rg -n -C 8 'openapi-filter|security|Cohort|basicAuth' api/openapi*.yml api/api/openapi.py api/app/settings --glob '*.yml' --glob '*.py'Repository: Flagsmith/flagsmith
Length of output: 50375
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- cohort authentication ---'
cat -n api/cohorts/authentication.py
printf '%s\n' '--- cohort sync view declarations ---'
rg -n -C 12 'class (AmplitudeCohortSyncViewSet|MixpanelCohortSyncView)|authentication_classes|permission_classes|extend_schema' api/cohorts/sync_views.py
printf '%s\n' '--- OpenAPI authentication extension ---'
sed -n '130,195p' api/api/openapi.py
printf '%s\n' '--- exact security arrays and schemes ---'
python3 - <<'PY'
import json
from pathlib import Path
def yaml_extract(path):
lines = Path(path).read_text().splitlines()
print(f'--- {path} ---')
for i, line in enumerate(lines):
if 'cohort-sync/' in line.lower() or 'mixpanel/webhook' in line.lower():
for j in range(i, min(i + 100, len(lines))):
if 'operationId:' in lines[j] or 'security:' in lines[j] or 'Cohort Sync Key' in lines[j] or 'basicAuth' in lines[j]:
print(f'{j+1}: {lines[j]}')
if j > i and lines[j].startswith(' /'):
break
for i, line in enumerate(lines):
if 'securitySchemes:' in line:
print(f'{i+1}: securitySchemes:')
for j in range(i + 1, min(i + 30, len(lines))):
if lines[j].startswith(' ') and (
'type:' in lines[j] or 'scheme:' in lines[j] or line.strip() == 'securitySchemes:'
):
print(f'{j+1}: {lines[j]}')
def json_extract(path):
data = json.loads(Path(path).read_text())
print(f'--- {path} ---')
for route, methods in data.get('paths', {}).items():
if 'cohort-sync/' in route.lower():
for method, operation in methods.items():
if isinstance(operation, dict):
print(route, method, 'operationId=', operation.get('operationId'), 'security=', operation.get('security'))
print('securitySchemes:')
for name, scheme in data.get('components', {}).get('securitySchemes', {}).items():
if name in {'Cohort Sync Key', 'basicAuth'}:
print(name, scheme)
yaml_extract('openapi.yaml')
yaml_extract('sdk/openapi.yaml')
json_extract('mcp/src/flagsmith_mcp/openapi.json')
PYRepository: Flagsmith/flagsmith
Length of output: 12777
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- Mixpanel operation in derived files ---'
rg -n -C 12 'mixpanel/webhook|api_v1_cohort_sync_mixpanel_webhook_create|Cohort Sync Key' \
mcp/src/flagsmith_mcp/openapi.json sdk/openapi.yaml openapi.yaml | head -240
printf '%s\n' '--- source generation and filtering rules ---'
sed -n '105,120p' api/Makefile
sed -n '175,195p' api/Makefile
cat api/openapi-filter-flagsmith-sdk.yml
cat api/openapi-filter-mcp.yaml
printf '%s\n' '--- all source-level schema customisation hooks ---'
rg -n -C 6 'get_security_requirement|get_security_definition|extend_schema\(|auth=' api --glob '*.py' | head -300Repository: Flagsmith/flagsmith
Length of output: 30738
Document Basic authentication for the Mixpanel webhook.
The operation accepts Basic credentials but advertises only bearer authentication. Add basicAuth as a separate security alternative, update the OpenAPI generation customisation, and regenerate all affected specifications.
📍 Affects 3 files
openapi.yaml#L1767-L1768(this comment)openapi.yaml#L29445-L29448mcp/src/flagsmith_mcp/openapi.json#L7758-L7762sdk/openapi.yaml#L547-L551
1 participant
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.