feat: emit transaction lifecycle events

[iloveagent57]

Emit openedx events (to the event bus) when transactions are created, committed, failed, and reversed.
Additionally fixes a bug in the write-reversals mgmt command that proceeded to write a reversal in case an external fulfillment could not successfully be canceled.
ENT-8761
ENT-9009

See proposed consumption: openedx/enterprise-access#502

Testing instructions

0. Run make app-logs in enterprise-subsidy, or have the local kafka control center running in your browser.
1. Find a combination of (policy, learner, content key) that can be redeemed.
2. Hit the policy redeem endpoint.
3. The response should be a committed transaction, and you should see events emitted to kafka in your logs.

Example of reversal signal being sent

From enterprise-subsidy:

enterprise-subsidy.app  | 2024-06-26 14:02:32,049 DEBUG 129 [urllib3.connectionpool] [user 6] [ip 172.18.0.1] [request_id None] connectionpool.py:546 - http://edx.devstack.schema-registry:8081 "POST /subjects/dev-enterprise-subsidy-ledger-transaction-lifecycle-org.openedx.enterprise.subsidy_ledger_transaction.reversed.v1.CloudEvent/versions?normalize=False HTTP/11" 200 8
...
enterprise-subsidy.app  | 2024-06-26 14:02:32,579 INFO 129 [edx_event_bus_kafka.internal.producer] [user None] [ip None] [request_id None] producer.py:244 - Message delivered to Kafka event bus: topic=dev-enterprise-subsidy-ledger-transaction-lifecycle, partition=0, offset=2, message_id=bb410e7a-33c4-11ef-a3f9-0242ac120013, key=b'\x00\x00\x00\x00\x01H3eacfe31-2453-4b2f-8ba5-392214dd8a98'

In local confluent:

The event data:

{
  "ledger_transaction": {
    "uuid": "3eacfe31-2453-4b2f-8ba5-392214dd8a98",
    "created": "2024-06-26T14:01:38.537264+00:00",
    "modified": "2024-06-26T14:01:39.489644+00:00",
    "idempotency_key": "ledger-for-subsidy-4c53f616-5ea0-42a0-9430-ba165cb2c916-ed691d7727be279ea98a40ff8517a219",
    "quantity": -14800,
    "state": "committed",
    "ledger_uuid": "2cc1f351-ee20-4c0f-92e8-55e6e442d22a",
    "subsidy_access_policy_uuid": "3cc0ad2c-ede1-476c-8271-97683855e41d",
    "lms_user_id": 2,
    "content_key": "course-v1:edX+DemoX+Demo_Course",
    "parent_content_key": {
      "string": "course-v1:edX+DemoX+Demo_Course"
    },
    "fulfillment_identifier": {
      "string": "1ab9b0ee-bcb2-4798-8d42-fd4176cb68e8"
    },
    "reversal": {
      "org.openedx.enterprise.subsidy_ledger_transaction.reversed.v1.LedgerTransactionReversal": {
        "uuid": "076e3be2-a702-4400-b3cf-27b1c077e740",
        "created": "2024-06-26T14:02:27.874315+00:00",
        "modified": "2024-06-26T14:02:27.874315+00:00",
        "idempotency_key": "admin-invoked-reverse-3eacfe31-2453-4b2f-8ba5-392214dd8a98",
        "quantity": 14800,
        "state": "committed"
      }
    }
  }
}

Merge checklist

• All reviewers approved
• CI build is green
• Documentation updated (not only docstrings)
• Commits are squashed

[macdiesel]

I left my thoughts. Looks good.

[macdiesel]

If the event bus send fails for these calls is the ledger_transaction also rolled back?

I think my overall question is this: When these event emissions fail to send, should the related DB writes also fail?

[iloveagent57]

No, because then we're making the communication pattern synchronous instead of asynchronous.

[macdiesel]

But if the event doesn't make it on to the bus, and the record is written, systems downstream of this message will not be able to act accordingly. Not having this event hit the bus means that whatever actions were supposed to happen because of this do not happen leaving the system in an inconsistent state.

I don't believe this makes it synchronous. It's more that the transaction isn't complete until the event is sent.

[macdiesel]

I would also suggest that this does not make it synchronous either as we are simply writing that the event happened. Not waiting for downstream systems to execute on the message.

I think that writing these messages on to the bus and ensuring that call succeeds is just as important now as ensuring the DB record is written.

[iloveagent57]

Synchronous was the wrong word, it's more like you'd be making the send of the message part of the "atomic block" of the flow.
Part of the goal of using an async event broker, though, is to move toward "eventual consistency", and force our services to adapt to inconsistent states. The ideal thing to do if kafka is down/unreachable/whatever isn't to fail the redemption flow; it's to have a mechanism for replaying events that occurred in a given time window, and to be able to rely on idempotent consumers to process those replayed events. Or to implement a dead-letter queue (here on the producer side).

[iloveagent57]

Now, having said all that, I should go through and check if this is already called from within a transaction.atomic() block.

[pwnage101]

LGTM!

The one thing I'm confused about is the purpose of event_key_field. I have no idea how it works...what dictionary has "ledger_transaction" as a top-level key? Why is the UUID of a transaction considered unique enough for an event? What happens when two events about the same transaction are emitted...wouldn't they conflict? The openedx-events documentation uses similarly confusing keys to me, so I know there's something I don't understand about this, and it's net-new anyway, so I don't want to block.

[iloveagent57]

@pwnage101 The signal's data is what maps a ledger_transaction top-level key to transaction event data: https://github.com/openedx/openedx-events/blob/c8fe32e309299c82cdecf7dd315edaeb9540bc92/openedx_events/enterprise/signals.py#L81-L86
My understanding of event_key_field is that it specifies which field of the event data can be used as a unique identifier for the event, our app-specific logic wouldn't reference it directly, it's only useful in the event bus layer.