vineyard

use std::collections::{BTreeSet, HashSet};

use std::fs::{File, OpenOptions, create_dir_all};

use std::future::Future;

use std::io::Error as IoError;

use std::iter::once;

use std::path::PathBuf;

use std::pin::Pin;

use std::sync::Arc;

use anyhow::Context as _;

use fs2::FileExt as _;

use nostr::event::{Event, EventId, Kind};

use nostr::filter::Filter;

use nostr::nips::nip65::extract_relay_list;

use nostr_database::error::Error;

use nostr_database::{DatabaseEventStatus, Features, NostrDatabase, SaveEventStatus};

use tokio::sync::Mutex;

use crate::config::{Config, relay_url};

use crate::journal::Journal;

type DatabaseFuture<'database, T> =

    Pin<Box<dyn Future<Output = Result<T, Error>> + Send + 'database>>;

#[derive(Debug)]

pub struct Store {

    pub(crate) journal: Journal,

    path: PathBuf,

    min_free_bytes: u64,

    max_query_bytes: usize,

    max_repositories: usize,

    max_discovered_peers: usize,

    excluded_peers: HashSet<String>,

    writer: Arc<Mutex<()>>,

    _lock: File,

}

impl Store {

    pub async fn open(config: &Config) -> anyhow::Result<Arc<Self>> {

        if let Some(parent) = config

            .database

            .parent()

            .filter(|parent| !parent.as_os_str().is_empty())

        {

            create_dir_all(parent).context("creating database directory")?;

        }

        let mut lock_name = config.database.as_os_str().to_os_string();

        lock_name.push(".lock");

        let mut lock_options = OpenOptions::new();

        lock_options

            .read(true)

            .write(true)

            .create(true)

            .truncate(false);

        #[cfg(unix)]

        {

            use std::os::unix::fs::OpenOptionsExt as _;

            lock_options.mode(0o600)

        };

        let lock = lock_options

            .open(PathBuf::from(lock_name))

            .context("opening database lock")?;

        lock.try_lock_exclusive()

            .context("database is already in use")?;

        let excluded_peers = once(&config.public_url)

            .chain(config.peers.iter().map(|peer| &peer.url))

            .map(|url| relay_url(url).map(|parsed| parsed.to_string()))

            .collect::<Option<HashSet<_>>>()

            .context("validated relay url became invalid")?;

        let writer = Arc::new(Mutex::new(()));

        let journal = Journal::open(

            config.database.clone(),

            config.limits.max_database_bytes,

            config

                .limits

                .max_discovered_peers

                .saturating_add(excluded_peers.len()),

            excluded_peers.iter().cloned().collect(),

            Arc::clone(&writer),

        )

        .await?;

        journal

            .forget_discovered(excluded_peers.iter().cloned().collect())

            .await?;

        tracing::info!(path = %config.database.display(), "opened event database");

        Ok(Arc::new(Self {

            journal,

            path: config.database.clone(),

            min_free_bytes: config.limits.min_free_bytes,

            max_query_bytes: config.limits.max_query_bytes,

            max_repositories: config.limits.max_repositories,

            max_discovered_peers: config.limits.max_discovered_peers,

            excluded_peers,

            writer,

            _lock: lock,

        }))

    }

    pub async fn healthy(&self) -> bool {

        self.journal.healthy().await

    }

    async fn query_events(&self, filter: Filter) -> Result<BTreeSet<Event>, Error> {

        self.journal

            .query(filter, self.max_query_bytes)

            .await

            .map_err(storage_error)

    }

}

fn storage_error(error: anyhow::Error) -> Error {

    Error::storage(IoError::other(error))

}

impl NostrDatabase for Store {

    fn backend(&self) -> &'static str {

        "vineyard-sqlite"

    }

    fn features(&self) -> Features {

        Features {

            persistent: true,

            event_expiration: false,

            full_text_search: false,

            request_to_vanish: false,

        }

    }

    fn save_event<'database>(

        &'database self,

        event: &'database Event,

    ) -> DatabaseFuture<'database, SaveEventStatus> {

        Box::pin(async move {

            let writer = self.writer.lock().await;

            if event.kind.as_u16() == 30617 {

                if let Some(identifier) = event.tags.identifier() {

                    let registered = self

                        .query_events(

                            Filter::new()

                                .kind(event.kind)

                                .author(event.pubkey)

                                .identifier(identifier)

                                .limit(1),

                        )

                        .await?

                        .len();

                    if registered == 0

                        && self

                            .query_events(

                                Filter::new()

                                    .kind(event.kind)

                                    .limit(self.max_repositories + 1),

                            )

                            .await?

                            .len()

                            >= self.max_repositories

                    {

                        return Err(Error::storage(IoError::other(

                            "repository directory is full",

                        )));

                    }

                }

            }

            if self.journal.status(event.id).await.map_err(storage_error)?

                != DatabaseEventStatus::Saved

            {

                let free = fs2::available_space(&self.path).map_err(Error::io)?;

                let event_bytes = u64::try_from(event.as_json().len())

                    .map_err(|error| Error::storage(IoError::other(error.to_string())))?;

                if free < self.min_free_bytes.saturating_add(event_bytes) {

                    return Err(Error::storage(IoError::other("storage reserve reached")));

                }

            }

            let status = self

                .journal

                .save_locked(event.clone(), &writer)

                .await

                .map_err(storage_error)?;

            if status.is_success() && event.kind == Kind::RelayList {

                let urls = extract_relay_list(event)

                    .map(|(url, _)| url.to_string())

                    .filter(|url| !self.excluded_peers.contains(url))

                    .collect();

                if let Err(error) = self

                    .journal

                    .replace_discovered_locked(

                        event.pubkey,

                        urls,

                        self.max_discovered_peers,

                        &writer,

                    )

                    .await

                {

                    tracing::error!(event = %event.id, %error, "saving discovered peers failed");

                }

            }

            drop(writer);

            Ok(status)

        })

    }

    fn check_id<'database>(

        &'database self,

        event_id: &'database EventId,

    ) -> DatabaseFuture<'database, DatabaseEventStatus> {

        let requested_id = *event_id;

        Box::pin(async move {

            self.journal

                .status(requested_id)

                .await

                .map_err(storage_error)

        })

    }

    fn event_by_id<'database>(

        &'database self,

        event_id: &'database EventId,

    ) -> DatabaseFuture<'database, Option<Event>> {

        let requested_id = *event_id;

        Box::pin(async move {

            self.journal

                .event(requested_id)

                .await

                .map_err(storage_error)

        })

    }

    fn count(&self, filter: Filter) -> DatabaseFuture<'_, usize> {

        Box::pin(async move { self.journal.count(filter).await.map_err(storage_error) })

    }

    fn query(&self, filter: Filter) -> DatabaseFuture<'_, BTreeSet<Event>> {

        Box::pin(self.query_events(filter))

    }

    fn delete(&self, _filter: Filter) -> DatabaseFuture<'_, ()> {

        Box::pin(async { Err(Error::unsupported("use signed deletion requests")) })

    }

    fn wipe(&self) -> DatabaseFuture<'_, ()> {

        Box::pin(async { Err(Error::unsupported("database wipe is disabled")) })

    }

}