feat(migration_event): setup trigger for migration event

api/app/settings/common.py

@@ -564,3 +564,6 @@
 EDGE_API_URL = env.str("EDGE_API_URL", None)
 # Used for signing forwarded request to edge
 EDGE_REQUEST_SIGNING_KEY = env.str("EDGE_REQUEST_SIGNING_KEY", None)
+
+# Aws Event bus used for sending identity migration events
+IDENTITY_MIGRATION_EVENT_BUS_NAME = env.str("IDENTITY_MIGRATION_EVENT_BUS_NAME", None)

api/edge_api/identities/events.py

@@ -0,0 +1,19 @@
+import json
+
+import boto3
+from django.conf import settings
+
+events_client = None
+if settings.IDENTITY_MIGRATION_EVENT_BUS_NAME:
+    events_client = boto3.client("events")
+
+
+def send_migration_event(project_id: int):
+    event = {
+        "EventBusName": settings.IDENTITY_MIGRATION_EVENT_BUS_NAME,
+        "Source": "flagsmith.api.migrate",
+        "DetailType": "Migrate Identities to dynamodb",
+        "Detail": json.dumps({"project_id": project_id}),
+    }
+
+    events_client.put_events(Entries=[event])

api/sales_dashboard/views.py

@@ -1,4 +1,5 @@
 import json
+from functools import partial
 
 from app_analytics.influxdb_wrapper import (
     get_event_list_for_organisation,
@@ -20,6 +21,7 @@
 from django.views.generic import ListView
 from django.views.generic.edit import FormView
 
+from edge_api.identities.events import send_migration_event
 from environments.dynamodb.migrator import IdentityMigrator
 from environments.identities.models import Identity
 from organisations.models import Organisation
@@ -30,6 +32,8 @@
 
 OBJECTS_PER_PAGE = 50
 
+MAX_MIGRATABLE_IDENTITIES_SYNC = 1000
+
 
 class OrganisationList(ListView):
     model = Organisation
@@ -193,8 +197,19 @@ def migrate_identities_to_edge(request, project_id):
     if not settings.PROJECT_METADATA_TABLE_NAME_DYNAMO:
         return HttpResponseBadRequest("DynamoDB is not enabled")
     identity_migrator = IdentityMigrator(project_id)
-    if identity_migrator.can_migrate:
-        identity_migrator.migrate()
+    if not identity_migrator.can_migrate:
+        return HttpResponseBadRequest(
+            "Migration is either already done or is in progress"
+        )
+
+    identity_count = Identity.objects.filter(environment__project_id=project_id).count()
+    migrator_function = (
+        identity_migrator.migrate
+        if identity_count < MAX_MIGRATABLE_IDENTITIES_SYNC
+        else partial(send_migration_event, project_id)
+    )
+
+    migrator_function()
     return HttpResponseRedirect(reverse("sales_dashboard:index"))
 
 

api/tests/integration/sales_dashboard/test_integration_sales_dashboard.py

@@ -57,7 +57,7 @@ def test_migrate_identities_to_edge_does_not_call_migrate_if_migration_is_alread
     response = superuser_authenticated_client.post(url)
 
     # Then
-    assert response.status_code == status.HTTP_302_FOUND
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
     mocked_identity_migrator.assert_called_with(project)
     mocked_identity_migrator.return_value.migrate.assert_not_called()
 
@@ -74,3 +74,32 @@ def test_migrate_identities_to_edge_returns_400_if_dynamodb_is_not_enabled(
 
     # Then
     assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
+def test_migrate_identities_to_edge_calls_send_migration_event_with_correct_arguments(
+    superuser_authenticated_client, mocker, project, settings, identity
+):
+    # Given
+    settings.PROJECT_METADATA_TABLE_NAME_DYNAMO = "project_metadata_table"
+    url = reverse("sales_dashboard:migrate_identities", args=[project])
+
+    # update the `MAX_MIGRATABLE_IDENTITIES_SYNC` to trigger send_migration_event
+    mocker.patch("sales_dashboard.views.MAX_MIGRATABLE_IDENTITIES_SYNC", 1)
+    mocked_send_migrate_event = mocker.patch(
+        "sales_dashboard.views.send_migration_event"
+    )
+    mocked_identity_migrator = mocker.patch(
+        "sales_dashboard.views.IdentityMigrator", spec=IdentityMigrator
+    )
+
+    mocked_identity_migrator.return_value.can_migrate = True
+
+    # When
+    response = superuser_authenticated_client.post(url)
+
+    # Then
+    assert response.status_code == status.HTTP_302_FOUND
+    mocked_send_migrate_event.assert_called_with(project)
+
+    mocked_identity_migrator.assert_called_with(project)
+    mocked_identity_migrator.return_value.migrate.assert_not_called()

api/tests/unit/edge_api/identities/test_events.py

@@ -0,0 +1,20 @@
+import json
+
+from edge_api.identities.events import send_migration_event
+
+
+def test_send_migration_event_calls_put_events_with_correct_arguments(mocker, settings):
+    # Given
+    project_id = 1
+    mocked_client = mocker.patch("edge_api.identities.events.events_client")
+    event_bus_name = "identity_migration"
+    settings.IDENTITY_MIGRATION_EVENT_BUS_NAME = event_bus_name
+
+    # When
+    send_migration_event(project_id)
+
+    # Then
+    args, kwargs = mocked_client.put_events.call_args
+    assert args == ()
+    assert kwargs["Entries"][0]["EventBusName"] == event_bus_name
+    assert kwargs["Entries"][0]["Detail"] == json.dumps({"project_id": project_id})