graphrefly/adapters/
bridge_protobuf.rs

1//! D497 canonical protobuf profile for D134/D496 wire bridge envelopes.
2//!
3//! This module is schema-specific on purpose: it validates the locked
4//! `protocol.proto` bridge envelope subset without introducing a protobuf
5//! codegen/build step or a production value codec registry.
6
7use std::collections::HashSet;
8use std::error::Error;
9use std::fmt;
10
11use super::bridge::WireBridgeMetadata as SemanticWireBridgeMetadata;
12
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14/// `CanonicalProtobufErrorCategory` variants.
15pub enum CanonicalProtobufErrorCategory {
16    /// `UnknownField` variant.
17    UnknownField,
18    /// `DuplicateSingular` variant.
19    DuplicateSingular,
20    /// `NoncanonicalBytes` variant.
21    NoncanonicalBytes,
22    /// `InvalidOneof` variant.
23    InvalidOneof,
24    /// `MissingRequired` variant.
25    MissingRequired,
26    /// `InvalidWireEdge` variant.
27    InvalidWireEdge,
28    /// `DefaultEmission` variant.
29    DefaultEmission,
30    /// `Malformed` variant.
31    Malformed,
32}
33
34impl CanonicalProtobufErrorCategory {
35    #[must_use]
36    /// Updates or reads `as_str`.
37    pub fn as_str(self) -> &'static str {
38        match self {
39            Self::UnknownField => "unknown_field",
40            Self::DuplicateSingular => "duplicate_singular",
41            Self::NoncanonicalBytes => "noncanonical_bytes",
42            Self::InvalidOneof => "invalid_oneof",
43            Self::MissingRequired => "missing_required",
44            Self::InvalidWireEdge => "invalid_wire_edge",
45            Self::DefaultEmission => "default_emission",
46            Self::Malformed => "malformed",
47        }
48    }
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
52/// `CanonicalProtobufError` data container.
53pub struct CanonicalProtobufError {
54    /// `category` field for category.
55    pub category: CanonicalProtobufErrorCategory,
56    message: String,
57}
58
59impl CanonicalProtobufError {
60    fn new(category: CanonicalProtobufErrorCategory, message: impl Into<String>) -> Self {
61        Self {
62            category,
63            message: message.into(),
64        }
65    }
66}
67
68impl fmt::Display for CanonicalProtobufError {
69    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
70        write!(f, "{}: {}", self.category.as_str(), self.message)
71    }
72}
73
74impl Error for CanonicalProtobufError {}
75
76#[derive(Debug, Clone, PartialEq, Eq)]
77/// `CanonicalWireBridgeMetadata` data container.
78pub struct CanonicalWireBridgeMetadata {
79    /// `seq` field for seq.
80    pub seq: u64,
81    /// `cursor` field for cursor.
82    pub cursor: u64,
83    /// `idempotency_key` field for idempotency key.
84    pub idempotency_key: String,
85    /// `attempt` field for attempt.
86    pub attempt: u32,
87    /// `max_attempts` field for max attempts.
88    pub max_attempts: u32,
89    /// `timestamp_ms` field for timestamp ms.
90    pub timestamp_ms: Option<u64>,
91    /// `ack_for_seq` field for ack for seq.
92    pub ack_for_seq: Option<u64>,
93    /// `request_id` field for request id.
94    pub request_id: Option<String>,
95}
96
97#[derive(Debug, Clone, PartialEq, Eq)]
98/// `CanonicalWireBridgeDataBody` variants.
99pub enum CanonicalWireBridgeDataBody {
100    /// `Value` variant.
101    Value(Vec<u8>),
102    /// `WireEdge` variant.
103    WireEdge(CanonicalWireEdgeFrame),
104}
105
106#[derive(Debug, Clone, PartialEq, Eq)]
107/// `CanonicalWireEdgeFrame` data container.
108pub struct CanonicalWireEdgeFrame {
109    /// `kind` field for kind.
110    pub kind: CanonicalWireEdgeKind,
111    /// `edge_id` field for edge id.
112    pub edge_id: String,
113    /// `cause_id` field for cause id.
114    pub cause_id: String,
115    /// `value` field for value.
116    pub value: Option<Vec<u8>>,
117}
118
119#[derive(Debug, Clone, Copy, PartialEq, Eq)]
120/// `CanonicalWireEdgeKind` variants.
121pub enum CanonicalWireEdgeKind {
122    /// `Dirty` variant.
123    Dirty,
124    /// `Data` variant.
125    Data,
126}
127
128#[derive(Debug, Clone, PartialEq, Eq)]
129/// `CanonicalWireBridgePayload` variants.
130pub enum CanonicalWireBridgePayload {
131    /// `Start` variant.
132    Start,
133    /// `Data` variant.
134    Data(CanonicalWireBridgeDataBody),
135    /// `Ack` variant.
136    Ack,
137    /// `Nack` variant.
138    Nack {
139        /// `error` field for `Nack`.
140        error: Option<Vec<u8>>,
141    },
142    /// `Status` variant.
143    Status {
144        /// `status` field for `Status`.
145        status: Vec<u8>,
146    },
147    /// `Error` variant.
148    Error {
149        /// `error` field for `Error`.
150        error: Vec<u8>,
151    },
152    /// `Close` variant.
153    Close {
154        /// `reason` field for `Close`.
155        reason: Option<Vec<u8>>,
156    },
157}
158
159#[derive(Debug, Clone, PartialEq, Eq)]
160/// `CanonicalWireBridgeEnvelope` data container.
161pub struct CanonicalWireBridgeEnvelope {
162    /// `session_id` field for session id.
163    pub session_id: String,
164    /// `metadata` field for metadata.
165    pub metadata: CanonicalWireBridgeMetadata,
166    /// `payload` field for payload.
167    pub payload: CanonicalWireBridgePayload,
168}
169
170#[derive(Debug, Clone, Copy, PartialEq, Eq)]
171/// `WireBridgeProtobufHelperShape` data container.
172pub struct WireBridgeProtobufHelperShape {
173    /// `byte_specific` field for byte specific.
174    pub byte_specific: bool,
175    /// `semantic_wire_bridge_dto` field for semantic wire bridge dto.
176    pub semantic_wire_bridge_dto: bool,
177    /// `core_wire_bridge_options` field for core wire bridge options.
178    pub core_wire_bridge_options: bool,
179    /// `protocol_surface` field for protocol surface.
180    pub protocol_surface: bool,
181    /// `value_codec_registry` field for value codec registry.
182    pub value_codec_registry: bool,
183}
184
185/// `constant` constant.
186pub const WIRE_BRIDGE_PROTOBUF_HELPER_SHAPE: WireBridgeProtobufHelperShape =
187    WireBridgeProtobufHelperShape {
188        byte_specific: true,
189        semantic_wire_bridge_dto: true,
190        core_wire_bridge_options: false,
191        protocol_surface: false,
192        value_codec_registry: false,
193    };
194
195#[derive(Debug, Clone, PartialEq, Eq)]
196/// `WireBridgeProtobufDataBody` variants.
197pub enum WireBridgeProtobufDataBody {
198    /// `Value` variant.
199    Value(Vec<u8>),
200    /// `WireEdge` variant.
201    WireEdge(CanonicalWireEdgeFrame),
202}
203
204#[derive(Debug, Clone, PartialEq, Eq)]
205/// `WireBridgeProtobufPayload` variants.
206pub enum WireBridgeProtobufPayload {
207    /// `Start` variant.
208    Start,
209    /// `Data` variant.
210    Data(WireBridgeProtobufDataBody),
211    /// `Ack` variant.
212    Ack,
213    /// `Nack` variant.
214    Nack {
215        /// `error` field for `Nack`.
216        error: Option<Vec<u8>>,
217    },
218    /// `Status` variant.
219    Status {
220        /// `status` field for `Status`.
221        status: Vec<u8>,
222    },
223    /// `Error` variant.
224    Error {
225        /// `error` field for `Error`.
226        error: Vec<u8>,
227    },
228    /// `Close` variant.
229    Close {
230        /// `reason` field for `Close`.
231        reason: Option<Vec<u8>>,
232    },
233}
234
235#[derive(Debug, Clone, PartialEq, Eq)]
236/// `WireBridgeProtobufEnvelope` data container.
237pub struct WireBridgeProtobufEnvelope {
238    /// `session_id` field for session id.
239    pub session_id: String,
240    /// `metadata` field for metadata.
241    pub metadata: SemanticWireBridgeMetadata,
242    /// `payload` field for payload.
243    pub payload: WireBridgeProtobufPayload,
244}
245
246#[derive(Debug, Clone, Copy, PartialEq, Eq)]
247/// `WireBridgeProtobufStatusKind` variants.
248pub enum WireBridgeProtobufStatusKind {
249    /// `Valid` variant.
250    Valid,
251    /// `Invalid` variant.
252    Invalid,
253}
254
255#[derive(Debug, Clone, PartialEq, Eq)]
256/// `WireBridgeProtobufStatus` data container.
257pub struct WireBridgeProtobufStatus {
258    /// `kind` field for kind.
259    pub kind: WireBridgeProtobufStatusKind,
260}
261
262#[derive(Debug, Clone, PartialEq, Eq)]
263/// `WireBridgeProtobufIssue` data container.
264pub struct WireBridgeProtobufIssue {
265    /// `category` field for category.
266    pub category: CanonicalProtobufErrorCategory,
267    /// `message` field for message.
268    pub message: String,
269}
270
271impl From<CanonicalProtobufError> for WireBridgeProtobufIssue {
272    fn from(error: CanonicalProtobufError) -> Self {
273        Self {
274            category: error.category,
275            message: error.to_string(),
276        }
277    }
278}
279
280#[derive(Debug, Clone, PartialEq, Eq)]
281/// `WireBridgeProtobufDecode` data container.
282pub struct WireBridgeProtobufDecode {
283    /// `envelope` field for envelope.
284    pub envelope: Option<WireBridgeProtobufEnvelope>,
285    /// `status` field for status.
286    pub status: WireBridgeProtobufStatus,
287    /// `issues` field for issues.
288    pub issues: Vec<WireBridgeProtobufIssue>,
289}
290
291#[derive(Debug, Clone, PartialEq, Eq)]
292/// `WireBridgeProtobufEncode` data container.
293pub struct WireBridgeProtobufEncode {
294    /// `bytes` field for bytes.
295    pub bytes: Option<Vec<u8>>,
296    /// `status` field for status.
297    pub status: WireBridgeProtobufStatus,
298    /// `issues` field for issues.
299    pub issues: Vec<WireBridgeProtobufIssue>,
300}
301
302/// Creates or computes `encode_canonical_wire_bridge_envelope`.
303pub fn encode_canonical_wire_bridge_envelope(
304    envelope: &CanonicalWireBridgeEnvelope,
305) -> Result<Vec<u8>, CanonicalProtobufError> {
306    validate_wire_bridge_envelope(envelope)?;
307    let mut out = Writer::default();
308    out.string_field(1, &envelope.session_id);
309    out.message_field(2, &encode_metadata(&envelope.metadata));
310    match &envelope.payload {
311        CanonicalWireBridgePayload::Start => out.message_field(3, &[]),
312        CanonicalWireBridgePayload::Data(body) => {
313            out.message_field(4, &encode_data_payload(body)?);
314        }
315        CanonicalWireBridgePayload::Ack => out.message_field(5, &[]),
316        CanonicalWireBridgePayload::Nack { error } => {
317            out.message_field(6, &encode_optional_bytes_message(1, error.as_deref())?);
318        }
319        CanonicalWireBridgePayload::Status { status } => {
320            out.message_field(7, &encode_required_bytes_message(1, status));
321        }
322        CanonicalWireBridgePayload::Error { error } => {
323            out.message_field(8, &encode_required_bytes_message(1, error));
324        }
325        CanonicalWireBridgePayload::Close { reason } => {
326            out.message_field(9, &encode_optional_bytes_message(1, reason.as_deref())?);
327        }
328    }
329    Ok(out.finish())
330}
331
332/// Creates or computes `decode_canonical_wire_bridge_envelope`.
333pub fn decode_canonical_wire_bridge_envelope(
334    bytes: &[u8],
335) -> Result<CanonicalWireBridgeEnvelope, CanonicalProtobufError> {
336    let envelope = parse_wire_bridge_envelope(bytes)?;
337    validate_wire_bridge_envelope(&envelope)?;
338    let canonical = encode_canonical_wire_bridge_envelope(&envelope)?;
339    if canonical != bytes {
340        return Err(CanonicalProtobufError::new(
341            CanonicalProtobufErrorCategory::NoncanonicalBytes,
342            "WireBridgeEnvelope bytes are not canonical deterministic protobuf",
343        ));
344    }
345    Ok(envelope)
346}
347
348/// Creates or computes `encode_canonical_wire_edge_frame`.
349pub fn encode_canonical_wire_edge_frame(
350    frame: &CanonicalWireEdgeFrame,
351) -> Result<Vec<u8>, CanonicalProtobufError> {
352    validate_wire_edge_frame(frame)?;
353    Ok(encode_wire_edge_frame(frame))
354}
355
356/// Creates or computes `decode_canonical_wire_edge_frame`.
357pub fn decode_canonical_wire_edge_frame(
358    bytes: &[u8],
359) -> Result<CanonicalWireEdgeFrame, CanonicalProtobufError> {
360    let frame = parse_wire_edge_frame(bytes)?;
361    validate_wire_edge_frame(&frame)?;
362    let canonical = encode_wire_edge_frame(&frame);
363    if canonical != bytes {
364        return Err(CanonicalProtobufError::new(
365            CanonicalProtobufErrorCategory::NoncanonicalBytes,
366            "WireEdgeFrame bytes are not canonical deterministic protobuf",
367        ));
368    }
369    Ok(frame)
370}
371
372#[must_use]
373/// Creates or computes `decode_wire_bridge_protobuf_bytes`.
374pub fn decode_wire_bridge_protobuf_bytes(bytes: &[u8]) -> WireBridgeProtobufDecode {
375    match decode_canonical_wire_bridge_envelope(bytes) {
376        Ok(envelope) => WireBridgeProtobufDecode {
377            envelope: Some(canonical_to_protobuf_envelope(envelope)),
378            status: WireBridgeProtobufStatus {
379                kind: WireBridgeProtobufStatusKind::Valid,
380            },
381            issues: Vec::new(),
382        },
383        Err(error) => WireBridgeProtobufDecode {
384            envelope: None,
385            status: WireBridgeProtobufStatus {
386                kind: WireBridgeProtobufStatusKind::Invalid,
387            },
388            issues: vec![error.into()],
389        },
390    }
391}
392
393#[must_use]
394/// Creates or computes `encode_wire_bridge_protobuf_bytes`.
395pub fn encode_wire_bridge_protobuf_bytes(
396    envelope: &WireBridgeProtobufEnvelope,
397) -> WireBridgeProtobufEncode {
398    match protobuf_to_canonical_envelope(envelope)
399        .and_then(|canonical| encode_canonical_wire_bridge_envelope(&canonical))
400    {
401        Ok(bytes) => WireBridgeProtobufEncode {
402            bytes: Some(bytes),
403            status: WireBridgeProtobufStatus {
404                kind: WireBridgeProtobufStatusKind::Valid,
405            },
406            issues: Vec::new(),
407        },
408        Err(error) => WireBridgeProtobufEncode {
409            bytes: None,
410            status: WireBridgeProtobufStatus {
411                kind: WireBridgeProtobufStatusKind::Invalid,
412            },
413            issues: vec![error.into()],
414        },
415    }
416}
417
418fn canonical_to_protobuf_envelope(
419    envelope: CanonicalWireBridgeEnvelope,
420) -> WireBridgeProtobufEnvelope {
421    WireBridgeProtobufEnvelope {
422        session_id: envelope.session_id,
423        metadata: SemanticWireBridgeMetadata {
424            seq: envelope.metadata.seq,
425            cursor: envelope.metadata.cursor,
426            idempotency_key: envelope.metadata.idempotency_key,
427            attempt: envelope.metadata.attempt,
428            max_attempts: envelope.metadata.max_attempts,
429            timestamp_ms: envelope.metadata.timestamp_ms,
430            ack_for_seq: envelope.metadata.ack_for_seq,
431            request_id: envelope.metadata.request_id,
432        },
433        payload: match envelope.payload {
434            CanonicalWireBridgePayload::Start => WireBridgeProtobufPayload::Start,
435            CanonicalWireBridgePayload::Data(CanonicalWireBridgeDataBody::Value(value)) => {
436                WireBridgeProtobufPayload::Data(WireBridgeProtobufDataBody::Value(value))
437            }
438            CanonicalWireBridgePayload::Data(CanonicalWireBridgeDataBody::WireEdge(frame)) => {
439                WireBridgeProtobufPayload::Data(WireBridgeProtobufDataBody::WireEdge(frame))
440            }
441            CanonicalWireBridgePayload::Ack => WireBridgeProtobufPayload::Ack,
442            CanonicalWireBridgePayload::Nack { error } => WireBridgeProtobufPayload::Nack { error },
443            CanonicalWireBridgePayload::Status { status } => {
444                WireBridgeProtobufPayload::Status { status }
445            }
446            CanonicalWireBridgePayload::Error { error } => {
447                WireBridgeProtobufPayload::Error { error }
448            }
449            CanonicalWireBridgePayload::Close { reason } => {
450                WireBridgeProtobufPayload::Close { reason }
451            }
452        },
453    }
454}
455
456fn protobuf_to_canonical_envelope(
457    envelope: &WireBridgeProtobufEnvelope,
458) -> Result<CanonicalWireBridgeEnvelope, CanonicalProtobufError> {
459    let canonical = CanonicalWireBridgeEnvelope {
460        session_id: envelope.session_id.clone(),
461        metadata: CanonicalWireBridgeMetadata {
462            seq: envelope.metadata.seq,
463            cursor: envelope.metadata.cursor,
464            idempotency_key: envelope.metadata.idempotency_key.clone(),
465            attempt: envelope.metadata.attempt,
466            max_attempts: envelope.metadata.max_attempts,
467            timestamp_ms: envelope.metadata.timestamp_ms,
468            ack_for_seq: envelope.metadata.ack_for_seq,
469            request_id: envelope.metadata.request_id.clone(),
470        },
471        payload: match &envelope.payload {
472            WireBridgeProtobufPayload::Start => CanonicalWireBridgePayload::Start,
473            WireBridgeProtobufPayload::Data(WireBridgeProtobufDataBody::Value(value)) => {
474                CanonicalWireBridgePayload::Data(CanonicalWireBridgeDataBody::Value(value.clone()))
475            }
476            WireBridgeProtobufPayload::Data(WireBridgeProtobufDataBody::WireEdge(frame)) => {
477                CanonicalWireBridgePayload::Data(CanonicalWireBridgeDataBody::WireEdge(
478                    frame.clone(),
479                ))
480            }
481            WireBridgeProtobufPayload::Ack => CanonicalWireBridgePayload::Ack,
482            WireBridgeProtobufPayload::Nack { error } => CanonicalWireBridgePayload::Nack {
483                error: error.clone(),
484            },
485            WireBridgeProtobufPayload::Status { status } => CanonicalWireBridgePayload::Status {
486                status: status.clone(),
487            },
488            WireBridgeProtobufPayload::Error { error } => CanonicalWireBridgePayload::Error {
489                error: error.clone(),
490            },
491            WireBridgeProtobufPayload::Close { reason } => CanonicalWireBridgePayload::Close {
492                reason: reason.clone(),
493            },
494        },
495    };
496    validate_wire_bridge_envelope(&canonical)?;
497    Ok(canonical)
498}
499
500fn parse_wire_bridge_envelope(
501    bytes: &[u8],
502) -> Result<CanonicalWireBridgeEnvelope, CanonicalProtobufError> {
503    let fields = read_fields(
504        bytes,
505        &[
506            (1, 2),
507            (2, 2),
508            (3, 2),
509            (4, 2),
510            (5, 2),
511            (6, 2),
512            (7, 2),
513            (8, 2),
514            (9, 2),
515        ],
516        "WireBridgeEnvelope",
517    )?;
518    let session = bytes_field(&fields, 1);
519    let metadata = bytes_field(&fields, 2);
520    let payload_fields: Vec<&Field> = fields
521        .iter()
522        .filter(|field| (3..=9).contains(&field.no))
523        .collect();
524    if session.is_none() || metadata.is_none() || payload_fields.is_empty() {
525        return Err(CanonicalProtobufError::new(
526            CanonicalProtobufErrorCategory::MissingRequired,
527            "WireBridgeEnvelope missing required fields",
528        ));
529    }
530    if payload_fields.len() != 1 {
531        return Err(CanonicalProtobufError::new(
532            CanonicalProtobufErrorCategory::InvalidOneof,
533            "WireBridgeEnvelope payload oneof has multiple cases",
534        ));
535    }
536    let payload_field = payload_fields[0];
537    let payload_bytes = payload_field.as_bytes()?;
538    Ok(CanonicalWireBridgeEnvelope {
539        session_id: utf8_string(session.unwrap(), "session_id")?,
540        metadata: parse_metadata(metadata.unwrap())?,
541        payload: parse_envelope_payload(payload_field.no, payload_bytes)?,
542    })
543}
544
545fn parse_envelope_payload(
546    field_no: u32,
547    bytes: &[u8],
548) -> Result<CanonicalWireBridgePayload, CanonicalProtobufError> {
549    match field_no {
550        3 => {
551            require_empty_message(bytes, "start")?;
552            Ok(CanonicalWireBridgePayload::Start)
553        }
554        4 => Ok(CanonicalWireBridgePayload::Data(parse_data_payload(bytes)?)),
555        5 => {
556            require_empty_message(bytes, "ack")?;
557            Ok(CanonicalWireBridgePayload::Ack)
558        }
559        6 => Ok(CanonicalWireBridgePayload::Nack {
560            error: parse_optional_bytes_payload(bytes, "nack")?,
561        }),
562        7 => Ok(CanonicalWireBridgePayload::Status {
563            status: parse_required_bytes_payload(bytes, "status")?,
564        }),
565        8 => Ok(CanonicalWireBridgePayload::Error {
566            error: parse_required_bytes_payload(bytes, "error")?,
567        }),
568        9 => Ok(CanonicalWireBridgePayload::Close {
569            reason: parse_optional_bytes_payload(bytes, "close")?,
570        }),
571        _ => Err(CanonicalProtobufError::new(
572            CanonicalProtobufErrorCategory::UnknownField,
573            "unknown envelope payload field",
574        )),
575    }
576}
577
578fn parse_metadata(bytes: &[u8]) -> Result<CanonicalWireBridgeMetadata, CanonicalProtobufError> {
579    let fields = read_fields(
580        bytes,
581        &[
582            (1, 0),
583            (2, 0),
584            (3, 2),
585            (4, 0),
586            (5, 0),
587            (6, 0),
588            (7, 0),
589            (8, 2),
590        ],
591        "WireBridgeMetadata",
592    )?;
593    let seq = uint_field(&fields, 1);
594    let cursor = uint_field(&fields, 2);
595    let key = bytes_field(&fields, 3);
596    let attempt = uint_field(&fields, 4);
597    let max_attempts = uint_field(&fields, 5);
598    if seq.is_none()
599        || cursor.is_none()
600        || key.is_none()
601        || attempt.is_none()
602        || max_attempts.is_none()
603    {
604        return Err(CanonicalProtobufError::new(
605            CanonicalProtobufErrorCategory::MissingRequired,
606            "WireBridgeMetadata missing required fields",
607        ));
608    }
609    let timestamp_ms = uint_field(&fields, 6);
610    let ack_for_seq = uint_field(&fields, 7);
611    let request_id = bytes_field(&fields, 8);
612    if timestamp_ms == Some(0) || ack_for_seq == Some(0) || request_id == Some(&[][..]) {
613        return Err(CanonicalProtobufError::new(
614            CanonicalProtobufErrorCategory::DefaultEmission,
615            "optional metadata default value was emitted",
616        ));
617    }
618    Ok(CanonicalWireBridgeMetadata {
619        seq: seq.unwrap(),
620        cursor: cursor.unwrap(),
621        idempotency_key: utf8_string(key.unwrap(), "idempotency_key")?,
622        attempt: uint32(attempt.unwrap(), "attempt")?,
623        max_attempts: uint32(max_attempts.unwrap(), "max_attempts")?,
624        timestamp_ms,
625        ack_for_seq,
626        request_id: request_id
627            .map(|value| utf8_string(value, "request_id"))
628            .transpose()?,
629    })
630}
631
632fn parse_data_payload(bytes: &[u8]) -> Result<CanonicalWireBridgeDataBody, CanonicalProtobufError> {
633    let fields = read_fields(bytes, &[(1, 2), (2, 2)], "WireBridgeDataPayload")?;
634    let value = bytes_field(&fields, 1);
635    let wire_edge = bytes_field(&fields, 2);
636    match (value, wire_edge) {
637        (Some(_), Some(_)) => Err(CanonicalProtobufError::new(
638            CanonicalProtobufErrorCategory::InvalidOneof,
639            "WireBridgeDataPayload body has multiple cases",
640        )),
641        (Some(value), None) => Ok(CanonicalWireBridgeDataBody::Value(value.to_vec())),
642        (None, Some(wire_edge)) => Ok(CanonicalWireBridgeDataBody::WireEdge(
643            parse_wire_edge_frame(wire_edge)?,
644        )),
645        (None, None) => Err(CanonicalProtobufError::new(
646            CanonicalProtobufErrorCategory::MissingRequired,
647            "WireBridgeDataPayload missing body",
648        )),
649    }
650}
651
652fn parse_wire_edge_frame(bytes: &[u8]) -> Result<CanonicalWireEdgeFrame, CanonicalProtobufError> {
653    let fields = read_fields(bytes, &[(1, 0), (2, 2), (3, 2), (4, 2)], "WireEdgeFrame")?;
654    let kind = uint_field(&fields, 1);
655    let edge = bytes_field(&fields, 2);
656    let cause = bytes_field(&fields, 3);
657    let value = bytes_field(&fields, 4);
658    if kind.is_none() || edge.is_none() || cause.is_none() {
659        return Err(CanonicalProtobufError::new(
660            CanonicalProtobufErrorCategory::MissingRequired,
661            "WireEdgeFrame missing required fields",
662        ));
663    }
664    let edge_id = utf8_string(edge.unwrap(), "edge_id")?;
665    let cause_id = utf8_string(cause.unwrap(), "cause_id")?;
666    match kind.unwrap() {
667        1 => {
668            if value.is_some() {
669                return Err(CanonicalProtobufError::new(
670                    CanonicalProtobufErrorCategory::InvalidWireEdge,
671                    "DIRTY WireEdgeFrame must not carry value",
672                ));
673            }
674            Ok(CanonicalWireEdgeFrame {
675                kind: CanonicalWireEdgeKind::Dirty,
676                edge_id,
677                cause_id,
678                value: None,
679            })
680        }
681        2 => {
682            let value = value.ok_or_else(|| {
683                CanonicalProtobufError::new(
684                    CanonicalProtobufErrorCategory::InvalidWireEdge,
685                    "DATA WireEdgeFrame requires value",
686                )
687            })?;
688            Ok(CanonicalWireEdgeFrame {
689                kind: CanonicalWireEdgeKind::Data,
690                edge_id,
691                cause_id,
692                value: Some(value.to_vec()),
693            })
694        }
695        _ => Err(CanonicalProtobufError::new(
696            CanonicalProtobufErrorCategory::InvalidWireEdge,
697            "WireEdgeFrame kind is invalid",
698        )),
699    }
700}
701
702fn validate_wire_bridge_envelope(
703    envelope: &CanonicalWireBridgeEnvelope,
704) -> Result<(), CanonicalProtobufError> {
705    if envelope.session_id.is_empty()
706        || envelope.metadata.seq == 0
707        || envelope.metadata.attempt == 0
708        || envelope.metadata.max_attempts < envelope.metadata.attempt
709        || envelope.metadata.idempotency_key.is_empty()
710    {
711        return Err(CanonicalProtobufError::new(
712            CanonicalProtobufErrorCategory::MissingRequired,
713            "WireBridgeEnvelope required semantics are invalid",
714        ));
715    }
716    if matches!(
717        envelope.payload,
718        CanonicalWireBridgePayload::Ack | CanonicalWireBridgePayload::Nack { .. }
719    ) && envelope.metadata.ack_for_seq.is_none()
720    {
721        return Err(CanonicalProtobufError::new(
722            CanonicalProtobufErrorCategory::MissingRequired,
723            "ACK/NACK requires metadata.ack_for_seq",
724        ));
725    }
726    if envelope.metadata.timestamp_ms == Some(0)
727        || envelope.metadata.ack_for_seq == Some(0)
728        || envelope.metadata.request_id.as_deref() == Some("")
729    {
730        return Err(CanonicalProtobufError::new(
731            CanonicalProtobufErrorCategory::DefaultEmission,
732            "optional metadata default value was emitted",
733        ));
734    }
735    if let CanonicalWireBridgePayload::Data(CanonicalWireBridgeDataBody::WireEdge(frame)) =
736        &envelope.payload
737    {
738        validate_wire_edge_frame(frame)?;
739    }
740    match &envelope.payload {
741        CanonicalWireBridgePayload::Nack { error } if error.as_deref() == Some(&[]) => {
742            return Err(CanonicalProtobufError::new(
743                CanonicalProtobufErrorCategory::DefaultEmission,
744                "nack optional bytes default value was emitted",
745            ));
746        }
747        CanonicalWireBridgePayload::Status { status } if status.is_empty() => {
748            return Err(CanonicalProtobufError::new(
749                CanonicalProtobufErrorCategory::MissingRequired,
750                "status payload bytes must be non-empty",
751            ));
752        }
753        CanonicalWireBridgePayload::Error { error } if error.is_empty() => {
754            return Err(CanonicalProtobufError::new(
755                CanonicalProtobufErrorCategory::MissingRequired,
756                "error payload bytes must be non-empty",
757            ));
758        }
759        CanonicalWireBridgePayload::Close { reason } if reason.as_deref() == Some(&[]) => {
760            return Err(CanonicalProtobufError::new(
761                CanonicalProtobufErrorCategory::DefaultEmission,
762                "close optional bytes default value was emitted",
763            ));
764        }
765        _ => {}
766    }
767    Ok(())
768}
769
770fn validate_wire_edge_frame(frame: &CanonicalWireEdgeFrame) -> Result<(), CanonicalProtobufError> {
771    if frame.edge_id.is_empty() || frame.cause_id.is_empty() {
772        return Err(CanonicalProtobufError::new(
773            CanonicalProtobufErrorCategory::MissingRequired,
774            "WireEdgeFrame edge_id/cause_id must be non-empty",
775        ));
776    }
777    match frame.kind {
778        CanonicalWireEdgeKind::Dirty if frame.value.is_some() => Err(CanonicalProtobufError::new(
779            CanonicalProtobufErrorCategory::InvalidWireEdge,
780            "DIRTY WireEdgeFrame must not carry value",
781        )),
782        CanonicalWireEdgeKind::Data if frame.value.is_none() => Err(CanonicalProtobufError::new(
783            CanonicalProtobufErrorCategory::InvalidWireEdge,
784            "DATA WireEdgeFrame requires value",
785        )),
786        _ => Ok(()),
787    }
788}
789
790fn encode_metadata(metadata: &CanonicalWireBridgeMetadata) -> Vec<u8> {
791    let mut out = Writer::default();
792    out.varint_field(1, metadata.seq);
793    out.varint_field(2, metadata.cursor);
794    out.string_field(3, &metadata.idempotency_key);
795    out.varint_field(4, u64::from(metadata.attempt));
796    out.varint_field(5, u64::from(metadata.max_attempts));
797    if let Some(timestamp_ms) = metadata.timestamp_ms {
798        out.varint_field(6, timestamp_ms);
799    }
800    if let Some(ack_for_seq) = metadata.ack_for_seq {
801        out.varint_field(7, ack_for_seq);
802    }
803    if let Some(request_id) = &metadata.request_id {
804        out.string_field(8, request_id);
805    }
806    out.finish()
807}
808
809fn encode_data_payload(
810    body: &CanonicalWireBridgeDataBody,
811) -> Result<Vec<u8>, CanonicalProtobufError> {
812    let mut out = Writer::default();
813    match body {
814        CanonicalWireBridgeDataBody::Value(value) => out.bytes_field(1, value),
815        CanonicalWireBridgeDataBody::WireEdge(frame) => {
816            out.message_field(2, &encode_canonical_wire_edge_frame(frame)?);
817        }
818    }
819    Ok(out.finish())
820}
821
822fn encode_wire_edge_frame(frame: &CanonicalWireEdgeFrame) -> Vec<u8> {
823    let mut out = Writer::default();
824    out.varint_field(
825        1,
826        match frame.kind {
827            CanonicalWireEdgeKind::Dirty => 1,
828            CanonicalWireEdgeKind::Data => 2,
829        },
830    );
831    out.string_field(2, &frame.edge_id);
832    out.string_field(3, &frame.cause_id);
833    if let Some(value) = &frame.value {
834        out.bytes_field(4, value);
835    }
836    out.finish()
837}
838
839fn encode_required_bytes_message(field_no: u32, value: &[u8]) -> Vec<u8> {
840    let mut out = Writer::default();
841    out.bytes_field(field_no, value);
842    out.finish()
843}
844
845fn encode_optional_bytes_message(
846    field_no: u32,
847    value: Option<&[u8]>,
848) -> Result<Vec<u8>, CanonicalProtobufError> {
849    let mut out = Writer::default();
850    if let Some(value) = value {
851        if value.is_empty() {
852            return Err(CanonicalProtobufError::new(
853                CanonicalProtobufErrorCategory::DefaultEmission,
854                "optional bytes default value must be omitted",
855            ));
856        }
857        out.bytes_field(field_no, value);
858    }
859    Ok(out.finish())
860}
861
862fn parse_required_bytes_payload(
863    bytes: &[u8],
864    name: &str,
865) -> Result<Vec<u8>, CanonicalProtobufError> {
866    let fields = read_fields(bytes, &[(1, 2)], &format!("WireBridge{name}Payload"))?;
867    let value = bytes_field(&fields, 1).ok_or_else(|| {
868        CanonicalProtobufError::new(
869            CanonicalProtobufErrorCategory::MissingRequired,
870            format!("{name} payload missing required bytes"),
871        )
872    })?;
873    if value.is_empty() {
874        return Err(CanonicalProtobufError::new(
875            CanonicalProtobufErrorCategory::MissingRequired,
876            format!("{name} payload bytes must be non-empty"),
877        ));
878    }
879    Ok(value.to_vec())
880}
881
882fn parse_optional_bytes_payload(
883    bytes: &[u8],
884    name: &str,
885) -> Result<Option<Vec<u8>>, CanonicalProtobufError> {
886    let fields = read_fields(bytes, &[(1, 2)], &format!("WireBridge{name}Payload"))?;
887    let value = bytes_field(&fields, 1);
888    if value == Some(&[]) {
889        return Err(CanonicalProtobufError::new(
890            CanonicalProtobufErrorCategory::DefaultEmission,
891            format!("{name} optional bytes default value was emitted"),
892        ));
893    }
894    Ok(value.map(<[u8]>::to_vec))
895}
896
897fn require_empty_message(bytes: &[u8], name: &str) -> Result<(), CanonicalProtobufError> {
898    if bytes.is_empty() {
899        Ok(())
900    } else {
901        Err(CanonicalProtobufError::new(
902            CanonicalProtobufErrorCategory::UnknownField,
903            format!("{name} payload must be empty"),
904        ))
905    }
906}
907
908#[derive(Debug, Clone, PartialEq, Eq)]
909struct Field {
910    no: u32,
911    wire_type: u8,
912    value: FieldValue,
913}
914
915impl Field {
916    fn as_bytes(&self) -> Result<&[u8], CanonicalProtobufError> {
917        match &self.value {
918            FieldValue::Bytes(value) => Ok(value),
919            FieldValue::Varint(_) => Err(CanonicalProtobufError::new(
920                CanonicalProtobufErrorCategory::Malformed,
921                "expected length-delimited field",
922            )),
923        }
924    }
925}
926
927#[derive(Debug, Clone, PartialEq, Eq)]
928enum FieldValue {
929    Varint(u64),
930    Bytes(Vec<u8>),
931}
932
933fn read_fields(
934    bytes: &[u8],
935    allowed: &[(u32, u8)],
936    message_name: &str,
937) -> Result<Vec<Field>, CanonicalProtobufError> {
938    let mut reader = Reader::new(bytes);
939    let mut fields = Vec::new();
940    let mut seen = HashSet::new();
941    while !reader.done() {
942        let key = reader.varint()?;
943        let field_no = u32::try_from(key >> 3).map_err(|_| {
944            CanonicalProtobufError::new(
945                CanonicalProtobufErrorCategory::Malformed,
946                "field number is too large",
947            )
948        })?;
949        let wire_type = u8::try_from(key & 7).expect("wire type fits in u8");
950        let expected = allowed
951            .iter()
952            .find_map(|(no, wt)| (*no == field_no).then_some(*wt));
953        let Some(expected_wire_type) = expected else {
954            return Err(CanonicalProtobufError::new(
955                CanonicalProtobufErrorCategory::UnknownField,
956                format!("{message_name} contains unknown field {field_no}"),
957            ));
958        };
959        if field_no == 0 || wire_type != expected_wire_type {
960            return Err(CanonicalProtobufError::new(
961                CanonicalProtobufErrorCategory::Malformed,
962                format!("{message_name} field {field_no} has wrong wire type"),
963            ));
964        }
965        if !seen.insert(field_no) {
966            return Err(CanonicalProtobufError::new(
967                CanonicalProtobufErrorCategory::DuplicateSingular,
968                format!("{message_name} field {field_no} is duplicated"),
969            ));
970        }
971        let value = match wire_type {
972            0 => FieldValue::Varint(reader.varint()?),
973            2 => FieldValue::Bytes(reader.read_bytes()?),
974            _ => {
975                return Err(CanonicalProtobufError::new(
976                    CanonicalProtobufErrorCategory::Malformed,
977                    "unsupported wire type",
978                ));
979            }
980        };
981        fields.push(Field {
982            no: field_no,
983            wire_type,
984            value,
985        });
986    }
987    Ok(fields)
988}
989
990fn uint_field(fields: &[Field], no: u32) -> Option<u64> {
991    fields.iter().find_map(|field| match field {
992        Field {
993            no: field_no,
994            value: FieldValue::Varint(value),
995            ..
996        } if *field_no == no => Some(*value),
997        _ => None,
998    })
999}
1000
1001fn bytes_field(fields: &[Field], no: u32) -> Option<&[u8]> {
1002    fields.iter().find_map(|field| match field {
1003        Field {
1004            no: field_no,
1005            value: FieldValue::Bytes(value),
1006            ..
1007        } if *field_no == no => Some(value.as_slice()),
1008        _ => None,
1009    })
1010}
1011
1012fn utf8_string(bytes: &[u8], field: &str) -> Result<String, CanonicalProtobufError> {
1013    let value = std::str::from_utf8(bytes).map_err(|_| {
1014        CanonicalProtobufError::new(
1015            CanonicalProtobufErrorCategory::Malformed,
1016            format!("{field} is not valid utf-8"),
1017        )
1018    })?;
1019    if value.is_empty() {
1020        return Err(CanonicalProtobufError::new(
1021            CanonicalProtobufErrorCategory::MissingRequired,
1022            format!("{field} must be non-empty"),
1023        ));
1024    }
1025    Ok(value.to_owned())
1026}
1027
1028fn uint32(value: u64, field: &str) -> Result<u32, CanonicalProtobufError> {
1029    u32::try_from(value).map_err(|_| {
1030        CanonicalProtobufError::new(
1031            CanonicalProtobufErrorCategory::Malformed,
1032            format!("{field} exceeds uint32"),
1033        )
1034    })
1035}
1036
1037struct Reader<'a> {
1038    bytes: &'a [u8],
1039    offset: usize,
1040}
1041
1042impl<'a> Reader<'a> {
1043    fn new(bytes: &'a [u8]) -> Self {
1044        Self { bytes, offset: 0 }
1045    }
1046
1047    fn done(&self) -> bool {
1048        self.offset == self.bytes.len()
1049    }
1050
1051    fn varint(&mut self) -> Result<u64, CanonicalProtobufError> {
1052        let mut shift = 0u32;
1053        let mut result = 0u64;
1054        for idx in 0..10 {
1055            let byte = *self.bytes.get(self.offset).ok_or_else(|| {
1056                CanonicalProtobufError::new(
1057                    CanonicalProtobufErrorCategory::Malformed,
1058                    "truncated varint",
1059                )
1060            })?;
1061            self.offset += 1;
1062            if idx == 9 && byte & 0xfe != 0 {
1063                return Err(CanonicalProtobufError::new(
1064                    CanonicalProtobufErrorCategory::Malformed,
1065                    "varint exceeds 64 bits",
1066                ));
1067            }
1068            result |= u64::from(byte & 0x7f) << shift;
1069            if byte & 0x80 == 0 {
1070                return Ok(result);
1071            }
1072            shift += 7;
1073        }
1074        Err(CanonicalProtobufError::new(
1075            CanonicalProtobufErrorCategory::Malformed,
1076            "varint exceeds 64 bits",
1077        ))
1078    }
1079
1080    fn read_bytes(&mut self) -> Result<Vec<u8>, CanonicalProtobufError> {
1081        let len = usize::try_from(self.varint()?).map_err(|_| {
1082            CanonicalProtobufError::new(
1083                CanonicalProtobufErrorCategory::Malformed,
1084                "length-delimited field is too large",
1085            )
1086        })?;
1087        let end = self.offset.checked_add(len).ok_or_else(|| {
1088            CanonicalProtobufError::new(
1089                CanonicalProtobufErrorCategory::Malformed,
1090                "length-delimited field length overflows",
1091            )
1092        })?;
1093        if end > self.bytes.len() {
1094            return Err(CanonicalProtobufError::new(
1095                CanonicalProtobufErrorCategory::Malformed,
1096                "length-delimited field is truncated",
1097            ));
1098        }
1099        let out = self.bytes[self.offset..end].to_vec();
1100        self.offset = end;
1101        Ok(out)
1102    }
1103}
1104
1105#[derive(Default)]
1106struct Writer {
1107    bytes: Vec<u8>,
1108}
1109
1110impl Writer {
1111    fn varint_field(&mut self, field_no: u32, value: u64) {
1112        self.tag(field_no, 0);
1113        self.varint(value);
1114    }
1115
1116    fn bytes_field(&mut self, field_no: u32, value: &[u8]) {
1117        self.tag(field_no, 2);
1118        self.varint(value.len() as u64);
1119        self.bytes.extend_from_slice(value);
1120    }
1121
1122    fn string_field(&mut self, field_no: u32, value: &str) {
1123        self.bytes_field(field_no, value.as_bytes());
1124    }
1125
1126    fn message_field(&mut self, field_no: u32, value: &[u8]) {
1127        self.bytes_field(field_no, value);
1128    }
1129
1130    fn finish(self) -> Vec<u8> {
1131        self.bytes
1132    }
1133
1134    fn tag(&mut self, field_no: u32, wire_type: u8) {
1135        self.varint(u64::from((field_no << 3) | u32::from(wire_type)));
1136    }
1137
1138    fn varint(&mut self, mut value: u64) {
1139        loop {
1140            let mut byte = (value & 0x7f) as u8;
1141            value >>= 7;
1142            if value != 0 {
1143                byte |= 0x80;
1144            }
1145            self.bytes.push(byte);
1146            if value == 0 {
1147                break;
1148            }
1149        }
1150    }
1151}
graphrefly/adapters/bridge_protobuf.rs

graphrefly/adapters/
bridge_protobuf.rs
