feat(replays): Add ReplayVideo envelope item type

CHANGELOG.md

@@ -11,6 +11,7 @@
 **Features**:
 
 - Add protobuf support for ingesting OpenTelemetry spans and use official `opentelemetry-proto` generated structs. ([#3044](https://github.com/getsentry/relay/pull/3044))
+- Adds ReplayVideo envelope-item type. ([#3105](https://github.com/getsentry/relay/pull/3105))
 
 **Internal**:
 

relay-replays/Cargo.toml

@@ -10,6 +10,7 @@ license-file = "../LICENSE.md"
 publish = false
 
 [dependencies]
+bytes = { version = "1.4.0" }
 flate2 = "1.0.19"
 once_cell = { workspace = true }
 relay-common = { path = "../relay-common" }

relay-server/Cargo.toml

@@ -97,6 +97,7 @@ reqwest = { version = "0.11.1", features = [
 rmp-serde = "1.1.1"
 rust-embed = { version = "8.0.0", optional = true }
 serde = { workspace = true }
+serde_bytes = { version = "0.11.14" }
 serde_json = { workspace = true }
 smallvec = { workspace = true, features = ["drain_filter"] }
 sqlx = { version = "0.7.3", features = [

relay-server/src/envelope.rs

@@ -113,6 +113,8 @@ pub enum ItemType {
     ReplayEvent,
     /// Replay Recording data.
     ReplayRecording,
+    /// Replay Video data.
+    ReplayVideo,
     /// Monitor check-in encoded as JSON.
     CheckIn,
     /// A standalone span.
@@ -167,6 +169,7 @@ impl ItemType {
             Self::Profile => "profile",
             Self::ReplayEvent => "replay_event",
             Self::ReplayRecording => "replay_recording",
+            Self::ReplayVideo => "replay_video",
             Self::CheckIn => "check_in",
             Self::Span => "span",
             Self::OtelSpan => "otel_span",
@@ -213,6 +216,7 @@ impl std::str::FromStr for ItemType {
             "profile" => Self::Profile,
             "replay_event" => Self::ReplayEvent,
             "replay_recording" => Self::ReplayRecording,
+            "replay_video" => Self::ReplayVideo,
             "check_in" => Self::CheckIn,
             "span" => Self::Span,
             "otel_span" => Self::OtelSpan,
@@ -509,6 +513,11 @@ pub struct ItemHeaders {
     #[serde(default, skip)]
     replay_combined_payload: bool,
 
+    /// The parsed replay-event and replay-recording payloads.
+    /// NOTE: This is internal-only and not exposed into the Envelope.
+    #[serde(default, skip)]
+    replay_video_events: Option<(Vec<u8>, Vec<u8>)>,
+
     /// Contains the amount of events this item was generated and aggregated from.
     ///
     /// A [metrics buckets](`ItemType::MetricBuckets`) item contains metrics extracted and
@@ -597,6 +606,7 @@ impl Item {
                 routing_hint: None,
                 rate_limited: false,
                 replay_combined_payload: false,
+                replay_video_events: None,
                 source_quantities: None,
                 sample_rates: None,
                 other: BTreeMap::new(),
@@ -652,7 +662,9 @@ impl Item {
             } else {
                 DataCategory::Profile
             }),
-            ItemType::ReplayEvent | ItemType::ReplayRecording => Some(DataCategory::Replay),
+            ItemType::ReplayEvent | ItemType::ReplayRecording | ItemType::ReplayVideo => {
+                Some(DataCategory::Replay)
+            }
             ItemType::ClientReport => None,
             ItemType::CheckIn => Some(DataCategory::Monitor),
             ItemType::Span | ItemType::OtelSpan => Some(if indexed {
@@ -771,6 +783,17 @@ impl Item {
         self.headers.replay_combined_payload = combined_payload;
     }
 
+    /// Returns the payload's replay video events.
+    #[cfg(feature = "processing")]
+    pub fn replay_video_events(&self) -> Option<(Vec<u8>, Vec<u8>)> {
+        self.headers.replay_video_events.clone()
+    }
+
+    /// Set the replay video events attribute for this item.
+    pub fn set_replay_video_events(&mut self, replay_event: Vec<u8>, replay_recording: Vec<u8>) {
+        self.headers.replay_video_events = Some((replay_event, replay_recording));
+    }
+
     /// Sets sample rates for this item.
     pub fn set_sample_rates(&mut self, sample_rates: Value) {
         if matches!(sample_rates, Value::Array(ref a) if !a.is_empty()) {
@@ -864,6 +887,7 @@ impl Item {
             | ItemType::ClientReport
             | ItemType::ReplayEvent
             | ItemType::ReplayRecording
+            | ItemType::ReplayVideo
             | ItemType::Profile
             | ItemType::CheckIn
             | ItemType::Span
@@ -898,6 +922,7 @@ impl Item {
             ItemType::MetricMeta => false,
             ItemType::ClientReport => false,
             ItemType::ReplayRecording => false,
+            ItemType::ReplayVideo => false,
             ItemType::Profile => true,
             ItemType::CheckIn => false,
             ItemType::Span => false,
@@ -1733,6 +1758,22 @@ mod tests {
         assert_eq!(items[0].ty(), &ItemType::ReplayRecording);
     }
 
+    #[test]
+    fn test_deserialize_envelope_replay_video() {
+        let bytes = Bytes::from(
+            "\
+             {\"event_id\":\"9ec79c33ec9942ab8353589fcb2e04dc\",\"dsn\":\"https://e12d836b15bb49d7bbf99e64295d995b:@sentry.io/42\"}\n\
+             {\"type\":\"replay_video\"}\n\
+             helloworld\n\
+             ",
+        );
+
+        let envelope = Envelope::parse_bytes(bytes).unwrap();
+        assert_eq!(envelope.len(), 1);
+        let items: Vec<_> = envelope.items().collect();
+        assert_eq!(items[0].ty(), &ItemType::ReplayVideo);
+    }
+
     #[test]
     fn test_deserialize_envelope_view_hierarchy() {
         let bytes = Bytes::from(

relay-server/src/services/outcome.rs

@@ -366,6 +366,7 @@ pub enum DiscardReason {
     InvalidReplayEventNoPayload,
     InvalidReplayEventPii,
     InvalidReplayRecordingEvent,
+    InvalidReplayVideoEvent,
 
     /// (Relay) Profiling related discard reasons
     Profiling(&'static str),
@@ -413,6 +414,7 @@ impl DiscardReason {
             DiscardReason::InvalidReplayEventNoPayload => "invalid_replay_no_payload",
             DiscardReason::InvalidReplayEventPii => "invalid_replay_pii_scrubber_failed",
             DiscardReason::InvalidReplayRecordingEvent => "invalid_replay_recording",
+            DiscardReason::InvalidReplayVideoEvent => "invalid_replay_video",
             DiscardReason::Profiling(reason) => reason,
             DiscardReason::InvalidSpan => "invalid_span",
         }

relay-server/src/services/processor.rs

@@ -204,7 +204,7 @@ impl ProcessingGroup {
         let replay_items = envelope.take_items_by(|item| {
             matches!(
                 item.ty(),
-                &ItemType::ReplayEvent | &ItemType::ReplayRecording
+                &ItemType::ReplayEvent | &ItemType::ReplayRecording | &ItemType::ReplayVideo
             )
         });
         if !replay_items.is_empty() {

relay-server/src/services/processor/event.rs

@@ -458,6 +458,7 @@ fn is_duplicate(item: &Item, processing_enabled: bool) -> bool {
         ItemType::Profile => false,
         ItemType::ReplayEvent => false,
         ItemType::ReplayRecording => false,
+        ItemType::ReplayVideo => false,
         ItemType::CheckIn => false,
         ItemType::Span => false,
         ItemType::OtelSpan => false,