Operation bulk recorder

[Ygohr]

Context

The transaction processing pipeline currently uses loop-based individual INSERT operations to persist operations to the database. Each operation INSERT incurs network overhead, transaction commit, and WAL write. For N operations, this results in O(N) database interactions, creating a significant performance bottleneck.

Problem/Motivation

The current architecture underutilizes PostgreSQL batch optimizations. Processing high volumes of operations results in:

• N round-trips per transaction (one per operation)
• N commits per transaction
• N WAL writes per transaction
• Throughput limited to ~1,000 ops/sec

This affects both latency and throughput for customers processing large batches of transactions.

Current Behavior

Message Queue --> Consumer --> Handler --> UseCase
                                              |
                                              v
                                       for each operation:
                                         repo.Create(op)
                                              |
                                              v
                                          PostgreSQL

INSERT op1 --> commit
INSERT op2 --> commit
INSERT op3 --> commit
...
INSERT opN --> commit

Result: N round-trips | N commits | N WAL writes | O(N) latency

Each operation is inserted individually, resulting in O(N) database interactions per message.

Goal

Replace loop-based individual INSERT operations with batch INSERT operations, reducing database interactions to O(1) per message:

Primary flow (batch):

Message Queue --> Consumer --> Handler --> UseCase
                                              |
                                              v
                                       repo.CreateBatch(operations[])
                                              |
                                              v
                                          PostgreSQL

INSERT INTO operations VALUES (op1), (op2), (op3), ..., (opN)
ON CONFLICT DO NOTHING
--> commit

Result: 1 round-trip | 1 commit | 1 WAL write | O(1) latency

Fallback flow (on batch failure):

CreateBatch(ops[]) --> Batch INSERT fails --> Fallback to individual INSERT
                                                      |
                                                      v
                                               for each op:
                                                 Create(op)
                                                 skip duplicates

Key deliverables:

• Add CreateBatch(ctx, operations[]) → (inserted, error) method to Operation Repository
• Modify CreateBalanceTransactionOperationsAsync use case to leverage batch method
• Implement fallback to individual INSERT on batch failure
• Use ON CONFLICT DO NOTHING for idempotency
• Sort operations by ID before INSERT for deadlock prevention

Success Metrics

• 100 operations processed in < 20ms (baseline ~100ms) — 5x improvement
• 1,000 operations processed in < 100ms (baseline ~1,000ms) — 10x improvement
• Throughput > 10,000 ops/sec (baseline ~1,000 ops/sec) — 10x improvement
• Zero breaking changes to message format or external APIs
• Fallback mechanism verified with integration tests
• Idempotency verified with retry tests