| 1 | use std::collections::{BTreeSet, HashSet}; |
| 2 | use std::fs::{File, OpenOptions, create_dir_all}; |
| 3 | use std::future::Future; |
| 4 | use std::io::Error as IoError; |
| 5 | use std::iter::once; |
| 6 | use std::path::PathBuf; |
| 7 | use std::pin::Pin; |
| 8 | use std::sync::Arc; |
| 9 | |
| 10 | use anyhow::Context as _; |
| 11 | use fs2::FileExt as _; |
| 12 | use nostr::event::{Event, EventId, Kind}; |
| 13 | use nostr::filter::Filter; |
| 14 | use nostr::nips::nip65::extract_relay_list; |
| 15 | use nostr_database::error::Error; |
| 16 | use nostr_database::{DatabaseEventStatus, Features, NostrDatabase, SaveEventStatus}; |
| 17 | use tokio::sync::Mutex; |
| 18 | |
| 19 | use crate::config::{Config, relay_url}; |
| 20 | use crate::journal::Journal; |
| 21 | |
| 22 | type DatabaseFuture<'database, T> = |
| 23 | Pin<Box<dyn Future<Output = Result<T, Error>> + Send + 'database>>; |
| 24 | |
| 25 | #[derive(Debug)] |
| 26 | pub struct Store { |
| 27 | pub(crate) journal: Journal, |
| 28 | path: PathBuf, |
| 29 | min_free_bytes: u64, |
| 30 | max_query_bytes: usize, |
| 31 | max_repositories: usize, |
| 32 | max_discovered_peers: usize, |
| 33 | excluded_peers: HashSet<String>, |
| 34 | writer: Arc<Mutex<()>>, |
| 35 | _lock: File, |
| 36 | } |
| 37 | |
| 38 | impl Store { |
| 39 | pub async fn open(config: &Config) -> anyhow::Result<Arc<Self>> { |
| 40 | if let Some(parent) = config |
| 41 | .database |
| 42 | .parent() |
| 43 | .filter(|parent| !parent.as_os_str().is_empty()) |
| 44 | { |
| 45 | create_dir_all(parent).context("creating database directory")?; |
| 46 | } |
| 47 | let mut lock_name = config.database.as_os_str().to_os_string(); |
| 48 | lock_name.push(".lock"); |
| 49 | let mut lock_options = OpenOptions::new(); |
| 50 | lock_options |
| 51 | .read(true) |
| 52 | .write(true) |
| 53 | .create(true) |
| 54 | .truncate(false); |
| 55 | #[cfg(unix)] |
| 56 | { |
| 57 | use std::os::unix::fs::OpenOptionsExt as _; |
| 58 | lock_options.mode(0o600) |
| 59 | }; |
| 60 | let lock = lock_options |
| 61 | .open(PathBuf::from(lock_name)) |
| 62 | .context("opening database lock")?; |
| 63 | lock.try_lock_exclusive() |
| 64 | .context("database is already in use")?; |
| 65 | |
| 66 | let excluded_peers = once(&config.public_url) |
| 67 | .chain(config.peers.iter().map(|peer| &peer.url)) |
| 68 | .map(|url| relay_url(url).map(|parsed| parsed.to_string())) |
| 69 | .collect::<Option<HashSet<_>>>() |
| 70 | .context("validated relay url became invalid")?; |
| 71 | let writer = Arc::new(Mutex::new(())); |
| 72 | let journal = Journal::open( |
| 73 | config.database.clone(), |
| 74 | config.limits.max_database_bytes, |
| 75 | config |
| 76 | .limits |
| 77 | .max_discovered_peers |
| 78 | .saturating_add(excluded_peers.len()), |
| 79 | excluded_peers.iter().cloned().collect(), |
| 80 | Arc::clone(&writer), |
| 81 | ) |
| 82 | .await?; |
| 83 | journal |
| 84 | .forget_discovered(excluded_peers.iter().cloned().collect()) |
| 85 | .await?; |
| 86 | tracing::info!(path = %config.database.display(), "opened event database"); |
| 87 | Ok(Arc::new(Self { |
| 88 | journal, |
| 89 | path: config.database.clone(), |
| 90 | min_free_bytes: config.limits.min_free_bytes, |
| 91 | max_query_bytes: config.limits.max_query_bytes, |
| 92 | max_repositories: config.limits.max_repositories, |
| 93 | max_discovered_peers: config.limits.max_discovered_peers, |
| 94 | excluded_peers, |
| 95 | writer, |
| 96 | _lock: lock, |
| 97 | })) |
| 98 | } |
| 99 | |
| 100 | pub async fn healthy(&self) -> bool { |
| 101 | self.journal.healthy().await |
| 102 | } |
| 103 | |
| 104 | async fn query_events(&self, filter: Filter) -> Result<BTreeSet<Event>, Error> { |
| 105 | self.journal |
| 106 | .query(filter, self.max_query_bytes) |
| 107 | .await |
| 108 | .map_err(storage_error) |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | fn storage_error(error: anyhow::Error) -> Error { |
| 113 | Error::storage(IoError::other(error)) |
| 114 | } |
| 115 | |
| 116 | impl NostrDatabase for Store { |
| 117 | fn backend(&self) -> &'static str { |
| 118 | "vineyard-sqlite" |
| 119 | } |
| 120 | |
| 121 | fn features(&self) -> Features { |
| 122 | Features { |
| 123 | persistent: true, |
| 124 | event_expiration: false, |
| 125 | full_text_search: false, |
| 126 | request_to_vanish: false, |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | fn save_event<'database>( |
| 131 | &'database self, |
| 132 | event: &'database Event, |
| 133 | ) -> DatabaseFuture<'database, SaveEventStatus> { |
| 134 | Box::pin(async move { |
| 135 | let writer = self.writer.lock().await; |
| 136 | if event.kind.as_u16() == 30617 { |
| 137 | if let Some(identifier) = event.tags.identifier() { |
| 138 | let registered = self |
| 139 | .query_events( |
| 140 | Filter::new() |
| 141 | .kind(event.kind) |
| 142 | .author(event.pubkey) |
| 143 | .identifier(identifier) |
| 144 | .limit(1), |
| 145 | ) |
| 146 | .await? |
| 147 | .len(); |
| 148 | if registered == 0 |
| 149 | && self |
| 150 | .query_events( |
| 151 | Filter::new() |
| 152 | .kind(event.kind) |
| 153 | .limit(self.max_repositories + 1), |
| 154 | ) |
| 155 | .await? |
| 156 | .len() |
| 157 | >= self.max_repositories |
| 158 | { |
| 159 | return Err(Error::storage(IoError::other( |
| 160 | "repository directory is full", |
| 161 | ))); |
| 162 | } |
| 163 | } |
| 164 | } |
| 165 | if self.journal.status(event.id).await.map_err(storage_error)? |
| 166 | != DatabaseEventStatus::Saved |
| 167 | { |
| 168 | let free = fs2::available_space(&self.path).map_err(Error::io)?; |
| 169 | let event_bytes = u64::try_from(event.as_json().len()) |
| 170 | .map_err(|error| Error::storage(IoError::other(error.to_string())))?; |
| 171 | if free < self.min_free_bytes.saturating_add(event_bytes) { |
| 172 | return Err(Error::storage(IoError::other("storage reserve reached"))); |
| 173 | } |
| 174 | } |
| 175 | let status = self |
| 176 | .journal |
| 177 | .save_locked(event.clone(), &writer) |
| 178 | .await |
| 179 | .map_err(storage_error)?; |
| 180 | if status.is_success() && event.kind == Kind::RelayList { |
| 181 | let urls = extract_relay_list(event) |
| 182 | .map(|(url, _)| url.to_string()) |
| 183 | .filter(|url| !self.excluded_peers.contains(url)) |
| 184 | .collect(); |
| 185 | if let Err(error) = self |
| 186 | .journal |
| 187 | .replace_discovered_locked( |
| 188 | event.pubkey, |
| 189 | urls, |
| 190 | self.max_discovered_peers, |
| 191 | &writer, |
| 192 | ) |
| 193 | .await |
| 194 | { |
| 195 | tracing::error!(event = %event.id, %error, "saving discovered peers failed"); |
| 196 | } |
| 197 | } |
| 198 | drop(writer); |
| 199 | Ok(status) |
| 200 | }) |
| 201 | } |
| 202 | |
| 203 | fn check_id<'database>( |
| 204 | &'database self, |
| 205 | event_id: &'database EventId, |
| 206 | ) -> DatabaseFuture<'database, DatabaseEventStatus> { |
| 207 | let requested_id = *event_id; |
| 208 | Box::pin(async move { |
| 209 | self.journal |
| 210 | .status(requested_id) |
| 211 | .await |
| 212 | .map_err(storage_error) |
| 213 | }) |
| 214 | } |
| 215 | |
| 216 | fn event_by_id<'database>( |
| 217 | &'database self, |
| 218 | event_id: &'database EventId, |
| 219 | ) -> DatabaseFuture<'database, Option<Event>> { |
| 220 | let requested_id = *event_id; |
| 221 | Box::pin(async move { |
| 222 | self.journal |
| 223 | .event(requested_id) |
| 224 | .await |
| 225 | .map_err(storage_error) |
| 226 | }) |
| 227 | } |
| 228 | |
| 229 | fn count(&self, filter: Filter) -> DatabaseFuture<'_, usize> { |
| 230 | Box::pin(async move { self.journal.count(filter).await.map_err(storage_error) }) |
| 231 | } |
| 232 | |
| 233 | fn query(&self, filter: Filter) -> DatabaseFuture<'_, BTreeSet<Event>> { |
| 234 | Box::pin(self.query_events(filter)) |
| 235 | } |
| 236 | |
| 237 | fn delete(&self, _filter: Filter) -> DatabaseFuture<'_, ()> { |
| 238 | Box::pin(async { Err(Error::unsupported("use signed deletion requests")) }) |
| 239 | } |
| 240 | |
| 241 | fn wipe(&self) -> DatabaseFuture<'_, ()> { |
| 242 | Box::pin(async { Err(Error::unsupported("database wipe is disabled")) }) |
| 243 | } |
| 244 | } |