wip: phased rollout

Author: gagantrivedi

api/audit/constants.py

@@ -77,3 +77,7 @@
     "Feature: %s removed from Release Pipeline: %s"
 )
 FEATURE_STATE_UPDATED_BY_RELEASE_PIPELINE_MESSAGE = "Flag state / Remote config updated for feature: %s by Release pipeline: %s (stage: %s)"
+PHASED_ROLLOUT_STATE_CREATED_MESSAGE = (
+    "Phased rollout created for feature: %s by release pipeline: %s (stage: %s)"
+)
+PHASED_ROLLOUT_STATE_UPDATED_MESSAGE = "Phased rollout split changed from '%s%%' to '%s%%' for feature '%s' by release pipeline '%s' (stage: '%s')"

api/features/release_pipelines/core/migrations/0002_add_phased_rollout.py

@@ -0,0 +1,96 @@
+# Generated by Django 4.2.22 on 2025-08-06 02:31
+
+import django.core.validators
+from django.db import migrations, models
+import django.db.models.deletion
+import django_lifecycle.mixins  # type: ignore[import-untyped]
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("segments", "0029_add_is_system_segment"),
+        ("release_pipelines_core", "0001_add_release_pipelines"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="pipelinestageaction",
+            name="action_type",
+            field=models.CharField(
+                choices=[
+                    ("TOGGLE_FEATURE", "Enable/Disable Feature for the environment"),
+                    (
+                        "UPDATE_FEATURE_VALUE",
+                        "Update Feature Value for the environment",
+                    ),
+                    (
+                        "TOGGLE_FEATURE_FOR_SEGMENT",
+                        "Enable/Disable Feature for a specific segment",
+                    ),
+                    (
+                        "UPDATE_FEATURE_VALUE_FOR_SEGMENT",
+                        "Update Feature Value for a specific segment",
+                    ),
+                    ("PHASED_ROLLOUT", "Create Phased Rollout"),
+                ],
+                default="TOGGLE_FEATURE",
+                max_length=50,
+            ),
+        ),
+        migrations.CreateModel(
+            name="PhasedRolloutState",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "initial_split",
+                    models.FloatField(
+                        validators=[
+                            django.core.validators.MinValueValidator(0.0),
+                            django.core.validators.MaxValueValidator(100.0),
+                        ]
+                    ),
+                ),
+                (
+                    "increase_by",
+                    models.FloatField(
+                        validators=[
+                            django.core.validators.MinValueValidator(0.0),
+                            django.core.validators.MaxValueValidator(100.0),
+                        ]
+                    ),
+                ),
+                ("increase_every", models.DurationField()),
+                (
+                    "current_split",
+                    models.FloatField(
+                        validators=[
+                            django.core.validators.MinValueValidator(0.0),
+                            django.core.validators.MaxValueValidator(100.0),
+                        ]
+                    ),
+                ),
+                ("is_rollout_complete", models.BooleanField(default=False)),
+                ("last_updated_at", models.DateTimeField(auto_now=True)),
+                (
+                    "rollout_segment",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="phased_rollout_state",
+                        to="segments.segment",
+                    ),
+                ),
+            ],
+            bases=(django_lifecycle.mixins.LifecycleModelMixin, models.Model),
+        ),
+    ]

api/features/release_pipelines/core/models.py

@@ -1,8 +1,15 @@
 import typing
+from datetime import datetime
 
-from django.core.validators import MaxValueValidator
+from django.core.validators import MaxValueValidator, MinValueValidator
 from django.db import models
+from django.db.models import Q, QuerySet
 from django.utils import timezone
+from django_lifecycle import (  # type: ignore[import-untyped]
+    BEFORE_CREATE,
+    LifecycleModelMixin,
+    hook,
+)
 
 from audit.constants import (
     RELEASE_PIPELINE_CREATED_MESSAGE,
@@ -39,6 +46,7 @@ class StageActionType(models.TextChoices):
         "UPDATE_FEATURE_VALUE_FOR_SEGMENT",
         "Update Feature Value for a specific segment",
     )
+    PHASED_ROLLOUT = ("PHASED_ROLLOUT", "Create Phased Rollout")
 
 
 class ReleasePipeline(
@@ -72,7 +80,7 @@ def publish(self, published_by: FFAdminUser) -> None:
         self.published_by = published_by
         self.save()
 
-    def unpublish(self, unpublished_by: FFAdminUser) -> None:
+    def unpublish(self) -> None:
         if self.published_at is None:
             raise InvalidPipelineStateError("Pipeline is not published.")
         self.published_at = None
@@ -95,10 +103,25 @@ def get_delete_log_message(
     ) -> typing.Optional[str]:
         return RELEASE_PIPELINE_DELETED_MESSAGE % self.name
 
+    def get_feature_versions_in_pipeline_qs(
+        self,
+    ) -> QuerySet[EnvironmentFeatureVersion]:
+        base_qs = EnvironmentFeatureVersion.objects.filter(
+            pipeline_stage__pipeline=self
+        )
+        phased_rollout_action_filter = Q(phased_rollout_state__isnull=False) & Q(
+            phased_rollout_state__is_rollout_complete=False
+        )
+        all_other_action_filters = Q(published_at__isnull=True)
+        qs: QuerySet[EnvironmentFeatureVersion] = base_qs.filter(
+            all_other_action_filters | phased_rollout_action_filter
+        )
+        return qs
+
     def has_feature_in_flight(self) -> bool:
-        has_feature_in_flight: bool = EnvironmentFeatureVersion.objects.filter(
-            published_at__isnull=True, pipeline_stage__in=self.stages.all()
-        ).exists()
+        has_feature_in_flight: bool = (
+            self.get_feature_versions_in_pipeline_qs().exists()
+        )
         return has_feature_in_flight
 
     def _get_project(self) -> Project:
@@ -134,6 +157,37 @@ def get_next_stage(self) -> "PipelineStage | None":
             .first()
         )
 
+    def get_phased_rollout_action(self) -> "PipelineStageAction | None":
+        return self.actions.filter(action_type=StageActionType.PHASED_ROLLOUT).first()
+
+    def get_in_stage_feature_versions_qs(self) -> QuerySet[EnvironmentFeatureVersion]:
+        phased_rollout_action_filter = Q(
+            phased_rollout_state__isnull=False,
+            phased_rollout_state__is_rollout_complete=False,
+        )
+        all_other_action_filters = Q(
+            published_at__isnull=True, phased_rollout_state__isnull=True
+        )
+
+        return self.environment_feature_versions.filter(
+            all_other_action_filters | phased_rollout_action_filter
+        )
+
+    def get_completed_feature_versions_qs(
+        self, completed_after: datetime = timezone.now()
+    ) -> QuerySet[EnvironmentFeatureVersion]:
+        phased_rollout_action_filter = Q(
+            phased_rollout_state__is_rollout_complete=True,
+            phased_rollout_state__last_updated_at__gte=completed_after,
+        )
+        all_other_action_filters = Q(
+            published_at__gte=completed_after, phased_rollout_state__isnull=True
+        )
+
+        return self.environment_feature_versions.filter(
+            all_other_action_filters | phased_rollout_action_filter
+        )
+
 
 class PipelineStageTrigger(models.Model):
     trigger_type = models.CharField(
@@ -162,3 +216,56 @@ class PipelineStageAction(models.Model):
         related_name="actions",
         on_delete=models.CASCADE,
     )
+
+
+class PhasedRolloutState(LifecycleModelMixin, models.Model):  # type: ignore[misc]
+    initial_split = models.FloatField(
+        validators=[
+            MinValueValidator(0.0),
+            MaxValueValidator(100.0),
+        ]
+    )
+    increase_by = models.FloatField(
+        validators=[
+            MinValueValidator(0.0),
+            MaxValueValidator(100.0),
+        ]
+    )
+    increase_every = models.DurationField()
+    current_split = models.FloatField(
+        validators=[
+            MinValueValidator(0.0),
+            MaxValueValidator(100.0),
+        ]
+    )
+    rollout_segment = models.ForeignKey(
+        "segments.Segment",
+        related_name="phased_rollout_state",
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+    )
+    is_rollout_complete = models.BooleanField(default=False)
+    last_updated_at = models.DateTimeField(auto_now=True)
+
+    @hook(BEFORE_CREATE)  # type: ignore[misc]
+    def set_initial_split(self) -> None:
+        if self.current_split is None:
+            self.current_split = self.initial_split
+
+    def increase_split(self) -> float:
+        self.current_split = min(self.current_split + self.increase_by, 100.0)
+        self.save()
+
+        # Update the segment value
+        condition = self.rollout_segment.rules.first().conditions.first()  # type: ignore[union-attr]
+        assert condition
+        condition.value = str(self.current_split)
+        condition.save()
+        return self.current_split
+
+    def complete_rollout(self) -> None:
+        assert self.rollout_segment is not None
+        self.rollout_segment.delete()
+        self.is_rollout_complete = True
+        self.save()

api/features/versioning/migrations/0007_add_phased_rollout.py

@@ -0,0 +1,38 @@
+# Generated by Django 4.2.22 on 2025-08-06 02:31
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("release_pipelines_core", "0002_add_phased_rollout"),
+        ("feature_versioning", "0006_add_pipeline_stage_to_envfeatureversion"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="environmentfeatureversion",
+            name="phased_rollout_state",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name="environment_feature_versions",
+                to="release_pipelines_core.phasedrolloutstate",
+            ),
+        ),
+        migrations.AddField(
+            model_name="historicalenvironmentfeatureversion",
+            name="phased_rollout_state",
+            field=models.ForeignKey(
+                blank=True,
+                db_constraint=False,
+                null=True,
+                on_delete=django.db.models.deletion.DO_NOTHING,
+                related_name="+",
+                to="release_pipelines_core.phasedrolloutstate",
+            ),
+        ),
+    ]

api/features/versioning/models.py

@@ -84,14 +84,21 @@ class EnvironmentFeatureVersion(  # type: ignore[django-manager-missing]
         null=True,
         blank=True,
     )
+
     pipeline_stage = models.ForeignKey(
         "release_pipelines_core.PipelineStage",
         related_name="environment_feature_versions",
         on_delete=models.CASCADE,
         null=True,
         blank=True,
     )
-
+    phased_rollout_state = models.ForeignKey(
+        "release_pipelines_core.PhasedRolloutState",
+        related_name="environment_feature_versions",
+        on_delete=models.CASCADE,
+        null=True,
+        blank=True,
+    )
     objects = EnvironmentFeatureVersionManager()  # type: ignore[misc]
 
     class Meta:

api/segments/migrations/0029_add_is_system_segment.py

@@ -0,0 +1,23 @@
+# Generated by Django 4.2.22 on 2025-08-06 03:18
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("segments", "0028_condition_property_required"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="historicalsegment",
+            name="is_system_segment",
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AddField(
+            model_name="segment",
+            name="is_system_segment",
+            field=models.BooleanField(default=False),
+        ),
+    ]

api/segments/models.py

@@ -79,6 +79,7 @@ class Segment(
 
     created_at = models.DateTimeField(null=True, auto_now_add=True)
     updated_at = models.DateTimeField(null=True, auto_now=True)
+    is_system_segment = models.BooleanField(default=False)
 
     objects = SegmentManager()  # type: ignore[misc]
 
@@ -92,6 +93,8 @@ def __str__(self):  # type: ignore[no-untyped-def]
         return "Segment - %s" % self.name
 
     def get_skip_create_audit_log(self) -> bool:
+        if self.is_system_segment:
+            return True
         try:
             if self.version_of_id and self.version_of_id != self.id:
                 return True
@@ -201,6 +204,8 @@ def get_skip_create_audit_log(self) -> bool:
             segment = self.get_segment()  # type: ignore[no-untyped-call]
             if segment.deleted_at:
                 return True
+            if segment.is_system_segment:
+                return True
             return segment.version_of_id != segment.id  # type: ignore[no-any-return]
         except (Segment.DoesNotExist, SegmentRule.DoesNotExist):
             # handle hard delete
@@ -346,6 +351,8 @@ def get_skip_create_audit_log(self) -> bool:
             segment = self.rule.get_segment()  # type: ignore[no-untyped-call]
             if segment.deleted_at:
                 return True
+            if segment.is_system_segment:
+                return True
 
             return segment.version_of_id != segment.id  # type: ignore[no-any-return]
         except (Segment.DoesNotExist, SegmentRule.DoesNotExist):

api/segments/views.py

@@ -51,7 +51,7 @@ def get_queryset(self):  # type: ignore[no-untyped-def]
         )
         project = get_object_or_404(permitted_projects, pk=self.kwargs["project_pk"])
 
-        queryset = Segment.live_objects.filter(project=project)
+        queryset = Segment.live_objects.filter(project=project, is_system_segment=False)
 
         if self.action == "list":
             # TODO: at the moment, the UI only shows the name and description of the segment in the list view.