TFRS: Create Transaction Message Queue for New Supplemental Acceptances

[AlexZorkin]

Description:
Implement a message queue in TFRS to trigger when a director approves a supplemental report, or when any balance change occurs. This message queue will send the organization ID and the balance change value (compliance units) to RabbitMQ for processing in LCFS. LCFS will then listen to this queue and create the corresponding transaction in its system.

Purpose:
This task enables seamless communication between TFRS and LCFS when compliance unit balances are updated. By sending a message with the organization ID and the balance change to RabbitMQ, LCFS can process this data and create a corresponding transaction. This automation reduces manual intervention and ensures accurate real-time data flow between systems.

Development Requirements:

• Implement message queue logic in TFRS using the existing RabbitMQ instance.
• Trigger message creation when a director approves a supplemental report, including the organization ID and compliance units amount in the message.
• Ensure the message is correctly formatted and placed on the RabbitMQ queue.
• Test message creation and delivery to ensure the message is properly received by LCFS.

Additional Context:

• TFRS already has RabbitMQ implemented, so focus on creating the logic to send the transaction message when needed.
• The message should include:
  • organization_id
  • compliance_units_amount
• This ticket is only for the TFRS side. LCFS will implement the logic to receive and process the message.

Code Snippet for Placing the Message on the Queue

You can use the pika library in Python to place a message on the RabbitMQ queue. Here’s an example implementation that would fit into the TFRS side when a supplemental report is approved:

import pika
import json

class MessageQueueService:
    def __init__(self, queue_name='transaction_queue'):
        self.queue_name = queue_name
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host='rabbitmq_url')  # replace with actual RabbitMQ URL
        )
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue=self.queue_name, durable=True)

    def send_transaction_message(self, organization_id, compliance_units_amount):
        message = {
            'organization_id': organization_id,
            'compliance_units_amount': compliance_units_amount
        }
        # Convert message to JSON string
        message_body = json.dumps(message)
        
        # Send the message to the queue
        self.channel.basic_publish(
            exchange='',  # Default exchange
            routing_key=self.queue_name,  # Queue name as routing key
            body=message_body,
            properties=pika.BasicProperties(
                delivery_mode=2,  # Make message persistent
            )
        )
        print(f" [x] Sent transaction message: {message_body}")
    
    def close_connection(self):
        self.connection.close()

# Usage Example in Supplemental Report Approval Logic
def on_supplemental_report_approved(organization_id, compliance_units_amount):
    mq_service = MessageQueueService()
    
    # Place the transaction message on the queue
    mq_service.send_transaction_message(organization_id, compliance_units_amount)
    
    # Close the connection
    mq_service.close_connection()

Explanation of the Code:

1. MessageQueueService:

   • This class sets up the connection to RabbitMQ and declares the queue where messages will be sent. It has a method send_transaction_message that formats the message with the organization_id and compliance_units_amount and sends it to the RabbitMQ queue.

2. on_supplemental_report_approved:

   • This function simulates the approval logic for a supplemental report, triggering the message to be sent. It instantiates the MessageQueueService, sends the transaction message, and then closes the connection.