ssp2/
decode.rs

1//! SSP2 envelope decoding (SPEC.md §1.2), request/response frame grammar
2//! (§1.5, §1.6), and the closed decode-error boundary of §1.7: exactly the
3//! listed violations are rejected here, nothing more.
4
5use crate::error::{DecodeError, Result};
6use crate::model::frame_type as ft;
7use crate::model::{
8    Change, Frame, MediaType, Message, MsgKind, Op, OpResult, Operation, PushStatus, SubStatus,
9};
10use crate::primitives::Reader;
11use crate::segment::decode_rows_segment;
12
13pub const SSP2_MAGIC: &[u8; 4] = b"SSP2";
14pub const WIRE_VERSION: u16 = 1;
15
16/// Decode a complete SSP2 message, enforcing envelope rules and the per-kind
17/// frame grammar. Unknown frame types are preserved byte-for-byte (§1.2
18/// rule 2).
19pub fn decode_message(bytes: &[u8]) -> Result<Message> {
20    let mut r = Reader::new(bytes);
21    let magic = r.take(4, "envelope magic")?;
22    if magic != SSP2_MAGIC {
23        return Err(DecodeError::invalid("bad envelope magic (expected SSP2)"));
24    }
25    let wire_version = r.u16("wireVersion")?;
26    if wire_version != WIRE_VERSION {
27        return Err(DecodeError::invalid(format!(
28            "unsupported wireVersion {wire_version}"
29        )));
30    }
31    let msg_kind = match r.u8("msgKind")? {
32        0x01 => MsgKind::Request,
33        0x02 => MsgKind::Response,
34        other => {
35            return Err(DecodeError::invalid(format!(
36                "unknown msgKind 0x{other:02x}"
37            )))
38        }
39    };
40    let flags = r.u8("flags")?;
41    if flags != 0 {
42        return Err(DecodeError::invalid(format!(
43            "envelope flags must be 0x00, got 0x{flags:02x}"
44        )));
45    }
46
47    let frames = match msg_kind {
48        MsgKind::Request => decode_request_frames(&mut r)?,
49        MsgKind::Response => decode_response_frames(&mut r)?,
50    };
51    if !r.is_empty() {
52        return Err(DecodeError::invalid(
53            "trailing bytes after the END frame (canonical encoding)",
54        ));
55    }
56    Ok(Message { msg_kind, frames })
57}
58
59/// Read one frame header + payload. Returns `None` for END.
60fn next_frame<'a>(r: &mut Reader<'a>) -> Result<Option<(u8, &'a [u8])>> {
61    if r.is_empty() {
62        return Err(DecodeError::invalid(
63            "truncated message: body ends without an END frame",
64        ));
65    }
66    let ty = r.u8("frameType")?;
67    let len = r.u32("frameLength")? as usize;
68    if len > r.remaining() {
69        return Err(DecodeError::invalid(format!(
70            "frame 0x{ty:02x} length {len} exceeds the {} remaining bytes",
71            r.remaining()
72        )));
73    }
74    let payload = r.take(len, "frame payload")?;
75    if ty == ft::END {
76        if len != 0 {
77            return Err(DecodeError::invalid("END frame must have frameLength 0"));
78        }
79        return Ok(None);
80    }
81    Ok(Some((ty, payload)))
82}
83
84/// "Unknown" per §1.2 rule 2: a frameType with no layout in this wire version
85/// for *either* message kind. A type registered for the other kind is a
86/// decode error, not unknown.
87fn classify(ty: u8, kind: MsgKind) -> Result<bool> {
88    let (own, other) = match kind {
89        MsgKind::Request => (ft::REQUEST_TYPES, ft::RESPONSE_TYPES),
90        MsgKind::Response => (ft::RESPONSE_TYPES, ft::REQUEST_TYPES),
91    };
92    if own.contains(&ty) {
93        Ok(true)
94    } else if other.contains(&ty) {
95        Err(DecodeError::invalid(format!(
96            "frame type 0x{ty:02x} is registered for the other message kind"
97        )))
98    } else {
99        Ok(false)
100    }
101}
102
103/// §1.5 grammar: REQ_HEADER, PUSH_COMMIT × N, PULL_HEADER 0|1,
104/// SUBSCRIPTION × M (only if PULL_HEADER present), END. A request with
105/// neither push nor pull frames is invalid.
106fn decode_request_frames(r: &mut Reader<'_>) -> Result<Vec<Frame>> {
107    let mut frames = Vec::new();
108    let mut seen_header = false;
109    let mut seen_pull = false;
110    let mut seen_push = false;
111
112    while let Some((ty, payload)) = next_frame(r)? {
113        let known = classify(ty, MsgKind::Request)?;
114        if !known {
115            if !seen_header {
116                return Err(DecodeError::invalid(
117                    "unknown frame before REQ_HEADER (unknown frames are only legal between the header frame and END)",
118                ));
119            }
120            frames.push(Frame::Unknown {
121                frame_type: ty,
122                payload: payload.to_vec(),
123            });
124            continue;
125        }
126        if !seen_header {
127            if ty != ft::REQ_HEADER {
128                return Err(DecodeError::invalid(
129                    "first frame of a request must be REQ_HEADER",
130                ));
131            }
132            frames.push(decode_req_header(payload)?);
133            seen_header = true;
134            continue;
135        }
136        match ty {
137            ft::REQ_HEADER => {
138                return Err(DecodeError::invalid("duplicate REQ_HEADER frame"));
139            }
140            ft::PUSH_COMMIT => {
141                if seen_pull {
142                    return Err(DecodeError::invalid(
143                        "PUSH_COMMIT after PULL_HEADER (out of grammar order)",
144                    ));
145                }
146                frames.push(decode_push_commit(payload)?);
147                seen_push = true;
148            }
149            ft::PULL_HEADER => {
150                if seen_pull {
151                    return Err(DecodeError::invalid("duplicate PULL_HEADER frame"));
152                }
153                frames.push(decode_pull_header(payload)?);
154                seen_pull = true;
155            }
156            ft::SUBSCRIPTION => {
157                if !seen_pull {
158                    return Err(DecodeError::invalid(
159                        "SUBSCRIPTION frame without a preceding PULL_HEADER",
160                    ));
161                }
162                frames.push(decode_subscription(payload)?);
163            }
164            _ => unreachable!("classified as known request type"),
165        }
166    }
167    if !seen_header {
168        return Err(DecodeError::invalid("request has no REQ_HEADER frame"));
169    }
170    if !seen_push && !seen_pull {
171        return Err(DecodeError::invalid(
172            "request has neither PUSH_COMMIT nor PULL_HEADER frames",
173        ));
174    }
175    Ok(frames)
176}
177
178/// §1.6 grammar: RESP_HEADER, PUSH_RESULT × N, then per subscription
179/// SUB_START (COMMIT × k | segment frames × m) SUB_END; ERROR 0|1 anywhere
180/// after RESP_HEADER, and if present the next frame MUST be END (no frame of
181/// any type, unknown included, may follow it).
182fn decode_response_frames(r: &mut Reader<'_>) -> Result<Vec<Frame>> {
183    let mut frames = Vec::new();
184    let mut seen_header = false;
185    let mut seen_lease = false; // §7.3.2: at most one LEASE, before the body
186    let mut seen_body = false; // any non-RESP_HEADER/non-LEASE frame
187    let mut past_results = false; // a SUB_START has been seen
188    let mut sub_open = false;
189    let mut sub_seen_commit = false;
190    let mut sub_seen_segment = false;
191    let mut error_seen = false;
192    let mut last_push_result: Option<(String, bool, Vec<i32>)> = None;
193    let mut last_push_result_has_details = false;
194
195    while let Some((ty, payload)) = next_frame(r)? {
196        if error_seen {
197            return Err(DecodeError::invalid(
198                "frame after ERROR: the next frame after ERROR must be END",
199            ));
200        }
201        let known = classify(ty, MsgKind::Response)?;
202        if !known {
203            if !seen_header {
204                return Err(DecodeError::invalid(
205                    "unknown frame before RESP_HEADER (unknown frames are only legal between the header frame and END)",
206                ));
207            }
208            frames.push(Frame::Unknown {
209                frame_type: ty,
210                payload: payload.to_vec(),
211            });
212            seen_body = true;
213            continue;
214        }
215        if !seen_header {
216            if ty != ft::RESP_HEADER {
217                return Err(DecodeError::invalid(
218                    "first frame of a response must be RESP_HEADER",
219                ));
220            }
221            frames.push(decode_resp_header(payload)?);
222            seen_header = true;
223            continue;
224        }
225        match ty {
226            ft::RESP_HEADER => {
227                return Err(DecodeError::invalid("duplicate RESP_HEADER frame"));
228            }
229            ft::LEASE => {
230                // §7.3.2: at most one LEASE, immediately after RESP_HEADER.
231                if seen_lease {
232                    return Err(DecodeError::invalid("duplicate LEASE frame"));
233                }
234                if seen_body {
235                    return Err(DecodeError::invalid(
236                        "LEASE frame must immediately follow RESP_HEADER",
237                    ));
238                }
239                frames.push(decode_lease(payload)?);
240                seen_lease = true;
241            }
242            ft::PUSH_RESULT => {
243                if past_results {
244                    return Err(DecodeError::invalid(
245                        "PUSH_RESULT after a SUB_START (out of grammar order)",
246                    ));
247                }
248                let frame = decode_push_result(payload)?;
249                if let Frame::PushResult {
250                    client_commit_id,
251                    status,
252                    results,
253                    ..
254                } = &frame
255                {
256                    let error_indexes = results
257                        .iter()
258                        .filter_map(|result| match result {
259                            OpResult::Error { op_index, .. } => Some(*op_index),
260                            _ => None,
261                        })
262                        .collect();
263                    last_push_result = Some((
264                        client_commit_id.clone(),
265                        *status == PushStatus::Rejected,
266                        error_indexes,
267                    ));
268                    last_push_result_has_details = false;
269                }
270                frames.push(frame);
271                seen_body = true;
272            }
273            ft::PUSH_RESULT_DETAILS => {
274                if past_results {
275                    return Err(DecodeError::invalid(
276                        "PUSH_RESULT_DETAILS after a SUB_START (out of grammar order)",
277                    ));
278                }
279                let frame = decode_push_result_details(payload)?;
280                let Some((result_commit_id, result_rejected, error_indexes)) = &last_push_result
281                else {
282                    return Err(DecodeError::invalid(
283                        "PUSH_RESULT_DETAILS without a preceding PUSH_RESULT",
284                    ));
285                };
286                if last_push_result_has_details {
287                    return Err(DecodeError::invalid(
288                        "duplicate PUSH_RESULT_DETAILS companion",
289                    ));
290                }
291                if let Frame::PushResultDetails {
292                    client_commit_id,
293                    entries,
294                } = &frame
295                {
296                    if !*result_rejected || client_commit_id != result_commit_id {
297                        return Err(DecodeError::invalid(
298                            "PUSH_RESULT_DETAILS does not match its rejected PUSH_RESULT",
299                        ));
300                    }
301                    if entries
302                        .iter()
303                        .any(|entry| !error_indexes.contains(&entry.op_index))
304                    {
305                        return Err(DecodeError::invalid(
306                            "PUSH_RESULT_DETAILS entry does not match an error result",
307                        ));
308                    }
309                }
310                last_push_result_has_details = true;
311                frames.push(frame);
312                seen_body = true;
313            }
314            ft::SUB_START => {
315                if sub_open {
316                    return Err(DecodeError::invalid(
317                        "SUB_START while a subscription context is already open",
318                    ));
319                }
320                frames.push(decode_sub_start(payload)?);
321                past_results = true;
322                seen_body = true;
323                sub_open = true;
324                sub_seen_commit = false;
325                sub_seen_segment = false;
326            }
327            ft::COMMIT => {
328                if !sub_open {
329                    return Err(DecodeError::invalid(
330                        "COMMIT frame outside an open subscription context",
331                    ));
332                }
333                if sub_seen_segment {
334                    return Err(DecodeError::invalid(
335                        "COMMIT and segment frames must not both appear for one subscription",
336                    ));
337                }
338                frames.push(decode_commit(payload)?);
339                sub_seen_commit = true;
340                seen_body = true;
341            }
342            ft::SEGMENT_REF | ft::SEGMENT_INLINE => {
343                if !sub_open {
344                    return Err(DecodeError::invalid(
345                        "segment frame outside an open subscription context",
346                    ));
347                }
348                if sub_seen_commit {
349                    return Err(DecodeError::invalid(
350                        "COMMIT and segment frames must not both appear for one subscription",
351                    ));
352                }
353                if ty == ft::SEGMENT_REF {
354                    frames.push(decode_segment_ref(payload)?);
355                } else {
356                    // §5.7: the payload must be a structurally valid rows
357                    // segment; the raw bytes are preserved for re-encoding.
358                    decode_rows_segment(payload)?;
359                    frames.push(Frame::SegmentInline {
360                        payload: payload.to_vec(),
361                    });
362                }
363                sub_seen_segment = true;
364                seen_body = true;
365            }
366            ft::SUB_END => {
367                if !sub_open {
368                    return Err(DecodeError::invalid(
369                        "SUB_END without an open subscription context",
370                    ));
371                }
372                frames.push(decode_sub_end(payload)?);
373                sub_open = false;
374                seen_body = true;
375            }
376            ft::ERROR => {
377                frames.push(decode_error_frame(payload)?);
378                error_seen = true;
379                seen_body = true;
380            }
381            _ => unreachable!("classified as known response type"),
382        }
383    }
384    if !seen_header {
385        return Err(DecodeError::invalid("response has no RESP_HEADER frame"));
386    }
387    if sub_open && !error_seen {
388        return Err(DecodeError::invalid(
389            "message ended with an open subscription context (missing SUB_END)",
390        ));
391    }
392    Ok(frames)
393}
394
395/// Run `parse` over a frame payload; trailing bytes inside a known frame are
396/// a decode error (§1.2 rule 3).
397fn frame_payload<T>(
398    payload: &[u8],
399    name: &str,
400    parse: impl FnOnce(&mut Reader<'_>) -> Result<T>,
401) -> Result<T> {
402    let mut r = Reader::new(payload);
403    let value = parse(&mut r)?;
404    if !r.is_empty() {
405        return Err(DecodeError::invalid(format!(
406            "trailing bytes inside {name} frame ({} bytes unconsumed)",
407            r.remaining()
408        )));
409    }
410    Ok(value)
411}
412
413fn decode_req_header(payload: &[u8]) -> Result<Frame> {
414    frame_payload(payload, "REQ_HEADER", |r| {
415        let client_id = r.str("clientId")?;
416        if client_id.is_empty() {
417            return Err(DecodeError::invalid("clientId must be non-empty"));
418        }
419        let schema_version = r.i32("schemaVersion")?;
420        if schema_version < 1 {
421            return Err(DecodeError::invalid(format!(
422                "schemaVersion must be ≥ 1, got {schema_version}"
423            )));
424        }
425        Ok(Frame::ReqHeader {
426            client_id,
427            schema_version,
428        })
429    })
430}
431
432fn decode_op(r: &mut Reader<'_>) -> Result<Op> {
433    match r.u8("op")? {
434        1 => Ok(Op::Upsert),
435        2 => Ok(Op::Delete),
436        other => Err(DecodeError::invalid(format!("unknown op byte {other}"))),
437    }
438}
439
440fn decode_push_commit(payload: &[u8]) -> Result<Frame> {
441    frame_payload(payload, "PUSH_COMMIT", |r| {
442        let client_commit_id = r.str("clientCommitId")?;
443        if client_commit_id.is_empty() {
444            return Err(DecodeError::invalid("clientCommitId must be non-empty"));
445        }
446        let count = r.u32("operations count")? as usize;
447        if count == 0 {
448            return Err(DecodeError::empty_commit(
449                "PUSH_COMMIT with zero operations",
450            ));
451        }
452        let mut operations = Vec::with_capacity(count.min(4096));
453        for _ in 0..count {
454            let table = r.str("operation table")?;
455            let row_id = r.str("operation rowId")?;
456            let op = decode_op(r)?;
457            let base_version = if r.presence("baseVersion")? {
458                Some(r.i64("baseVersion")?)
459            } else {
460                None
461            };
462            let payload_bytes = if r.presence("payload")? {
463                Some(r.bytes("payload")?)
464            } else {
465                None
466            };
467            // §6.1 presence tie: payload present iff upsert.
468            match op {
469                Op::Upsert if payload_bytes.is_none() => {
470                    return Err(DecodeError::invalid(
471                        "upsert operation without a payload (presence invariant)",
472                    ));
473                }
474                Op::Delete if payload_bytes.is_some() => {
475                    return Err(DecodeError::invalid(
476                        "delete operation with a payload (presence invariant)",
477                    ));
478                }
479                _ => {}
480            }
481            operations.push(Operation {
482                table,
483                row_id,
484                op,
485                base_version,
486                payload: payload_bytes,
487            });
488        }
489        Ok(Frame::PushCommit {
490            client_commit_id,
491            operations,
492        })
493    })
494}
495
496fn decode_pull_header(payload: &[u8]) -> Result<Frame> {
497    frame_payload(payload, "PULL_HEADER", |r| {
498        let limit_commits = r.i32("limitCommits")?;
499        let limit_snapshot_rows = r.i32("limitSnapshotRows")?;
500        let max_snapshot_pages = r.i32("maxSnapshotPages")?;
501        let accept = r.u8("accept")?;
502        if accept & 0xF0 != 0 {
503            return Err(DecodeError::invalid(format!(
504                "accept bits 4–7 must be zero, got 0b{accept:08b}"
505            )));
506        }
507        Ok(Frame::PullHeader {
508            limit_commits,
509            limit_snapshot_rows,
510            max_snapshot_pages,
511            accept,
512        })
513    })
514}
515
516fn decode_subscription(payload: &[u8]) -> Result<Frame> {
517    frame_payload(payload, "SUBSCRIPTION", |r| {
518        let id = r.str("id")?;
519        let table = r.str("table")?;
520        let scopes = r.scope_map("scopes")?;
521        let params = if r.presence("params")? {
522            Some(r.json("params")?)
523        } else {
524            None
525        };
526        let cursor = r.i64("cursor")?;
527        let bootstrap_state = if r.presence("bootstrapState")? {
528            Some(r.json("bootstrapState")?)
529        } else {
530            None
531        };
532        Ok(Frame::Subscription {
533            id,
534            table,
535            scopes,
536            params,
537            cursor,
538            bootstrap_state,
539        })
540    })
541}
542
543fn decode_resp_header(payload: &[u8]) -> Result<Frame> {
544    frame_payload(payload, "RESP_HEADER", |r| {
545        let required_schema_version = if r.presence("requiredSchemaVersion")? {
546            Some(r.i32("requiredSchemaVersion")?)
547        } else {
548            None
549        };
550        let latest_schema_version = if r.presence("latestSchemaVersion")? {
551            Some(r.i32("latestSchemaVersion")?)
552        } else {
553            None
554        };
555        Ok(Frame::RespHeader {
556            required_schema_version,
557            latest_schema_version,
558        })
559    })
560}
561
562fn decode_lease(payload: &[u8]) -> Result<Frame> {
563    frame_payload(payload, "LEASE", |r| {
564        let lease_id = r.str("leaseId")?;
565        if lease_id.is_empty() {
566            return Err(DecodeError::invalid("LEASE.leaseId must be non-empty"));
567        }
568        let expires_at_ms = r.i64("expiresAtMs")?;
569        Ok(Frame::Lease {
570            lease_id,
571            expires_at_ms,
572        })
573    })
574}
575
576fn decode_push_result(payload: &[u8]) -> Result<Frame> {
577    frame_payload(payload, "PUSH_RESULT", |r| {
578        let client_commit_id = r.str("clientCommitId")?;
579        let status = match r.u8("status")? {
580            1 => PushStatus::Applied,
581            2 => PushStatus::Cached,
582            3 => PushStatus::Rejected,
583            other => {
584                return Err(DecodeError::invalid(format!(
585                    "unknown PUSH_RESULT status byte {other}"
586                )))
587            }
588        };
589        let commit_seq = if r.presence("commitSeq")? {
590            Some(r.i64("commitSeq")?)
591        } else {
592            None
593        };
594        // §6.3 presence tie: commitSeq present for applied/cached, absent for
595        // rejected.
596        match status {
597            PushStatus::Rejected if commit_seq.is_some() => {
598                return Err(DecodeError::invalid(
599                    "commitSeq present on a rejected PUSH_RESULT (presence invariant)",
600                ));
601            }
602            PushStatus::Applied | PushStatus::Cached if commit_seq.is_none() => {
603                return Err(DecodeError::invalid(
604                    "commitSeq absent on an applied/cached PUSH_RESULT (presence invariant)",
605                ));
606            }
607            _ => {}
608        }
609        let count = r.u32("results count")? as usize;
610        let mut results = Vec::with_capacity(count.min(4096));
611        for _ in 0..count {
612            let op_index = r.i32("opIndex")?;
613            results.push(match r.u8("result status")? {
614                1 => OpResult::Applied { op_index },
615                2 => OpResult::Conflict {
616                    op_index,
617                    code: r.str("conflict code")?,
618                    message: r.str("conflict message")?,
619                    server_version: r.i64("serverVersion")?,
620                    server_row: r.bytes("serverRow")?,
621                },
622                3 => OpResult::Error {
623                    op_index,
624                    code: r.str("error code")?,
625                    message: r.str("error message")?,
626                    retryable: r.bool("retryable")?,
627                },
628                other => {
629                    return Err(DecodeError::invalid(format!(
630                        "unknown result record status byte {other}"
631                    )))
632                }
633            });
634        }
635        Ok(Frame::PushResult {
636            client_commit_id,
637            status,
638            commit_seq,
639            results,
640        })
641    })
642}
643
644fn decode_push_result_details(payload: &[u8]) -> Result<Frame> {
645    frame_payload(payload, "PUSH_RESULT_DETAILS", |r| {
646        let client_commit_id = r.str("clientCommitId")?;
647        if client_commit_id.is_empty() {
648            return Err(DecodeError::invalid(
649                "PUSH_RESULT_DETAILS.clientCommitId must be non-empty",
650            ));
651        }
652        let count = r.u32("entries count")? as usize;
653        if count == 0 {
654            return Err(DecodeError::invalid(
655                "PUSH_RESULT_DETAILS must carry at least one entry",
656            ));
657        }
658        let mut entries = Vec::with_capacity(count.min(64));
659        let mut seen = std::collections::BTreeSet::new();
660        for _ in 0..count {
661            let op_index = r.i32("opIndex")?;
662            if op_index < 0 || !seen.insert(op_index) {
663                return Err(DecodeError::invalid(
664                    "PUSH_RESULT_DETAILS opIndex values must be unique and non-negative",
665                ));
666            }
667            entries.push(crate::model::PushResultDetail {
668                op_index,
669                details: r.json("details")?,
670            });
671        }
672        Ok(Frame::PushResultDetails {
673            client_commit_id,
674            entries,
675        })
676    })
677}
678
679fn decode_sub_start(payload: &[u8]) -> Result<Frame> {
680    frame_payload(payload, "SUB_START", |r| {
681        let id = r.str("id")?;
682        let status = match r.u8("status")? {
683            1 => SubStatus::Active,
684            2 => SubStatus::Revoked,
685            3 => SubStatus::Reset,
686            other => {
687                return Err(DecodeError::invalid(format!(
688                    "unknown SUB_START status byte {other}"
689                )))
690            }
691        };
692        let reason_code = r.str("reasonCode")?;
693        let effective_scopes = r.scope_map("effectiveScopes")?;
694        let bootstrap = r.bool("bootstrap")?;
695        Ok(Frame::SubStart {
696            id,
697            status,
698            reason_code,
699            effective_scopes,
700            bootstrap,
701        })
702    })
703}
704
705fn decode_commit(payload: &[u8]) -> Result<Frame> {
706    frame_payload(payload, "COMMIT", |r| {
707        let commit_seq = r.i64("commitSeq")?;
708        let created_at_ms = r.i64("createdAtMs")?;
709        let actor_id = r.str("actorId")?;
710        let table_count = r.u32("tables count")? as usize;
711        let mut tables = Vec::with_capacity(table_count.min(4096));
712        for _ in 0..table_count {
713            tables.push(r.str("table")?);
714        }
715        let change_count = r.u32("changes count")? as usize;
716        let mut changes = Vec::with_capacity(change_count.min(4096));
717        for _ in 0..change_count {
718            let table_index = r.u16("tableIndex")?;
719            if table_index as usize >= tables.len() {
720                return Err(DecodeError::invalid(format!(
721                    "change tableIndex {table_index} out of range (tables: {})",
722                    tables.len()
723                )));
724            }
725            let row_id = r.str("rowId")?;
726            let op = decode_op(r)?;
727            let row_version = if r.presence("rowVersion")? {
728                Some(r.i64("rowVersion")?)
729            } else {
730                None
731            };
732            let scopes = r.str_map("change scopes")?;
733            let row = if r.presence("row")? {
734                Some(r.bytes("row")?)
735            } else {
736                None
737            };
738            // §4.5 presence ties: rowVersion and row present iff upsert.
739            let ok = match op {
740                Op::Upsert => row_version.is_some() && row.is_some(),
741                Op::Delete => row_version.is_none() && row.is_none(),
742            };
743            if !ok {
744                return Err(DecodeError::invalid(
745                    "change rowVersion/row presence does not match op (presence invariant)",
746                ));
747            }
748            changes.push(Change {
749                table_index,
750                row_id,
751                op,
752                row_version,
753                scopes,
754                row,
755            });
756        }
757        Ok(Frame::Commit {
758            commit_seq,
759            created_at_ms,
760            actor_id,
761            tables,
762            changes,
763        })
764    })
765}
766
767fn decode_segment_ref(payload: &[u8]) -> Result<Frame> {
768    frame_payload(payload, "SEGMENT_REF", |r| {
769        let segment_id = r.str("segmentId")?;
770        let media_type = match r.u8("mediaType")? {
771            1 => MediaType::Rows,
772            2 => MediaType::Sqlite,
773            other => {
774                return Err(DecodeError::invalid(format!(
775                    "unknown mediaType byte {other}"
776                )))
777            }
778        };
779        let table = r.str("table")?;
780        let byte_length = r.i64("byteLength")?;
781        let row_count = r.i64("rowCount")?;
782        let as_of_commit_seq = r.i64("asOfCommitSeq")?;
783        let scope_digest = r.str("scopeDigest")?;
784        let row_cursor = if r.presence("rowCursor")? {
785            Some(r.str("rowCursor")?)
786        } else {
787            None
788        };
789        let next_row_cursor = if r.presence("nextRowCursor")? {
790            Some(r.str("nextRowCursor")?)
791        } else {
792            None
793        };
794        let url = if r.presence("url")? {
795            Some(r.str("url")?)
796        } else {
797            None
798        };
799        let url_expires_at_ms = if r.presence("urlExpiresAtMs")? {
800            Some(r.i64("urlExpiresAtMs")?)
801        } else {
802            None
803        };
804        // §5.4 presence tie: urlExpiresAtMs present iff url is.
805        if url.is_some() != url_expires_at_ms.is_some() {
806            return Err(DecodeError::invalid(
807                "url and urlExpiresAtMs must be present together (presence invariant)",
808            ));
809        }
810        Ok(Frame::SegmentRef {
811            segment_id,
812            media_type,
813            table,
814            byte_length,
815            row_count,
816            as_of_commit_seq,
817            scope_digest,
818            row_cursor,
819            next_row_cursor,
820            url,
821            url_expires_at_ms,
822        })
823    })
824}
825
826fn decode_sub_end(payload: &[u8]) -> Result<Frame> {
827    frame_payload(payload, "SUB_END", |r| {
828        let next_cursor = r.i64("nextCursor")?;
829        let bootstrap_state = if r.presence("bootstrapState")? {
830            Some(r.json("bootstrapState")?)
831        } else {
832            None
833        };
834        Ok(Frame::SubEnd {
835            next_cursor,
836            bootstrap_state,
837        })
838    })
839}
840
841fn decode_error_frame(payload: &[u8]) -> Result<Frame> {
842    frame_payload(payload, "ERROR", |r| {
843        let code = r.str("code")?;
844        let message = r.str("message")?;
845        let category = r.str("category")?;
846        let retryable = r.bool("retryable")?;
847        let recommended_action = r.str("recommendedAction")?;
848        let details = if r.presence("details")? {
849            Some(r.json("details")?)
850        } else {
851            None
852        };
853        Ok(Frame::Error {
854            code,
855            message,
856            category,
857            retryable,
858            recommended_action,
859            details,
860        })
861    })
862}
ssp2/decode.rs

ssp2/
decode.rs
