ssp2/
encode.rs

1//! Canonical SSP2 encoding (SPEC.md §1.2). Re-encoding a decoded message
2//! reproduces the input byte-for-byte: raw `json` strings, map entry order,
3//! and unknown frames are all preserved by the model.
4
5use crate::decode::{MINIMUM_WIRE_VERSION, SSP2_MAGIC, WIRE_VERSION};
6use crate::model::frame_type as ft;
7use crate::model::{Frame, MediaType, Message, MsgKind, Op, OpResult, PushStatus, SubStatus};
8use crate::primitives::Writer;
9
10pub fn encode_message(msg: &Message) -> Vec<u8> {
11    assert!(
12        (MINIMUM_WIRE_VERSION..=WIRE_VERSION).contains(&msg.wire_version),
13        "unsupported wireVersion {}",
14        msg.wire_version
15    );
16    let mut w = Writer::new();
17    w.raw(SSP2_MAGIC);
18    w.u16(msg.wire_version);
19    w.u8(match msg.msg_kind {
20        MsgKind::Request => 0x01,
21        MsgKind::Response => 0x02,
22    });
23    w.u8(0x00); // flags
24    for frame in &msg.frames {
25        let (ty, payload) = encode_frame(frame, msg.wire_version);
26        w.u8(ty);
27        w.u32(payload.len() as u32);
28        w.raw(&payload);
29    }
30    w.u8(ft::END);
31    w.u32(0);
32    w.into_bytes()
33}
34
35fn op_byte(op: Op) -> u8 {
36    match op {
37        Op::Upsert => 1,
38        Op::Delete => 2,
39    }
40}
41
42/// Every frame type with a defined layout in wire versions 1 and 2 (§1.2
43/// registry), `END` included.
44fn is_registered_frame_type(ty: u8) -> bool {
45    ty == ft::END || ft::REQUEST_TYPES.contains(&ty) || ft::RESPONSE_TYPES.contains(&ty)
46}
47
48fn encode_frame(frame: &Frame, wire_version: u16) -> (u8, Vec<u8>) {
49    let mut w = Writer::new();
50    let ty = match frame {
51        Frame::ReqHeader {
52            client_id,
53            schema_version,
54            log_epoch,
55        } => {
56            w.str(client_id);
57            w.i32(*schema_version);
58            if wire_version >= 2 {
59                assert!(log_epoch.as_ref().is_none_or(|value| !value.is_empty()));
60                w.opt(log_epoch, |w, value| w.str(value));
61            } else {
62                assert!(
63                    log_epoch.is_none(),
64                    "REQ_HEADER.logEpoch requires wireVersion 2"
65                );
66            }
67            ft::REQ_HEADER
68        }
69        Frame::PushCommit {
70            client_commit_id,
71            operations,
72        } => {
73            w.str(client_commit_id);
74            w.u32(operations.len() as u32);
75            for op in operations {
76                w.str(&op.table);
77                w.str(&op.row_id);
78                w.u8(op_byte(op.op));
79                w.opt(&op.base_version, |w, v| w.i64(*v));
80                w.opt(&op.payload, |w, p| w.bytes(p));
81            }
82            ft::PUSH_COMMIT
83        }
84        Frame::PullHeader {
85            limit_commits,
86            limit_snapshot_rows,
87            max_snapshot_pages,
88            accept,
89        } => {
90            w.i32(*limit_commits);
91            w.i32(*limit_snapshot_rows);
92            w.i32(*max_snapshot_pages);
93            w.u8(*accept);
94            ft::PULL_HEADER
95        }
96        Frame::Subscription {
97            id,
98            table,
99            scopes,
100            params,
101            cursor,
102            bootstrap_state,
103        } => {
104            w.str(id);
105            w.str(table);
106            w.scope_map(scopes);
107            w.opt(params, |w, j| w.str(&j.0));
108            w.i64(*cursor);
109            w.opt(bootstrap_state, |w, j| w.str(&j.0));
110            ft::SUBSCRIPTION
111        }
112        Frame::RespHeader {
113            required_schema_version,
114            latest_schema_version,
115            log_epoch,
116            reset_required,
117        } => {
118            w.opt(required_schema_version, |w, v| w.i32(*v));
119            w.opt(latest_schema_version, |w, v| w.i32(*v));
120            if wire_version >= 2 {
121                let value = log_epoch
122                    .as_ref()
123                    .filter(|value| !value.is_empty())
124                    .expect("RESP_HEADER.logEpoch is required and non-empty");
125                w.str(value);
126                w.bool(reset_required.expect("RESP_HEADER.resetRequired is required"));
127            } else {
128                assert!(
129                    log_epoch.is_none() && reset_required.is_none(),
130                    "RESP_HEADER epoch fields require wireVersion 2"
131                );
132            }
133            ft::RESP_HEADER
134        }
135        Frame::Lease {
136            lease_id,
137            expires_at_ms,
138        } => {
139            w.str(lease_id);
140            w.i64(*expires_at_ms);
141            ft::LEASE
142        }
143        Frame::PushResult {
144            client_commit_id,
145            status,
146            commit_seq,
147            results,
148        } => {
149            w.str(client_commit_id);
150            w.u8(match status {
151                PushStatus::Applied => 1,
152                PushStatus::Cached => 2,
153                PushStatus::Rejected => 3,
154            });
155            w.opt(commit_seq, |w, v| w.i64(*v));
156            w.u32(results.len() as u32);
157            for result in results {
158                match result {
159                    OpResult::Applied { op_index } => {
160                        w.i32(*op_index);
161                        w.u8(1);
162                    }
163                    OpResult::Conflict {
164                        op_index,
165                        code,
166                        message,
167                        server_version,
168                        server_row,
169                        conflict_columns,
170                    } => {
171                        w.i32(*op_index);
172                        w.u8(2);
173                        w.str(code);
174                        w.str(message);
175                        w.i64(*server_version);
176                        w.bytes(server_row);
177                        w.bytes(conflict_columns);
178                    }
179                    OpResult::Error {
180                        op_index,
181                        code,
182                        message,
183                        retryable,
184                    } => {
185                        w.i32(*op_index);
186                        w.u8(3);
187                        w.str(code);
188                        w.str(message);
189                        w.bool(*retryable);
190                    }
191                }
192            }
193            ft::PUSH_RESULT
194        }
195        Frame::PushResultDetails {
196            client_commit_id,
197            entries,
198        } => {
199            w.str(client_commit_id);
200            w.u32(entries.len() as u32);
201            for entry in entries {
202                w.i32(entry.op_index);
203                w.str(&entry.details.0);
204            }
205            ft::PUSH_RESULT_DETAILS
206        }
207        Frame::SubStart {
208            id,
209            status,
210            reason_code,
211            effective_scopes,
212            bootstrap,
213        } => {
214            w.str(id);
215            w.u8(match status {
216                SubStatus::Active => 1,
217                SubStatus::Revoked => 2,
218                SubStatus::Reset => 3,
219            });
220            w.str(reason_code);
221            w.scope_map(effective_scopes);
222            w.bool(*bootstrap);
223            ft::SUB_START
224        }
225        Frame::Commit {
226            commit_seq,
227            created_at_ms,
228            actor_id,
229            tables,
230            changes,
231        } => {
232            w.i64(*commit_seq);
233            w.i64(*created_at_ms);
234            w.str(actor_id);
235            w.u32(tables.len() as u32);
236            for table in tables {
237                w.str(table);
238            }
239            w.u32(changes.len() as u32);
240            for change in changes {
241                w.u16(change.table_index);
242                w.str(&change.row_id);
243                w.u8(op_byte(change.op));
244                w.opt(&change.row_version, |w, v| w.i64(*v));
245                w.str_map(&change.scopes);
246                w.opt(&change.row, |w, row| w.bytes(row));
247            }
248            ft::COMMIT
249        }
250        Frame::SegmentRef {
251            segment_id,
252            media_type,
253            table,
254            byte_length,
255            row_count,
256            as_of_commit_seq,
257            scope_digest,
258            row_cursor,
259            next_row_cursor,
260            url,
261            url_expires_at_ms,
262        } => {
263            w.str(segment_id);
264            w.u8(match media_type {
265                MediaType::Rows => 1,
266                MediaType::Sqlite => 2,
267            });
268            w.str(table);
269            w.i64(*byte_length);
270            w.i64(*row_count);
271            w.i64(*as_of_commit_seq);
272            w.str(scope_digest);
273            w.opt(row_cursor, |w, s| w.str(s));
274            w.opt(next_row_cursor, |w, s| w.str(s));
275            w.opt(url, |w, s| w.str(s));
276            w.opt(url_expires_at_ms, |w, v| w.i64(*v));
277            ft::SEGMENT_REF
278        }
279        Frame::SegmentInline { payload } => {
280            w.raw(payload);
281            ft::SEGMENT_INLINE
282        }
283        Frame::SubEnd {
284            next_cursor,
285            bootstrap_state,
286        } => {
287            w.i64(*next_cursor);
288            w.opt(bootstrap_state, |w, j| w.str(&j.0));
289            ft::SUB_END
290        }
291        Frame::Error {
292            code,
293            message,
294            category,
295            retryable,
296            recommended_action,
297            details,
298        } => {
299            w.str(code);
300            w.str(message);
301            w.str(category);
302            w.bool(*retryable);
303            w.str(recommended_action);
304            w.opt(details, |w, j| w.str(&j.0));
305            ft::ERROR
306        }
307        Frame::Unknown {
308            frame_type,
309            payload,
310        } => {
311            // §1.2: an encoder MUST NOT emit an unknown frame under a
312            // frameType registered in this wire version. Preserved unknown
313            // frames come only from decoding, which never produces such a
314            // frame — hitting this is encoder misuse, not a decode error.
315            assert!(
316                !is_registered_frame_type(*frame_type),
317                "UNKNOWN frame type 0x{frame_type:02x} collides with a registered frame type"
318            );
319            w.raw(payload);
320            *frame_type
321        }
322    };
323    (ty, w.into_bytes())
324}
ssp2/encode.rs

ssp2/
encode.rs
