feat(experimentation): provision per-org ingestion infra on external warehouse connect

api/experimentation/ingestion_infra_service.py

@@ -206,3 +206,30 @@ def provision_ingestion_infrastructure(
         organisation_id=organisation_id,
     )
     return IngestionInfrastructure(bucket_name=bucket_name, stream_name=stream_name)
+
+
+def _delete_events_bucket(bucket_name: str) -> None:
+    s3 = _get_s3_client()
+    paginator = s3.get_paginator("list_objects_v2")
+    for page in paginator.paginate(Bucket=bucket_name):
+        objects = [{"Key": obj["Key"]} for obj in page.get("Contents", [])]
+        if objects:
+            s3.delete_objects(Bucket=bucket_name, Delete={"Objects": objects})
+    s3.delete_bucket(Bucket=bucket_name)
+
+
+def deprovision_ingestion_infrastructure(organisation_id: int) -> None:
+    bucket_name = get_bucket_name(organisation_id)
+    stream_name = get_stream_name(organisation_id)
+
+    _get_firehose_client().delete_delivery_stream(
+        DeliveryStreamName=stream_name,
+        AllowForceDelete=True,
+    )
+    _delete_events_bucket(bucket_name)
+    logger.info(
+        "ingestion_infra.deprovisioned",
+        organisation__id=organisation_id,
+        bucket__name=bucket_name,
+        stream__name=stream_name,
+    )

api/experimentation/migrations/0011_organisation_ingestion_infrastructure.py

@@ -0,0 +1,59 @@
+# Generated by Django 5.2.16 on 2026-07-21 07:36
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("experimentation", "0010_warehouse_connection_credentials_and_status_detail"),
+        ("organisations", "0058_update_audit_and_history_limits_in_sub_cache"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="OrganisationIngestionInfrastructure",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "status",
+                    models.CharField(
+                        choices=[
+                            ("pending", "Pending"),
+                            ("created", "Created"),
+                            ("errored", "Errored"),
+                        ],
+                        default="pending",
+                        max_length=50,
+                    ),
+                ),
+                (
+                    "bucket_name",
+                    models.CharField(blank=True, max_length=255, null=True),
+                ),
+                (
+                    "stream_name",
+                    models.CharField(blank=True, max_length=255, null=True),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                (
+                    "organisation",
+                    models.OneToOneField(
+                        on_delete=django.db.models.deletion.DO_NOTHING,
+                        related_name="ingestion_infrastructure",
+                        to="organisations.organisation",
+                    ),
+                ),
+            ],
+        ),
+    ]

api/experimentation/models.py

@@ -78,9 +78,18 @@ class Meta:
 
     @hook(AFTER_CREATE)  # type: ignore[misc]
     def sync_to_ingestion_on_create(self) -> None:
-        from experimentation.tasks import write_environment_ingestion_keys
+        from experimentation.tasks import (
+            provision_external_warehouse_ingestion_infrastructure,
+            write_environment_ingestion_keys,
+        )
+
+        if self.warehouse_type == WarehouseType.FLAGSMITH:
+            write_environment_ingestion_keys.delay(
+                kwargs={"environment_id": self.environment_id},
+            )
+            return
 
-        write_environment_ingestion_keys.delay(
+        provision_external_warehouse_ingestion_infrastructure.delay(
             kwargs={"environment_id": self.environment_id},
         )
 
@@ -93,6 +102,29 @@ def sync_to_ingestion_on_delete(self) -> None:
         )
 
 
+class IngestionInfrastructureStatus(models.TextChoices):
+    PENDING = "pending", "Pending"
+    CREATED = "created", "Created"
+    ERRORED = "errored", "Errored"
+
+
+class OrganisationIngestionInfrastructure(models.Model):
+    organisation = models.OneToOneField(
+        "organisations.Organisation",
+        on_delete=models.DO_NOTHING,
+        related_name="ingestion_infrastructure",
+    )
+    status = models.CharField(
+        max_length=50,
+        choices=IngestionInfrastructureStatus.choices,
+        default=IngestionInfrastructureStatus.PENDING,
+    )
+    bucket_name = models.CharField(max_length=255, null=True, blank=True)
+    stream_name = models.CharField(max_length=255, null=True, blank=True)
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+
 class ExperimentStatus(models.TextChoices):
     CREATED = "created", "Created"
     RUNNING = "running", "Running"

api/experimentation/organisation_ingestion_service.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import typing
+
+import structlog
+
+from experimentation.ingestion_infra_service import (
+    deprovision_ingestion_infrastructure,
+    provision_ingestion_infrastructure,
+)
+from experimentation.models import (
+    IngestionInfrastructureStatus,
+    OrganisationIngestionInfrastructure,
+)
+
+if typing.TYPE_CHECKING:
+    from organisations.models import Organisation
+
+logger = structlog.get_logger("experimentation")
+
+
+def enable_ingestion_for_organisation(
+    organisation: "Organisation",
+) -> OrganisationIngestionInfrastructure:
+    infrastructure = OrganisationIngestionInfrastructure.objects.filter(
+        organisation=organisation
+    ).first()
+    if (
+        infrastructure is not None
+        and infrastructure.status == IngestionInfrastructureStatus.CREATED
+    ):
+        return infrastructure
+    if infrastructure is None:
+        infrastructure = OrganisationIngestionInfrastructure(organisation=organisation)
+
+    try:
+        result = provision_ingestion_infrastructure(organisation.id)
+    except Exception as exc:
+        infrastructure.status = IngestionInfrastructureStatus.ERRORED
+        infrastructure.save()
+        logger.error(
+            "ingestion_infra.provision_failed",
+            exc_info=exc,
+            organisation__id=organisation.id,
+        )
+        raise
+
+    infrastructure.bucket_name = result.bucket_name
+    infrastructure.stream_name = result.stream_name
+    infrastructure.status = IngestionInfrastructureStatus.CREATED
+    infrastructure.save()
+    logger.info(
+        "ingestion_infra.provisioned",
+        organisation__id=organisation.id,
+        bucket__name=result.bucket_name,
+        stream__name=result.stream_name,
+    )
+    return infrastructure
+
+
+def disable_ingestion_for_organisation(organisation_id: int) -> None:
+    infrastructure = OrganisationIngestionInfrastructure.objects.filter(
+        organisation_id=organisation_id
+    ).first()
+    if infrastructure is None:
+        return
+
+    if infrastructure.status == IngestionInfrastructureStatus.CREATED:
+        deprovision_ingestion_infrastructure(organisation_id)
+        logger.info(
+            "ingestion_infra.torn_down",
+            organisation__id=organisation_id,
+        )
+    infrastructure.delete()

api/experimentation/tasks.py

@@ -5,6 +5,10 @@
 from environments.models import Environment, EnvironmentAPIKey
 from experimentation import ingestion_sync_service
 from experimentation.models import Experiment, ExperimentExposures, ExperimentResults
+from experimentation.organisation_ingestion_service import (
+    disable_ingestion_for_organisation,
+    enable_ingestion_for_organisation,
+)
 from experimentation.services import compute_exposures_summary, compute_results_summary
 
 logger = structlog.get_logger("experimentation")
@@ -48,6 +52,25 @@ def remove_environment_ingestion_keys(environment_id: int) -> None:
         ingestion_sync_service.delete_ingestion_key(api_key.key)
 
 
+@register_task_handler()
+def provision_external_warehouse_ingestion_infrastructure(environment_id: int) -> None:
+    environment = (
+        Environment.objects.select_related("project__organisation")
+        .filter(id=environment_id)
+        .first()
+    )
+    if environment is None:
+        return
+
+    enable_ingestion_for_organisation(environment.project.organisation)
+    write_environment_ingestion_keys(environment_id)
+
+
+@register_task_handler()
+def teardown_organisation_ingestion_infrastructure(organisation_id: int) -> None:
+    disable_ingestion_for_organisation(organisation_id)
+
+
 @register_task_handler()
 def write_environment_ingestion_key(environment_api_key_id: int) -> None:
     api_key = (

api/organisations/models.py

@@ -10,6 +10,7 @@
 from django.utils import timezone
 from django_lifecycle import (  # type: ignore[import-untyped]
     AFTER_CREATE,
+    AFTER_DELETE,
     AFTER_SAVE,
     BEFORE_DELETE,
     LifecycleModelMixin,
@@ -150,6 +151,19 @@ def cancel_subscription(self):  # type: ignore[no-untyped-def]
         if self.has_paid_subscription():
             self.subscription.prepare_for_cancel()
 
+    @hook(AFTER_DELETE)
+    def teardown_ingestion_infrastructure(self):  # type: ignore[no-untyped-def]
+        if not hasattr(self, "ingestion_infrastructure"):
+            return
+
+        from experimentation.tasks import (
+            teardown_organisation_ingestion_infrastructure,
+        )
+
+        teardown_organisation_ingestion_infrastructure.delay(
+            kwargs={"organisation_id": self.id},
+        )
+
     @hook(AFTER_CREATE)
     def create_subscription(self):  # type: ignore[no-untyped-def]
         Subscription.objects.create(organisation=self)

api/tests/unit/experimentation/conftest.py

@@ -26,6 +26,17 @@ def mock_ingestion_redis_client(mocker: MockerFixture) -> None:
     mocker.patch("experimentation.ingestion_sync_service.RedisCluster.from_url")
 
 
+@pytest.fixture(autouse=True)
+def mock_provision_external_warehouse_ingestion_infrastructure(
+    mocker: MockerFixture,
+) -> None:
+    # Creating an external warehouse connection enqueues provisioning, which
+    # runs synchronously under test; stub it so tests don't reach AWS.
+    mocker.patch(
+        "experimentation.tasks.provision_external_warehouse_ingestion_infrastructure",
+    )
+
+
 @pytest.fixture()
 def warehouse_connection(environment: Environment) -> WarehouseConnection:
     connection: WarehouseConnection = WarehouseConnection.objects.create(

api/tests/unit/experimentation/test_ingestion_infra_service.py

@@ -234,3 +234,57 @@ def test_provision_ingestion_infrastructure__stream_creation_fails__propagates_c
     # When / Then
     with pytest.raises(ClientError, match="LimitExceededException"):
         ingestion_infra_service.provision_ingestion_infrastructure(organisation_id=42)
+
+
+def test_deprovision_ingestion_infrastructure__existing_resources__deletes_bucket_and_stream(
+    ingestion_infra_settings: SettingsWrapper,
+    aws_backends: None,
+    log: StructuredLogCapture,
+) -> None:
+    # Given
+    result = ingestion_infra_service.provision_ingestion_infrastructure(
+        organisation_id=42,
+    )
+
+    # When
+    ingestion_infra_service.deprovision_ingestion_infrastructure(organisation_id=42)
+
+    # Then
+    s3 = boto3.client("s3", region_name="eu-west-2")
+    bucket_names = [bucket["Name"] for bucket in s3.list_buckets()["Buckets"]]
+    assert result.bucket_name not in bucket_names
+
+    firehose = boto3.client("firehose", region_name="eu-west-2")
+    with pytest.raises(ClientError):
+        firehose.describe_delivery_stream(DeliveryStreamName=result.stream_name)
+
+    assert {
+        "level": "info",
+        "event": "ingestion_infra.deprovisioned",
+        "organisation__id": 42,
+        "bucket__name": result.bucket_name,
+        "stream__name": result.stream_name,
+    } in log.events
+
+
+def test_deprovision_ingestion_infrastructure__bucket_with_objects__empties_and_deletes_bucket(
+    ingestion_infra_settings: SettingsWrapper,
+    aws_backends: None,
+) -> None:
+    # Given
+    result = ingestion_infra_service.provision_ingestion_infrastructure(
+        organisation_id=42,
+    )
+    s3 = boto3.client("s3", region_name="eu-west-2")
+    s3.put_object(
+        Bucket=result.bucket_name,
+        Key="events/env_key=abc/data.json.gz",
+        Body=b"{}",
+    )
+
+    # When
+    ingestion_infra_service.deprovision_ingestion_infrastructure(organisation_id=42)
+
+    # Then
+    bucket_names = [bucket["Name"] for bucket in s3.list_buckets()["Buckets"]]
+    assert result.bucket_name not in bucket_names

api/tests/unit/experimentation/test_models.py

@@ -47,6 +47,32 @@ def test_warehouse_connection__after_create__enqueues_ingestion_write_task(
     )
 
 
+def test_warehouse_connection__after_create_external_type__enqueues_provision_task(
+    environment: Environment,
+    mocker: MockerFixture,
+) -> None:
+    # Given
+    mock_provision = mocker.patch(
+        "experimentation.tasks.provision_external_warehouse_ingestion_infrastructure",
+    )
+    mock_write_keys = mocker.patch(
+        "experimentation.tasks.write_environment_ingestion_keys",
+    )
+
+    # When
+    WarehouseConnection.objects.create(
+        environment=environment,
+        warehouse_type=WarehouseType.CLICKHOUSE,
+        name="external warehouse",
+    )
+
+    # Then the per-org infrastructure is provisioned, which chains the key sync
+    mock_provision.delay.assert_called_once_with(
+        kwargs={"environment_id": environment.id},
+    )
+    mock_write_keys.delay.assert_not_called()
+
+
 def test_warehouse_connection__after_delete__enqueues_ingestion_remove_task(
     warehouse_connection: WarehouseConnection,
     mocker: MockerFixture,