Transfers

message CreateCashMovementRequest {
  string idempotency_key;
  oneof intent { Transfer transfer; }
}

message Transfer {
  TransferReason reason;          // DEPOSIT | WITHDRAWAL | VENDOR_FEES
  string participant_account_id;  // the participant account — the other side is always your funding account
  string amount;                  // exact decimal
  string currency;                // ISO 4217, must be USD
  string external_reference;      // your reference (required)
  string memo;                    // free text (optional)
}

enum TransferReason {
  TRANSFER_REASON_UNSPECIFIED = 0;
  TRANSFER_REASON_DEPOSIT = 1;
  TRANSFER_REASON_WITHDRAWAL = 2;
  TRANSFER_REASON_VENDOR_FEES = 3;
}

message CreateCashMovementResponse { CashMovement cash_movement; }

message GetCashMovementRequest { string workflow_id; }
message GetCashMovementResponse { CashMovement cash_movement; }

message CashMovement {
  string workflow_id;
  string idempotency_key;
  string intent_type;             // "transfer"
  CashMovementStatus status;      // PENDING, CONFIRMED, REJECTED, AMBIGUOUS
  string amount;
  string currency;
  string source_account_id;       // derived from the reason
  string destination_account_id;  // derived from the reason
  string dco_transfer_id;
  string rejection_reason;
  google.protobuf.Timestamp created_at;
  google.protobuf.Timestamp updated_at;
  TransferReason reason;
}

# Paced drain: one dispatcher, 4 transfers/second, deposits first
import time, queue

deposit_q, batch_q = queue.Queue(), queue.Queue()

def dispatch_loop(stub, metadata):
    while True:
        item = None
        try:
            item = deposit_q.get_nowait()          # deposits preempt
        except queue.Empty:
            try:
                item = batch_q.get(timeout=1.0)    # then batch work (fees, withdrawals)
            except queue.Empty:
                continue
        try:
            stub.CreateCashMovement(item.request, metadata=metadata)
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.RESOURCE_EXHAUSTED:
                item.backoff = min(item.backoff * 2, 60)
                schedule_retry(item, delay=item.backoff * jitter())  # same idempotency_key
            else:
                raise
        time.sleep(0.25)                           # 4/s, headroom under the 5/s cap