feat(scheduled flag changes): implement scheduling of changes

posthog/celery.py

@@ -271,6 +271,13 @@ def setup_periodic_tasks(sender: Celery, **kwargs):
         name="recalculate cohorts",
     )
 
+    add_periodic_task_with_expiry(
+        sender,
+        120,
+        process_scheduled_changes.s(),
+        name="process scheduled changes",
+    )
+
     if clear_clickhouse_crontab := get_crontab(settings.CLEAR_CLICKHOUSE_REMOVED_DATA_SCHEDULE_CRON):
         sender.add_periodic_task(
             clear_clickhouse_crontab,
@@ -871,6 +878,13 @@ def calculate_cohort():
     calculate_cohorts()
 
 
+@app.task(ignore_result=True)
+def process_scheduled_changes():
+    from posthog.tasks.process_scheduled_changes import process_scheduled_changes
+
+    process_scheduled_changes()
+
+
 @app.task(ignore_result=True)
 def sync_insight_cache_states_task():
     from posthog.caching.insight_caching_state import sync_insight_cache_states

posthog/models/feature_flag/feature_flag.py

@@ -297,6 +297,17 @@ def get_cohort_ids(
 
         return list(cohort_ids)
 
+    def scheduled_changes_dispatcher(self, payload):
+        if "field" not in payload or "value" not in payload:
+            raise Exception("Invalid payload")
+        elif payload["field"] == "active":
+            self.active = payload["value"]
+        elif payload["field"] == "filters":
+            for group in payload["value"]["groups"]:
+                self.filters["groups"].append(group)
+
+        self.save()
+
     @property
     def uses_cohorts(self) -> bool:
         for condition in self.conditions:

posthog/tasks/__init__.py

@@ -7,6 +7,7 @@
     demo_create_data,
     email,
     exporter,
+    process_scheduled_changes,
     split_person,
     sync_all_organization_available_features,
     usage_report,
@@ -20,6 +21,7 @@
     "demo_create_data",
     "email",
     "exporter",
+    "process_scheduled_changes",
     "split_person",
     "sync_all_organization_available_features",
     "user_identify",

posthog/tasks/process_scheduled_changes.py

@@ -0,0 +1,35 @@
+from posthog.models import ScheduledChange
+from django.utils import timezone
+from posthog.models import FeatureFlag
+from django.db import transaction, OperationalError
+
+models = {"FeatureFlag": FeatureFlag}
+
+
+def process_scheduled_changes() -> None:
+    try:
+        with transaction.atomic():
+            scheduled_changes = ScheduledChange.objects.select_for_update(nowait=True).filter(
+                executed_at__isnull=True,
+                scheduled_at__lt=timezone.now(),
+            )
+
+            for scheduled_change in scheduled_changes.iterator(chunk_size=10000):
+                try:
+                    # Execute the change on the model instance
+                    model = models[scheduled_change.model_name]
+                    instance = model.objects.get(id=scheduled_change.record_id)
+                    instance.scheduled_changes_dispatcher(scheduled_change.payload)
+
+                    # Mark scheduled change completed
+                    scheduled_change.executed_at = timezone.now()
+                    scheduled_change.save()
+
+                except Exception as e:
+                    # Store the failure reason
+                    scheduled_change.failure_reason = str(e)
+                    scheduled_change.executed_at = timezone.now()
+                    scheduled_change.save()
+    except OperationalError:
+        # Failed to obtain the lock
+        pass

posthog/tasks/test/test_process_scheduled_changes.py

@@ -0,0 +1,91 @@
+from datetime import datetime, timedelta, timezone
+from posthog.models import ScheduledChange, FeatureFlag
+from posthog.test.base import APIBaseTest
+from posthog.tasks.process_scheduled_changes import process_scheduled_changes
+
+
+class TestProcessScheduledChanges(APIBaseTest):
+    def test_schedule_feature_flag_set_active(self):
+        feature_flag = FeatureFlag.objects.create(
+            name="Flag 1",
+            key="flag-1",
+            active=False,
+            filters={"groups": [{"properties": [], "rollout_percentage": None}]},
+            team=self.team,
+            created_by=self.user,
+        )
+
+        ScheduledChange.objects.create(
+            team=self.team,
+            record_id=feature_flag.id,
+            model_name="FeatureFlag",
+            payload={"field": "active", "value": True},
+            scheduled_at=(datetime.now(timezone.utc) - timedelta(seconds=30)).isoformat(),
+        )
+
+        process_scheduled_changes()
+
+        updated_flag = FeatureFlag.objects.get(key="flag-1")
+        self.assertEqual(updated_flag.active, True)
+
+    def test_schedule_feature_flag_add_release_condition(self):
+        feature_flag = FeatureFlag.objects.create(
+            name="Flag 1",
+            key="flag-1",
+            active=False,
+            filters={"groups": []},
+            team=self.team,
+            created_by=self.user,
+        )
+
+        new_release_condition = {
+            "variant": None,
+            "properties": [{"key": "$browser", "type": "person", "value": ["Chrome"], "operator": "exact"}],
+            "rollout_percentage": 30,
+        }
+
+        payload = {
+            "field": "filters",
+            "value": {"groups": [new_release_condition], "payloads": {}, "multivariate": None},
+        }
+
+        ScheduledChange.objects.create(
+            team=self.team,
+            record_id=feature_flag.id,
+            model_name="FeatureFlag",
+            payload=payload,
+            scheduled_at=(datetime.now(timezone.utc) - timedelta(seconds=30)).isoformat(),
+        )
+
+        process_scheduled_changes()
+
+        updated_flag = FeatureFlag.objects.get(key="flag-1")
+        self.assertEqual(updated_flag.filters["groups"][0], new_release_condition)
+
+    def test_schedule_feature_flag_invalid_payload(self):
+        feature_flag = FeatureFlag.objects.create(
+            name="Flag 1",
+            key="flag-1",
+            active=False,
+            filters={"groups": []},
+            team=self.team,
+            created_by=self.user,
+        )
+
+        payload = {"foo": "bar"}
+
+        scheduled_change = ScheduledChange.objects.create(
+            team=self.team,
+            record_id=feature_flag.id,
+            model_name="FeatureFlag",
+            payload=payload,
+            scheduled_at=(datetime.now(timezone.utc) - timedelta(seconds=30)).isoformat(),
+        )
+
+        process_scheduled_changes()
+
+        updated_flag = FeatureFlag.objects.get(key="flag-1")
+        self.assertEqual(updated_flag.filters["groups"], [])
+
+        updated_scheduled_change = ScheduledChange.objects.get(id=scheduled_change.id)
+        self.assertEqual(updated_scheduled_change.failure_reason, "Invalid payload")