dk_protocol/
watch.rs

1use tokio::sync::{broadcast, mpsc};
2use tonic::Status;
3
4use crate::server::ProtocolServer;
5use crate::{WatchEvent, WatchRequest};
6
7/// Long-running handler for the WATCH server-streaming RPC.
8///
9/// Subscribes to the shared [`EventBus`] and forwards every event to the
10/// client via the provided `mpsc::Sender`.  The loop terminates when:
11///
12/// * The client disconnects (send fails).
13/// * The event bus is dropped (channel closed).
14///
15/// Lagged receivers (slow consumers) log a warning and continue.
16pub async fn handle_watch(
17    server: &ProtocolServer,
18    req: WatchRequest,
19    tx: mpsc::Sender<Result<WatchEvent, Status>>,
20) {
21    let session = match server.validate_session(&req.session_id) {
22        Ok(s) => s,
23        Err(e) => {
24            let _ = tx.send(Err(e)).await;
25            return;
26        }
27    };
28    if let Err(e) = crate::require_live_session::require_live_session(server, &req.session_id).await {
29        let _ = tx.send(Err(e)).await;
30        return;
31    }
32
33    // Resolve repo_id: prefer the one from the request, fall back to
34    // resolving it from the session's codebase.
35    let repo_id = if !req.repo_id.is_empty() {
36        req.repo_id.clone()
37    } else {
38        // Try to resolve repo_id from the session's codebase name.
39        match server.engine().get_repo(&session.codebase).await {
40            Ok((rid, _git_repo)) => rid.to_string(),
41            Err(_) => {
42                // Fall back to subscribe_all if we can't resolve the repo.
43                String::new()
44            }
45        }
46    };
47
48    // Subscribe to the repo-specific channel, or all events if no repo resolved.
49    let mut rx = if repo_id.is_empty() {
50        server.event_bus().subscribe_all()
51    } else {
52        server.event_bus().subscribe(&repo_id)
53    };
54
55    let filter = &req.filter;
56    let self_session_id = req.session_id.clone();
57
58    loop {
59        match rx.recv().await {
60            Ok(event) => {
61                // Filter out events from the requesting session itself.
62                if event.session_id == self_session_id {
63                    continue;
64                }
65
66                if matches_filter(&event.event_type, filter)
67                    && tx.send(Ok(event)).await.is_err()
68                {
69                    break;
70                }
71            }
72            Err(broadcast::error::RecvError::Lagged(n)) => {
73                tracing::warn!("watch stream lagged by {} events", n);
74            }
75            Err(broadcast::error::RecvError::Closed) => break,
76        }
77    }
78}
79
80/// Check if an event matches a glob-style filter.
81///
82/// Supported patterns:
83/// - Empty or "*" matches everything
84/// - "changeset.*" matches "changeset.submitted", "changeset.merged", etc.
85/// - "*.merged" matches "changeset.merged", "branch.merged", etc.
86/// - Exact match: "changeset.submitted" matches only that event type
87fn matches_filter(event_type: &str, filter: &str) -> bool {
88    if filter.is_empty() || filter == "*" {
89        return true;
90    }
91
92    if let Some(prefix) = filter.strip_suffix(".*") {
93        event_type.starts_with(prefix)
94            && event_type.as_bytes().get(prefix.len()) == Some(&b'.')
95    } else if let Some(suffix) = filter.strip_prefix("*.") {
96        event_type.ends_with(suffix)
97            && event_type.len() > suffix.len()
98            && event_type.as_bytes()[event_type.len() - suffix.len() - 1] == b'.'
99    } else {
100        event_type == filter
101    }
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107
108    #[test]
109    fn empty_filter_matches_all() {
110        assert!(matches_filter("changeset.submitted", ""));
111        assert!(matches_filter("anything", ""));
112    }
113
114    #[test]
115    fn star_matches_all() {
116        assert!(matches_filter("changeset.submitted", "*"));
117        assert!(matches_filter("anything", "*"));
118    }
119
120    #[test]
121    fn prefix_glob() {
122        assert!(matches_filter("changeset.submitted", "changeset.*"));
123        assert!(matches_filter("changeset.merged", "changeset.*"));
124        assert!(matches_filter("changeset.verified", "changeset.*"));
125        assert!(!matches_filter("branch.created", "changeset.*"));
126        assert!(!matches_filter("changesetx.foo", "changeset.*"));
127    }
128
129    #[test]
130    fn suffix_glob() {
131        assert!(matches_filter("changeset.merged", "*.merged"));
132        assert!(matches_filter("branch.merged", "*.merged"));
133        assert!(!matches_filter("changeset.submitted", "*.merged"));
134        assert!(!matches_filter("xmerged", "*.merged"));
135    }
136
137    #[test]
138    fn exact_match() {
139        assert!(matches_filter("changeset.submitted", "changeset.submitted"));
140        assert!(!matches_filter("changeset.merged", "changeset.submitted"));
141    }
142}
dk_protocol/watch.rs

dk_protocol/
watch.rs
