nexus_watcher/
builder.rs

1use crate::events::processor::EventProcessor;
2use nexus_common::db::{DatabaseConfig, PubkyClient};
3use nexus_common::file::ConfigReader;
4use nexus_common::types::DynError;
5use nexus_common::{DaemonConfig, Level, StackConfig};
6use nexus_common::{StackManager, WatcherConfig};
7use pubky_app_specs::PubkyId;
8use std::path::PathBuf;
9use tokio::time::Duration;
10use tokio::{pin, signal};
11use tracing::{debug, error, info};
12
13#[derive(Debug, Default)]
14pub struct NexusWatcherBuilder(pub WatcherConfig);
15
16impl NexusWatcherBuilder {
17    /// Creates a `NexusWatcherBuilder` instance with the given configuration and stack settings.
18    pub fn with_stack(mut config: WatcherConfig, stack: &StackConfig) -> Self {
19        config.stack = stack.clone();
20        Self(config)
21    }
22
23    /// Sets the service name for observability (tracing, logging, monitoring)
24    pub fn name(&mut self, name: String) -> &mut Self {
25        self.0.name = name;
26
27        self
28    }
29
30    /// Configures the logging level for the service, determining verbosity and log output
31    pub fn log_level(&mut self, log_level: Level) -> &mut Self {
32        self.0.stack.log_level = log_level;
33
34        self
35    }
36
37    pub fn testnet(&mut self, testnet: bool) -> &mut Self {
38        self.0.testnet = testnet;
39
40        self
41    }
42
43    pub fn homeserver(&mut self, homeserver: PubkyId) -> &mut Self {
44        self.0.homeserver = homeserver;
45
46        self
47    }
48
49    /// Sets the directory for storing static files on the server
50    pub fn files_path(&mut self, files_path: PathBuf) -> &mut Self {
51        self.0.stack.files_path = files_path;
52
53        self
54    }
55
56    /// Sets the OpenTelemetry endpoint for tracing and monitoring
57    pub fn otlp_endpoint(&mut self, otlp_endpoint: Option<String>) -> &mut Self {
58        self.0.stack.otlp_endpoint = otlp_endpoint;
59
60        self
61    }
62
63    /// Sets the database configuration, including graph database and Redis settings
64    pub fn db(&mut self, db: DatabaseConfig) -> &mut Self {
65        self.0.stack.db = db;
66
67        self
68    }
69
70    /// Opens ddbb connections and initialises tracing layer (if provided in config)
71    pub async fn init_stack(&self) -> Result<(), DynError> {
72        StackManager::setup(&self.0.name, &self.0.stack).await?;
73        let _ = PubkyClient::initialise(self.0.testnet).await;
74        Ok(())
75    }
76
77    /// Initializes the watcher integration test stack
78    pub async fn init_test_stack(&self) -> Result<(), DynError> {
79        StackManager::setup(&self.0.name, &self.0.stack).await?;
80        Ok(())
81    }
82
83    /// Initializes the service stack and starts the NexusWatcher event loop
84    pub async fn start(self) -> Result<(), DynError> {
85        self.init_stack().await?;
86        NexusWatcher::start(self.0).await
87    }
88}
89
90pub struct NexusWatcher {}
91
92impl NexusWatcher {
93    /// Creates a new instance with default configuration
94    pub fn builder() -> NexusWatcherBuilder {
95        NexusWatcherBuilder::default()
96    }
97
98    /// Loads the configuration from a file and starts the Watcher
99    pub async fn start_from_path(config_dir: PathBuf) -> Result<(), DynError> {
100        let config = WatcherConfig::read_config_file(config_dir).await?;
101        NexusWatcherBuilder(config).start().await
102    }
103
104    /// Loads the configuration from nexusd service and starts the Watcher
105    pub async fn start_from_daemon(config_dir: PathBuf) -> Result<(), DynError> {
106        let config = DaemonConfig::read_config_file(config_dir).await?;
107        NexusWatcherBuilder(Into::<WatcherConfig>::into(config))
108            .start()
109            .await
110    }
111
112    pub async fn start(config: WatcherConfig) -> Result<(), DynError> {
113        debug!(?config, "Running NexusWatcher with ");
114        let mut event_processor = EventProcessor::from_config(&config).await?;
115
116        let shutdown_signal = signal::ctrl_c();
117        pin!(shutdown_signal);
118        // If we wanted to handle SIGTERM too
119        // let mut sigterm = signal::unix::signal(signal::unix::SignalKind::terminate())?;
120        // Now we only catch SIGINT
121
122        let mut interval = tokio::time::interval(Duration::from_millis(config.watcher_sleep));
123
124        // TODO: This lets you cancel the underlying future instead of waiting for it to complete
125        // To achieve low-latency shutdown (i.e. abort in-flight processing immediately on Ctrl+C),
126        // consider offloading `event_processor.run()` into its own cancellable Tokio task (or spawn_blocking thread),
127        // keeping its `JoinHandle`, and invoking `handle.abort()` when the shutdown (ctlr + c) future resolves
128        loop {
129            tokio::select! {
130                _ = &mut shutdown_signal => {
131                    info!("SIGINT received, starting graceful shutdown...");
132                    break;
133                }
134                _ = interval.tick() => {
135                    info!("Fetching events…");
136                    if let Err(e) = event_processor.run().await {
137                        error!("Error while processing events: {:?}", e);
138                    }
139                }
140            }
141        }
142        info!("service shut down gracefully");
143        Ok(())
144    }
145}
nexus_watcher/builder.rs

nexus_watcher/
builder.rs
