# Current System Architecture

This document explains the current `unrip` runtime at a high level:

- which components exist
- what responsibilities each component owns
- which signals move through Kafka/Redpanda
- what gets persisted
- which attributes are available in the stored data
- what kinds of analysis and operator workflows the data supports today

The system is intentionally bus-first and data-first:

- components are separate services
- the core trading flow moves through Kafka-compatible topics
- PostgreSQL is the durable analytical and operator history store
- some safety-critical operational state is also persisted in local/PVC-backed JSON files
- the browser reads only the operator dashboard backend, not Kafka or PostgreSQL directly

## Design Rules

The current architecture follows these rules:

- venue inputs, strategy, execution, alerting, persistence, and the dashboard are separate services
- the core trading path is event-driven, not direct service-to-service RPC
- raw and normalized events are both preserved
- decisions and execution attempts are auditable after the fact
- execution safety depends on persisted idempotency state, not Kafka consumer groups alone

## System At A Glance

```mermaid
flowchart LR
    subgraph External Sources
        NearWS["NEAR Intents solver relay websocket"]
        RefData["Kraken + CoinGecko"]
        Bridge["Bridge RPC"]
        Verifier["Verifier / NEAR RPC"]
        BtcObs["BTC address observer"]
    end

    subgraph Runtime Services
        Ingest["near-intents-ingest"]
        RefIngest["market-reference-ingest"]
        Liquidity["liquidity-manager"]
        Inventory["inventory-sync"]
        Strategy["strategy-engine"]
        Executor["trade-executor"]
        Sentinel["ops-sentinel"]
        History["history-writer"]
        Dashboard["operator-dashboard"]
    end

    Bus[("Redpanda / Kafka")]
    Postgres[("PostgreSQL")]
    Browser["Operator browser"]

    NearWS --> Ingest
    RefData --> RefIngest
    Bridge --> Liquidity
    Verifier --> Liquidity
    BtcObs --> Liquidity
    Verifier --> Inventory
    Bridge --> Inventory

    Ingest -->|"raw.near_intents.quote"| Bus
    Ingest -->|"norm.swap_demand"| Bus
    RefIngest -->|"ref.market_price"| Bus
    Liquidity -->|"ops.liquidity_action"| Bus
    Liquidity -->|"ops.funding_observation"| Bus
    Inventory -->|"state.intent_inventory"| Bus
    Strategy -->|"decision.trade_decision"| Bus
    Strategy -->|"cmd.execute_trade"| Bus
    Executor -->|"exec.trade_result"| Bus
    Sentinel -->|"ops.alert"| Bus

    Bus --> Inventory
    Bus --> Strategy
    Bus --> Executor
    Bus --> Sentinel
    Bus --> History
    Bus --> Dashboard

    History --> Postgres
    Postgres --> Dashboard
    Dashboard --> Browser
    Dashboard -. "/state + /healthz" .-> Ingest
    Dashboard -. "/state + /healthz" .-> RefIngest
    Dashboard -. "/state + /healthz" .-> Liquidity
    Dashboard -. "/state + /healthz" .-> Inventory
    Dashboard -. "/state + /healthz" .-> History
    Dashboard -. "/state + /healthz" .-> Sentinel
    Dashboard -. "/state + /healthz" .-> Strategy
    Dashboard -. "/state + /healthz" .-> Executor
```

## Components And Responsibilities

### `near-intents-ingest`

Responsibility:

- connect to the NEAR Intents websocket
- subscribe to quote traffic
- preserve the raw upstream payload
- normalize quote requests into the canonical `swap_demand` shape
- apply pair filtering before publishing

Produces:

- `raw.near_intents.quote`
- `norm.swap_demand`

### `market-reference-ingest`

Responsibility:

- fetch BTC/EUR reference prices
- prefer Kraken and fall back to CoinGecko
- publish the reference price used for valuation and strategy checks

Produces:

- `ref.market_price`

### `liquidity-manager`

Responsibility:

- track bridge deposit handles
- observe deposits and withdrawals
- observe pre-credit funding activity
- estimate and optionally submit withdrawals
- publish treasury and funding operations signals

Produces:

- `ops.liquidity_action`
- `ops.funding_observation`

### `inventory-sync`

Responsibility:

- combine verifier balances with deposit and withdrawal side inputs
- publish the current inventory snapshot
- separate spendable balances from pending inbound and pending outbound funds

Consumes:

- `ops.liquidity_action`
- `ops.funding_observation`

Produces:

- `state.intent_inventory`

### `strategy-engine`

Responsibility:

- consume normalized demand
- read the latest reference price and inventory snapshot
- evaluate whether a quote is actionable
- record why it skipped or accepted a quote
- emit an execution command when the strategy is armed and the quote passes checks

Consumes:

- `norm.swap_demand`
- `ref.market_price`
- `state.intent_inventory`

Produces:

- `decision.trade_decision`
- `cmd.execute_trade`

### `trade-executor`

Responsibility:

- consume execution commands
- enforce arming state
- enforce command idempotency
- submit a quote response back to the venue
- publish the execution result

Consumes:

- `cmd.execute_trade`

Produces:

- `exec.trade_result`

### `ops-sentinel`

Responsibility:

- watch price freshness, inventory freshness, funding progression, liquidity events, and execution outcomes
- turn those signals into operator alert transitions

Consumes:

- `ref.market_price`
- `state.intent_inventory`
- `ops.liquidity_action`
- `ops.funding_observation`
- `exec.trade_result`

Produces:

- `ops.alert`

### `history-writer`

Responsibility:

- consume the major event topics
- route each topic into a PostgreSQL history table
- compute portfolio metrics from the durable event history

Consumes:

- `raw.near_intents.quote`
- `norm.swap_demand`
- `ref.market_price`
- `state.intent_inventory`
- `ops.liquidity_action`
- `ops.funding_observation`
- `ops.alert`
- `decision.trade_decision`
- `cmd.execute_trade`
- `exec.trade_result`

Produces:

- rows in PostgreSQL event tables
- rows in `portfolio_metrics_snapshots`

### `operator-dashboard`

Responsibility:

- expose one browser-facing API and WebSocket endpoint
- read durable history and metrics from PostgreSQL
- read live service state from `/state` and `/healthz`
- consume a small live Kafka topic set to avoid polling in the browser
- fan live updates to the frontend

Consumes:

- PostgreSQL
- service control APIs
- `norm.swap_demand`
- `ref.market_price`
- `state.intent_inventory`
- `ops.alert`
- `exec.trade_result`

Produces:

- `/api/session`
- `/api/bootstrap`
- `/api/trades`
- `/api/control/...`
- `/ws`

## Topic Classes And Signals

The topic naming scheme is meaningful:

- `raw.*` means source-native truth
- `norm.*` means canonical normalized inputs
- `ref.*` means reference market data
- `state.*` means derived state snapshots
- `ops.*` means operational or treasury state transitions
- `decision.*` means strategy output explaining what the system thought
- `cmd.*` means requested execution work
- `exec.*` means execution outcome

| Topic | Producer | Primary meaning | Main consumers |
| --- | --- | --- | --- |
| `raw.near_intents.quote` | `near-intents-ingest` | Raw venue quote frame | `history-writer` |
| `norm.swap_demand` | `near-intents-ingest` | Canonical demand signal from quote requests | `strategy-engine`, `history-writer`, `operator-dashboard` |
| `ref.market_price` | `market-reference-ingest` | Reference BTC/EUR pricing for valuation and strategy | `strategy-engine`, `ops-sentinel`, `history-writer`, `operator-dashboard` |
| `state.intent_inventory` | `inventory-sync` | Current spendable and pending inventory state | `strategy-engine`, `ops-sentinel`, `history-writer`, `operator-dashboard` |
| `ops.liquidity_action` | `liquidity-manager` | Deposit handle, deposit, withdrawal, and treasury actions | `inventory-sync`, `ops-sentinel`, `history-writer` |
| `ops.funding_observation` | `liquidity-manager` | Pre-credit inbound funding observations | `inventory-sync`, `ops-sentinel`, `history-writer` |
| `ops.alert` | `ops-sentinel` | Raised or cleared operator alerts | `history-writer`, `operator-dashboard` |
| `decision.trade_decision` | `strategy-engine` | Strategy explanation for skip or action | `history-writer` |
| `cmd.execute_trade` | `strategy-engine` | Instruction for the executor | `trade-executor`, `history-writer` |
| `exec.trade_result` | `trade-executor` | Outcome of the execution attempt | `ops-sentinel`, `history-writer`, `operator-dashboard` |

## Signal Flow

This is the high-level flow from upstream quote to operator visibility:

```mermaid
flowchart TD
    A["Quote request arrives from NEAR Intents"] --> B["near-intents-ingest"]
    B --> C["raw.near_intents.quote"]
    B --> D["norm.swap_demand"]

    E["Reference price refresh"] --> F["market-reference-ingest"]
    F --> G["ref.market_price"]

    H["Bridge / verifier / BTC observer state"] --> I["liquidity-manager"]
    I --> J["ops.liquidity_action"]
    I --> K["ops.funding_observation"]
    J --> L["inventory-sync"]
    K --> L
    L --> M["state.intent_inventory"]

    D --> N["strategy-engine"]
    G --> N
    M --> N
    N --> O["decision.trade_decision"]
    N --> P["cmd.execute_trade"]

    P --> Q["trade-executor"]
    Q --> R["exec.trade_result"]

    G --> S["ops-sentinel"]
    M --> S
    J --> S
    K --> S
    R --> S
    S --> T["ops.alert"]

    C --> U["history-writer"]
    D --> U
    G --> U
    M --> U
    J --> U
    K --> U
    O --> U
    P --> U
    R --> U
    T --> U
    U --> V["PostgreSQL"]

    V --> W["operator-dashboard bootstrap reads"]
    D --> X["operator-dashboard live consumer"]
    G --> X
    M --> X
    T --> X
    R --> X
    W --> Y["Browser REST bootstrap"]
    X --> Z["Browser WebSocket live updates"]
```

## Sequence Diagrams

### Quote To Decision To Execution

```mermaid
sequenceDiagram
    participant Venue as NEAR Intents WS
    participant Ingest as near-intents-ingest
    participant Bus as Redpanda/Kafka
    participant Strategy as strategy-engine
    participant Executor as trade-executor
    participant Sentinel as ops-sentinel
    participant History as history-writer
    participant DB as PostgreSQL

    Venue->>Ingest: quote event frame
    Ingest->>Bus: raw.near_intents.quote
    Ingest->>Bus: norm.swap_demand

    Bus->>Strategy: norm.swap_demand
    Bus->>History: raw + norm events
    History->>DB: insert raw_near_intents_quotes / swap_demand_events

    Note over Strategy: Strategy already tracks latest ref.market_price and state.intent_inventory

    Strategy->>Bus: decision.trade_decision
    alt actionable and armed
        Strategy->>Bus: cmd.execute_trade
        Bus->>Executor: cmd.execute_trade
        Executor->>Venue: quote_response submission
        Venue-->>Executor: ack / response
        Executor->>Bus: exec.trade_result
    else rejected
        Note over Strategy: No command emitted
    end

    Bus->>Sentinel: exec.trade_result
    Sentinel->>Bus: ops.alert (if alert state changes)

    Bus->>History: decision / command / result / alert
    History->>DB: insert trade_decisions / execute_trade_commands / trade_execution_results / ops_alerts
```

### Dashboard Bootstrap And Live Updates

```mermaid
sequenceDiagram
    participant Browser
    participant Dashboard as operator-dashboard
    participant DB as PostgreSQL
    participant Services as internal /state + /healthz
    participant Bus as Redpanda/Kafka

    Browser->>Dashboard: GET /api/bootstrap
    Dashboard->>DB: load portfolio metric, price, inventory, quotes, trades, alerts
    Dashboard->>Services: GET /state and /healthz across services
    DB-->>Dashboard: durable history + metrics
    Services-->>Dashboard: live service state
    Dashboard-->>Browser: bootstrap JSON

    Browser->>Dashboard: WebSocket /ws
    Dashboard-->>Browser: session.ready + initial live state

    Bus->>Dashboard: norm.swap_demand
    Dashboard-->>Browser: quotes.recent

    Bus->>Dashboard: ref.market_price / state.intent_inventory / ops.alert / exec.trade_result
    Dashboard-->>Browser: status_bar.updated / alerts.updated
```

## Event Envelope

Every bus event shares the same outer shape:

```json
{
  "event_id": "string",
  "event_type": "string",
  "venue": "string",
  "source": "string|null",
  "schema_version": 1,
  "observed_at": "ISO-8601|null",
  "ingested_at": "ISO-8601",
  "payload": {},
  "raw": {}
}
```

This matters because it gives every downstream consumer a stable minimum contract:

- unique event identity
- event type
- venue/source attribution
- two timestamps
- the domain payload
- optional raw payload preservation

## What Is Persisted

There are three different persistence layers in the current system.

### 1. Kafka / Redpanda Retention

Kafka is the operational event backbone:

- it retains the event stream for replay and decoupling
- it is not the main analytical store
- it lets multiple consumers read the same event flow independently

What this layer is good for:

- fan-out
- short-horizon replay
- decoupled service design
- live dashboard updates without browser polling

### 2. PostgreSQL Durable History

PostgreSQL is the main durable queryable history layer for operator views and analysis.

All event tables share the same durable columns:

| Column | Meaning |
| --- | --- |
| `event_id` | Unique event id from the envelope |
| `topic` | Kafka topic name |
| `venue` | Venue or source family |
| `source` | Producing service or adapter |
| `event_type` | Domain event type |
| `observed_at` | When the upstream thing happened, if known |
| `ingested_at` | When this service emitted the event |
| `quote_id` | Quote correlation key when available |
| `pair` | Pair string when available |
| `decision_key` | Topic-specific correlation key such as decision id or command id |
| `payload` | The structured event payload |
| `raw` | The raw upstream payload when preserved |

#### `raw_near_intents_quotes`

Source topic:

- `raw.near_intents.quote`

Key payload attributes:

- `message`: original upstream frame

Useful for:

- schema drift inspection
- replay or re-normalization
- debugging unexpected parser behavior
- checking what fields the venue actually sent

#### `swap_demand_events`

Source topic:

- `norm.swap_demand`

Key payload attributes:

- `quote_id`
- `asset_in`
- `asset_out`
- `pair`
- `request_kind`
- `amount_in`
- `amount_out`
- `min_deadline_ms`

Useful for:

- quote frequency by time
- pair mix and direction analysis
- size distributions
- exact-in vs exact-out behavior
- quote freshness analysis

#### `market_price_events`

Source topic:

- `ref.market_price`

Key payload attributes:

- `price_id`
- `pair`
- `eur_per_btc`
- `eure_per_btc`
- `btc_per_eur`
- `btc_per_eure`
- `source_used`
- `fallback_in_use`
- `divergence_pct`
- `eure_per_eur_assumption`
- `kraken.price`
- `kraken.observed_at`
- `kraken.age_ms`
- `kraken.healthy`
- `kraken.error`
- `coingecko.price`
- `coingecko.observed_at`
- `coingecko.age_ms`
- `coingecko.healthy`
- `coingecko.error`

Useful for:

- valuation at a point in time
- price freshness tracking
- fallback source usage
- disagreement between Kraken and CoinGecko
- linking strategy decisions to the exact price snapshot they saw

#### `intent_inventory_snapshots`

Source topic:

- `state.intent_inventory`

Key payload attributes:

- `inventory_id`
- `account_id`
- `synced_at`
- `spendable`
- `pending_inbound`
- `pending_outbound`
- `deposits`
- `withdrawals`
- `reconciliation_status`

Deposit record attributes:

- `tx_hash`
- `chain`
- `asset_id`
- `amount`
- `address`
- `status`
- `decimals`

Withdrawal record attributes:

- `withdrawal_hash`
- `asset_id`
- `chain`
- `amount`
- `status`
- `address`

Useful for:

- spendable balance history
- pending transfer tracking
- treasury state over time
- deposit and withdrawal visibility
- linking a decision to the inventory snapshot it used

#### `liquidity_actions`

Source topic:

- `ops.liquidity_action`

Key payload attributes:

- `liquidity_action_id`
- `account_id`
- `action_type`
- `status`
- `chain`
- `asset_id`
- `details`

The `details` object varies by action type. Common examples include:

- refreshed deposit addresses
- observed deposit records
- tracked withdrawal objects
- changed withdrawal status objects
- submitted withdrawal details

Useful for:

- treasury workflow timelines
- deposit handle rotation history
- withdrawal lifecycle tracking
- identifying when operator actions changed state

#### `funding_observations`

Source topic:

- `ops.funding_observation`

Key payload attributes:

- `funding_observation_id`
- `account_id`
- `asset_id`
- `chain`
- `funding_handle`
- `source`
- `tx_hash`
- `status`
- `amount`
- `confirmations`
- `first_seen_at`
- `last_seen_at`
- `credited_at`
- `bridge_deposit_tx_hash`
- `bridge_status`
- `credit_correlation`

Useful for:

- pre-credit deposit detection
- funding aging and stuck funding analysis
- credit lag measurement
- correlation between observed chain transactions and bridge crediting

#### `ops_alerts`

Source topic:

- `ops.alert`

Key payload attributes:

- `alert_event_id`
- `alert_code`
- `status`
- `severity`
- `reason`
- `service_scope`
- `pair`
- `asset_id`
- `raised_at`
- `cleared_at`
- `details`

Useful for:

- alert timelines
- mean time to recovery
- stale data and treasury issue analysis
- operator incident review

#### `trade_decisions`

Source topic:

- `decision.trade_decision`

Key payload attributes:

- `decision_id`
- `quote_id`
- `pair`
- `direction`
- `request_kind`
- `decision`
- `decision_reason`
- `threshold_pct`
- `max_notional_eure`
- `strategy_armed`
- `assumptions.eure_per_eur`
- `price_freshness_ms`
- `inventory_freshness_ms`
- `reference_rate`
- `implied_rate`
- `gross_edge_pct`
- `inventory_asset`
- `inventory_required`
- `inventory_available`
- `inventory_id`
- `price_id`
- `eure_notional`
- `proposed_amount_in`
- `proposed_amount_out`
- `pending_inbound`

Useful for:

- understanding why quotes were skipped
- measuring how many opportunities reached the actionable state
- analyzing insufficient inventory vs stale data vs threshold rejects
- reconstructing the exact decision context used by the strategy

#### `execute_trade_commands`

Source topic:

- `cmd.execute_trade`

Key payload attributes:

- `command_id`
- `decision_id`
- `idempotency_key`
- `execution_key`
- `quote_id`
- `pair`
- `asset_in`
- `asset_out`
- `amount_in`
- `amount_out`
- `request_kind`
- `min_deadline_ms`
- `quote_output.amount_in`
- `quote_output.amount_out`
- `proposed_amount_in`
- `proposed_amount_out`

Useful for:

- seeing exactly what execution was asked to do
- duplicate detection analysis
- comparing command intent to result outcome
- joining strategy decisions to actual execution attempts

#### `trade_execution_results`

Source topic:

- `exec.trade_result`

Key payload attributes:

- `command_id`
- `decision_id`
- `idempotency_key`
- `execution_key`
- `quote_id`
- `pair`
- `status`
- `result_code`
- `note`
- `venue_response`
- `error`

Useful for:

- execution success and failure rates
- venue response analysis
- distinguishing disarmed rejects from submission failures
- command-to-result conversion tracking

#### `portfolio_metrics_snapshots`

This table is derived from durable history rather than copied directly from a Kafka event.

Top-level columns:

- `metric_id`
- `computed_at`
- `baseline_anchor_at`
- `baseline_status`
- `payload`

Key payload attributes:

- `metric_version`
- `baseline_status`
- `command_count`
- `result_count`
- `current_price.price_id`
- `current_price.observed_at`
- `current_price.eure_per_btc`
- `current_inventory.inventory_id`
- `current_inventory.synced_at`
- `current_inventory.btc_units`
- `current_inventory.btc`
- `current_inventory.eure_units`
- `current_inventory.eure`
- `current_portfolio_value_eure`
- `current_btc_mark_value_eure`
- `current_eure_cash_value_eure`
- `trade_pnl_eure`
- `mark_to_market_pnl_eure`
- `price_move_pnl_eure`
- `baseline_portfolio_value_eure_at_baseline_price`
- `baseline_portfolio_value_eure_at_current_price`
- `current_portfolio_value_eure_at_baseline_price`
- `inventory_delta.btc_units`
- `inventory_delta.btc`
- `inventory_delta.eure_units`
- `inventory_delta.eure`
- `baseline.anchor`
- `baseline.command_at`
- `baseline.price`
- `baseline.inventory`

Useful for:

- current operator PnL display
- separating trade contribution from price movement
- baseline vs current comparison
- current portfolio valuation using the current price snapshot

### 3. JSON / PVC-Backed Operational State

Some important state is persisted outside PostgreSQL because it is operational control state, not analytical history.

#### Strategy armed state

Stored in:

- `strategyStateDir/strategy-engine-control.json`

Attributes:

- `armed`
- `updated_at`

Purpose:

- preserve strategy arming/disarming across restarts

#### Executor armed state

Stored in:

- `executorStateDir/trade-executor-control.json`

Attributes:

- `armed`
- `updated_at`

Purpose:

- preserve executor arming/disarming across restarts

#### Executor command idempotency state

Stored in:

- `executorStateDir/trade-executor-commands.json`

Attributes per command id:

- `quote_id`
- `idempotency_key`
- `execution_key`
- `status`
- `updated_at`
- result or error metadata when available

Purpose:

- prevent duplicate execution across restart or replay scenarios

#### Liquidity manager state

Stored in:

- `liquidityStateDir/liquidity.json`

Key attributes:

- `paused`
- `funding_observer_paused`
- `withdrawals_frozen`
- `deposit_addresses`
- `deposits`
- `tracked_withdrawals`
- `supported_tokens`
- `funding_observations`
- `funding_observations_by_handle`
- `funding_visibility_by_asset`
- `uncredited_funding_total_by_asset`
- `credit_correlation`
- `observer_health`
- `last_refresh_at`
- `last_funding_observation_at`
- `funding_observer_last_refresh_at`
- `funding_observer_last_error`
- `last_error`
- `last_withdrawal_request`
- `last_withdrawal_result`
- `publish_count`
- `funding_publish_count`

Purpose:

- preserve treasury and observer operational state across restarts
- back service `/state` responses

## What The Data Lets You Do

With the currently persisted data, you can already answer questions like:

- How many quotes arrived, at what sizes, and in which direction?
- Which quotes were rejected because of stale price, stale inventory, insufficient inventory, pending deposits, unsupported pair, or max notional?
- Which decisions became actual commands?
- Which commands became successful submitted results?
- What was the reference price and inventory snapshot seen by the strategy when it made a decision?
- How often did the pricing source fall back from Kraken to CoinGecko?
- How much pre-credit funding was visible before bridge crediting happened?
- How long did funding remain in a pending or stuck state?
- What alert transitions happened and how long did they remain raised?
- What is the current portfolio value, and how much of the change looks like price move vs trade contribution?

## Current Limits

A few important limits are worth naming plainly:

- Kafka retention is not the long-term historical warehouse; PostgreSQL is the main durable query surface
- raw venue schemas can drift, so raw quote storage matters for parser changes
- execution results currently represent submission outcomes, not full realized settlement economics
- fee coverage is not yet complete enough to treat all PnL numbers as full realized net profit
- some safety-critical state lives in JSON/PVC-backed operational stores, not only in PostgreSQL

## Recommended Reading Order

If you want to trace the system from code:

1. `src/lib/config.mjs`
2. `src/venues/near-intents/ws.mjs`
3. `src/apps/strategy-engine.mjs`
4. `src/apps/trade-executor.mjs`
5. `src/apps/history-writer.mjs`
6. `src/core/history-records.mjs`
7. `src/lib/postgres.mjs`
8. `src/apps/operator-dashboard.mjs`