nexus_watcher/events/
processor.rs

1use super::moderation::Moderation;
2use super::Event;
3use crate::events::errors::EventProcessorError;
4use crate::events::retry::event::RetryEvent;
5use nexus_common::db::PubkyClient;
6use nexus_common::models::homeserver::Homeserver;
7use nexus_common::types::DynError;
8use nexus_common::{WatcherConfig, FILES_DIR_TEST};
9use opentelemetry::trace::{FutureExt, Span, TraceContextExt, Tracer};
10use opentelemetry::{global, Context, KeyValue};
11use pubky_app_specs::PubkyId;
12use std::error::Error;
13use std::path::PathBuf;
14use tracing::{debug, error, info};
15
16pub struct EventProcessor {
17    pub homeserver: Homeserver,
18    limit: u32,
19    pub files_path: PathBuf,
20    pub tracer_name: String,
21    pub moderation: Moderation,
22}
23
24impl EventProcessor {
25    /// Creates a new `EventProcessor` instance for testing purposes.
26    ///
27    /// This function initializes an `EventProcessor` configured with:
28    /// - A mock homeserver constructed using the provided `homeserver_url` and `homeserver_pubky`.
29    /// - A default configuration, including an HTTP client, a limit of 1000 events, and a sender channel.
30    ///
31    /// It is designed for use in integration tests, benchmarking scenarios, or other test environments
32    /// where a controlled and predictable `EventProcessor` instance is required.
33    ///
34    /// # Parameters
35    /// - `homeserver_id`: A `String` representing the URL of the homeserver to be used in the test environment.
36    /// - `tx`: A `RetryManagerSenderChannel` used to handle outgoing messages or events.
37    pub async fn test(homeserver_id: String) -> Self {
38        let id = PubkyId::try_from(&homeserver_id).expect("Homeserver ID should be valid");
39        let homeserver = Homeserver::new(id).await.unwrap();
40
41        // hardcoded nexus-watcher/tests/utils/moderator_key.pkarr public key used by the moderator user on tests
42        let moderation = Moderation {
43            id: PubkyId::try_from("uo7jgkykft4885n8cruizwy6khw71mnu5pq3ay9i8pw1ymcn85ko")
44                .expect("Hardcoded test moderation key should be valid"),
45            tags: Vec::from(["label_to_moderate".to_string()]),
46        };
47
48        info!(
49            "Watcher static files PATH during tests are stored inside of the watcher crate: {:?}",
50            PathBuf::from(FILES_DIR_TEST)
51        );
52        Self {
53            homeserver,
54            limit: 1000,
55            files_path: PathBuf::from(FILES_DIR_TEST),
56            tracer_name: String::from("watcher.test"),
57            moderation,
58        }
59    }
60
61    pub async fn from_config(config: &WatcherConfig) -> Result<Self, DynError> {
62        let homeserver = Homeserver::from_config(config.homeserver.clone()).await?;
63        let limit = config.events_limit;
64        let files_path = config.stack.files_path.clone();
65        let tracer_name = config.name.clone();
66
67        let moderation = Moderation {
68            id: config.moderation_id.clone(),
69            tags: config.moderated_tags.clone(),
70        };
71
72        info!(
73            "Initialized Event Processor for homeserver: {:?}",
74            homeserver
75        );
76
77        Ok(Self {
78            homeserver,
79            limit,
80            files_path,
81            tracer_name,
82            moderation,
83        })
84    }
85
86    pub async fn run(&mut self) -> Result<(), DynError> {
87        let lines = {
88            let tracer = global::tracer(self.tracer_name.clone());
89            let span = tracer.start("Polling Events");
90            let cx = Context::new().with_span(span);
91            self.poll_events().with_context(cx).await
92        };
93
94        match lines {
95            Err(e) => {
96                error!("Error polling events: {:?}", e);
97                return Err(e);
98            }
99            Ok(None) => {
100                info!("No new events");
101            }
102            Ok(Some(lines)) => {
103                self.process_event_lines(lines).await?;
104            }
105        }
106
107        Ok(())
108    }
109
110    /// Polls new events from the homeserver.
111    ///
112    /// It sends a GET request to the homeserver's events endpoint
113    /// using the current cursor and a specified limit. It retrieves new event
114    /// URIs in a newline-separated format, processes it into a vector of strings,
115    /// and returns the result.
116    async fn poll_events(&mut self) -> Result<Option<Vec<String>>, DynError> {
117        debug!("Polling new events from homeserver");
118
119        let response_text = {
120            let pubky_client =
121                PubkyClient::get().map_err(|e| EventProcessorError::PubkyClientError {
122                    message: e.to_string(),
123                })?;
124            let url = format!(
125                "https://{}/events/?cursor={}&limit={}",
126                self.homeserver.id, self.homeserver.cursor, self.limit
127            );
128
129            let response = pubky_client.get(url).send().await.map_err(|e| {
130                Box::new(EventProcessorError::PubkyClientError {
131                    message: format!("{:?}", e.source()),
132                })
133            })?;
134
135            response.text().await?
136        };
137
138        let lines: Vec<String> = response_text.trim().lines().map(String::from).collect();
139        debug!("Homeserver response lines {:?}", lines);
140
141        if lines.is_empty() || (lines.len() == 1 && lines[0].is_empty()) {
142            return Ok(None);
143        }
144
145        Ok(Some(lines))
146    }
147
148    /// Processes a batch of event lines retrieved from the homeserver.
149    ///
150    /// This function iterates over a vector of event URIs, handling each line based on its content:
151    /// - Lines starting with `cursor:` update the cursor for the homeserver and save it to the index.
152    /// - Other lines are parsed into events and processed accordingly. If parsing fails, an error is logged.
153    ///
154    /// # Parameters
155    /// - `lines`: A vector of strings representing event lines retrieved from the homeserver.
156    pub async fn process_event_lines(&mut self, lines: Vec<String>) -> Result<(), DynError> {
157        for line in &lines {
158            if line.starts_with("cursor:") {
159                if let Some(cursor) = line.strip_prefix("cursor: ") {
160                    self.homeserver.cursor = cursor.to_string();
161                    self.homeserver.put_to_index().await?;
162                    info!("Cursor for the next request: {}", cursor);
163                }
164            } else {
165                let event = match Event::parse_event(line, self.files_path.clone()) {
166                    Ok(event) => event,
167                    Err(e) => {
168                        error!("{}", e);
169                        None
170                    }
171                };
172                if let Some(event) = event {
173                    let tracer = global::tracer(self.tracer_name.clone());
174                    let mut span = tracer.start(event.parsed_uri.resource.to_string());
175                    span.set_attribute(KeyValue::new("event.uri", event.uri.clone()));
176                    span.set_attribute(KeyValue::new("event.type", event.event_type.to_string()));
177                    span.set_attribute(KeyValue::new(
178                        "event.user_id",
179                        event.parsed_uri.user_id.to_string(),
180                    ));
181                    span.set_attribute(KeyValue::new(
182                        "event.resource_id",
183                        event.parsed_uri.resource.id().unwrap_or("".to_string()),
184                    ));
185                    let cx = Context::new().with_span(span);
186                    debug!("Processing event: {:?}", event);
187                    self.handle_event(event).with_context(cx).await?;
188                }
189            }
190        }
191
192        Ok(())
193    }
194
195    /// Processes an event and track the fail event it if necessary
196    /// # Parameters:
197    /// - `event`: The event to be processed
198    async fn handle_event(&mut self, event: Event) -> Result<(), DynError> {
199        if let Err(e) = event.clone().handle(&self.moderation).await {
200            if let Some((index_key, retry_event)) = extract_retry_event_info(&event, e) {
201                error!("{}, {}", retry_event.error_type, index_key);
202                if let Err(err) = retry_event.put_to_index(index_key).await {
203                    error!("Failed to put event to retry index: {}", err);
204                }
205            }
206        }
207        Ok(())
208    }
209}
210
211/// Extracts retry-related information from an event and its associated error
212///
213/// # Parameters
214/// - `event`: Reference to the event for which retry information is being extracted
215/// - `error`: Determines whether the event is eligible for a retry or should be discarded
216fn extract_retry_event_info(event: &Event, error: DynError) -> Option<(String, RetryEvent)> {
217    let retry_event = match error.downcast_ref::<EventProcessorError>() {
218        Some(EventProcessorError::InvalidEventLine { message }) => {
219            error!("{}", message);
220            return None;
221        }
222        Some(event_processor_error) => RetryEvent::new(event_processor_error.clone()),
223        // Others errors must be logged at least for now
224        None => {
225            error!("Unhandled error type for URI: {}, {:?}", event.uri, error);
226            return None;
227        }
228    };
229
230    // Generate a compress index to save in the cache
231    let index = match RetryEvent::generate_index_key(&event.uri) {
232        Some(retry_index) => retry_index,
233        None => {
234            return None;
235        }
236    };
237    Some((format!("{}:{}", event.event_type, index), retry_event))
238}
nexus_watcher/events/processor.rs

nexus_watcher/events/
processor.rs
