1use 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
16pub 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
59fn 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
84fn 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
103fn 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
178fn 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; let mut seen_body = false; let mut past_results = false; 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
193 while let Some((ty, payload)) = next_frame(r)? {
194 if error_seen {
195 return Err(DecodeError::invalid(
196 "frame after ERROR: the next frame after ERROR must be END",
197 ));
198 }
199 let known = classify(ty, MsgKind::Response)?;
200 if !known {
201 if !seen_header {
202 return Err(DecodeError::invalid(
203 "unknown frame before RESP_HEADER (unknown frames are only legal between the header frame and END)",
204 ));
205 }
206 frames.push(Frame::Unknown {
207 frame_type: ty,
208 payload: payload.to_vec(),
209 });
210 seen_body = true;
211 continue;
212 }
213 if !seen_header {
214 if ty != ft::RESP_HEADER {
215 return Err(DecodeError::invalid(
216 "first frame of a response must be RESP_HEADER",
217 ));
218 }
219 frames.push(decode_resp_header(payload)?);
220 seen_header = true;
221 continue;
222 }
223 match ty {
224 ft::RESP_HEADER => {
225 return Err(DecodeError::invalid("duplicate RESP_HEADER frame"));
226 }
227 ft::LEASE => {
228 if seen_lease {
230 return Err(DecodeError::invalid("duplicate LEASE frame"));
231 }
232 if seen_body {
233 return Err(DecodeError::invalid(
234 "LEASE frame must immediately follow RESP_HEADER",
235 ));
236 }
237 frames.push(decode_lease(payload)?);
238 seen_lease = true;
239 }
240 ft::PUSH_RESULT => {
241 if past_results {
242 return Err(DecodeError::invalid(
243 "PUSH_RESULT after a SUB_START (out of grammar order)",
244 ));
245 }
246 frames.push(decode_push_result(payload)?);
247 seen_body = true;
248 }
249 ft::SUB_START => {
250 if sub_open {
251 return Err(DecodeError::invalid(
252 "SUB_START while a subscription context is already open",
253 ));
254 }
255 frames.push(decode_sub_start(payload)?);
256 past_results = true;
257 seen_body = true;
258 sub_open = true;
259 sub_seen_commit = false;
260 sub_seen_segment = false;
261 }
262 ft::COMMIT => {
263 if !sub_open {
264 return Err(DecodeError::invalid(
265 "COMMIT frame outside an open subscription context",
266 ));
267 }
268 if sub_seen_segment {
269 return Err(DecodeError::invalid(
270 "COMMIT and segment frames must not both appear for one subscription",
271 ));
272 }
273 frames.push(decode_commit(payload)?);
274 sub_seen_commit = true;
275 seen_body = true;
276 }
277 ft::SEGMENT_REF | ft::SEGMENT_INLINE => {
278 if !sub_open {
279 return Err(DecodeError::invalid(
280 "segment frame outside an open subscription context",
281 ));
282 }
283 if sub_seen_commit {
284 return Err(DecodeError::invalid(
285 "COMMIT and segment frames must not both appear for one subscription",
286 ));
287 }
288 if ty == ft::SEGMENT_REF {
289 frames.push(decode_segment_ref(payload)?);
290 } else {
291 decode_rows_segment(payload)?;
294 frames.push(Frame::SegmentInline {
295 payload: payload.to_vec(),
296 });
297 }
298 sub_seen_segment = true;
299 seen_body = true;
300 }
301 ft::SUB_END => {
302 if !sub_open {
303 return Err(DecodeError::invalid(
304 "SUB_END without an open subscription context",
305 ));
306 }
307 frames.push(decode_sub_end(payload)?);
308 sub_open = false;
309 seen_body = true;
310 }
311 ft::ERROR => {
312 frames.push(decode_error_frame(payload)?);
313 error_seen = true;
314 seen_body = true;
315 }
316 _ => unreachable!("classified as known response type"),
317 }
318 }
319 if !seen_header {
320 return Err(DecodeError::invalid("response has no RESP_HEADER frame"));
321 }
322 if sub_open && !error_seen {
323 return Err(DecodeError::invalid(
324 "message ended with an open subscription context (missing SUB_END)",
325 ));
326 }
327 Ok(frames)
328}
329
330fn frame_payload<T>(
333 payload: &[u8],
334 name: &str,
335 parse: impl FnOnce(&mut Reader<'_>) -> Result<T>,
336) -> Result<T> {
337 let mut r = Reader::new(payload);
338 let value = parse(&mut r)?;
339 if !r.is_empty() {
340 return Err(DecodeError::invalid(format!(
341 "trailing bytes inside {name} frame ({} bytes unconsumed)",
342 r.remaining()
343 )));
344 }
345 Ok(value)
346}
347
348fn decode_req_header(payload: &[u8]) -> Result<Frame> {
349 frame_payload(payload, "REQ_HEADER", |r| {
350 let client_id = r.str("clientId")?;
351 if client_id.is_empty() {
352 return Err(DecodeError::invalid("clientId must be non-empty"));
353 }
354 let schema_version = r.i32("schemaVersion")?;
355 if schema_version < 1 {
356 return Err(DecodeError::invalid(format!(
357 "schemaVersion must be ≥ 1, got {schema_version}"
358 )));
359 }
360 Ok(Frame::ReqHeader {
361 client_id,
362 schema_version,
363 })
364 })
365}
366
367fn decode_op(r: &mut Reader<'_>) -> Result<Op> {
368 match r.u8("op")? {
369 1 => Ok(Op::Upsert),
370 2 => Ok(Op::Delete),
371 other => Err(DecodeError::invalid(format!("unknown op byte {other}"))),
372 }
373}
374
375fn decode_push_commit(payload: &[u8]) -> Result<Frame> {
376 frame_payload(payload, "PUSH_COMMIT", |r| {
377 let client_commit_id = r.str("clientCommitId")?;
378 if client_commit_id.is_empty() {
379 return Err(DecodeError::invalid("clientCommitId must be non-empty"));
380 }
381 let count = r.u32("operations count")? as usize;
382 if count == 0 {
383 return Err(DecodeError::empty_commit(
384 "PUSH_COMMIT with zero operations",
385 ));
386 }
387 let mut operations = Vec::with_capacity(count.min(4096));
388 for _ in 0..count {
389 let table = r.str("operation table")?;
390 let row_id = r.str("operation rowId")?;
391 let op = decode_op(r)?;
392 let base_version = if r.presence("baseVersion")? {
393 Some(r.i64("baseVersion")?)
394 } else {
395 None
396 };
397 let payload_bytes = if r.presence("payload")? {
398 Some(r.bytes("payload")?)
399 } else {
400 None
401 };
402 match op {
404 Op::Upsert if payload_bytes.is_none() => {
405 return Err(DecodeError::invalid(
406 "upsert operation without a payload (presence invariant)",
407 ));
408 }
409 Op::Delete if payload_bytes.is_some() => {
410 return Err(DecodeError::invalid(
411 "delete operation with a payload (presence invariant)",
412 ));
413 }
414 _ => {}
415 }
416 operations.push(Operation {
417 table,
418 row_id,
419 op,
420 base_version,
421 payload: payload_bytes,
422 });
423 }
424 Ok(Frame::PushCommit {
425 client_commit_id,
426 operations,
427 })
428 })
429}
430
431fn decode_pull_header(payload: &[u8]) -> Result<Frame> {
432 frame_payload(payload, "PULL_HEADER", |r| {
433 let limit_commits = r.i32("limitCommits")?;
434 let limit_snapshot_rows = r.i32("limitSnapshotRows")?;
435 let max_snapshot_pages = r.i32("maxSnapshotPages")?;
436 let accept = r.u8("accept")?;
437 if accept & 0xF0 != 0 {
438 return Err(DecodeError::invalid(format!(
439 "accept bits 4–7 must be zero, got 0b{accept:08b}"
440 )));
441 }
442 Ok(Frame::PullHeader {
443 limit_commits,
444 limit_snapshot_rows,
445 max_snapshot_pages,
446 accept,
447 })
448 })
449}
450
451fn decode_subscription(payload: &[u8]) -> Result<Frame> {
452 frame_payload(payload, "SUBSCRIPTION", |r| {
453 let id = r.str("id")?;
454 let table = r.str("table")?;
455 let scopes = r.scope_map("scopes")?;
456 let params = if r.presence("params")? {
457 Some(r.json("params")?)
458 } else {
459 None
460 };
461 let cursor = r.i64("cursor")?;
462 let bootstrap_state = if r.presence("bootstrapState")? {
463 Some(r.json("bootstrapState")?)
464 } else {
465 None
466 };
467 Ok(Frame::Subscription {
468 id,
469 table,
470 scopes,
471 params,
472 cursor,
473 bootstrap_state,
474 })
475 })
476}
477
478fn decode_resp_header(payload: &[u8]) -> Result<Frame> {
479 frame_payload(payload, "RESP_HEADER", |r| {
480 let required_schema_version = if r.presence("requiredSchemaVersion")? {
481 Some(r.i32("requiredSchemaVersion")?)
482 } else {
483 None
484 };
485 let latest_schema_version = if r.presence("latestSchemaVersion")? {
486 Some(r.i32("latestSchemaVersion")?)
487 } else {
488 None
489 };
490 Ok(Frame::RespHeader {
491 required_schema_version,
492 latest_schema_version,
493 })
494 })
495}
496
497fn decode_lease(payload: &[u8]) -> Result<Frame> {
498 frame_payload(payload, "LEASE", |r| {
499 let lease_id = r.str("leaseId")?;
500 if lease_id.is_empty() {
501 return Err(DecodeError::invalid("LEASE.leaseId must be non-empty"));
502 }
503 let expires_at_ms = r.i64("expiresAtMs")?;
504 Ok(Frame::Lease {
505 lease_id,
506 expires_at_ms,
507 })
508 })
509}
510
511fn decode_push_result(payload: &[u8]) -> Result<Frame> {
512 frame_payload(payload, "PUSH_RESULT", |r| {
513 let client_commit_id = r.str("clientCommitId")?;
514 let status = match r.u8("status")? {
515 1 => PushStatus::Applied,
516 2 => PushStatus::Cached,
517 3 => PushStatus::Rejected,
518 other => {
519 return Err(DecodeError::invalid(format!(
520 "unknown PUSH_RESULT status byte {other}"
521 )))
522 }
523 };
524 let commit_seq = if r.presence("commitSeq")? {
525 Some(r.i64("commitSeq")?)
526 } else {
527 None
528 };
529 match status {
532 PushStatus::Rejected if commit_seq.is_some() => {
533 return Err(DecodeError::invalid(
534 "commitSeq present on a rejected PUSH_RESULT (presence invariant)",
535 ));
536 }
537 PushStatus::Applied | PushStatus::Cached if commit_seq.is_none() => {
538 return Err(DecodeError::invalid(
539 "commitSeq absent on an applied/cached PUSH_RESULT (presence invariant)",
540 ));
541 }
542 _ => {}
543 }
544 let count = r.u32("results count")? as usize;
545 let mut results = Vec::with_capacity(count.min(4096));
546 for _ in 0..count {
547 let op_index = r.i32("opIndex")?;
548 results.push(match r.u8("result status")? {
549 1 => OpResult::Applied { op_index },
550 2 => OpResult::Conflict {
551 op_index,
552 code: r.str("conflict code")?,
553 message: r.str("conflict message")?,
554 server_version: r.i64("serverVersion")?,
555 server_row: r.bytes("serverRow")?,
556 },
557 3 => OpResult::Error {
558 op_index,
559 code: r.str("error code")?,
560 message: r.str("error message")?,
561 retryable: r.bool("retryable")?,
562 },
563 other => {
564 return Err(DecodeError::invalid(format!(
565 "unknown result record status byte {other}"
566 )))
567 }
568 });
569 }
570 Ok(Frame::PushResult {
571 client_commit_id,
572 status,
573 commit_seq,
574 results,
575 })
576 })
577}
578
579fn decode_sub_start(payload: &[u8]) -> Result<Frame> {
580 frame_payload(payload, "SUB_START", |r| {
581 let id = r.str("id")?;
582 let status = match r.u8("status")? {
583 1 => SubStatus::Active,
584 2 => SubStatus::Revoked,
585 3 => SubStatus::Reset,
586 other => {
587 return Err(DecodeError::invalid(format!(
588 "unknown SUB_START status byte {other}"
589 )))
590 }
591 };
592 let reason_code = r.str("reasonCode")?;
593 let effective_scopes = r.scope_map("effectiveScopes")?;
594 let bootstrap = r.bool("bootstrap")?;
595 Ok(Frame::SubStart {
596 id,
597 status,
598 reason_code,
599 effective_scopes,
600 bootstrap,
601 })
602 })
603}
604
605fn decode_commit(payload: &[u8]) -> Result<Frame> {
606 frame_payload(payload, "COMMIT", |r| {
607 let commit_seq = r.i64("commitSeq")?;
608 let created_at_ms = r.i64("createdAtMs")?;
609 let actor_id = r.str("actorId")?;
610 let table_count = r.u32("tables count")? as usize;
611 let mut tables = Vec::with_capacity(table_count.min(4096));
612 for _ in 0..table_count {
613 tables.push(r.str("table")?);
614 }
615 let change_count = r.u32("changes count")? as usize;
616 let mut changes = Vec::with_capacity(change_count.min(4096));
617 for _ in 0..change_count {
618 let table_index = r.u16("tableIndex")?;
619 if table_index as usize >= tables.len() {
620 return Err(DecodeError::invalid(format!(
621 "change tableIndex {table_index} out of range (tables: {})",
622 tables.len()
623 )));
624 }
625 let row_id = r.str("rowId")?;
626 let op = decode_op(r)?;
627 let row_version = if r.presence("rowVersion")? {
628 Some(r.i64("rowVersion")?)
629 } else {
630 None
631 };
632 let scopes = r.str_map("change scopes")?;
633 let row = if r.presence("row")? {
634 Some(r.bytes("row")?)
635 } else {
636 None
637 };
638 let ok = match op {
640 Op::Upsert => row_version.is_some() && row.is_some(),
641 Op::Delete => row_version.is_none() && row.is_none(),
642 };
643 if !ok {
644 return Err(DecodeError::invalid(
645 "change rowVersion/row presence does not match op (presence invariant)",
646 ));
647 }
648 changes.push(Change {
649 table_index,
650 row_id,
651 op,
652 row_version,
653 scopes,
654 row,
655 });
656 }
657 Ok(Frame::Commit {
658 commit_seq,
659 created_at_ms,
660 actor_id,
661 tables,
662 changes,
663 })
664 })
665}
666
667fn decode_segment_ref(payload: &[u8]) -> Result<Frame> {
668 frame_payload(payload, "SEGMENT_REF", |r| {
669 let segment_id = r.str("segmentId")?;
670 let media_type = match r.u8("mediaType")? {
671 1 => MediaType::Rows,
672 2 => MediaType::Sqlite,
673 other => {
674 return Err(DecodeError::invalid(format!(
675 "unknown mediaType byte {other}"
676 )))
677 }
678 };
679 let table = r.str("table")?;
680 let byte_length = r.i64("byteLength")?;
681 let row_count = r.i64("rowCount")?;
682 let as_of_commit_seq = r.i64("asOfCommitSeq")?;
683 let scope_digest = r.str("scopeDigest")?;
684 let row_cursor = if r.presence("rowCursor")? {
685 Some(r.str("rowCursor")?)
686 } else {
687 None
688 };
689 let next_row_cursor = if r.presence("nextRowCursor")? {
690 Some(r.str("nextRowCursor")?)
691 } else {
692 None
693 };
694 let url = if r.presence("url")? {
695 Some(r.str("url")?)
696 } else {
697 None
698 };
699 let url_expires_at_ms = if r.presence("urlExpiresAtMs")? {
700 Some(r.i64("urlExpiresAtMs")?)
701 } else {
702 None
703 };
704 if url.is_some() != url_expires_at_ms.is_some() {
706 return Err(DecodeError::invalid(
707 "url and urlExpiresAtMs must be present together (presence invariant)",
708 ));
709 }
710 Ok(Frame::SegmentRef {
711 segment_id,
712 media_type,
713 table,
714 byte_length,
715 row_count,
716 as_of_commit_seq,
717 scope_digest,
718 row_cursor,
719 next_row_cursor,
720 url,
721 url_expires_at_ms,
722 })
723 })
724}
725
726fn decode_sub_end(payload: &[u8]) -> Result<Frame> {
727 frame_payload(payload, "SUB_END", |r| {
728 let next_cursor = r.i64("nextCursor")?;
729 let bootstrap_state = if r.presence("bootstrapState")? {
730 Some(r.json("bootstrapState")?)
731 } else {
732 None
733 };
734 Ok(Frame::SubEnd {
735 next_cursor,
736 bootstrap_state,
737 })
738 })
739}
740
741fn decode_error_frame(payload: &[u8]) -> Result<Frame> {
742 frame_payload(payload, "ERROR", |r| {
743 let code = r.str("code")?;
744 let message = r.str("message")?;
745 let category = r.str("category")?;
746 let retryable = r.bool("retryable")?;
747 let recommended_action = r.str("recommendedAction")?;
748 let details = if r.presence("details")? {
749 Some(r.json("details")?)
750 } else {
751 None
752 };
753 Ok(Frame::Error {
754 code,
755 message,
756 category,
757 retryable,
758 recommended_action,
759 details,
760 })
761 })
762}