player_plugin/
hook.rs

1use std::collections::BTreeMap;
2
3use serde::{Deserialize, Serialize};
4use thiserror::Error;
5
6use crate::{
7    MAX_PLUGIN_DIAGNOSTICS, MAX_PLUGIN_ERROR_MESSAGE_BYTES, MAX_PLUGIN_EVENT_ID_BYTES,
8    MAX_PLUGIN_EVENT_NAME_BYTES, MAX_PLUGIN_MEASUREMENTS, MAX_PLUGIN_PLATFORM_BYTES,
9    MAX_PLUGIN_PROTOCOL_BYTES, MAX_PLUGIN_RESOURCE_IDENTITY_BYTES, MAX_PLUGIN_THREAD_BYTES,
10    PluginDiagnostic, PluginMeasurement, PluginProtocolViolation,
11    protocol::{validate_attributes, validate_optional_text, validate_text},
12};
13
14/// One bounded, transport-neutral event emitted by a host pipeline.
15#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
16#[serde(rename_all = "camelCase")]
17pub struct PipelineEvent {
18    pub run_id: String,
19    pub session_id: String,
20    pub platform: String,
21    pub protocol: Option<String>,
22    pub event_name: String,
23    pub timestamp_ns: u64,
24    pub thread: Option<String>,
25    pub resource_identity: Option<String>,
26    #[serde(default)]
27    pub attributes: BTreeMap<String, String>,
28    pub diagnostic: Option<PluginDiagnostic>,
29}
30
31impl PipelineEvent {
32    pub fn validate(&self) -> Result<(), PipelineEventHookError> {
33        validate_text(
34            "pipeline_event.run_id",
35            &self.run_id,
36            MAX_PLUGIN_EVENT_ID_BYTES,
37        )?;
38        validate_text(
39            "pipeline_event.session_id",
40            &self.session_id,
41            MAX_PLUGIN_EVENT_ID_BYTES,
42        )?;
43        validate_text(
44            "pipeline_event.platform",
45            &self.platform,
46            MAX_PLUGIN_PLATFORM_BYTES,
47        )?;
48        validate_optional_text(
49            "pipeline_event.protocol",
50            self.protocol.as_deref(),
51            MAX_PLUGIN_PROTOCOL_BYTES,
52        )?;
53        validate_text(
54            "pipeline_event.event_name",
55            &self.event_name,
56            MAX_PLUGIN_EVENT_NAME_BYTES,
57        )?;
58        validate_optional_text(
59            "pipeline_event.thread",
60            self.thread.as_deref(),
61            MAX_PLUGIN_THREAD_BYTES,
62        )?;
63        validate_optional_text(
64            "pipeline_event.resource_identity",
65            self.resource_identity.as_deref(),
66            MAX_PLUGIN_RESOURCE_IDENTITY_BYTES,
67        )?;
68        validate_attributes(&self.attributes)?;
69        if let Some(diagnostic) = &self.diagnostic {
70            diagnostic.validate()?;
71        }
72        Ok(())
73    }
74}
75
76#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
77#[serde(rename_all = "camelCase")]
78pub struct PipelineEventHookOutcome {
79    pub accepted: bool,
80    #[serde(default)]
81    pub measurements: Vec<PluginMeasurement>,
82    #[serde(default)]
83    pub diagnostics: Vec<PluginDiagnostic>,
84}
85
86impl PipelineEventHookOutcome {
87    pub fn accepted() -> Self {
88        Self {
89            accepted: true,
90            ..Self::default()
91        }
92    }
93
94    pub fn validate(&self) -> Result<(), PipelineEventHookError> {
95        if self.measurements.len() > MAX_PLUGIN_MEASUREMENTS {
96            return Err(PipelineEventHookError::ProtocolViolation(format!(
97                "event hook returned more than {MAX_PLUGIN_MEASUREMENTS} measurements"
98            )));
99        }
100        if self.diagnostics.len() > MAX_PLUGIN_DIAGNOSTICS {
101            return Err(PipelineEventHookError::ProtocolViolation(format!(
102                "event hook returned more than {MAX_PLUGIN_DIAGNOSTICS} diagnostics"
103            )));
104        }
105        for measurement in &self.measurements {
106            measurement.validate()?;
107        }
108        for diagnostic in &self.diagnostics {
109            diagnostic.validate()?;
110        }
111        Ok(())
112    }
113}
114
115#[derive(Debug, Error, Clone, PartialEq, Eq, Serialize, Deserialize)]
116#[serde(rename_all = "camelCase", tag = "code", content = "message")]
117pub enum PipelineEventHookError {
118    #[error("event hook rejected invalid input: {0}")]
119    InvalidInput(String),
120    #[error("payload codec error: {0}")]
121    PayloadCodec(String),
122    #[error("plugin ABI violation: {0}")]
123    AbiViolation(String),
124    #[error("event hook rejected the event: {0}")]
125    Rejected(String),
126    #[error("event hook failed: {0}")]
127    Failed(String),
128    #[error("event hook protocol violation: {0}")]
129    ProtocolViolation(String),
130}
131
132impl PipelineEventHookError {
133    pub fn validate_author_failure(&self) -> Result<(), Self> {
134        let message = match self {
135            Self::InvalidInput(message) | Self::Rejected(message) | Self::Failed(message) => {
136                message
137            }
138            Self::PayloadCodec(_) | Self::AbiViolation(_) | Self::ProtocolViolation(_) => {
139                return Err(Self::ProtocolViolation(
140                    "plugin returned a host-owned event-hook error kind".to_owned(),
141                ));
142            }
143        };
144        validate_text(
145            "event_hook.error.message",
146            message,
147            MAX_PLUGIN_ERROR_MESSAGE_BYTES,
148        )
149        .map_err(Self::from)
150    }
151}
152
153impl From<PluginProtocolViolation> for PipelineEventHookError {
154    fn from(value: PluginProtocolViolation) -> Self {
155        Self::ProtocolViolation(value.to_string())
156    }
157}
158
159pub trait PipelineEventHook: Send + Sync {
160    fn on_event(
161        &self,
162        event: &PipelineEvent,
163    ) -> Result<PipelineEventHookOutcome, PipelineEventHookError>;
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169
170    fn event() -> PipelineEvent {
171        PipelineEvent {
172            run_id: "run-1".to_owned(),
173            session_id: "session-1".to_owned(),
174            platform: "test".to_owned(),
175            protocol: Some("hls".to_owned()),
176            event_name: "download.completed".to_owned(),
177            timestamp_ns: 1,
178            thread: None,
179            resource_identity: Some("download-task:1".to_owned()),
180            attributes: BTreeMap::new(),
181            diagnostic: None,
182        }
183    }
184
185    #[test]
186    fn event_validates_transport_fields_and_structured_diagnostic() {
187        let mut event = event();
188        assert_eq!(event.validate(), Ok(()));
189
190        event.platform.clear();
191        assert!(matches!(
192            event.validate(),
193            Err(PipelineEventHookError::ProtocolViolation(_))
194        ));
195
196        event.platform = "test".to_owned();
197        event.diagnostic = Some(PluginDiagnostic {
198            code: "x".repeat(65),
199            severity: crate::PluginDiagnosticSeverity::Error,
200            message: "invalid diagnostic".to_owned(),
201            attributes: BTreeMap::new(),
202        });
203        assert!(matches!(
204            event.validate(),
205            Err(PipelineEventHookError::ProtocolViolation(_))
206        ));
207    }
208
209    #[test]
210    fn event_accepts_awkward_valid_utf8_without_rewriting_identity() {
211        let mut event = event();
212        event.resource_identity = Some("opaque:资源 identity/with spaces".to_owned());
213        assert_eq!(event.validate(), Ok(()));
214        assert_eq!(
215            event.resource_identity.as_deref(),
216            Some("opaque:资源 identity/with spaces")
217        );
218    }
219
220    #[test]
221    fn outcome_rejects_non_finite_measurement() {
222        let outcome = PipelineEventHookOutcome {
223            accepted: true,
224            measurements: vec![PluginMeasurement {
225                name: "latency".to_owned(),
226                value: f64::INFINITY,
227                unit: "ms".to_owned(),
228                attributes: Default::default(),
229            }],
230            diagnostics: Vec::new(),
231        };
232        assert!(matches!(
233            outcome.validate(),
234            Err(PipelineEventHookError::ProtocolViolation(_))
235        ));
236    }
237}
player_plugin/hook.rs

player_plugin/
hook.rs
