headroom

public

Star0

Watch0

6 commits 1 branch 0 tags 0 releases521.6 KiB

//! protocol message types. spec: `IPC.md`.

use std::collections::BTreeMap;

use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::error::ProtoError;

/// subscription topic.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
#[non_exhaustive]
pub enum Topic {
    /// loudness / peak / GR telemetry.
    Meters,
    Profile,
    Routing,
    Daemon,
    /// synthetic, for `hello`; clients never subscribe.
    Control,
}

impl Topic {
    #[must_use]
    pub fn as_str(self) -> &'static str {
        match self {
            Topic::Meters => "meters",
            Topic::Profile => "profile",
            Topic::Routing => "routing",
            Topic::Daemon => "daemon",
            Topic::Control => "control",
        }
    }

    /// topics clients may subscribe to (excludes `Control`).
    #[must_use]
    pub const fn subscribable() -> &'static [Topic] {
        &[Topic::Meters, Topic::Profile, Topic::Routing, Topic::Daemon]
    }
}

impl std::fmt::Display for Topic {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

/// routing decision for one app or stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Route {
    /// through the processed (filtered) sink.
    Processed,
    /// straight to the real hardware sink; no processing, no extra graph hop.
    Bypass,
}

impl Route {
    #[must_use]
    pub fn as_str(self) -> &'static str {
        match self {
            Route::Processed => "processed",
            Route::Bypass => "bypass",
        }
    }
}

impl std::fmt::Display for Route {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

/// all operations. wire form `{ "op": "<name>", "args": <args> }`; arg-less ops omit `args`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "op", content = "args")]
#[non_exhaustive]
pub enum Op {
    #[serde(rename = "status")]
    Status,

    #[serde(rename = "profile.list")]
    ProfileList,

    #[serde(rename = "profile.use")]
    ProfileUse { name: String },

    /// `name` omitted ⇒ active profile.
    #[serde(rename = "profile.show")]
    ProfileShow {
        #[serde(default, skip_serializing_if = "Option::is_none")]
        name: Option<String>,
    },

    /// reload all profile files from disk.
    #[serde(rename = "profile.reload")]
    ProfileReload,

    #[serde(rename = "route.list")]
    RouteList,

    /// add/replace a persistent routing rule.
    #[serde(rename = "route.set")]
    RouteSet {
        /// typically `application.process.binary`.
        app: String,
        to: Route,
    },

    #[serde(rename = "route.unset")]
    RouteUnset { app: String },

    /// one-shot reroute of a live stream.
    #[serde(rename = "route.stream")]
    RouteStream { node_id: u32, to: Route },

    #[serde(rename = "setting.get")]
    SettingGet {
        /// dotted key, e.g. `compressor.threshold_db`.
        key: String,
    },

    #[serde(rename = "setting.set")]
    SettingSet { key: String, value: Value },

    #[serde(rename = "setting.list")]
    SettingList,

    /// revert one key to the active profile's value.
    #[serde(rename = "setting.clear")]
    SettingClear { key: String },

    /// revert all settings to the active profile; leaves route/per-app/bypass overrides intact.
    #[serde(rename = "setting.reset")]
    SettingReset,

    /// global bypass kill switch.
    #[serde(rename = "bypass.set")]
    BypassSet { enabled: bool },

    /// per-app (Layer A) controller state for managed streams.
    #[serde(rename = "per-app.list")]
    LayerAList,

    /// Layer A enable for one app (persistent overlay override).
    #[serde(rename = "per-app.set")]
    PerAppSet {
        /// process_binary or application_name.
        app: String,
        enabled: bool,
    },

    /// Layer A master switch (persistent overlay override).
    #[serde(rename = "per-app.master")]
    PerAppMaster { enabled: bool },

    /// clear a stream's deference lock (user-ceiling / strict-mode) so the controller resumes.
    #[serde(rename = "per-app.reset")]
    LayerAReset { node_id: u32 },

    #[serde(rename = "subscribe")]
    Subscribe { topics: Vec<Topic> },

    #[serde(rename = "unsubscribe")]
    Unsubscribe { topics: Vec<Topic> },

    /// optional client protocol handshake
    #[serde(rename = "hello")]
    Hello { protocol: u32 },
}

impl Op {
    /// canonical wire name.
    #[must_use]
    pub fn name(&self) -> &'static str {
        match self {
            Op::Status => "status",
            Op::ProfileList => "profile.list",
            Op::ProfileUse { .. } => "profile.use",
            Op::ProfileShow { .. } => "profile.show",
            Op::ProfileReload => "profile.reload",
            Op::RouteList => "route.list",
            Op::RouteSet { .. } => "route.set",
            Op::RouteUnset { .. } => "route.unset",
            Op::RouteStream { .. } => "route.stream",
            Op::SettingGet { .. } => "setting.get",
            Op::SettingSet { .. } => "setting.set",
            Op::SettingList => "setting.list",
            Op::SettingClear { .. } => "setting.clear",
            Op::SettingReset => "setting.reset",
            Op::BypassSet { .. } => "bypass.set",
            Op::LayerAList => "per-app.list",
            Op::PerAppSet { .. } => "per-app.set",
            Op::PerAppMaster { .. } => "per-app.master",
            Op::LayerAReset { .. } => "per-app.reset",
            Op::Subscribe { .. } => "subscribe",
            Op::Unsubscribe { .. } => "unsubscribe",
            Op::Hello { .. } => "hello",
        }
    }
}

/// client-to-server request.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Request {
    /// client-chosen; echoed in the paired response.
    pub id: u64,
    /// flattened: contributes `op` and optional `args` alongside `id`.
    #[serde(flatten)]
    pub op: Op,
}

impl Request {
    #[must_use]
    pub fn new(id: u64, op: Op) -> Self {
        Self { id, op }
    }
}

/// server-to-client response, paired to a request by `id`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Response {
    pub id: u64,
    #[serde(flatten)]
    pub payload: ResponsePayload,
}

impl Response {
    /// # Errors
    /// serde error if `value` fails to serialize.
    pub fn ok<T: Serialize>(id: u64, value: &T) -> Result<Self, serde_json::Error> {
        Ok(Self {
            id,
            payload: ResponsePayload::Ok {
                result: serde_json::to_value(value)?,
            },
        })
    }

    #[must_use]
    pub fn err(id: u64, error: ProtoError) -> Self {
        Self {
            id,
            payload: ResponsePayload::Err { error },
        }
    }
}

/// body of a `Response`. on the wire: a single `result` or `error` field inlined alongside `id`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum ResponsePayload {
    Ok { result: Value },
    Err { error: ProtoError },
}

/// server-to-client event on a subscribed topic.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Event {
    /// e.g. `tick`, `changed`.
    pub event: String,
    pub topic: Topic,
    /// shape depends on `topic` and `event`.
    pub data: Value,
}

impl Event {
    /// # Errors
    /// serde error if `data` fails to serialize.
    pub fn new<T: Serialize>(
        topic: Topic,
        event: impl Into<String>,
        data: &T,
    ) -> Result<Self, serde_json::Error> {
        Ok(Self {
            event: event.into(),
            topic,
            data: serde_json::to_value(data)?,
        })
    }
}

/// one frame as received from the server: paired response or fan-out event.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum ServerFrame {
    Response(Response),
    Event(Event),
}

/// result body of `status`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Status {
    /// semver.
    pub version: String,
    pub protocol: u32,
    pub uptime_s: u64,
    pub profile: String,
    pub bypass: bool,
    /// Layer A master switch (per-app level control, global).
    #[serde(default)]
    pub per_app: bool,
    pub sinks: Sinks,
    pub streams: Vec<StreamRoute>,
    /// per-app (Layer A) controller state; empty when not managing anything.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub layer_a: Vec<LayerASnapshot>,
    /// non-fatal warnings (typically profile-load: TOML parse errors, active profile missing on disk).
    /// reflects the last successful load/reload; empty when healthy.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub warnings: Vec<String>,
    /// user-overlay setting overrides (dotted key → value). these shadow the active profile and
    /// persist across `profile.use`, so a stale `headroom set` can silently override it
    /// (e.g. `agc.enabled=false`). surfaced so clients can flag it.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub setting_overrides: BTreeMap<String, Value>,
}

#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct Sinks {
    /// the processed virtual sink; the only sink headroom creates.
    pub processed: SinkInfo,
    /// hardware sink headroom forwards to, and where bypassed streams go directly.
    pub real: SinkInfo,
}

#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct SinkInfo {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub node_id: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    /// linked and accepting audio.
    #[serde(default)]
    pub ready: bool,
    /// native rate (Hz). filter matches the *real* sink's rate to skip the output-edge resample;
    /// the processed sink advertises whatever rate the filter currently runs at.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sample_rate: Option<u32>,
}

/// one playback stream and its route.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct StreamRoute {
    pub node_id: u32,
    /// typically `application.process.binary`.
    pub app: String,
    pub route: Route,
}

/// per-app (Layer A) controller state for one managed stream; on `status` and `per-app.list`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct LayerASnapshot {
    pub node_id: u32,
    pub app: String,
    /// true while a tap + controller is actively managing the stream.
    pub managed: bool,
    /// last linear volume written (1.0 = unity).
    pub volume_lin: f32,
    /// asserted gain reduction in dB (`>= 0`; `0` = no cut).
    pub reduction_db: f32,
    /// user-set ceiling (linear) when ceiling-mode deference is active.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub user_ceiling_lin: Option<f32>,
    /// strict-mode deference has locked the controller until `per-app.reset`.
    pub deferred: bool,
}

/// entry in `profile.list`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ProfileInfo {
    pub name: String,
    pub active: bool,
    pub description: String,
}

/// result body of `route.list`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RouteList {
    /// in evaluation order.
    pub rules: Vec<RouteRule>,
    pub current: Vec<StreamRoute>,
    /// fallback when no rule matches.
    pub default_route: Route,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RouteRule {
    #[serde(rename = "match")]
    pub match_: RouteRuleMatch,
    pub route: Route,
}

/// present fields are AND'd; values within a field are OR'd; empty match is true.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct RouteRuleMatch {
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub process_binary: Vec<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub application_name: Vec<String>,
    /// `pipewire.access.portal.app_id` (Flatpak).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub portal_app_id: Vec<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub media_role: Vec<String>,
}

/// `hello` payload (sent on connect).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct HelloData {
    /// always `"headroom"`.
    pub daemon: String,
    /// semver.
    pub version: String,
    pub protocol: u32,
}

/// `meters` tick payload.
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
pub struct MeterTick {
    /// BS.1770 M, 400 ms, LUFS.
    pub momentary_lufs: f32,
    /// BS.1770 S, 3 s, LUFS.
    pub shortterm_lufs: f32,
    /// BS.1770 I, gated, LUFS.
    pub integrated_lufs: f32,
    /// max true peak across channels, dBTP.
    pub true_peak_dbtp: f32,
    /// compressor + limiter, dB.
    pub gain_reduction_db: f32,
    pub compressor_gr_db: f32,
    pub limiter_gr_db: f32,
    /// AGC contribution (positive = boost), dB.
    pub agc_gain_db: f32,
    /// when `false`, AGC is bypassed and `agc_gain_db` is inert (0) — render as "disabled",
    /// not an active 0 dB. absent ⇒ `true`.
    #[serde(default = "default_true")]
    pub agc_enabled: bool,
    /// same as [`Self::agc_enabled`] for `compressor_gr_db`. (limiter has no flag — always-on backstop.)
    #[serde(default = "default_true")]
    pub compressor_enabled: bool,
}

/// absent ⇒ enabled, matching the always-on assumption before these fields existed.
fn default_true() -> bool {
    true
}

/// `profile` topic events.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "event", rename_all = "snake_case")]
#[non_exhaustive]
pub enum ProfileEvent {
    Changed { name: String, previous: String },
    /// `changed`: profiles whose definitions changed on disk.
    Reloaded { changed: Vec<String> },
}

/// `routing` topic events.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "event", rename_all = "snake_case")]
#[non_exhaustive]
pub enum RoutingEvent {
    StreamRouted {
        node_id: u32,
        app: String,
        to: Route,
    },
    /// stream's PipeWire node disappeared; drop state keyed by `node_id`.
    StreamRemoved { node_id: u32 },
    /// Layer A tap attached; daemon now manages `Props.channelVolumes` and publishes `meters/layer_a_level`.
    LayerAAttached { node_id: u32, app: String },
    /// Layer A tap torn down (usually stream gone); drop Layer A state for `node_id`.
    LayerADetached { node_id: u32 },
    RuleChanged,
}

/// `meters/layer_a_level` payload — emitted when the Layer A controller writes a new `channelVolumes`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct LayerALevel {
    pub node_id: u32,
    pub app: String,
    /// linear volume written (1.0 = unity).
    pub volume_lin: f32,
    /// asserted gain reduction, dB; ≤ 0 when reducing.
    pub reduction_db: f32,
}

/// `daemon` topic events.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "event", rename_all = "snake_case")]
#[non_exhaustive]
pub enum DaemonEvent {
    Started { version: String },
    Shutdown,
    /// events dropped on this connection.
    Overflow {
        lost_topic: Topic,
        /// lost in this batch.
        lost: u32,
        /// total lost on this connection.
        total_lost: u64,
    },
    /// non-fatal error.
    Error {
        /// matches `ErrorCode` when applicable.
        code: String,
        message: String,
    },
}

#[cfg(test)]
mod tests {
    use super::*;

    fn roundtrip<T>(v: &T) -> T
    where
        T: Serialize + for<'de> Deserialize<'de>,
    {
        let s = serde_json::to_string(v).expect("serialize");
        serde_json::from_str(&s).expect("deserialize")
    }

    #[test]
    fn op_status_serializes_without_args() {
        let req = Request::new(1, Op::Status);
        let s = serde_json::to_string(&req).unwrap();
        // Must be the flat form, no `kind` wrapper, no `args` field.
        assert_eq!(s, r#"{"id":1,"op":"status"}"#);

        let back: Request = serde_json::from_str(&s).unwrap();
        assert_eq!(back, req);
    }

    #[test]
    fn op_profile_use_round_trips_flat() {
        let req = Request::new(
            7,
            Op::ProfileUse {
                name: "night".into(),
            },
        );
        let s = serde_json::to_string(&req).unwrap();
        assert_eq!(s, r#"{"id":7,"op":"profile.use","args":{"name":"night"}}"#);

        let back: Request = serde_json::from_str(&s).unwrap();
        assert_eq!(back, req);
    }

    #[test]
    fn route_set_serializes_canonical() {
        let req = Request::new(
            12,
            Op::RouteSet {
                app: "firefox".into(),
                to: Route::Processed,
            },
        );
        let s = serde_json::to_string(&req).unwrap();
        assert_eq!(
            s,
            r#"{"id":12,"op":"route.set","args":{"app":"firefox","to":"processed"}}"#
        );
    }

    #[test]
    fn response_ok_shape() {
        let resp = Response::ok(3, &serde_json::json!({ "name": "default" })).unwrap();
        let s = serde_json::to_string(&resp).unwrap();
        assert_eq!(s, r#"{"id":3,"result":{"name":"default"}}"#);
    }

    #[test]
    fn response_err_shape() {
        let resp = Response::err(4, ProtoError::new(crate::ErrorCode::NotFound, "missing"));
        let s = serde_json::to_string(&resp).unwrap();
        assert_eq!(
            s,
            r#"{"id":4,"error":{"code":"NOT_FOUND","message":"missing"}}"#
        );
    }

    #[test]
    fn server_frame_distinguishes_response_from_event() {
        let resp = Response::ok(1, &serde_json::json!(null)).unwrap();
        let s = serde_json::to_string(&resp).unwrap();
        let frame: ServerFrame = serde_json::from_str(&s).unwrap();
        assert!(matches!(frame, ServerFrame::Response(_)));

        let ev = Event::new(Topic::Meters, "tick", &serde_json::json!({})).unwrap();
        let s = serde_json::to_string(&ev).unwrap();
        let frame: ServerFrame = serde_json::from_str(&s).unwrap();
        assert!(matches!(frame, ServerFrame::Event(_)));
    }

    #[test]
    fn meter_tick_roundtrip() {
        let m = MeterTick {
            momentary_lufs: -19.3,
            shortterm_lufs: -20.1,
            integrated_lufs: -19.8,
            true_peak_dbtp: -1.4,
            gain_reduction_db: -2.1,
            compressor_gr_db: -0.8,
            limiter_gr_db: -1.3,
            agc_gain_db: 0.5,
            agc_enabled: true,
            compressor_enabled: true,
        };
        let back = roundtrip(&m);
        assert_eq!(back, m);
    }

    #[test]
    fn topic_string_canonical() {
        assert_eq!(Topic::Meters.as_str(), "meters");
        assert_eq!(serde_json::to_string(&Topic::Meters).unwrap(), "\"meters\"");
        let t: Topic = serde_json::from_str("\"profile\"").unwrap();
        assert_eq!(t, Topic::Profile);
    }

    #[test]
    fn route_string_canonical() {
        let r: Route = serde_json::from_str("\"bypass\"").unwrap();
        assert_eq!(r, Route::Bypass);
        assert_eq!(serde_json::to_string(&r).unwrap(), "\"bypass\"");
    }

    #[test]
    fn error_code_screaming_snake() {
        let s = serde_json::to_string(&crate::ErrorCode::InvalidFrame).unwrap();
        assert_eq!(s, "\"INVALID_FRAME\"");
        let c: crate::ErrorCode = serde_json::from_str("\"UNKNOWN_OP\"").unwrap();
        assert_eq!(c, crate::ErrorCode::UnknownOp);
    }

    #[test]
    fn subscribe_op_roundtrip() {
        let req = Request::new(
            5,
            Op::Subscribe {
                topics: vec![Topic::Meters, Topic::Profile],
            },
        );
        let back: Request = serde_json::from_str(&serde_json::to_string(&req).unwrap()).unwrap();
        assert_eq!(back, req);
    }
}

1	//! protocol message types. spec: `IPC.md`.
2
3	use std::collections::BTreeMap;
4
5	use serde::{Deserialize, Serialize};
6	use serde_json::Value;
7
8	use crate::error::ProtoError;
9
10	/// subscription topic.
11	#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
12	#[serde(rename_all = "kebab-case")]
13	#[non_exhaustive]
14	pub enum Topic {
15	/// loudness / peak / GR telemetry.
16	Meters,
17	Profile,
18	Routing,
19	Daemon,
20	/// synthetic, for `hello`; clients never subscribe.
21	Control,
22	}
23
24	impl Topic {
25	#[must_use]
26	pub fn as_str(self) -> &'static str {
27	match self {
28	Topic::Meters => "meters",
29	Topic::Profile => "profile",
30	Topic::Routing => "routing",
31	Topic::Daemon => "daemon",
32	Topic::Control => "control",
33	}
34	}
35
36	/// topics clients may subscribe to (excludes `Control`).
37	#[must_use]
38	pub const fn subscribable() -> &'static [Topic] {
39	&[Topic::Meters, Topic::Profile, Topic::Routing, Topic::Daemon]
40	}
41	}
42
43	impl std::fmt::Display for Topic {
44	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45	f.write_str(self.as_str())
46	}
47	}
48
49	/// routing decision for one app or stream.
50	#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
51	#[serde(rename_all = "lowercase")]
52	pub enum Route {
53	/// through the processed (filtered) sink.
54	Processed,
55	/// straight to the real hardware sink; no processing, no extra graph hop.
56	Bypass,
57	}
58
59	impl Route {
60	#[must_use]
61	pub fn as_str(self) -> &'static str {
62	match self {
63	Route::Processed => "processed",
64	Route::Bypass => "bypass",
65	}
66	}
67	}
68
69	impl std::fmt::Display for Route {
70	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71	f.write_str(self.as_str())
72	}
73	}
74
75	/// all operations. wire form `{ "op": "<name>", "args": <args> }`; arg-less ops omit `args`.
76	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
77	#[serde(tag = "op", content = "args")]
78	#[non_exhaustive]
79	pub enum Op {
80	#[serde(rename = "status")]
81	Status,
82
83	#[serde(rename = "profile.list")]
84	ProfileList,
85
86	#[serde(rename = "profile.use")]
87	ProfileUse { name: String },
88
89	/// `name` omitted ⇒ active profile.
90	#[serde(rename = "profile.show")]
91	ProfileShow {
92	#[serde(default, skip_serializing_if = "Option::is_none")]
93	name: Option<String>,
94	},
95
96	/// reload all profile files from disk.
97	#[serde(rename = "profile.reload")]
98	ProfileReload,
99
100	#[serde(rename = "route.list")]
101	RouteList,
102
103	/// add/replace a persistent routing rule.
104	#[serde(rename = "route.set")]
105	RouteSet {
106	/// typically `application.process.binary`.
107	app: String,
108	to: Route,
109	},
110
111	#[serde(rename = "route.unset")]
112	RouteUnset { app: String },
113
114	/// one-shot reroute of a live stream.
115	#[serde(rename = "route.stream")]
116	RouteStream { node_id: u32, to: Route },
117
118	#[serde(rename = "setting.get")]
119	SettingGet {
120	/// dotted key, e.g. `compressor.threshold_db`.
121	key: String,
122	},
123
124	#[serde(rename = "setting.set")]
125	SettingSet { key: String, value: Value },
126
127	#[serde(rename = "setting.list")]
128	SettingList,
129
130	/// revert one key to the active profile's value.
131	#[serde(rename = "setting.clear")]
132	SettingClear { key: String },
133
134	/// revert all settings to the active profile; leaves route/per-app/bypass overrides intact.
135	#[serde(rename = "setting.reset")]
136	SettingReset,
137
138	/// global bypass kill switch.
139	#[serde(rename = "bypass.set")]
140	BypassSet { enabled: bool },
141
142	/// per-app (Layer A) controller state for managed streams.
143	#[serde(rename = "per-app.list")]
144	LayerAList,
145
146	/// Layer A enable for one app (persistent overlay override).
147	#[serde(rename = "per-app.set")]
148	PerAppSet {
149	/// process_binary or application_name.
150	app: String,
151	enabled: bool,
152	},
153
154	/// Layer A master switch (persistent overlay override).
155	#[serde(rename = "per-app.master")]
156	PerAppMaster { enabled: bool },
157
158	/// clear a stream's deference lock (user-ceiling / strict-mode) so the controller resumes.
159	#[serde(rename = "per-app.reset")]
160	LayerAReset { node_id: u32 },
161
162	#[serde(rename = "subscribe")]
163	Subscribe { topics: Vec<Topic> },
164
165	#[serde(rename = "unsubscribe")]
166	Unsubscribe { topics: Vec<Topic> },
167
168	/// optional client protocol handshake
169	#[serde(rename = "hello")]
170	Hello { protocol: u32 },
171	}
172
173	impl Op {
174	/// canonical wire name.
175	#[must_use]
176	pub fn name(&self) -> &'static str {
177	match self {
178	Op::Status => "status",
179	Op::ProfileList => "profile.list",
180	Op::ProfileUse { .. } => "profile.use",
181	Op::ProfileShow { .. } => "profile.show",
182	Op::ProfileReload => "profile.reload",
183	Op::RouteList => "route.list",
184	Op::RouteSet { .. } => "route.set",
185	Op::RouteUnset { .. } => "route.unset",
186	Op::RouteStream { .. } => "route.stream",
187	Op::SettingGet { .. } => "setting.get",
188	Op::SettingSet { .. } => "setting.set",
189	Op::SettingList => "setting.list",
190	Op::SettingClear { .. } => "setting.clear",
191	Op::SettingReset => "setting.reset",
192	Op::BypassSet { .. } => "bypass.set",
193	Op::LayerAList => "per-app.list",
194	Op::PerAppSet { .. } => "per-app.set",
195	Op::PerAppMaster { .. } => "per-app.master",
196	Op::LayerAReset { .. } => "per-app.reset",
197	Op::Subscribe { .. } => "subscribe",
198	Op::Unsubscribe { .. } => "unsubscribe",
199	Op::Hello { .. } => "hello",
200	}
201	}
202	}
203
204	/// client-to-server request.
205	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
206	pub struct Request {
207	/// client-chosen; echoed in the paired response.
208	pub id: u64,
209	/// flattened: contributes `op` and optional `args` alongside `id`.
210	#[serde(flatten)]
211	pub op: Op,
212	}
213
214	impl Request {
215	#[must_use]
216	pub fn new(id: u64, op: Op) -> Self {
217	Self { id, op }
218	}
219	}
220
221	/// server-to-client response, paired to a request by `id`.
222	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
223	pub struct Response {
224	pub id: u64,
225	#[serde(flatten)]
226	pub payload: ResponsePayload,
227	}
228
229	impl Response {
230	/// # Errors
231	/// serde error if `value` fails to serialize.
232	pub fn ok<T: Serialize>(id: u64, value: &T) -> Result<Self, serde_json::Error> {
233	Ok(Self {
234	id,
235	payload: ResponsePayload::Ok {
236	result: serde_json::to_value(value)?,
237	},
238	})
239	}
240
241	#[must_use]
242	pub fn err(id: u64, error: ProtoError) -> Self {
243	Self {
244	id,
245	payload: ResponsePayload::Err { error },
246	}
247	}
248	}
249
250	/// body of a `Response`. on the wire: a single `result` or `error` field inlined alongside `id`.
251	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
252	#[serde(untagged)]
253	pub enum ResponsePayload {
254	Ok { result: Value },
255	Err { error: ProtoError },
256	}
257
258	/// server-to-client event on a subscribed topic.
259	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
260	pub struct Event {
261	/// e.g. `tick`, `changed`.
262	pub event: String,
263	pub topic: Topic,
264	/// shape depends on `topic` and `event`.
265	pub data: Value,
266	}
267
268	impl Event {
269	/// # Errors
270	/// serde error if `data` fails to serialize.
271	pub fn new<T: Serialize>(
272	topic: Topic,
273	event: impl Into<String>,
274	data: &T,
275	) -> Result<Self, serde_json::Error> {
276	Ok(Self {
277	event: event.into(),
278	topic,
279	data: serde_json::to_value(data)?,
280	})
281	}
282	}
283
284	/// one frame as received from the server: paired response or fan-out event.
285	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
286	#[serde(untagged)]
287	pub enum ServerFrame {
288	Response(Response),
289	Event(Event),
290	}
291
292	/// result body of `status`.
293	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
294	pub struct Status {
295	/// semver.
296	pub version: String,
297	pub protocol: u32,
298	pub uptime_s: u64,
299	pub profile: String,
300	pub bypass: bool,
301	/// Layer A master switch (per-app level control, global).
302	#[serde(default)]
303	pub per_app: bool,
304	pub sinks: Sinks,
305	pub streams: Vec<StreamRoute>,
306	/// per-app (Layer A) controller state; empty when not managing anything.
307	#[serde(default, skip_serializing_if = "Vec::is_empty")]
308	pub layer_a: Vec<LayerASnapshot>,
309	/// non-fatal warnings (typically profile-load: TOML parse errors, active profile missing on disk).
310	/// reflects the last successful load/reload; empty when healthy.
311	#[serde(default, skip_serializing_if = "Vec::is_empty")]
312	pub warnings: Vec<String>,
313	/// user-overlay setting overrides (dotted key → value). these shadow the active profile and
314	/// persist across `profile.use`, so a stale `headroom set` can silently override it
315	/// (e.g. `agc.enabled=false`). surfaced so clients can flag it.
316	#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
317	pub setting_overrides: BTreeMap<String, Value>,
318	}
319
320	#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
321	pub struct Sinks {
322	/// the processed virtual sink; the only sink headroom creates.
323	pub processed: SinkInfo,
324	/// hardware sink headroom forwards to, and where bypassed streams go directly.
325	pub real: SinkInfo,
326	}
327
328	#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
329	pub struct SinkInfo {
330	#[serde(default, skip_serializing_if = "Option::is_none")]
331	pub node_id: Option<u32>,
332	#[serde(default, skip_serializing_if = "Option::is_none")]
333	pub name: Option<String>,
334	/// linked and accepting audio.
335	#[serde(default)]
336	pub ready: bool,
337	/// native rate (Hz). filter matches the real sink's rate to skip the output-edge resample;
338	/// the processed sink advertises whatever rate the filter currently runs at.
339	#[serde(default, skip_serializing_if = "Option::is_none")]
340	pub sample_rate: Option<u32>,
341	}
342
343	/// one playback stream and its route.
344	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
345	pub struct StreamRoute {
346	pub node_id: u32,
347	/// typically `application.process.binary`.
348	pub app: String,
349	pub route: Route,
350	}
351
352	/// per-app (Layer A) controller state for one managed stream; on `status` and `per-app.list`.
353	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
354	pub struct LayerASnapshot {
355	pub node_id: u32,
356	pub app: String,
357	/// true while a tap + controller is actively managing the stream.
358	pub managed: bool,
359	/// last linear volume written (1.0 = unity).
360	pub volume_lin: f32,
361	/// asserted gain reduction in dB (`>= 0`; `0` = no cut).
362	pub reduction_db: f32,
363	/// user-set ceiling (linear) when ceiling-mode deference is active.
364	#[serde(default, skip_serializing_if = "Option::is_none")]
365	pub user_ceiling_lin: Option<f32>,
366	/// strict-mode deference has locked the controller until `per-app.reset`.
367	pub deferred: bool,
368	}
369
370	/// entry in `profile.list`.
371	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
372	pub struct ProfileInfo {
373	pub name: String,
374	pub active: bool,
375	pub description: String,
376	}
377
378	/// result body of `route.list`.
379	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
380	pub struct RouteList {
381	/// in evaluation order.
382	pub rules: Vec<RouteRule>,
383	pub current: Vec<StreamRoute>,
384	/// fallback when no rule matches.
385	pub default_route: Route,
386	}
387
388	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
389	pub struct RouteRule {
390	#[serde(rename = "match")]
391	pub match_: RouteRuleMatch,
392	pub route: Route,
393	}
394
395	/// present fields are AND'd; values within a field are OR'd; empty match is true.
396	#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
397	pub struct RouteRuleMatch {
398	#[serde(default, skip_serializing_if = "Vec::is_empty")]
399	pub process_binary: Vec<String>,
400	#[serde(default, skip_serializing_if = "Vec::is_empty")]
401	pub application_name: Vec<String>,
402	/// `pipewire.access.portal.app_id` (Flatpak).
403	#[serde(default, skip_serializing_if = "Vec::is_empty")]
404	pub portal_app_id: Vec<String>,
405	#[serde(default, skip_serializing_if = "Vec::is_empty")]
406	pub media_role: Vec<String>,
407	}
408
409	/// `hello` payload (sent on connect).
410	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
411	pub struct HelloData {
412	/// always `"headroom"`.
413	pub daemon: String,
414	/// semver.
415	pub version: String,
416	pub protocol: u32,
417	}
418
419	/// `meters` tick payload.
420	#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
421	pub struct MeterTick {
422	/// BS.1770 M, 400 ms, LUFS.
423	pub momentary_lufs: f32,
424	/// BS.1770 S, 3 s, LUFS.
425	pub shortterm_lufs: f32,
426	/// BS.1770 I, gated, LUFS.
427	pub integrated_lufs: f32,
428	/// max true peak across channels, dBTP.
429	pub true_peak_dbtp: f32,
430	/// compressor + limiter, dB.
431	pub gain_reduction_db: f32,
432	pub compressor_gr_db: f32,
433	pub limiter_gr_db: f32,
434	/// AGC contribution (positive = boost), dB.
435	pub agc_gain_db: f32,
436	/// when `false`, AGC is bypassed and `agc_gain_db` is inert (0) — render as "disabled",
437	/// not an active 0 dB. absent ⇒ `true`.
438	#[serde(default = "default_true")]
439	pub agc_enabled: bool,
440	/// same as [`Self::agc_enabled`] for `compressor_gr_db`. (limiter has no flag — always-on backstop.)
441	#[serde(default = "default_true")]
442	pub compressor_enabled: bool,
443	}
444
445	/// absent ⇒ enabled, matching the always-on assumption before these fields existed.
446	fn default_true() -> bool {
447	true
448	}
449
450	/// `profile` topic events.
451	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
452	#[serde(tag = "event", rename_all = "snake_case")]
453	#[non_exhaustive]
454	pub enum ProfileEvent {
455	Changed { name: String, previous: String },
456	/// `changed`: profiles whose definitions changed on disk.
457	Reloaded { changed: Vec<String> },
458	}
459
460	/// `routing` topic events.
461	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
462	#[serde(tag = "event", rename_all = "snake_case")]
463	#[non_exhaustive]
464	pub enum RoutingEvent {
465	StreamRouted {
466	node_id: u32,
467	app: String,
468	to: Route,
469	},
470	/// stream's PipeWire node disappeared; drop state keyed by `node_id`.
471	StreamRemoved { node_id: u32 },
472	/// Layer A tap attached; daemon now manages `Props.channelVolumes` and publishes `meters/layer_a_level`.
473	LayerAAttached { node_id: u32, app: String },
474	/// Layer A tap torn down (usually stream gone); drop Layer A state for `node_id`.
475	LayerADetached { node_id: u32 },
476	RuleChanged,
477	}
478
479	/// `meters/layer_a_level` payload — emitted when the Layer A controller writes a new `channelVolumes`.
480	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
481	pub struct LayerALevel {
482	pub node_id: u32,
483	pub app: String,
484	/// linear volume written (1.0 = unity).
485	pub volume_lin: f32,
486	/// asserted gain reduction, dB; ≤ 0 when reducing.
487	pub reduction_db: f32,
488	}
489
490	/// `daemon` topic events.
491	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
492	#[serde(tag = "event", rename_all = "snake_case")]
493	#[non_exhaustive]
494	pub enum DaemonEvent {
495	Started { version: String },
496	Shutdown,
497	/// events dropped on this connection.
498	Overflow {
499	lost_topic: Topic,
500	/// lost in this batch.
501	lost: u32,
502	/// total lost on this connection.
503	total_lost: u64,
504	},
505	/// non-fatal error.
506	Error {
507	/// matches `ErrorCode` when applicable.
508	code: String,
509	message: String,
510	},
511	}
512
513	#[cfg(test)]
514	mod tests {
515	use super::*;
516
517	fn roundtrip<T>(v: &T) -> T
518	where
519	T: Serialize + for<'de> Deserialize<'de>,
520	{
521	let s = serde_json::to_string(v).expect("serialize");
522	serde_json::from_str(&s).expect("deserialize")
523	}
524
525	#[test]
526	fn op_status_serializes_without_args() {
527	let req = Request::new(1, Op::Status);
528	let s = serde_json::to_string(&req).unwrap();
529	// Must be the flat form, no `kind` wrapper, no `args` field.
530	assert_eq!(s, r#"{"id":1,"op":"status"}"#);
531
532	let back: Request = serde_json::from_str(&s).unwrap();
533	assert_eq!(back, req);
534	}
535
536	#[test]
537	fn op_profile_use_round_trips_flat() {
538	let req = Request::new(
539	7,
540	Op::ProfileUse {
541	name: "night".into(),
542	},
543	);
544	let s = serde_json::to_string(&req).unwrap();
545	assert_eq!(s, r#"{"id":7,"op":"profile.use","args":{"name":"night"}}"#);
546
547	let back: Request = serde_json::from_str(&s).unwrap();
548	assert_eq!(back, req);
549	}
550
551	#[test]
552	fn route_set_serializes_canonical() {
553	let req = Request::new(
554	12,
555	Op::RouteSet {
556	app: "firefox".into(),
557	to: Route::Processed,
558	},
559	);
560	let s = serde_json::to_string(&req).unwrap();
561	assert_eq!(
562	s,
563	r#"{"id":12,"op":"route.set","args":{"app":"firefox","to":"processed"}}"#
564	);
565	}
566
567	#[test]
568	fn response_ok_shape() {
569	let resp = Response::ok(3, &serde_json::json!({ "name": "default" })).unwrap();
570	let s = serde_json::to_string(&resp).unwrap();
571	assert_eq!(s, r#"{"id":3,"result":{"name":"default"}}"#);
572	}
573
574	#[test]
575	fn response_err_shape() {
576	let resp = Response::err(4, ProtoError::new(crate::ErrorCode::NotFound, "missing"));
577	let s = serde_json::to_string(&resp).unwrap();
578	assert_eq!(
579	s,
580	r#"{"id":4,"error":{"code":"NOT_FOUND","message":"missing"}}"#
581	);
582	}
583
584	#[test]
585	fn server_frame_distinguishes_response_from_event() {
586	let resp = Response::ok(1, &serde_json::json!(null)).unwrap();
587	let s = serde_json::to_string(&resp).unwrap();
588	let frame: ServerFrame = serde_json::from_str(&s).unwrap();
589	assert!(matches!(frame, ServerFrame::Response(_)));
590
591	let ev = Event::new(Topic::Meters, "tick", &serde_json::json!({})).unwrap();
592	let s = serde_json::to_string(&ev).unwrap();
593	let frame: ServerFrame = serde_json::from_str(&s).unwrap();
594	assert!(matches!(frame, ServerFrame::Event(_)));
595	}
596
597	#[test]
598	fn meter_tick_roundtrip() {
599	let m = MeterTick {
600	momentary_lufs: -19.3,
601	shortterm_lufs: -20.1,
602	integrated_lufs: -19.8,
603	true_peak_dbtp: -1.4,
604	gain_reduction_db: -2.1,
605	compressor_gr_db: -0.8,
606	limiter_gr_db: -1.3,
607	agc_gain_db: 0.5,
608	agc_enabled: true,
609	compressor_enabled: true,
610	};
611	let back = roundtrip(&m);
612	assert_eq!(back, m);
613	}
614
615	#[test]
616	fn topic_string_canonical() {
617	assert_eq!(Topic::Meters.as_str(), "meters");
618	assert_eq!(serde_json::to_string(&Topic::Meters).unwrap(), "\"meters\"");
619	let t: Topic = serde_json::from_str("\"profile\"").unwrap();
620	assert_eq!(t, Topic::Profile);
621	}
622
623	#[test]
624	fn route_string_canonical() {
625	let r: Route = serde_json::from_str("\"bypass\"").unwrap();
626	assert_eq!(r, Route::Bypass);
627	assert_eq!(serde_json::to_string(&r).unwrap(), "\"bypass\"");
628	}
629
630	#[test]
631	fn error_code_screaming_snake() {
632	let s = serde_json::to_string(&crate::ErrorCode::InvalidFrame).unwrap();
633	assert_eq!(s, "\"INVALID_FRAME\"");
634	let c: crate::ErrorCode = serde_json::from_str("\"UNKNOWN_OP\"").unwrap();
635	assert_eq!(c, crate::ErrorCode::UnknownOp);
636	}
637
638	#[test]
639	fn subscribe_op_roundtrip() {
640	let req = Request::new(
641	5,
642	Op::Subscribe {
643	topics: vec![Topic::Meters, Topic::Profile],
644	},
645	);
646	let back: Request = serde_json::from_str(&serde_json::to_string(&req).unwrap()).unwrap();
647	assert_eq!(back, req);
648	}
649	}