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

ssp2/
encode.rs
