lit/events/
subscription.rs

1//! Event subscription and notification engine
2//!
3//! Stores subscriptions as JSON in .lit/events/subscriptions/ and emits
4//! events to a log that can be tailed or queried.
5
6use crate::errors::LitError;
7use serde::{Deserialize, Serialize};
8use std::fs;
9use std::path::Path;
10
11/// Types of events that can be subscribed to
12#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
13pub enum EventType {
14    CommitPushed,
15    BranchCreated,
16    BranchDeleted,
17    BranchUpdated,
18    MergeCompleted,
19    TagCreated,
20    IssueOpened,
21    IssueClosed,
22    PrOpened,
23    PrMerged,
24    PrClosed,
25    AgentJoined,
26    AgentLeft,
27    TaskDelegated,
28    TaskCompleted,
29    TrustChanged,
30    UcanIssued,
31    UcanRevoked,
32    All,
33}
34
35impl std::fmt::Display for EventType {
36    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        write!(f, "{:?}", self)
38    }
39}
40
41impl std::str::FromStr for EventType {
42    type Err = String;
43    fn from_str(s: &str) -> Result<Self, Self::Err> {
44        match s.to_lowercase().as_str() {
45            "commitpushed" | "commit" => Ok(EventType::CommitPushed),
46            "branchcreated" | "branch-created" => Ok(EventType::BranchCreated),
47            "branchdeleted" | "branch-deleted" => Ok(EventType::BranchDeleted),
48            "branchupdated" | "branch-updated" => Ok(EventType::BranchUpdated),
49            "mergecompleted" | "merge" => Ok(EventType::MergeCompleted),
50            "tagcreated" | "tag" => Ok(EventType::TagCreated),
51            "issueopened" | "issue-opened" => Ok(EventType::IssueOpened),
52            "issueclosed" | "issue-closed" => Ok(EventType::IssueClosed),
53            "propened" | "pr-opened" => Ok(EventType::PrOpened),
54            "prmerged" | "pr-merged" => Ok(EventType::PrMerged),
55            "prclosed" | "pr-closed" => Ok(EventType::PrClosed),
56            "agentjoined" | "agent-joined" => Ok(EventType::AgentJoined),
57            "agentleft" | "agent-left" => Ok(EventType::AgentLeft),
58            "taskdelegated" | "task-delegated" => Ok(EventType::TaskDelegated),
59            "taskcompleted" | "task-completed" => Ok(EventType::TaskCompleted),
60            "trustchanged" | "trust-changed" => Ok(EventType::TrustChanged),
61            "ucanissued" | "ucan-issued" => Ok(EventType::UcanIssued),
62            "ucanrevoked" | "ucan-revoked" => Ok(EventType::UcanRevoked),
63            "all" | "*" => Ok(EventType::All),
64            _ => Err(format!("Unknown event type: {}", s)),
65        }
66    }
67}
68
69/// A subscription entry
70#[derive(Debug, Clone, Serialize, Deserialize)]
71pub struct EventSubscription {
72    /// Unique subscription ID
73    pub id: String,
74    /// Subscriber DID (or local user if no DID)
75    pub subscriber: String,
76    /// Event types to subscribe to
77    pub event_types: Vec<EventType>,
78    /// Optional branch filter
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub branch_filter: Option<String>,
81    /// Created timestamp
82    pub created: String,
83    /// Whether the subscription is active
84    pub active: bool,
85}
86
87/// A recorded event
88#[derive(Debug, Clone, Serialize, Deserialize)]
89pub struct Event {
90    /// Event type
91    pub event_type: EventType,
92    /// Timestamp
93    pub timestamp: String,
94    /// Actor DID or identifier
95    pub actor: String,
96    /// Event-specific payload
97    pub payload: serde_json::Value,
98    /// Optional branch context
99    #[serde(skip_serializing_if = "Option::is_none")]
100    pub branch: Option<String>,
101}
102
103fn subscriptions_dir(repo_root: &Path) -> std::path::PathBuf {
104    repo_root.join(".lit").join("events").join("subscriptions")
105}
106
107fn events_log_path(repo_root: &Path) -> std::path::PathBuf {
108    repo_root.join(".lit").join("events").join("log.jsonl")
109}
110
111/// Subscribe to event types
112pub fn subscribe(
113    repo_root: &Path,
114    subscriber: &str,
115    event_types: Vec<EventType>,
116    branch_filter: Option<String>,
117) -> Result<EventSubscription, LitError> {
118    let dir = subscriptions_dir(repo_root);
119    fs::create_dir_all(&dir)
120        .map_err(|e| LitError::io(format!("Failed to create subscriptions dir: {}", e)))?;
121
122    let id = format!(
123        "{:016x}",
124        chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0)
125    );
126    let sub = EventSubscription {
127        id: id.clone(),
128        subscriber: subscriber.to_string(),
129        event_types,
130        branch_filter,
131        created: chrono::Utc::now().to_rfc3339(),
132        active: true,
133    };
134
135    let path = dir.join(format!("{}.json", id));
136    let json = serde_json::to_string_pretty(&sub)
137        .map_err(|e| LitError::general(format!("Serialize error: {}", e)))?;
138    fs::write(&path, json).map_err(|e| LitError::io(format!("Write error: {}", e)))?;
139
140    Ok(sub)
141}
142
143/// List all subscriptions
144pub fn list_subscriptions(repo_root: &Path) -> Result<Vec<EventSubscription>, LitError> {
145    let dir = subscriptions_dir(repo_root);
146    if !dir.exists() {
147        return Ok(Vec::new());
148    }
149
150    let mut subs = Vec::new();
151    for entry in fs::read_dir(&dir).map_err(|e| LitError::io(format!("IO: {}", e)))? {
152        let entry = entry.map_err(|e| LitError::io(format!("IO: {}", e)))?;
153        if entry.path().extension().is_some_and(|e| e == "json") {
154            if let Ok(json) = fs::read_to_string(entry.path()) {
155                if let Ok(sub) = serde_json::from_str::<EventSubscription>(&json) {
156                    subs.push(sub);
157                }
158            }
159        }
160    }
161    Ok(subs)
162}
163
164/// Remove a subscription
165pub fn unsubscribe(repo_root: &Path, sub_id: &str) -> Result<(), LitError> {
166    let path = subscriptions_dir(repo_root).join(format!("{}.json", sub_id));
167    if path.exists() {
168        fs::remove_file(&path).map_err(|e| LitError::io(format!("Remove error: {}", e)))?;
169        Ok(())
170    } else {
171        Err(LitError::general(format!(
172            "Subscription not found: {}",
173            sub_id
174        )))
175    }
176}
177
178/// Emit an event to the event log
179pub fn emit_event(repo_root: &Path, event: &Event) -> Result<(), LitError> {
180    let log_path = events_log_path(repo_root);
181    if let Some(parent) = log_path.parent() {
182        fs::create_dir_all(parent).map_err(|e| LitError::io(format!("IO: {}", e)))?;
183    }
184
185    let line = serde_json::to_string(event)
186        .map_err(|e| LitError::general(format!("Serialize error: {}", e)))?;
187
188    use std::io::Write;
189    let mut file = fs::OpenOptions::new()
190        .create(true)
191        .append(true)
192        .open(&log_path)
193        .map_err(|e| LitError::io(format!("IO: {}", e)))?;
194    writeln!(file, "{}", line).map_err(|e| LitError::io(format!("Write error: {}", e)))?;
195
196    Ok(())
197}
198
199/// Read recent events from the log, optionally filtered
200pub fn read_events(
201    repo_root: &Path,
202    event_type: Option<&EventType>,
203    limit: usize,
204) -> Result<Vec<Event>, LitError> {
205    let log_path = events_log_path(repo_root);
206    if !log_path.exists() {
207        return Ok(Vec::new());
208    }
209
210    let content = fs::read_to_string(&log_path).map_err(|e| LitError::io(format!("IO: {}", e)))?;
211
212    let mut events: Vec<Event> = content
213        .lines()
214        .filter_map(|line| serde_json::from_str::<Event>(line).ok())
215        .filter(|e| event_type.is_none_or(|et| *et == EventType::All || e.event_type == *et))
216        .collect();
217
218    // Return most recent first
219    events.reverse();
220    events.truncate(limit);
221    Ok(events)
222}
223
224#[cfg(test)]
225mod tests {
226    use super::*;
227    use std::path::PathBuf;
228    use std::sync::atomic::{AtomicU32, Ordering};
229
230    static COUNTER: AtomicU32 = AtomicU32::new(0);
231
232    /// Per-test scratch directory.
233    ///
234    /// Subscriptions and the event log live under one repo root, and each test
235    /// removes that root when it finishes. Sharing a root across concurrently
236    /// running tests would let one test's cleanup delete another's data
237    /// mid-run, so every test gets its own.
238    fn tmp_dir() -> PathBuf {
239        let n = COUNTER.fetch_add(1, Ordering::SeqCst);
240        let dir =
241            std::env::temp_dir().join(format!("lit_events_test_{}_{}", std::process::id(), n));
242        let _ = fs::remove_dir_all(&dir);
243        fs::create_dir_all(&dir).unwrap();
244        dir
245    }
246
247    #[test]
248    fn test_subscribe_and_list() {
249        let dir = tmp_dir();
250        let sub = subscribe(
251            &dir,
252            "did:lit:user1",
253            vec![EventType::CommitPushed, EventType::MergeCompleted],
254            Some("main".to_string()),
255        )
256        .unwrap();
257
258        assert!(sub.active);
259        assert_eq!(sub.event_types.len(), 2);
260
261        let subs = list_subscriptions(&dir).unwrap();
262        assert_eq!(subs.len(), 1);
263
264        let _ = fs::remove_dir_all(&dir);
265    }
266
267    #[test]
268    fn test_emit_and_read() {
269        let dir = tmp_dir();
270        let event = Event {
271            event_type: EventType::CommitPushed,
272            timestamp: chrono::Utc::now().to_rfc3339(),
273            actor: "did:lit:agent1".to_string(),
274            payload: serde_json::json!({"hash": "abc123"}),
275            branch: Some("main".to_string()),
276        };
277        emit_event(&dir, &event).unwrap();
278
279        let events = read_events(&dir, None, 10).unwrap();
280        assert_eq!(events.len(), 1);
281        assert_eq!(events[0].event_type, EventType::CommitPushed);
282
283        let _ = fs::remove_dir_all(&dir);
284    }
285
286    #[test]
287    fn test_event_type_parse() {
288        assert_eq!(
289            "commit".parse::<EventType>().unwrap(),
290            EventType::CommitPushed
291        );
292        assert_eq!("all".parse::<EventType>().unwrap(), EventType::All);
293        assert!("invalid".parse::<EventType>().is_err());
294    }
295}
lit/events/subscription.rs

lit/events/
subscription.rs
