feat: Add data migration to backfill legacy initial deposits

This needs to be located in the Subsidy app instead of openedx_ledger
because it relies on having access to the SubsidyReferenceChoices class.

ENT-9075

enterprise_subsidy/apps/subsidy/migration_utils.py

@@ -0,0 +1,40 @@
+"""
+Helper functions for data migrations.
+"""
+from django.db.models import F, Window
+from django.db.models.functions import FirstValue
+# This utils module is imported by migrations, so never import any models directly here, nor any modules that import
+# models in turn.
+from openedx_ledger.constants import INITIAL_DEPOSIT_TRANSACTION_SLUG
+
+
+def find_legacy_initial_transactions(transaction_cls):
+    """
+    Heuristic to identify "legacy" initial transactions.
+
+    An initial transaction is one that has the following traits:
+    * Is chronologically the first transaction for a Ledger.
+    * Contains a hint in its idempotency key which indicates that it is an initial deposit.
+    * Has a positive quantity.
+
+    A legacy initial transaction is one that has the following additional traits:
+    * does not have a related Deposit.
+    """
+    # All transactions which are chronologically the first in their respective ledgers.
+    first_transactions = transaction_cls.objects.annotate(
+        first_tx_uuid=Window(
+            expression=FirstValue('uuid'),
+            partition_by=['ledger'],
+            order_by=F('created').asc(),  # "first chronologically" above means first created.
+        ),
+    ).filter(uuid=F('first_tx_uuid'))
+
+    # Further filter first_transactions to find ones that qualify as _initial_ and _legacy_.
+    legacy_initial_transactions = first_transactions.filter(
+        # Traits of an _initial_ deposit:
+        idempotency_key__contains=INITIAL_DEPOSIT_TRANSACTION_SLUG,
+        quantity__gte=0,
+        # Traits of a _legacy_ initial deposit:
+        deposit__isnull=True,
+    )
+    return legacy_initial_transactions

enterprise_subsidy/apps/subsidy/migrations/0022_backfill_initial_deposits.py

@@ -0,0 +1,78 @@
+"""
+Backfill initial deposits.
+
+Necessarily, this also backfills SalesContractReferenceProvider objects based on the values currently defined via
+SubsidyReferenceChoices.
+
+Note this has no reverse migration logic. Attempts to rollback the deployment which includes this PR will not delete
+(un-backfill) the deposits created during the forward migration.
+"""
+from datetime import datetime
+from django.db import migrations
+
+from enterprise_subsidy.apps.subsidy.migration_utils import find_legacy_initial_transactions
+from enterprise_subsidy.apps.subsidy.models import SubsidyReferenceChoices
+
+
+def forwards_func(apps, schema_editor):
+    """
+    The core logic of this migration.
+    """
+    # We get the models from the versioned app registry; if we directly import it, it'll be the wrong version.
+    Transaction = apps.get_model("openedx_ledger", "Transaction")
+    Deposit = apps.get_model("openedx_ledger", "Deposit")
+    HistoricalDeposit = apps.get_model("openedx_ledger", "HistoricalDeposit")
+    SalesContractReferenceProvider = apps.get_model("openedx_ledger", "SalesContractReferenceProvider")
+
+    # Idempotently duplicate all SubsidyReferenceChoices into SalesContractReferenceProvider.
+    sales_contract_reference_providers = {}
+    for slug, name in SubsidyReferenceChoices.CHOICES:
+        sales_contract_reference_providers[slug] = SalesContractReferenceProvider.objects.get_or_create(
+            slug=slug,
+            defaults={"name": name},
+        )
+
+    # Fetch all "legacy" initial transactions.
+    legacy_initial_transactions = find_legacy_initial_transactions(Transaction)
+
+    # Construct all missing Deposits and HistoricalDeposits to backfill, but do not save them yet.
+    #
+    # Note: The reason we need to manually create historical objects is that Django's bulk_create() built-in does not
+    # call post_save hooks, which is normally where history objects are created. Next you might ask why we don't just
+    # use django-simple-history's bulk_create_with_history() utility function: that's because it attempts to access the
+    # custom simple history model manager, but unfortunately custom model attributes are unavailable from migrations.
+    deposits_to_backfill = []
+    historical_deposits_to_backfill = []
+    for tx in legacy_initial_transactions:
+        deposit_fields = {
+            "ledger": tx.ledger,
+            "transaction": tx,
+            "desired_deposit_quantity": tx.quantity,
+            "sales_contract_reference_id": tx.ledger.subsidy.reference_id,
+            "sales_contract_reference_provider": sales_contract_reference_providers[tx.ledger.subsidy.reference_type],
+        }
+        deposit = Deposit(**deposit_fields)
+        historical_deposit = HistoricalDeposit(
+            uuid=deposit.uuid,
+            history_date=datetime.now()
+            **deposit_fields,
+        )
+        deposits_to_backfill.append(deposit)
+        historical_deposits_to_backfill.append(historical_deposit)
+
+    # Finally, save the missing Deposits and HistoricalDeposits in bulk.
+    Deposit.objects.bulk_create(deposits_to_backfill, batch_size=50)
+    HistoricalDeposit.objects.bulk_create(historical_deposits_to_backfill, batch_size=50)
+
+
+class Migration(migrations.Migration):
+    """
+    Migration for backfilling initial deposits.
+    """
+    dependencies = [
+        ("subsidy", "0021_alter_historicalsubsidy_options"),
+    ]
+
+    operations = [
+        migrations.RunPython(forwards_func),
+    ]

enterprise_subsidy/apps/subsidy/tests/test_migration_utils.py

@@ -0,0 +1,53 @@
+"""
+Tests for the migration_utils.py module.
+"""
+from django.test import TestCase
+from openedx_ledger.constants import INITIAL_DEPOSIT_TRANSACTION_SLUG
+from openedx_ledger.models import Transaction
+from openedx_ledger.test_utils.factories import AdjustmentFactory, DepositFactory, LedgerFactory, TransactionFactory
+
+from enterprise_subsidy.apps.subsidy import migration_utils
+
+
+class MigrationUtilsTests(TestCase):
+    """
+    Tests for utils used for migrations.
+    """
+    def test_find_legacy_initial_transactions(self):
+        """
+        Test find_legacy_initial_transactions(), used for a data migration to backfill initial deposits.
+        """
+        ledgers = [
+            (LedgerFactory(), False),
+            (LedgerFactory(), False),
+            (LedgerFactory(), True),
+            (LedgerFactory(), False),
+            (LedgerFactory(), True),
+        ]
+        expected_legacy_initial_transactions = []
+        for ledger, create_initial_deposit in ledgers:
+            # Simulate a legacy initial transaction (i.e. transaction WITHOUT a deposit).
+            initial_transaction = TransactionFactory(
+                ledger=ledger,
+                idempotency_key=INITIAL_DEPOSIT_TRANSACTION_SLUG,
+                quantity=100,
+            )
+            if create_initial_deposit:
+                # Make it a modern initial deposit by creating a related Deposit.
+                DepositFactory(
+                    ledger=ledger,
+                    transaction=initial_transaction,
+                    desired_deposit_quantity=initial_transaction.quantity,
+                )
+            else:
+                # Keep it a legacy initial deposit by NOT creating a related Deposit.
+                expected_legacy_initial_transactions.append(initial_transaction)
+            # Throw in a few spend, deposit, and adjustment transactions for fun.
+            TransactionFactory(ledger=ledger, quantity=-10)
+            TransactionFactory(ledger=ledger, quantity=-10)
+            DepositFactory(ledger=ledger, desired_deposit_quantity=50)
+            tx_to_adjust = TransactionFactory(ledger=ledger, quantity=-5)
+            AdjustmentFactory(ledger=ledger, adjustment_quantity=5, transaction_of_interest=tx_to_adjust)
+
+        actual_legacy_initial_transactions = migration_utils.find_legacy_initial_transactions(Transaction)
+        assert set(actual_legacy_initial_transactions) == set(expected_legacy_initial_transactions)