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::PushResultDetails {
164            client_commit_id,
165            entries,
166        } => {
167            w.str(client_commit_id);
168            w.u32(entries.len() as u32);
169            for entry in entries {
170                w.i32(entry.op_index);
171                w.str(&entry.details.0);
172            }
173            ft::PUSH_RESULT_DETAILS
174        }
175        Frame::SubStart {
176            id,
177            status,
178            reason_code,
179            effective_scopes,
180            bootstrap,
181        } => {
182            w.str(id);
183            w.u8(match status {
184                SubStatus::Active => 1,
185                SubStatus::Revoked => 2,
186                SubStatus::Reset => 3,
187            });
188            w.str(reason_code);
189            w.scope_map(effective_scopes);
190            w.bool(*bootstrap);
191            ft::SUB_START
192        }
193        Frame::Commit {
194            commit_seq,
195            created_at_ms,
196            actor_id,
197            tables,
198            changes,
199        } => {
200            w.i64(*commit_seq);
201            w.i64(*created_at_ms);
202            w.str(actor_id);
203            w.u32(tables.len() as u32);
204            for table in tables {
205                w.str(table);
206            }
207            w.u32(changes.len() as u32);
208            for change in changes {
209                w.u16(change.table_index);
210                w.str(&change.row_id);
211                w.u8(op_byte(change.op));
212                w.opt(&change.row_version, |w, v| w.i64(*v));
213                w.str_map(&change.scopes);
214                w.opt(&change.row, |w, row| w.bytes(row));
215            }
216            ft::COMMIT
217        }
218        Frame::SegmentRef {
219            segment_id,
220            media_type,
221            table,
222            byte_length,
223            row_count,
224            as_of_commit_seq,
225            scope_digest,
226            row_cursor,
227            next_row_cursor,
228            url,
229            url_expires_at_ms,
230        } => {
231            w.str(segment_id);
232            w.u8(match media_type {
233                MediaType::Rows => 1,
234                MediaType::Sqlite => 2,
235            });
236            w.str(table);
237            w.i64(*byte_length);
238            w.i64(*row_count);
239            w.i64(*as_of_commit_seq);
240            w.str(scope_digest);
241            w.opt(row_cursor, |w, s| w.str(s));
242            w.opt(next_row_cursor, |w, s| w.str(s));
243            w.opt(url, |w, s| w.str(s));
244            w.opt(url_expires_at_ms, |w, v| w.i64(*v));
245            ft::SEGMENT_REF
246        }
247        Frame::SegmentInline { payload } => {
248            w.raw(payload);
249            ft::SEGMENT_INLINE
250        }
251        Frame::SubEnd {
252            next_cursor,
253            bootstrap_state,
254        } => {
255            w.i64(*next_cursor);
256            w.opt(bootstrap_state, |w, j| w.str(&j.0));
257            ft::SUB_END
258        }
259        Frame::Error {
260            code,
261            message,
262            category,
263            retryable,
264            recommended_action,
265            details,
266        } => {
267            w.str(code);
268            w.str(message);
269            w.str(category);
270            w.bool(*retryable);
271            w.str(recommended_action);
272            w.opt(details, |w, j| w.str(&j.0));
273            ft::ERROR
274        }
275        Frame::Unknown {
276            frame_type,
277            payload,
278        } => {
279            // §1.2: an encoder MUST NOT emit an unknown frame under a
280            // frameType registered in this wire version. Preserved unknown
281            // frames come only from decoding, which never produces such a
282            // frame — hitting this is encoder misuse, not a decode error.
283            assert!(
284                !is_registered_frame_type(*frame_type),
285                "UNKNOWN frame type 0x{frame_type:02x} collides with a registered frame type"
286            );
287            w.raw(payload);
288            *frame_type
289        }
290    };
291    (ty, w.into_bytes())
292}
ssp2/encode.rs

ssp2/
encode.rs
