nexus_watcher/events/
mod.rs

1use errors::EventProcessorError;
2use moderation::Moderation;
3use nexus_common::db::PubkyClient;
4use nexus_common::types::DynError;
5use pubky_app_specs::{ParsedUri, PubkyAppObject, Resource};
6use serde::{Deserialize, Serialize};
7use std::{fmt, path::PathBuf};
8use tracing::debug;
9
10pub mod errors;
11pub mod handlers;
12pub mod moderation;
13pub mod processor;
14pub mod retry;
15
16// Look for the end pattern after the start index, or use the end of the string if not found
17#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
18pub enum EventType {
19    Put,
20    Del,
21}
22
23impl fmt::Display for EventType {
24    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
25        let upper_case_str = match self {
26            EventType::Put => "PUT",
27            EventType::Del => "DEL",
28        };
29        write!(f, "{upper_case_str}")
30    }
31}
32
33#[derive(Debug, Clone)]
34pub struct Event {
35    pub uri: String,
36    pub event_type: EventType,
37    pub parsed_uri: ParsedUri,
38    pub files_path: PathBuf,
39}
40
41impl Event {
42    pub fn parse_event(line: &str, files_path: PathBuf) -> Result<Option<Self>, DynError> {
43        debug!("New event: {}", line);
44        let parts: Vec<&str> = line.split(' ').collect();
45        if parts.len() != 2 {
46            return Err(EventProcessorError::InvalidEventLine {
47                message: format!("Malformed event line, {line}"),
48            }
49            .into());
50        }
51
52        let event_type = match parts[0] {
53            "PUT" => EventType::Put,
54            "DEL" => EventType::Del,
55            other => {
56                return Err(EventProcessorError::InvalidEventLine {
57                    message: format!("Unknown event type: {other}"),
58                }
59                .into())
60            }
61        };
62
63        // Validate and parse the URI using pubky-app-specs
64        let uri = parts[1].to_string();
65        let parsed_uri = ParsedUri::try_from(uri.as_str()).map_err(|e| {
66            {
67                EventProcessorError::InvalidEventLine {
68                    message: format!("Cannot parse event URI: {e}"),
69                }
70            }
71        })?;
72
73        match parsed_uri.resource {
74            // Unknown resource
75            Resource::Unknown => {
76                return Err(EventProcessorError::InvalidEventLine {
77                    message: format!("Unknown resource in URI: {uri}"),
78                }
79                .into())
80            }
81            // Known resources not handled by Nexus
82            Resource::LastRead | Resource::Feed(_) | Resource::Blob(_) => return Ok(None),
83            _ => (),
84        };
85
86        Ok(Some(Event {
87            uri,
88            event_type,
89            parsed_uri,
90            files_path,
91        }))
92    }
93
94    pub async fn handle(self, moderation: &Moderation) -> Result<(), DynError> {
95        match self.event_type {
96            EventType::Put => self.handle_put_event(moderation).await,
97            EventType::Del => self.handle_del_event().await,
98        }
99    }
100
101    /// Handles a PUT event by fetching the blob from the homeserver
102    /// and using the importer to convert it to a PubkyAppObject.
103    pub async fn handle_put_event(self, moderation: &Moderation) -> Result<(), DynError> {
104        debug!("Handling PUT event for URI: {}", self.uri);
105
106        let response;
107        {
108            let pubky_client =
109                PubkyClient::get().map_err(|e| EventProcessorError::PubkyClientError {
110                    message: e.to_string(),
111                })?;
112
113            response = match pubky_client.get(&self.uri).send().await {
114                Ok(response) => response,
115                Err(e) => {
116                    return Err(EventProcessorError::PubkyClientError {
117                        message: format!("{e}"),
118                    }
119                    .into())
120                }
121            };
122        } // drop the pubky_client lock
123
124        let blob = response.bytes().await?;
125        let resource = self.parsed_uri.resource;
126
127        // Use the new importer from pubky-app-specs
128        let pubky_object = PubkyAppObject::from_resource(&resource, &blob).map_err(|e| {
129            EventProcessorError::PubkyClientError {
130                message: format!(
131                    "The importer could not create PubkyAppObject from Uri and Blob: {e}"
132                ),
133            }
134        })?;
135
136        let user_id = self.parsed_uri.user_id;
137        match (pubky_object, resource) {
138            (PubkyAppObject::User(user), Resource::User) => {
139                handlers::user::sync_put(user, user_id).await?
140            }
141            (PubkyAppObject::Post(post), Resource::Post(post_id)) => {
142                handlers::post::sync_put(post, user_id, post_id).await?
143            }
144            (PubkyAppObject::Follow(_follow), Resource::Follow(followee_id)) => {
145                handlers::follow::sync_put(user_id, followee_id).await?
146            }
147            (PubkyAppObject::Mute(_mute), Resource::Mute(muted_id)) => {
148                handlers::mute::sync_put(user_id, muted_id).await?
149            }
150            (PubkyAppObject::Bookmark(bookmark), Resource::Bookmark(bookmark_id)) => {
151                handlers::bookmark::sync_put(user_id, bookmark, bookmark_id).await?
152            }
153            (PubkyAppObject::Tag(tag), Resource::Tag(tag_id)) => {
154                if moderation.should_delete(&tag, user_id.clone()).await {
155                    Moderation::apply_moderation(tag, self.files_path).await?
156                } else {
157                    handlers::tag::sync_put(tag, user_id, tag_id).await?
158                }
159            }
160            (PubkyAppObject::File(file), Resource::File(file_id)) => {
161                handlers::file::sync_put(file, self.uri, user_id, file_id, self.files_path).await?
162            }
163            other => {
164                debug!("Event type not handled, Resource: {:?}", other);
165            }
166        }
167        Ok(())
168    }
169
170    pub async fn handle_del_event(self) -> Result<(), DynError> {
171        debug!("Handling DEL event for URI: {}", self.uri);
172
173        let user_id = self.parsed_uri.user_id;
174        match self.parsed_uri.resource {
175            Resource::User => handlers::user::del(user_id).await?,
176            Resource::Post(post_id) => handlers::post::del(user_id, post_id).await?,
177            Resource::Follow(followee_id) => handlers::follow::del(user_id, followee_id).await?,
178            Resource::Mute(muted_id) => handlers::mute::del(user_id, muted_id).await?,
179            Resource::Bookmark(bookmark_id) => {
180                handlers::bookmark::del(user_id, bookmark_id).await?
181            }
182            Resource::Tag(tag_id) => handlers::tag::del(user_id, tag_id).await?,
183            Resource::File(file_id) => {
184                handlers::file::del(&user_id, file_id, self.files_path).await?
185            }
186            other => {
187                debug!("DEL event type not handled for resource: {:?}", other);
188            }
189        }
190        Ok(())
191    }
192}
nexus_watcher/events/mod.rs

nexus_watcher/events/
mod.rs
