mcpr_integrations/store/
event.rs

1//! Event types sent from the proxy hot path to the background storage writer.
2//!
3//! These are the "write side" of the storage engine. The proxy constructs events
4//! from MCP request/response data and sends them through an mpsc channel —
5//! never blocking the hot path.
6//!
7//! # Separation from McprEvent
8//!
9//! `mcpr-integrations` has its own `McprEvent` for cloud/stdout emission.
10//! `StoreEvent` is intentionally independent — different schema, different
11//! lifecycle, no coupling between crates. The conversion happens at the
12//! proxy's emit stage (`mcpr_core::proxy::pipeline::emit`) where both are
13//! built from the same locals.
14
15/// Top-level event enum sent through the storage channel.
16///
17/// The background writer dispatches on this to decide which table to write to
18/// and whether to INSERT, UPDATE, or close a session.
19#[derive(Debug)]
20pub enum StoreEvent {
21    /// A completed MCP request — one row in the `requests` table.
22    Request(RequestEvent),
23
24    /// A new MCP session from an `initialize` handshake — one row in `sessions`.
25    Session(SessionEvent),
26
27    /// A session was cleanly closed (transport disconnect).
28    /// Updates `ended_at` on the existing session row.
29    SessionClosed {
30        session_id: String,
31        /// Unix milliseconds (UTC) when the close was detected.
32        ended_at: i64,
33    },
34
35    /// A new `SchemaVersion` was created by the proxy's SchemaManager.
36    /// Writer UPSERTs `server_schema` and appends change rows.
37    SchemaVersion(SchemaVersionEvent),
38}
39
40/// One completed MCP request, ready to be written to the `requests` table.
41///
42/// Built from the same data the proxy already computes for logging and cloud
43/// emission (method, tool name, latency, status, sizes). No extra parsing needed.
44#[derive(Debug)]
45pub struct RequestEvent {
46    /// UUIDv7 generated by mcpr — time-ordered, globally unique.
47    /// Used for cloud sink correlation and as the primary lookup key.
48    pub request_id: String,
49
50    /// Unix milliseconds (UTC) when the request was received.
51    /// Millisecond resolution is sufficient for latency tracking and avoids
52    /// i64 overflow concerns that nanosecond timestamps would introduce.
53    pub ts: i64,
54
55    /// Proxy name from config (e.g., "api-server").
56    /// Tags every row so a shared database can hold data from multiple proxies.
57    pub proxy: String,
58
59    /// MCP session ID from the `mcp-session-id` header.
60    /// Nullable: pre-handshake probes or malformed clients may not have one.
61    /// Soft foreign key to `sessions.session_id` (no hard FK constraint —
62    /// avoids ordering edge cases in the async writer).
63    pub session_id: Option<String>,
64
65    /// MCP JSON-RPC method (e.g., "tools/call", "resources/read", "initialize").
66    /// Stored as-is from the protocol layer — no normalization.
67    pub method: String,
68
69    /// Tool name for `tools/call` requests, None for other methods.
70    /// Extracted from the JSON-RPC params by the proxy's MCP parser.
71    pub tool: Option<String>,
72
73    /// Wall-clock time from proxy receiving the request to getting the upstream response.
74    /// Includes network round-trip to upstream — this is what the AI client experiences.
75    /// Stored in microseconds for sub-millisecond precision.
76    pub latency_us: i64,
77
78    /// Whether the request succeeded, failed, or timed out.
79    pub status: RequestStatus,
80
81    /// MCP error code if `status` is `Error` (e.g., "-32600", "-32601").
82    /// None for successful requests.
83    pub error_code: Option<String>,
84
85    /// Human-readable error message, truncated to 512 chars at the call site.
86    /// None for successful requests.
87    pub error_msg: Option<String>,
88
89    /// Request payload size in bytes. None if not measured.
90    pub bytes_in: Option<i64>,
91
92    /// Response payload size in bytes. None if not measured.
93    pub bytes_out: Option<i64>,
94}
95
96/// Emitted once per MCP session, when the proxy intercepts the `initialize` handshake.
97///
98/// The `clientInfo` field in the MCP `initialize` request is the authoritative source
99/// of client identity — we don't guess from headers or user-agent strings.
100#[derive(Debug)]
101pub struct SessionEvent {
102    /// MCP session ID — primary key in the `sessions` table.
103    pub session_id: String,
104
105    /// Which proxy this session connected through.
106    pub proxy: String,
107
108    /// Unix milliseconds (UTC) of the `initialize` request.
109    pub started_at: i64,
110
111    /// Client name from `clientInfo.name` (e.g., "claude-desktop", "cursor").
112    /// None if the client omits `clientInfo` (non-compliant but possible).
113    pub client_name: Option<String>,
114
115    /// Client version from `clientInfo.version` (e.g., "1.2.0").
116    pub client_version: Option<String>,
117
118    /// Normalized platform identifier derived from client_name.
119    /// Values: "claude", "chatgpt", "vscode", "cursor", "unknown".
120    /// Normalization happens at write time via a static lookup table.
121    pub client_platform: Option<String>,
122}
123
124/// A new `SchemaVersion` produced by the proxy's `SchemaManager`.
125///
126/// The writer trusts the event: `SchemaManager` guarantees the payload
127/// differs from the previous version (unchanged ingests return `None`
128/// and produce no event). The writer only diffs against the prior row
129/// to populate `schema_changes`.
130#[derive(Debug)]
131pub struct SchemaVersionEvent {
132    pub ts: i64,
133    /// Proxy name from config (upstream identity).
134    pub proxy: String,
135    pub upstream_url: String,
136    pub method: String,
137    /// Merged `result` payload as JSON string.
138    pub payload: String,
139    /// SHA-256 hex digest of the canonical payload.
140    pub content_hash: String,
141}
142
143/// Outcome of a single MCP request.
144///
145/// Maps to the `status` CHECK constraint in the `requests` table:
146/// `CHECK (status IN ('ok', 'error', 'timeout'))`.
147#[derive(Debug, Clone, Copy, PartialEq, Eq)]
148pub enum RequestStatus {
149    /// Upstream returned a valid MCP response (may still contain application-level errors).
150    Ok,
151    /// Upstream returned an MCP error response (JSON-RPC error object).
152    Error,
153    /// The request timed out before the upstream responded.
154    Timeout,
155}
156
157impl RequestStatus {
158    /// SQL-safe string representation, matching the CHECK constraint values.
159    pub fn as_str(&self) -> &'static str {
160        match self {
161            RequestStatus::Ok => "ok",
162            RequestStatus::Error => "error",
163            RequestStatus::Timeout => "timeout",
164        }
165    }
166}
mcpr_integrations/store/event.rs

mcpr_integrations/store/
event.rs
