# PIX & SPB NATS JetStream Integration Guide

> **For PIX and SPB subsystem teams:** This document describes exactly what Core expects to receive and what it publishes, so your subsystems can integrate with full resilience.

**Last updated:** 2026-02-09

---

## Architecture Overview

```
┌─────────────────────────────────────────────────────────────────────┐
│                        NATS JetStream Cluster                       │
│              10.10.40.5:4222 / 10.10.40.7:4222 / 10.10.40.4:4222  │
│                                                                     │
│  Streams:                                                           │
│    MONETARIE_PIX  (monetarie.spi.*, monetarie.settlement.session.*)         │
│    MONETARIE_SPB  (monetarie.spb.*)                                      │
│    MONETARIE_NPC  (monetarie.npc.*)                                      │
│    MONETARIE_STA  (monetarie.sta.*)                                      │
└──────────────────────────┬──────────────────────────────────────────┘
                           │
            ┌──────────────┼──────────────┐
            │              │              │
     ┌──────▼──────┐ ┌────▼────┐ ┌──────▼──────┐
     │    Core     │ │   PIX   │ │    SPB      │
     │  (Gateway)  │ │ (SPI)   │ │ (Messaging) │
     └─────────────┘ └─────────┘ └─────────────┘
```

**Core publishes** on `ledger.*` subjects (outbound to subsystems).
**Core consumes** from `monetarie.*` subjects (inbound from subsystems).

---

## 1. What PIX Subsystem Must Implement

### 1.1 Streams PIX Must Create

Create the `MONETARIE_PIX` stream in JetStream with these subjects:

```bash
nats stream add MONETARIE_PIX \
  --subjects "monetarie.spi.transaction.*,monetarie.spi.payment.*,monetarie.spi.return.*,monetarie.settlement.session.*" \
  --retention limits \
  --max-msgs -1 \
  --max-bytes -1 \
  --max-age 72h \
  --storage file \
  --replicas 3 \
  --discard old
```

### 1.2 Subjects PIX Must Publish To

Core's `PixConsumer` (durable: `core-pix-consumer`) listens on these subjects:

| Subject | Handler | What Core Does |
|---------|---------|---------------|
| `monetarie.spi.transaction.*` | `PixHandler.handle_transaction` | Audit + COSIF + TigerBeetle transfer + balance update |
| `monetarie.spi.payment.*` | `PixHandler.handle_payment` | Audit + COSIF + TigerBeetle transfer + balance update |
| `monetarie.spi.return.*` | `PixHandler.handle_return` | Audit + COSIF + TigerBeetle transfer + balance update |
| `monetarie.settlement.session.credit` | `SettlementHandler.handle_settlement_credit` | Deposits each credit into target accounts via TigerBeetle |
| `monetarie.settlement.session.*` (other) | `PixHandler.handle_settlement` | Audit log only |

#### 1.2.1 Transaction Payload (`monetarie.spi.transaction.*`)

Publish when a PIX transaction status changes (confirmed, rejected, returned, etc).

```json
{
  "transaction_id": "E00000001202602091200ABC123",
  "type": "credit",
  "direction": "inbound",
  "amount": 15000,
  "account_id": "100",
  "end_to_end_id": "E00000001202602091200ABC123",
  "status": "confirmed",
  "payer_name": "Maria Santos",
  "payer_document": "98765432100",
  "description": "Pagamento PIX"
}
```

**Required fields:**
| Field | Type | Description |
|-------|------|-------------|
| `transaction_id` | string | Unique transaction identifier (used for idempotency) |
| `type` | string | `"credit"` or `"debit"` — determines deposit vs withdrawal in TigerBeetle |
| `direction` | string | `"inbound"` or `"outbound"` |
| `amount` | integer | Amount in **centavos** (R$ 150.00 = 15000) |
| `account_id` | string/integer | TigerBeetle account ID (e.g., `"100"` for joao.silva) |

**Optional fields:** `end_to_end_id`, `status`, `payer_name`, `payer_document`, `description`

**How Core processes it:**
1. Creates audit log (`PIX_TRANSACTION`)
2. Creates COSIF journal entry (transfer code 6 for PIX)
3. Executes TigerBeetle deposit (if `type` contains "credit"/"inbound") or withdrawal (if "debit"/"outbound")
4. Publishes `ledger.balance.updated` with the account_id

#### 1.2.2 Payment Payload (`monetarie.spi.payment.*`)

Same structure as transaction. Used for payment confirmations.

#### 1.2.3 Return Payload (`monetarie.spi.return.*`)

```json
{
  "return_id": "D00000001202602091200RET456",
  "type": "credit",
  "direction": "inbound",
  "amount": 15000,
  "account_id": "100",
  "end_to_end_id": "E00000001202602091200ABC123",
  "original_end_to_end_id": "E00000001202602091200ABC123",
  "status": "returned"
}
```

Uses `return_id` instead of `transaction_id` for idempotency key.

#### 1.2.4 Settlement Credit Payload (`monetarie.settlement.session.credit`)

Publish when CLST settlement credits are confirmed for user accounts.

```json
{
  "session_id": "SETT-2026-02-09-001",
  "type": "settlement_credit",
  "total_credits": 3,
  "total_amount": 500000,
  "credits": [
    {
      "account_id": "100",
      "amount": 200000,
      "reference_id": "SETT-REF-001"
    },
    {
      "account_id": "101",
      "amount": 150000,
      "reference_id": "SETT-REF-002"
    },
    {
      "account_id": "102",
      "amount": 150000,
      "reference_id": "SETT-REF-003"
    }
  ]
}
```

**How Core processes it:**
- Iterates over `credits[]` array
- For each credit: deposits `amount` into `account_id` via TigerBeetle (transfer code 4 = settlement_credit)
- Creates COSIF journal entry per credit
- Uses deterministic TX ID: `phash2({"settlement", session_id, account_id})`

### 1.3 DICT Lookup — Request/Reply

Core sends DICT lookups via NATS request/reply (NOT pub/sub).

**Subject:** `monetarie.spi.dict.lookup`

**PIX must listen and reply on this subject.**

**Request payload from Core:**
```json
{
  "key": "12345678901",
  "key_type": "cpf",
  "merchant_id": "1",
  "requested_at": "2026-02-09T12:00:00Z"
}
```

Key types: `"cpf"`, `"cnpj"`, `"email"`, `"phone"`, `"evp"`

**Expected reply from PIX (JSON):**
```json
{
  "found": true,
  "account": {
    "participant_ispb": "00000000",
    "branch": "0001",
    "account_number": "123456",
    "account_type": "CACC",
    "holder_name": "João da Silva",
    "holder_document": "12345678901",
    "holder_type": "NATURAL_PERSON"
  },
  "key": {
    "key_type": "cpf",
    "key_value": "12345678901",
    "created_at": "2025-01-15T10:00:00Z"
  }
}
```

**Timeout:** Core waits 5 seconds. If no reply, returns HTTP 504 to the user.

### 1.4 Subjects PIX Must Consume From

Core publishes PIX transaction requests on these subjects (via Oban outbox):

| Subject | When Published | What PIX Should Do |
|---------|---------------|-------------------|
| `ledger.transaction.pix` | User initiates PIX send | Process the PIX payment via SPI |
| `ledger.transaction.pix_payment` | User initiates PIX out | Initiate payment request via SPI |

**Payload from Core (`ledger.transaction.pix`):**
```json
{
  "type": "pix_transaction",
  "data": {
    "transaction_id": "PIXOUT20260209000000001",
    "end_to_end_id": "E00000001202602091200000001",
    "type": "pix_out",
    "direction": "outbound",
    "amount": 10000,
    "account_id": "100",
    "hold_id": "a1b2c3d4e5f6",
    "recipient_key": "12345678901",
    "recipient_key_type": "cpf",
    "description": "PIX enviado",
    "sender_id": "1",
    "status": "processing",
    "initiated_at": "2026-02-09T12:00:00Z"
  },
  "published_at": "2026-02-09T12:00:01Z",
  "source": "core"
}
```

**What PIX should do:**
1. Process via SPI (DICT lookup, payment initiation)
2. On success: publish `monetarie.spi.transaction.confirmed` with result
3. On failure: publish `monetarie.spi.transaction.failed` with error

---

## 2. What SPB Subsystem Must Implement

### 2.1 Streams SPB Must Create

```bash
nats stream add MONETARIE_SPB \
  --subjects "monetarie.spb.messages.*,monetarie.spb.transactions.*" \
  --retention limits \
  --max-age 72h \
  --storage file \
  --replicas 3
```

### 2.2 Subjects SPB Must Publish To

Core's `SpbConsumer` (durable: `core-spb-consumer`) listens on:

| Subject | Handler | What Core Does |
|---------|---------|---------------|
| `monetarie.spb.messages.*` | `SpbHandler.handle_message_confirmed` | Audit log only |
| `monetarie.spb.transactions.*` | `SpbHandler.handle_transaction_updated` | Audit + COSIF + TigerBeetle + balance update |

#### 2.2.1 Transaction Updated (`monetarie.spb.transactions.*`)

```json
{
  "transaction_id": "TED20260209000000001",
  "type": "credit",
  "direction": "inbound",
  "amount": 50000,
  "account_id": "100",
  "message_type": "STR0004",
  "status": "confirmed"
}
```

**Transfer codes used:** Code 1 (deposit) for inbound, Code 8 (TED) for outbound.

#### 2.2.2 Message Confirmed (`monetarie.spb.messages.*`)

```json
{
  "message_id": "MSG20260209000001",
  "message_type": "STR0004",
  "status": "confirmed",
  "timestamp": "2026-02-09T12:00:00Z"
}
```

Audit log only — no ledger activity.

### 2.3 Subjects SPB Must Consume From

| Subject | When Published | What SPB Should Do |
|---------|---------------|-------------------|
| `ledger.transaction.spb` | User initiates TED transfer | Process the TED via SPB |

**Payload:**
```json
{
  "type": "spb_transaction",
  "data": {
    "type": "spb_transaction",
    "transaction_type": "TED",
    "transaction_id": "TED20260209000000001",
    "amount": 50000,
    "account_id": "100",
    "hold_id": "a1b2c3d4",
    "sender_id": "1",
    "recipient_ispb": "00000000",
    "recipient_name": "Maria Santos",
    "recipient_document": "98765432100",
    "recipient_agency": "0001",
    "recipient_account": "654321",
    "description": "Transferencia TED",
    "status": "processing",
    "initiated_at": "2026-02-09T12:00:00Z"
  },
  "published_at": "2026-02-09T12:00:01Z",
  "source": "core"
}
```

---

## 3. Resilience Requirements

### 3.1 Idempotency (Critical)

Core handles duplicate messages via three layers. **Your subsystems must set proper IDs to enable this:**

| Layer | Mechanism | Requirement |
|-------|-----------|-------------|
| **TigerBeetle** | Rejects duplicate transfer IDs (`:exists` = success) | Always include `transaction_id` / `return_id` in payload |
| **COSIF** | Unique index on `(reference_type, reference_id)` | Same reference = no-op on retry |
| **Deterministic TX ID** | `phash2({"monetarie_core", reference})` | Same message reference → same TB transfer ID |

**Rule:** If you retry a message with the same `transaction_id`, Core will process it as a no-op (idempotent). This is safe for at-least-once delivery.

### 3.2 JetStream ACK Protocol

Core uses **explicit ACK** with `max_deliver: 4` (3 retries).

- On successful processing: Core ACKs the message (`+ACK` to reply_to)
- On failure: Core does NOT NAK — message expires and is redelivered by JetStream
- After 4 failed deliveries: Message goes to DLQ (if configured) or is dropped

**Your subsystem should also use explicit ACK when consuming from `ledger.*` subjects.**

### 3.3 Outbox Pattern (Core → Subsystems)

Core uses an Oban-backed outbox for all `ledger.*` publishes:

1. Controller inserts Oban job with subject + payload into PostgreSQL
2. `PublishJob` worker picks it up (queue: `nats_publish`, 10 concurrent)
3. Publishes via `Gnat.pub/3`
4. If NATS is down: Oban retries with exponential backoff (1s→512s, 10 attempts)

**Guarantee:** At-least-once delivery from Core to NATS. Your consumer must be idempotent.

### 3.4 Connection Resilience

- Core auto-reconnects to NATS every 5 seconds on connection loss
- JetStream durable consumers resume from last ACKed position on restart
- If JetStream is unavailable, consumers fall back to plain NATS subscriptions (no replay guarantee)

---

## 4. TigerBeetle Account IDs

These are the seeded accounts Core uses:

| ID | Type | Code | Purpose |
|----|------|------|---------|
| 1 | System | 10 (cash_asset) | Cash asset account |
| 2 | System | 40 (revenue_equity) | Fee collection / revenue |
| 3 | System | 10 | Provider settlement pool |
| 5 | System | 50 (capital_equity) | System capital |
| 100 | Client | 30 (client_liability) | joao.silva |
| 101 | Client | 30 | maria.santos |
| 102 | Client | 30 | pedro.oliveira |
| 103 | Client | 30 | ana.costa |
| 104 | Client | 30 | carlos.ferreira |

**When publishing events, use the client account ID (100-104) as `account_id`.**

### Transfer Codes

| Code | Type | Used When |
|------|------|-----------|
| 1 | deposit | Inbound credit to client account |
| 2 | withdrawal | Outbound debit from client account |
| 3 | internal_transfer | Between two client accounts |
| 4 | settlement_credit | CLST settlement credit |
| 5 | fee | Fee deduction |
| 6 | pix | PIX-specific transfer |
| 7 | boleto | Boleto payment |
| 8 | ted | TED transfer |

---

## 5. Testing Integration

### 5.1 Test PIX → Core

```bash
# Publish a test PIX transaction to NATS
nats pub monetarie.spi.transaction.confirmed '{
  "transaction_id": "TEST-PIX-001",
  "type": "credit",
  "direction": "inbound",
  "amount": 1000,
  "account_id": "100",
  "end_to_end_id": "E00000001202602091200TEST001",
  "status": "confirmed"
}'
```

**Verify in Core logs:**
- `[PIX Consumer] Processing PIX event: monetarie.spi.transaction.confirmed`
- `[PIX Handler] Processing transaction: TEST-PIX-001`
- COSIF journal entry created
- TigerBeetle transfer executed
- `ledger.balance.updated` published

### 5.2 Test Core → PIX

```bash
# Subscribe to see Core's outbound messages
nats sub "ledger.transaction.pix"
```

Then send a PIX via the Core API:
```bash
curl -X POST https://coreapi-dev.fluxiq.com.br/api/v2/merchants/1/pix/send \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"amount": 1000, "recipient_key": "98765432100", "description": "Test PIX"}'
```

### 5.3 Test DICT Lookup

```bash
# Subscribe as PIX (simulated)
nats sub monetarie.spi.dict.lookup --reply '{"found":true,"account":{"holder_name":"Test User"}}'
```

Then from Core API:
```bash
curl "https://coreapi-dev.fluxiq.com.br/api/v2/merchants/1/pix/dict-lookup?key=12345678901"
```

### 5.4 Test Settlement Credit

```bash
nats pub monetarie.settlement.session.credit '{
  "session_id": "TEST-SETT-001",
  "type": "settlement_credit",
  "credits": [
    {"account_id": "100", "amount": 5000, "reference_id": "REF-001"},
    {"account_id": "101", "amount": 3000, "reference_id": "REF-002"}
  ]
}'
```

---

## 6. Checklist for PIX Integration

- [ ] Create `MONETARIE_PIX` JetStream stream with correct subjects
- [ ] Consume `ledger.transaction.pix` and `ledger.transaction.pix_payment` from Core
- [ ] Publish `monetarie.spi.transaction.*` on transaction status changes
- [ ] Publish `monetarie.spi.payment.*` on payment confirmations
- [ ] Publish `monetarie.spi.return.*` on return/refund events
- [ ] Publish `monetarie.settlement.session.credit` with credits array
- [ ] Implement `monetarie.spi.dict.lookup` request/reply handler
- [ ] Include `transaction_id` (or `return_id`) in every payload for idempotency
- [ ] Use explicit JetStream ACK on successful processing
- [ ] Test with deterministic IDs to verify idempotent retry handling
- [ ] Amounts always in **centavos** (integer)

## 7. Checklist for SPB Integration

- [ ] Create `MONETARIE_SPB` JetStream stream with correct subjects
- [ ] Consume `ledger.transaction.spb` from Core
- [ ] Publish `monetarie.spb.transactions.*` on TED/STR status changes
- [ ] Publish `monetarie.spb.messages.*` on message confirmations
- [ ] Include `transaction_id` or `message_id` in every payload
- [ ] Use explicit JetStream ACK
- [ ] Amounts in centavos