Skip to content

message_formats.md

Latest commit

 

History

History
292 lines (229 loc) · 7.65 KB

message_formats.md

File metadata and controls

292 lines (229 loc) · 7.65 KB
Raw

Message Formats

This document describes the JSON message formats generated by the processor and sent to SQS.

Client Message

Generated once per client, aggregating taxes paid across all their portfolios/accounts.

{
  "type": "client_message",
  "client_reference": "9e40659b-8b9f-4fc4-814b-5a7b5a23b64",
  "tax_free_allowance": 801,
  "taxes_paid": 0.00
}

Fields

Field Type Description
type string Always "client_message"
client_reference string Unique client identifier (UUID)
tax_free_allowance number Client's tax-free allowance amount
taxes_paid number Total taxes paid across all client's accounts

Business Logic

  • taxes_paid is the sum of taxes_paid from all accounts belonging to the client's portfolios
  • One message generated per unique client in the clients CSV
  • If a client has no portfolios, taxes_paid will be 0

Portfolio Message

Generated once per portfolio with transaction metrics and account balance.

{
  "type": "portfolio_message",
  "portfolio_reference": "90755e32-7438-4354-ad37-ad900e297844",
  "cash_balance": 15000.00,
  "number_of_transactions": 15,
  "sum_of_deposits": 5000.00
}

Fields

Field Type Description
type string Always "portfolio_message"
portfolio_reference string Unique portfolio identifier (UUID)
cash_balance number Current cash balance from linked account
number_of_transactions integer Total count of all transactions
sum_of_deposits number Sum of all transactions with keyword "DEPOSIT"

Business Logic

  • cash_balance comes from the account linked to the portfolio
  • number_of_transactions counts all transaction records for the account
  • sum_of_deposits sums only transactions where keyword = "DEPOSIT"
  • One message per portfolio in the portfolios CSV

Error Message

Generated when data validation fails or expected relationships are missing.

{
  "type": "error_message",
  "client_reference": "f4a0cc2c-d0b4-4f14-b202-c8a5e45e90e7",
  "portfolio_reference": "439695b4-508d-4562-8576-670e70024627",
  "message": "Account 123456789 not found for portfolio"
}

Fields

Field Type Description
type string Always "error_message"
client_reference string or null Related client reference if known
portfolio_reference string or null Related portfolio reference if known
message string Human-readable error description

Error Scenarios

Error messages are generated for:

  1. Missing Client Reference

    • Portfolio references a client that doesn't exist in clients CSV
    {
  "type": "error_message",
  "client_reference": "unknown-client-id",
  "portfolio_reference": "port-123",
  "message": "Client reference unknown-client-id not found for portfolio"
}

  2. Missing Account

    • Portfolio references an account that doesn't exist in accounts CSV
    {
  "type": "error_message",
  "client_reference": "client-123",
  "portfolio_reference": "port-456",
  "message": "Account 987654321 not found for portfolio"
}

  3. Invalid Data Format

    • CSV record has invalid or missing required fields
    {
  "type": "error_message",
  "client_reference": "client-123",
  "portfolio_reference": null,
  "message": "Error processing client record: could not convert string to float: 'invalid'"
}

  4. Processing Errors

    • Unexpected errors during data processing
    {
  "type": "error_message",
  "client_reference": null,
  "portfolio_reference": null,
  "message": "Error processing transaction record: 'account_number'"
}

Message Processing

Message Order

  • Messages are sent to SQS in batches (up to 10 per batch)
  • Order is not guaranteed (Standard SQS queue)
  • All client messages typically sent first, then portfolio messages, then errors

Message Attributes

No custom message attributes are currently set. Each message only contains the JSON body.

Batch Processing

The processor sends messages in batches using send_message_batch:

  • Up to 10 messages per API call
  • Reduces SQS API costs
  • Failed messages in a batch are retried

Dead Letter Queue

Messages that fail processing after 3 attempts are moved to the DLQ:

  • Queue: {environment}-portfolio-messages-dlq
  • Retention: 14 days
  • Requires manual investigation and replay

Example Processing Output

For the example data provided, the complete output would be:

// Client message for Frida Muller
{
  "type": "client_message",
  "client_reference": "9e40659b-8b9f-4fc4-814b-5a7b5a23b64",
  "tax_free_allowance": 801,
  "taxes_paid": 0.00
}

// Client message for Fritz Maier
{
  "type": "client_message",
  "client_reference": "f4a0cc2c-d0b4-4f14-b202-c8a5e45e90e7",
  "tax_free_allowance": 0,
  "taxes_paid": 789.56
}

// Portfolio message for first portfolio
{
  "type": "portfolio_message",
  "portfolio_reference": "90755e32-7438-4354-ad37-ad900e297844",
  "cash_balance": 15000.00,
  "number_of_transactions": 1,
  "sum_of_deposits": 5000.00
}

// Portfolio message for second portfolio
{
  "type": "portfolio_message",
  "portfolio_reference": "439695b4-508d-4562-8576-670e70024627",
  "cash_balance": -56.00,
  "number_of_transactions": 1,
  "sum_of_deposits": 0.00
}

Consuming Messages

SQS Receive Example

aws sqs receive-message \
  --queue-url https://sqs.us-east-1.amazonaws.com/123456789/staging-portfolio-messages \
  --max-number-of-messages 10 \
  --wait-time-seconds 20

Python Consumer Example

import boto3
import json

sqs = boto3.client('sqs')
queue_url = 'your-queue-url'

while True:
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=20
    )

    if 'Messages' in response:
        for message in response['Messages']:
            body = json.loads(message['Body'])

            if body['type'] == 'client_message':
                process_client_message(body)
            elif body['type'] == 'portfolio_message':
                process_portfolio_message(body)
            elif body['type'] == 'error_message':
                handle_error_message(body)

            # Delete message after processing
            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=message['ReceiptHandle']
            )

Message Validation Schema

For consumers that want to validate messages:

from typing import Literal, Optional
from pydantic import BaseModel

class ClientMessage(BaseModel):
    type: Literal["client_message"]
    client_reference: str
    tax_free_allowance: float
    taxes_paid: float

class PortfolioMessage(BaseModel):
    type: Literal["portfolio_message"]
    portfolio_reference: str
    cash_balance: float
    number_of_transactions: int
    sum_of_deposits: float

class ErrorMessage(BaseModel):
    type: Literal["error_message"]
    client_reference: Optional[str]
    portfolio_reference: Optional[str]
    message: str

Testing Message Generation

Use the provided test data:

# Upload test files
make upload-test-data

# Or manually
aws s3 cp example_data/clients_20200130.csv s3://your-bucket/
aws s3 cp example_data/portfolios_20200130.csv s3://your-bucket/
aws s3 cp example_data/accounts_20200130.csv s3://your-bucket/
aws s3 cp example_data/transactions_20200130.csv s3://your-bucket/

# Check CloudWatch Logs
aws logs tail /aws/lambda/staging-csv-processor --follow

# Receive messages
aws sqs receive-message --queue-url your-queue-url --max-number-of-messages 10