nostralink 0.2.1

//! nostr-database backend

use super::prelude::*;
use crate::niri::ToNamedNode;
use crate::querydb::nrq_get;
use nostr::{Event, Filter, Kind};
use nostr_database::prelude::*;
use nostr_database::{NostrDatabase, NostrEventsDatabase};
use std::str::FromStr;
use std::time::Duration;
use tokio::time::sleep;

impl RdfEventsStore {
    /// Creates a SparQL query for a given nostr filter
    fn sparqlify_filter(&self, filter: &Filter) -> (String, String) {
        let mut ckey = String::new();
        let mut q =
            self.prepare_query(&nrq_get("events_matching_filter").unwrap());

        // IDs
        if let Some(ref ids) = filter.ids {
            let values = ids
                .iter()
                .map(|x| format!(r#"'{}'"#, x.to_hex()))
                .collect::<Vec<_>>()
                .join(",");
            q = q.replace(
                "@IDS@",
                &format!("FILTER(?event_id IN ({}))", values),
            );
            ckey.push_str(&values);
        } else {
            q = q.replace("@IDS@", "");
        }

        // Kinds
        if let Some(ref kinds) = filter.kinds {
            let kvl = kinds
                .iter()
                .map(|x| x.to_string())
                .collect::<Vec<_>>()
                .join(",");
            q = q.replace("@KINDS@", &format!("FILTER(?kind IN ({}))", kvl));
            ckey.push_str(&kvl);
        } else {
            q = q.replace("@KINDS@", "");
        }

        // Authors
        if let Some(ref authors) = filter.authors {
            let pk = authors
                .iter()
                .map(|x| format!(r#"'{}'"#, x.to_hex()))
                .collect::<Vec<_>>()
                .join(",");
            q = q.replace("@PUBKS@", &format!("FILTER(?pubk IN ({}))", pk));
            ckey.push_str(&pk);
        } else {
            q = q.replace("@PUBKS@", "");
        }

        // Since timestamp
        if let Some(ref ts) = filter.since {
            q = q.replace(
                "@SINCE@",
                &format!("FILTER(?created_at > {})", ts.as_u64()),
            );
            ckey.push_str(&format!("{}", ts.as_u64()));
        } else {
            q = q.replace("@SINCE@", "");
        }

        // Until timestamp
        if let Some(ref ts) = filter.until {
            q = q.replace(
                "@UNTIL@",
                &format!("FILTER(?created_at < {})", ts.as_u64()),
            );
            ckey.push_str(&format!("{}", ts.as_u64()));
        } else {
            q = q.replace("@UNTIL@", "");
        }

        (q, ckey)
    }
}

/// Returns the event channel priority for an event
pub fn event_chan_prio(event: &Event) -> i32 {
    match event.kind {
        // Notes
        Kind::TextNote | Kind::LongFormTextNote => 100,
        // Reposts
        Kind::Repost | Kind::GenericRepost => 50,
        // Metadata
        Kind::Metadata => 220,
        // Relay lists
        Kind::RelayList | Kind::InboxRelays => 200,
        Kind::ContactList => 200,
        Kind::Reaction => 10,
        // Follow packs
        Kind::Custom(39089) => 90,
        _ => 0,
    }
}

impl NostrDatabase for RdfEventsStore {
    /// Custom backend type: RDF
    fn backend(&self) -> Backend {
        Backend::Custom("RDF".to_string())
    }
}

impl NostrEventsDatabase for RdfEventsStore {
    fn save_event<'a>(
        &'a self,
        event: &'a Event,
    ) -> BoxedFuture<'a, Result<SaveEventStatus, DatabaseError>> {
        Box::pin(async move {
            // Restrict by event kind to prevent bloating the store
            match event.kind {
                Kind::TextNote
                | Kind::LongFormTextNote
                | Kind::ContactList
                | Kind::Repost
                | Kind::Metadata
                | Kind::Reaction
                | Kind::ZapReceipt
                | Kind::Custom(39089)
                | Kind::Custom(7101)
                | Kind::Custom(7102)
                | Kind::Custom(7103) => {
                    match self.database_save_mode {
                        // Queue mode: queue the event for processing by the threadpool
                        DatabaseEventsSaveMode::Queue => {
                            self.process_event(
                                event.clone(),
                                Some(event_chan_prio(&event)),
                            );
                            Ok(SaveEventStatus::Success)
                        }
                        // Direct mode: store it straight away
                        DatabaseEventsSaveMode::Direct => {
                            match self.insert_event(&event) {
                                Ok(_) => Ok(SaveEventStatus::Success),
                                Err(_) => Ok(SaveEventStatus::Rejected(
                                    RejectedReason::Other,
                                )),
                            }
                        }
                    }
                }
                _ => Ok(SaveEventStatus::Rejected(RejectedReason::Other)),
            }
        })
    }

    fn check_id<'a>(
        &'a self,
        event_id: &'a EventId,
    ) -> BoxedFuture<'a, Result<DatabaseEventStatus, DatabaseError>> {
        Box::pin(async move {
            // Turn event id to a NamedNode
            let enn = event_id.named_node().map_err(|_| {
                DatabaseError::Backend(Box::from("Cannot parse event id"))
            })?;

            // Check if there are quads for this event id
            if self
                .store
                .quads_for_pattern(Some((&enn).into()), None, None, None)
                .count()
                > 0
            {
                return Ok(DatabaseEventStatus::Saved);
            } else {
                return Ok(DatabaseEventStatus::NotExistent);
            }
        })
    }

    fn has_coordinate_been_deleted<'a>(
        &'a self,
        _coordinate: &'a CoordinateBorrow<'a>,
        _timestamp: &'a Timestamp,
    ) -> BoxedFuture<'a, Result<bool, DatabaseError>> {
        Box::pin(async move { Ok(false) })
    }

    /// Get an event by its id
    fn event_by_id<'a>(
        &'a self,
        event_id: &'a EventId,
    ) -> BoxedFuture<'a, Result<Option<Event>, DatabaseError>> {
        Box::pin(async move {
            Ok(self.query(Filter::new().id(*event_id)).await?.first_owned())
        })
    }

    /// Count
    fn count(
        &self,
        filter: Filter,
    ) -> BoxedFuture<Result<usize, DatabaseError>> {
        Box::pin(async move { Ok(self.query(filter).await?.len()) })
    }

    /// Query
    fn query(
        &self,
        filter: Filter,
    ) -> BoxedFuture<Result<Events, DatabaseError>> {
        Box::pin(async move {
            let mut events: Events = Events::new(&filter);
            let (q, ckey) = self.sparqlify_filter(&filter);

            if let Ok(set) = self.run_query(&q, [], Some(ckey)) {
                for row in &set.rows {
                    events.insert(Event::new(
                        row.get(SPVars::EVENT_ID)
                            .unwrap()
                            .value
                            .to_event_id()
                            .map_err(|_| {
                                DatabaseError::Backend(Box::from(
                                    "Invalid event ID",
                                ))
                            })?,
                        row.get(SPVars::PUBK)
                            .unwrap()
                            .value
                            .to_public_key()
                            .map_err(|_| {
                                DatabaseError::Backend(Box::from(
                                    "Invalid pubk",
                                ))
                            })?,
                        row.get(SPVars::CREATED_AT)
                            .unwrap()
                            .try_into()
                            .map_err(|_| {
                                DatabaseError::Backend(Box::from("Invalid TS"))
                            })?,
                        Kind::from_str(
                            &row.get(SPVars::KIND).unwrap().to_string(),
                        )
                        .unwrap(),
                        vec![], // wrong
                        row.get(SPVars::CONTENT).unwrap().to_string(),
                        Signature::from_str(
                            &row.get(SPVars::SIG).unwrap().to_string(),
                        )
                        .map_err(|_| {
                            DatabaseError::Backend(Box::from(
                                "Invalid signature",
                            ))
                        })?,
                    ));

                    sleep(Duration::from_millis(10)).await;
                }
            }
            Ok(events)
        })
    }

    /// Return events matching this filter for the negentropy reconciliation
    fn negentropy_items(
        &self,
        filter: Filter,
    ) -> BoxedFuture<Result<Vec<(EventId, Timestamp)>, DatabaseError>> {
        Box::pin(async move {
            let (q, ckey) = self.sparqlify_filter(&filter);

            if let Ok(results) = self.run_query(&q, [], Some(ckey)) {
                Ok(results
                    .rows
                    .iter()
                    .filter_map(|r| {
                        let Ok(event_id) = r
                            .get(SPVars::EVENT_ID)
                            .unwrap()
                            .value
                            .to_event_id()
                        else {
                            return None;
                        };

                        let Ok(ts) =
                            r.get(SPVars::CREATED_AT).unwrap().try_into()
                        else {
                            return None;
                        };

                        Some((event_id, ts))
                    })
                    .collect())
            } else {
                Err(DatabaseError::Backend(Box::from(
                    "Error running SparQL query",
                )))
            }
        })
    }

    /// Delete: not supported yet
    fn delete(
        &self,
        _filter: Filter,
    ) -> BoxedFuture<Result<(), DatabaseError>> {
        Box::pin(async move { Err(DatabaseError::NotSupported) })
    }
}

impl NostrDatabaseWipe for RdfEventsStore {
    #[inline]
    /// wipe: not supported yet
    fn wipe(&self) -> BoxedFuture<Result<(), DatabaseError>> {
        Box::pin(async move { Err(DatabaseError::NotSupported) })
    }
}