shrike 0.1.1

AT Protocol library for Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

use crate::cbor::Cid;
use crate::syntax::{Did, Nsid, RecordKey};
use serde::Deserialize;

use crate::streaming::StreamError;

/// Jetstream event (JSON protocol — separate from CBOR firehose).
#[derive(Debug)]
pub enum JetstreamEvent {
    Commit {
        did: Did,
        time_us: i64,
        collection: Nsid,
        rkey: RecordKey,
        operation: JetstreamCommit,
    },
    Identity {
        did: Did,
        time_us: i64,
    },
    Account {
        did: Did,
        time_us: i64,
        active: bool,
    },
}

/// The commit operation for a Jetstream commit event.
#[derive(Debug)]
pub enum JetstreamCommit {
    Create { cid: Cid, record: serde_json::Value },
    Update { cid: Cid, record: serde_json::Value },
    Delete,
}

// ---------------------------------------------------------------------------
// Internal serde types for JSON parsing
// ---------------------------------------------------------------------------

#[derive(Deserialize)]
pub(crate) struct RawJetstreamEvent {
    pub did: String,
    pub time_us: i64,
    pub kind: String,
    pub commit: Option<RawCommit>,
    pub account: Option<RawAccount>,
}

#[derive(Deserialize)]
pub(crate) struct RawCommit {
    pub operation: String,
    pub collection: String,
    pub rkey: String,
    pub cid: Option<String>,
    pub record: Option<serde_json::Value>,
}

#[derive(Deserialize)]
pub(crate) struct RawAccount {
    pub active: bool,
}

// ---------------------------------------------------------------------------
// Parsing
// ---------------------------------------------------------------------------

/// Parse a single Jetstream JSON message into a [`JetstreamEvent`].
pub fn parse_jetstream_message(json: &str) -> Result<JetstreamEvent, StreamError> {
    let raw: RawJetstreamEvent =
        serde_json::from_str(json).map_err(|e| StreamError::ParseJson(e.to_string()))?;

    let did = Did::try_from(raw.did.as_str())
        .map_err(|e| StreamError::ParseJson(format!("invalid DID: {e}")))?;

    match raw.kind.as_str() {
        "commit" => {
            let commit = raw
                .commit
                .ok_or_else(|| StreamError::ParseJson("commit kind missing commit field".into()))?;

            let collection = Nsid::try_from(commit.collection.as_str())
                .map_err(|e| StreamError::ParseJson(format!("invalid collection NSID: {e}")))?;

            let rkey = RecordKey::try_from(commit.rkey.as_str())
                .map_err(|e| StreamError::ParseJson(format!("invalid rkey: {e}")))?;

            let operation = match commit.operation.as_str() {
                "create" => {
                    let cid_str = commit.cid.ok_or_else(|| {
                        StreamError::ParseJson("create commit missing cid".into())
                    })?;
                    let cid = cid_str
                        .parse::<Cid>()
                        .map_err(|e| StreamError::ParseJson(format!("invalid CID: {e}")))?;
                    let record = commit.record.ok_or_else(|| {
                        StreamError::ParseJson("create commit missing record".into())
                    })?;
                    JetstreamCommit::Create { cid, record }
                }
                "update" => {
                    let cid_str = commit.cid.ok_or_else(|| {
                        StreamError::ParseJson("update commit missing cid".into())
                    })?;
                    let cid = cid_str
                        .parse::<Cid>()
                        .map_err(|e| StreamError::ParseJson(format!("invalid CID: {e}")))?;
                    let record = commit.record.ok_or_else(|| {
                        StreamError::ParseJson("update commit missing record".into())
                    })?;
                    JetstreamCommit::Update { cid, record }
                }
                "delete" => JetstreamCommit::Delete,
                other => {
                    return Err(StreamError::ParseJson(format!(
                        "unknown commit operation: {other:?}"
                    )));
                }
            };

            Ok(JetstreamEvent::Commit {
                did,
                time_us: raw.time_us,
                collection,
                rkey,
                operation,
            })
        }
        "identity" => Ok(JetstreamEvent::Identity {
            did,
            time_us: raw.time_us,
        }),
        "account" => {
            let account = raw.account.ok_or_else(|| {
                StreamError::ParseJson("account kind missing account field".into())
            })?;
            Ok(JetstreamEvent::Account {
                did,
                time_us: raw.time_us,
                active: account.active,
            })
        }
        other => Err(StreamError::ParseJson(format!(
            "unknown event kind: {other:?}"
        ))),
    }
}