Data migration to backfill session expiration contact fires

temba/contacts/migrations/0205_create_session_expires_fires.py

@@ -0,0 +1,65 @@
+# Generated by Django 5.1.4 on 2025-02-26 16:27
+
+import random
+from datetime import timedelta
+
+from django.db import migrations
+from django.db.models import Exists, OuterRef
+
+
+def create_session_expires_fires(apps, schema_editor):
+    Contact = apps.get_model("contacts", "Contact")
+    ContactFire = apps.get_model("contacts", "ContactFire")
+    FlowSession = apps.get_model("flows", "FlowSession")
+
+    num_created = 0
+
+    while True:
+        # find contacts with waiting sessions that don't have a corresponding session expiration fire
+        batch = list(
+            Contact.objects.filter(current_session_uuid__isnull=False)
+            .filter(~Exists(ContactFire.objects.filter(contact=OuterRef("pk"), fire_type="S")))
+            .only("id", "org_id", "current_session_uuid")[:1000]
+        )
+        if not batch:
+            break
+
+        sessions = FlowSession.objects.filter(uuid__in=[c.current_session_uuid for c in batch]).only(
+            "uuid", "created_on"
+        )
+        created_on_by_uuid = {s.uuid: s.created_on for s in sessions}
+
+        to_create = []
+        for contact in batch:
+            session_created_on = created_on_by_uuid[contact.current_session_uuid]
+            to_create.append(
+                ContactFire(
+                    org_id=contact.org_id,
+                    contact=contact,
+                    fire_type="S",
+                    scope="",
+                    fire_on=session_created_on + timedelta(days=30) + timedelta(seconds=random.randint(0, 86400)),
+                    session_uuid=contact.current_session_uuid,
+                )
+            )
+
+        ContactFire.objects.bulk_create(to_create)
+        num_created += len(to_create)
+        print(f"Created {num_created} session expiration fires")
+
+
+def apply_manual():  # pragma: no cover
+    from django.apps import apps
+
+    create_session_expires_fires(apps, None)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("contacts", "0204_alter_contactfire_fire_type"),
+    ]
+
+    operations = [
+        migrations.RunPython(create_session_expires_fires, migrations.RunPython.noop),
+    ]

temba/contacts/tests/test_migrations.py

@@ -0,0 +1,72 @@
+from datetime import timedelta
+
+from django.utils import timezone
+
+from temba.contacts.models import ContactFire
+from temba.flows.models import FlowSession
+from temba.tests import MigrationTest
+from temba.utils.uuid import uuid4
+
+
+class CreateSessionExpiresFiresTest(MigrationTest):
+    app = "contacts"
+    migrate_from = "0204_alter_contactfire_fire_type"
+    migrate_to = "0205_create_session_expires_fires"
+
+    def setUpBeforeMigration(self, apps):
+        def create_contact_and_sessions(name, phone, current_session_uuid):
+            contact = self.create_contact(name, phone=phone, current_session_uuid=current_session_uuid)
+            FlowSession.objects.create(
+                uuid=uuid4(),
+                contact=contact,
+                status=FlowSession.STATUS_COMPLETED,
+                output_url="http://sessions.com/123.json",
+                created_on=timezone.now(),
+                ended_on=timezone.now(),
+            )
+            FlowSession.objects.create(
+                uuid=current_session_uuid,
+                contact=contact,
+                status=FlowSession.STATUS_WAITING,
+                output_url="http://sessions.com/123.json",
+                created_on=timezone.now(),
+            )
+            return contact
+
+        # contacts with waiting sessions but no session expiration fire
+        self.contact1 = create_contact_and_sessions("Ann", "+1234567001", "a0e707ef-ae06-4e39-a9b1-49eed0273dae")
+        self.contact2 = create_contact_and_sessions("Bob", "+1234567002", "4a675e5d-ebc1-4fe7-be74-0450f550f8ee")
+
+        # contact with waiting session and already has a session expiration fire
+        self.contact3 = create_contact_and_sessions("Cat", "+1234567003", "a83a82f4-6a25-4662-a8e1-b53ee7d259a2")
+        ContactFire.objects.create(
+            org=self.org,
+            contact=self.contact3,
+            fire_type="S",
+            scope="",
+            fire_on=timezone.now() + timedelta(days=30),
+            session_uuid="a83a82f4-6a25-4662-a8e1-b53ee7d259a2",
+        )
+
+        # contact with no waiting session
+        self.contact4 = self.create_contact("Dan", phone="+1234567004")
+
+    def test_migration(self):
+        def assert_session_expire(contact):
+            self.assertTrue(contact.fires.exists())
+
+            session = contact.sessions.filter(status="W").get()
+            fire = contact.fires.get()
+
+            self.assertEqual(fire.org, contact.org)
+            self.assertEqual(fire.fire_type, "S")
+            self.assertEqual(fire.scope, "")
+            self.assertGreaterEqual(fire.fire_on, session.created_on + timedelta(days=30))
+            self.assertLess(fire.fire_on, session.created_on + timedelta(days=31))
+            self.assertEqual(fire.session_uuid, session.uuid)
+
+        assert_session_expire(self.contact1)
+        assert_session_expire(self.contact2)
+        assert_session_expire(self.contact3)
+
+        self.assertFalse(self.contact4.fires.exists())

temba/tests/base.py

@@ -210,7 +210,7 @@ def create_contact(
         org=None,
         user=None,
         status=Contact.STATUS_ACTIVE,
-        last_seen_on=None,
+        **kwargs,
     ):
         """
         Create a new contact
@@ -229,7 +229,7 @@ def create_contact(
             fields or {},
             group_uuids=[],
             status=status,
-            last_seen_on=last_seen_on,
+            **kwargs,
         )
 
     def create_group(self, name, contacts=(), query=None, org=None):

temba/tests/mailroom.py

@@ -608,7 +608,7 @@ def contact_resolve(org, phone: str) -> tuple:
 
 
 def create_contact_locally(
-    org, user, name, language, urns, fields, group_uuids, status=Contact.STATUS_ACTIVE, last_seen_on=None
+    org, user, name, language, urns, fields, group_uuids, status=Contact.STATUS_ACTIVE, last_seen_on=None, **kwargs
 ):
     orphaned_urns = {}
 
@@ -629,6 +629,7 @@ def create_contact_locally(
         created_on=timezone.now(),
         status=status,
         last_seen_on=last_seen_on,
+        **kwargs,
     )
     update_urns_locally(contact, urns)
     update_fields_locally(user, contact, fields)