choreo_content/
indexer.rs

1//! Synchronous client for the `acuity-index` WebSocket JSON-RPC API.
2//!
3//! This speaks the indexer's wire protocol directly (no `acuity-index-api-rs`
4//! dependency) using `tungstenite` in synchronous mode over a plain `TcpStream`
5//! — no async runtime, matching the crate's "only subxt uses the sidecar"
6//! threading rule.
7//!
8//! Queries are keyed by the declared query keys from `acuity.toml`. The common
9//! ones this crate builds — `item_id`, `account_id`, `ipfs_hash`, and the
10//! composite `item_id_revision_id` — are exposed as [`QueryKey`] constructors.
11//!
12//! The wire envelope is JSON-RPC 2.0. The `key` object is
13//! `{"type":"Custom","value":{"name":...,"kind":...,"value":...}}` where
14//! `kind` matches the `CustomValue` tag (`bytes32`, `u32`, `composite`, ...).
15
16use serde::Deserialize;
17use serde_json::Value;
18use std::time::Duration;
19use tungstenite::Message;
20use tungstenite::client::IntoClientRequest;
21use tungstenite::stream::MaybeTlsStream;
22
23use crate::ContentError;
24use crate::config::INDEXER_WS_URL;
25use crate::encode::{bytes_to_hex, hex_to_bytes};
26
27/// A query key for the indexer's `acuity_getEvents`.
28#[derive(Clone, Debug, PartialEq, Eq)]
29pub enum QueryKey {
30    /// Exact `item_id` (32 bytes) match.
31    ItemId([u8; 32]),
32    /// Exact `account_id` (32 bytes) match.
33    AccountId([u8; 32]),
34    /// Exact `ipfs_hash` (32 bytes) match.
35    IpfsHash([u8; 32]),
36    /// Composite `item_id` + `revision_id` match.
37    ItemRevision { item_id: [u8; 32], revision_id: u32 },
38    /// Raw custom key with an explicit name and kind/value (for the arbitrary
39    /// declared keys, e.g. `index`-style keys not wrapped by this crate).
40    Raw { name: String, value: Value },
41}
42
43impl QueryKey {
44    /// Serialize into the wire `CustomKey` object (`name` + flattened value).
45    fn to_custom_key(&self) -> Value {
46        match self {
47            QueryKey::ItemId(b) => custom_key("item_id", "bytes32", &Value::from(bytes_to_hex(b))),
48            QueryKey::AccountId(b) => {
49                custom_key("account_id", "bytes32", &Value::from(bytes_to_hex(b)))
50            }
51            QueryKey::IpfsHash(b) => {
52                custom_key("ipfs_hash", "bytes32", &Value::from(bytes_to_hex(b)))
53            }
54            QueryKey::ItemRevision {
55                item_id,
56                revision_id,
57            } => custom_key(
58                "item_id_revision_id",
59                "composite",
60                &Value::Array(vec![
61                    custom_scalar("bytes32", &Value::from(bytes_to_hex(item_id))),
62                    custom_scalar("u32", &Value::from(*revision_id)),
63                ]),
64            ),
65            QueryKey::Raw { name, value } => {
66                // A raw key carries an already-shaped CustomValue object.
67                let kind = value["kind"].clone();
68                let inner = value["value"].clone();
69                custom_key(name, kind.as_str().unwrap_or("bytes32"), &inner)
70            }
71        }
72    }
73}
74
75/// Build the wire `Key` object for `acuity_getEvents`/`acuity_subscribeEvents`.
76/// The indexer's `Key` enum is tagged `#[serde(tag = "type", content = "value")]`,
77/// so the `CustomKey` payload must be wrapped as `{"type":"Custom","value":…}`;
78/// sending the bare `CustomKey` fails to deserialize with `invalid_key`.
79fn wire_key(custom: &Value) -> Value {
80    serde_json::json!({ "type": "Custom", "value": custom })
81}
82
83/// Build the inner `CustomKey` object `{"name":..., "kind":..., "value":...}`.
84fn custom_key(name: &str, kind: &str, value: &Value) -> Value {
85    serde_json::json!({ "name": name, "kind": kind, "value": value })
86}
87
88/// Build a nested `CustomValue` object (used inside a composite).
89fn custom_scalar(kind: &str, value: &Value) -> Value {
90    serde_json::json!({ "kind": kind, "value": value })
91}
92
93/// A hydrated event returned by `acuity_getEvents`.
94#[derive(Clone, Debug, Deserialize, serde::Serialize)]
95#[serde(rename_all = "camelCase")]
96pub struct DecodedEvent {
97    pub block_number: u32,
98    pub event_index: u32,
99    /// Milliseconds since Unix epoch (from the block's `Timestamp::Now`).
100    pub timestamp: u64,
101    /// The decoded event object (`pallet_name`, `event_name`, `fields`, ...).
102    pub event: StoredEvent,
103}
104
105/// The decoded event shape. The indexer serializes this camelCase
106/// (`palletName`, `eventName`, …) per its documented `acuity_getEvents`
107/// response format, so we must mirror that here.
108#[derive(Clone, Debug, Deserialize, serde::Serialize)]
109#[serde(rename_all = "camelCase")]
110pub struct StoredEvent {
111    pub pallet_name: String,
112    pub event_name: String,
113    pub pallet_index: u8,
114    pub variant_index: u8,
115    pub event_index: u8,
116    /// Free-form field map; string keys map event params to their values.
117    pub fields: Value,
118}
119
120impl DecodedEvent {
121    /// The pallet name of this event.
122    #[must_use]
123    pub fn pallet_name(&self) -> &str {
124        &self.event.pallet_name
125    }
126    /// The event variant name.
127    #[must_use]
128    pub fn event_name(&self) -> &str {
129        &self.event.event_name
130    }
131    /// Look up a field by name, returning the JSON value.
132    #[must_use]
133    pub fn field(&self, name: &str) -> Option<&Value> {
134        self.event.fields.get(name)
135    }
136    /// Look up a string-valued field.
137    pub fn field_str(&self, name: &str) -> Option<&str> {
138        self.field(name).and_then(Value::as_str)
139    }
140    /// Look up a numeric field (accepts integer or numeric string; the indexer
141    /// renders some scalars as strings).
142    #[must_use]
143    pub fn field_u64(&self, name: &str) -> Option<u64> {
144        self.field(name)
145            .and_then(|v| v.as_u64().or_else(|| v.as_str()?.parse().ok()))
146    }
147}
148
149/// Paged result for `acuity_getEvents`.
150#[derive(Clone, Debug, Deserialize, serde::Serialize)]
151#[serde(rename_all = "camelCase")]
152pub struct GetEventsResult {
153    pub events: Vec<DecodedEvent>,
154}
155
156/// Status result for `acuity_indexStatus`.
157#[derive(Clone, Debug, Deserialize, serde::Serialize)]
158pub struct IndexStatusResult {
159    pub spans: Vec<Span>,
160}
161
162/// An indexed span (block range).
163#[derive(Clone, Debug, Deserialize, serde::Serialize)]
164pub struct Span {
165    pub start: u32,
166    pub end: u32,
167}
168
169/// Opened indexer connection (synchronous WebSocket).
170struct Connection {
171    ws: tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>,
172    next_id: u64,
173}
174
175impl Connection {
176    fn open() -> Result<Self, ContentError> {
177        let request = INDEXER_WS_URL
178            .into_client_request()
179            .map_err(|e| ContentError::Indexer(format!("invalid indexer url: {e}")))?;
180        let (ws, _resp) = tungstenite::connect(request)
181            .map_err(|e| ContentError::Indexer(format!("failed to connect to indexer: {e}")))?;
182        // Bound socket reads/writes so a hung indexer cannot block a daemon tool
183        // thread indefinitely. The indexer is reached over plain `ws://`, so
184        // the underlying stream is always a plain TcpStream (not TLS).
185        if let MaybeTlsStream::Plain(tcp) = ws.get_ref() {
186            let _ = tcp.set_read_timeout(Some(Duration::from_secs(15)));
187            let _ = tcp.set_write_timeout(Some(Duration::from_secs(10)));
188        }
189        Ok(Self { ws, next_id: 1 })
190    }
191
192    fn request(&mut self, method: &str, params: &Value) -> Result<Value, ContentError> {
193        let id = self.next_id;
194        self.next_id += 1;
195        let req =
196            serde_json::json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params });
197        self.ws
198            .send(Message::Text(req.to_string().into()))
199            .map_err(|e| ContentError::Indexer(format!("write failed: {e}")))?;
200        loop {
201            let msg = self
202                .ws
203                .read()
204                .map_err(|e| ContentError::Indexer(format!("read failed: {e}")))?;
205            match msg {
206                Message::Text(text) => {
207                    let v: Value = serde_json::from_str(&text)
208                        .map_err(|e| ContentError::Indexer(format!("bad json: {e}")))?;
209                    // Match only our request id (ignore unrelated notifications).
210                    if v.get("id").and_then(Value::as_u64) == Some(id) {
211                        return if v.get("error").is_some() {
212                            Err(ContentError::Indexer(format!(
213                                "indexer error for {method}: {v}"
214                            )))
215                        } else {
216                            Ok(v.get("result").cloned().unwrap_or_default())
217                        };
218                    }
219                }
220                Message::Ping(data) => {
221                    let _ = self.ws.send(Message::Pong(data));
222                }
223                Message::Close(_) => {
224                    return Err(ContentError::Indexer("indexer closed connection".into()));
225                }
226                _ => {}
227            }
228        }
229    }
230
231    fn close(&mut self) {
232        let _ = self.ws.close(None);
233    }
234}
235
236/// Query the indexer for events matching `key`, newest-first, up to `limit`.
237///
238/// # Errors
239///
240/// Fails with [`ContentError::Indexer`] when the WebSocket connection to the
241/// indexer cannot be opened, the JSON-RPC request write/read fails, the
242/// indexer closes the connection or returns a JSON-RPC error, or the result
243/// cannot be decoded as [`GetEventsResult`].
244pub fn get_events(
245    key: &QueryKey,
246    limit: u16,
247    before: Option<(u32, u32)>,
248) -> Result<Vec<DecodedEvent>, ContentError> {
249    let mut conn = Connection::open()?;
250    let key_json = wire_key(&key.to_custom_key());
251    let params = serde_json::json!({
252        "key": key_json,
253        "limit": limit,
254        "before": before.map(|(b, e)| serde_json::json!({ "blockNumber": b, "eventIndex": e })),
255    });
256    let result = conn.request("acuity_getEvents", &params)?;
257    conn.close();
258    let parsed: GetEventsResult = serde_json::from_value(result)
259        .map_err(|e| ContentError::Indexer(format!("failed to decode get_events result: {e}")))?;
260    Ok(parsed.events)
261}
262
263/// Query the current indexer status (indexed spans).
264///
265/// # Errors
266///
267/// Fails with [`ContentError::Indexer`] when the WebSocket connection or the
268/// `acuity_indexStatus` JSON-RPC round-trip fails, or the result cannot be
269/// decoded as [`IndexStatusResult`].
270pub fn index_status() -> Result<IndexStatusResult, ContentError> {
271    let mut conn = Connection::open()?;
272    let result = conn.request("acuity_indexStatus", &serde_json::json!({}))?;
273    conn.close();
274    serde_json::from_value(result)
275        .map_err(|e| ContentError::Indexer(format!("failed to decode index status: {e}")))
276}
277
278/// Convert a hex item id string (or `0x` hex) to a [`QueryKey::ItemId`].
279///
280/// # Errors
281///
282/// Fails with [`ContentError::Cid`] (via [`hex_to_bytes`]) when
283/// `item_id_hex` is not exactly 32 bytes of hex.
284pub fn item_id_key(item_id_hex: &str) -> Result<QueryKey, ContentError> {
285    Ok(QueryKey::ItemId(hex_to_bytes(item_id_hex)?))
286}
287
288/// Build a composite [`QueryKey::ItemRevision`] from an item id + revision id.
289///
290/// # Errors
291///
292/// Fails with [`ContentError::Cid`] (via [`hex_to_bytes`]) when
293/// `item_id_hex` is not exactly 32 bytes of hex.
294pub fn item_revision_key(item_id_hex: &str, revision_id: u32) -> Result<QueryKey, ContentError> {
295    Ok(QueryKey::ItemRevision {
296        item_id: hex_to_bytes(item_id_hex)?,
297        revision_id,
298    })
299}
300
301#[cfg(test)]
302mod tests {
303    use super::*;
304
305    #[test]
306    fn wire_key_wraps_custom_key() {
307        // The get_events params must carry the tagged `Key` shape the indexer
308        // deserializes, not the bare CustomKey.
309        let item_id = [0x11u8; 32];
310        let key = wire_key(&QueryKey::ItemId(item_id).to_custom_key());
311        assert_eq!(key["type"], "Custom");
312        assert_eq!(key["value"]["name"], "item_id");
313        assert_eq!(key["value"]["kind"], "bytes32");
314        assert_eq!(key["value"]["value"], bytes_to_hex(&item_id));
315    }
316
317    #[test]
318    fn query_key_wire_shapes() {
319        // item_id -> {"name":"item_id","kind":"bytes32","value":"0x..."}
320        let item_id = [0x11u8; 32];
321        let key = QueryKey::ItemId(item_id).to_custom_key();
322        assert_eq!(key["name"], "item_id");
323        assert_eq!(key["kind"], "bytes32");
324        assert_eq!(key["value"], bytes_to_hex(&item_id));
325
326        // composite revision key
327        let rev = QueryKey::ItemRevision {
328            item_id,
329            revision_id: 7,
330        }
331        .to_custom_key();
332        assert_eq!(rev["name"], "item_id_revision_id");
333        assert_eq!(rev["kind"], "composite");
334        assert_eq!(rev["value"][0]["kind"], "bytes32");
335        assert_eq!(rev["value"][1]["kind"], "u32");
336        assert_eq!(rev["value"][1]["value"], 7);
337    }
338
339    #[test]
340    fn decoded_event_field_helpers() {
341        let ev = DecodedEvent {
342            block_number: 1,
343            event_index: 2,
344            timestamp: 123,
345            event: StoredEvent {
346                pallet_name: "Content".into(),
347                event_name: "PublishRevision".into(),
348                pallet_index: 7,
349                variant_index: 3,
350                event_index: 2,
351                fields: serde_json::json!({
352                    "ipfs_hash": "0xaa",
353                    "revision_id": "7",
354                    "n": 42,
355                }),
356            },
357        };
358        assert_eq!(ev.pallet_name(), "Content");
359        assert_eq!(ev.event_name(), "PublishRevision");
360        assert_eq!(ev.field_str("ipfs_hash"), Some("0xaa"));
361        // revision_id arrives as a numeric string.
362        assert_eq!(ev.field_u64("revision_id"), Some(7));
363        assert_eq!(ev.field_u64("n"), Some(42));
364        assert_eq!(ev.field_u64("missing"), None);
365    }
366}
choreo_content/indexer.rs

choreo_content/
indexer.rs
