bezant_server/events/
mod.rs

1//! Event capture + REST surface for streaming CPAPI topics.
2//!
3//! `bezant-server` runs an internal [`bezant::WsClient`] that subscribes
4//! to order, PnL, and (lazily) market-data feeds. Decoded frames go into
5//! per-topic ring buffers behind cursor-based REST endpoints
6//! (`/events/{topic}?since=<cursor>`), so polling consumers can read every
7//! event the socket has seen since their last visit — no events lost
8//! between client polls, no need to keep a long-lived WS open from the
9//! consumer side.
10//!
11//! See the architecture sketch:
12//!
13//! ```text
14//! CPGateway WS  ─► bezant_core::WsClient  ─► Connector (tokio task)
15//!                                              │
16//!                                              ▼
17//!                                       per-topic TopicRing
18//!                                              │
19//!                                              ▼
20//!                                  GET /events/{topic}?since=<cursor>
21//! ```
22
23pub mod connector;
24pub mod persistence;
25pub mod ring;
26
27use std::collections::BTreeMap;
28
29use serde::Serialize;
30
31pub use connector::{spawn_connector, ConnectorCfg, EventsHandle, TestSink};
32pub use persistence::{EventLog, RetentionPolicy};
33pub use ring::{ReadResult, TopicRing};
34
35/// One captured event. Wire-shape returned by `/events/{topic}` endpoints.
36#[derive(Clone, Debug, Serialize)]
37pub struct ObservedEvent {
38    /// Server-assigned monotonic cursor within `(topic, reset_epoch)`.
39    /// Use it as the `since=` parameter on the next poll.
40    pub cursor: u64,
41    /// Topic name — `"orders"`, `"pnl"`, `"marketdata:265598"`, `"gap"`.
42    pub topic: String,
43    /// RFC 3339 timestamp at which the connector pushed this into the ring.
44    pub received_at: String,
45    /// Increments every time the underlying WS reconnects or the server
46    /// restarts. Clients use it to detect "the cursor space reset under me".
47    pub reset_epoch: u64,
48    /// The decoded JSON frame. Shape depends on the topic — see the
49    /// generated `bezant-client` TS types for the typed views.
50    pub payload: serde_json::Value,
51}
52
53/// Snapshot of the connector's state, returned by `GET /events/_status`.
54#[derive(Clone, Debug, Default, Serialize)]
55pub struct EventsStatus {
56    /// `true` if the underlying WS is currently connected.
57    pub connected: bool,
58    /// RFC 3339 timestamp of the last frame the connector received, if any.
59    pub last_message_at: Option<String>,
60    /// How many times the connector has reconnected since the process
61    /// started. Bumps on every successful reconnect, not on each retry.
62    pub reconnect_count: u64,
63    /// Wall-clock seconds since the connector task spawned.
64    pub uptime_seconds: u64,
65    /// Current `reset_epoch` — bumps on each reconnect or process restart.
66    pub reset_epoch: u64,
67    /// Topics currently subscribed at the upstream WS. Always includes
68    /// `"orders"` and `"pnl"`; market data topics appear when a client
69    /// has polled `/events/marketdata?conid=…` recently.
70    pub topics_subscribed: Vec<String>,
71    /// Per-topic ring buffer occupancy. Useful for "are we close to
72    /// wraparound?" capacity planning.
73    pub buffer_sizes: BTreeMap<String, usize>,
74    /// Whether CPAPI has actually honoured each standing subscription
75    /// (`orders`, `pnl`). `topics_subscribed` above records what we ASKED
76    /// for; this records what we GOT. The difference is the whole story of
77    /// a fill that never arrived: CPAPI answers `sor+{}` with
78    /// `{"error":"unable to subscribe"}` more often than not, and a socket
79    /// that outlives a Gateway re-login keeps heartbeating with every
80    /// subscription dead. A consumer confirming fills should treat anything
81    /// but `subscribed` on `orders` as "the stream will not tell you".
82    pub subscriptions: BTreeMap<String, SubscriptionState>,
83    /// How many subscribe refusals CPAPI has sent since the process started.
84    pub subscribe_refusals: u64,
85    /// How many times the connector tore its socket down because the
86    /// Gateway's session id changed underneath it (a re-login).
87    pub session_rollovers: u64,
88}
89
90/// Where a standing subscription (`orders`, `pnl`) stands at the upstream WS.
91#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
92#[serde(rename_all = "snake_case")]
93pub enum SubscriptionState {
94    /// Subscribe sent; CPAPI has neither honoured nor refused it yet.
95    Pending,
96    /// CPAPI has sent at least one real frame on the topic this connection.
97    Subscribed,
98    /// CPAPI answered the subscribe with an error; the connector is retrying.
99    Refused,
100}
101
102/// Reason a synthetic [`ObservedEvent`] of topic `"gap"` was injected.
103/// Surfaced to the client so they know they missed something.
104#[derive(Clone, Copy, Debug, Serialize)]
105#[serde(rename_all = "snake_case")]
106pub enum GapReason {
107    /// The connector reconnected — anything that happened while
108    /// disconnected is permanently gone from the upstream feed.
109    ReconnectedAfterDisconnect,
110    /// Process restarted — ring buffers were freshly initialised.
111    ProcessRestart,
112}
bezant_server/events/mod.rs

bezant_server/events/
mod.rs
