mirror of
https://github.com/scsibug/nostr-rs-relay.git
synced 2024-11-12 22:29:07 -05:00
543 lines
21 KiB
Rust
543 lines
21 KiB
Rust
//! Event persistence and querying
|
|
use crate::config::SETTINGS;
|
|
use crate::error::Error;
|
|
use crate::error::Result;
|
|
use crate::event::Event;
|
|
use crate::hexrange::hex_range;
|
|
use crate::hexrange::HexSearch;
|
|
use crate::nip05;
|
|
use crate::schema::{upgrade_db, STARTUP_SQL};
|
|
use crate::subscription::Subscription;
|
|
use crate::utils::is_hex;
|
|
use governor::clock::Clock;
|
|
use governor::{Quota, RateLimiter};
|
|
use hex;
|
|
use log::*;
|
|
use r2d2;
|
|
use r2d2_sqlite::SqliteConnectionManager;
|
|
use rusqlite::params;
|
|
use rusqlite::types::ToSql;
|
|
use rusqlite::Connection;
|
|
use rusqlite::OpenFlags;
|
|
use std::path::Path;
|
|
use std::thread;
|
|
use std::time::Duration;
|
|
use std::time::Instant;
|
|
use tokio::task;
|
|
|
|
pub type SqlitePool = r2d2::Pool<r2d2_sqlite::SqliteConnectionManager>;
|
|
pub type PooledConnection = r2d2::PooledConnection<r2d2_sqlite::SqliteConnectionManager>;
|
|
|
|
/// Events submitted from a client, with a return channel for notices
|
|
pub struct SubmittedEvent {
|
|
pub event: Event,
|
|
pub notice_tx: tokio::sync::mpsc::Sender<String>,
|
|
}
|
|
|
|
/// Database file
|
|
pub const DB_FILE: &str = "nostr.db";
|
|
|
|
/// Build a database connection pool.
|
|
pub fn build_pool(
|
|
name: &str,
|
|
flags: OpenFlags,
|
|
min_size: u32,
|
|
max_size: u32,
|
|
wait_for_db: bool,
|
|
) -> SqlitePool {
|
|
let settings = SETTINGS.read().unwrap();
|
|
|
|
let db_dir = &settings.database.data_directory;
|
|
let full_path = Path::new(db_dir).join(DB_FILE);
|
|
// small hack; if the database doesn't exist yet, that means the
|
|
// writer thread hasn't finished. Give it a chance to work. This
|
|
// is only an issue with the first time we run.
|
|
while !full_path.exists() && wait_for_db {
|
|
debug!("Database reader pool is waiting on the database to be created...");
|
|
thread::sleep(Duration::from_millis(500));
|
|
}
|
|
let manager = SqliteConnectionManager::file(&full_path)
|
|
.with_flags(flags)
|
|
.with_init(|c| c.execute_batch(STARTUP_SQL));
|
|
let pool: SqlitePool = r2d2::Pool::builder()
|
|
.test_on_check_out(true) // no noticeable performance hit
|
|
.min_idle(Some(min_size))
|
|
.max_size(max_size)
|
|
.build(manager)
|
|
.unwrap();
|
|
info!(
|
|
"Built a connection pool {:?} (min={}, max={})",
|
|
name, min_size, max_size
|
|
);
|
|
pool
|
|
}
|
|
|
|
/// Build a single database connection, with provided flags
|
|
pub fn build_conn(flags: OpenFlags) -> Result<Connection> {
|
|
let settings = SETTINGS.read().unwrap();
|
|
let db_dir = &settings.database.data_directory;
|
|
let full_path = Path::new(db_dir).join(DB_FILE);
|
|
// create a connection
|
|
Ok(Connection::open_with_flags(&full_path, flags)?)
|
|
}
|
|
|
|
/// Spawn a database writer that persists events to the SQLite store.
|
|
pub async fn db_writer(
|
|
mut event_rx: tokio::sync::mpsc::Receiver<SubmittedEvent>,
|
|
bcast_tx: tokio::sync::broadcast::Sender<Event>,
|
|
metadata_tx: tokio::sync::broadcast::Sender<Event>,
|
|
mut shutdown: tokio::sync::broadcast::Receiver<()>,
|
|
) -> tokio::task::JoinHandle<Result<()>> {
|
|
let settings = SETTINGS.read().unwrap();
|
|
|
|
// are we performing NIP-05 checking?
|
|
let nip05_active = settings.verified_users.is_active();
|
|
// are we requriing NIP-05 user verification?
|
|
let nip05_enabled = settings.verified_users.is_enabled();
|
|
|
|
task::spawn_blocking(move || {
|
|
// get database configuration settings
|
|
let settings = SETTINGS.read().unwrap();
|
|
let db_dir = &settings.database.data_directory;
|
|
let full_path = Path::new(db_dir).join(DB_FILE);
|
|
// create a connection pool
|
|
let pool = build_pool(
|
|
"event writer",
|
|
OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE,
|
|
1,
|
|
4,
|
|
false,
|
|
);
|
|
info!("opened database {:?} for writing", full_path);
|
|
upgrade_db(&mut pool.get()?)?;
|
|
|
|
// Make a copy of the whitelist
|
|
let whitelist = &settings.authorization.pubkey_whitelist.clone();
|
|
|
|
// get rate limit settings
|
|
let rps_setting = settings.limits.messages_per_sec;
|
|
let mut most_recent_rate_limit = Instant::now();
|
|
let mut lim_opt = None;
|
|
let clock = governor::clock::QuantaClock::default();
|
|
if let Some(rps) = rps_setting {
|
|
if rps > 0 {
|
|
info!("Enabling rate limits for event creation ({}/sec)", rps);
|
|
let quota = core::num::NonZeroU32::new(rps * 60).unwrap();
|
|
lim_opt = Some(RateLimiter::direct(Quota::per_minute(quota)));
|
|
}
|
|
}
|
|
loop {
|
|
if shutdown.try_recv().is_ok() {
|
|
info!("shutting down database writer");
|
|
break;
|
|
}
|
|
// call blocking read on channel
|
|
let next_event = event_rx.blocking_recv();
|
|
// if the channel has closed, we will never get work
|
|
if next_event.is_none() {
|
|
break;
|
|
}
|
|
let mut event_write = false;
|
|
let subm_event = next_event.unwrap();
|
|
let event = subm_event.event;
|
|
let notice_tx = subm_event.notice_tx;
|
|
// check if this event is authorized.
|
|
if let Some(allowed_addrs) = whitelist {
|
|
// if the event address is not in allowed_addrs.
|
|
if !allowed_addrs.contains(&event.pubkey) {
|
|
info!(
|
|
"Rejecting event {}, unauthorized author",
|
|
event.get_event_id_prefix()
|
|
);
|
|
notice_tx
|
|
.try_send("pubkey is not allowed to publish to this relay".to_owned())
|
|
.ok();
|
|
continue;
|
|
}
|
|
}
|
|
|
|
// send any metadata events to the NIP-05 verifier
|
|
if nip05_active && event.is_kind_metadata() {
|
|
// we are sending this prior to even deciding if we
|
|
// persist it. this allows the nip05 module to
|
|
// inspect it, update if necessary, or persist a new
|
|
// event and broadcast it itself.
|
|
metadata_tx.send(event.clone()).ok();
|
|
}
|
|
|
|
// check for NIP-05 verification
|
|
if nip05_enabled {
|
|
match nip05::query_latest_user_verification(pool.get()?, event.pubkey.to_owned()) {
|
|
Ok(uv) => {
|
|
if uv.is_valid() {
|
|
info!(
|
|
"new event from verified author ({:?},{:?})",
|
|
uv.name.to_string(),
|
|
event.get_author_prefix()
|
|
);
|
|
} else {
|
|
info!("rejecting event, author ({:?} / {:?}) verification invalid (expired/wrong domain)",
|
|
uv.name.to_string(),
|
|
event.get_author_prefix()
|
|
);
|
|
notice_tx
|
|
.try_send(
|
|
"NIP-05 verification is no longer valid (expired/wrong domain)"
|
|
.to_owned(),
|
|
)
|
|
.ok();
|
|
continue;
|
|
}
|
|
}
|
|
Err(Error::SqlError(rusqlite::Error::QueryReturnedNoRows)) => {
|
|
debug!(
|
|
"no verification records found for pubkey: {:?}",
|
|
event.get_author_prefix()
|
|
);
|
|
notice_tx
|
|
.try_send("NIP-05 verification needed to publish events".to_owned())
|
|
.ok();
|
|
continue;
|
|
}
|
|
Err(e) => {
|
|
warn!("checking nip05 verification status failed: {:?}", e);
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
// TODO: cache recent list of authors to remove a DB call.
|
|
let start = Instant::now();
|
|
match write_event(&mut pool.get()?, &event) {
|
|
Ok(updated) => {
|
|
if updated == 0 {
|
|
trace!("ignoring duplicate event");
|
|
} else {
|
|
info!(
|
|
"persisted event {:?} from {:?} in {:?}",
|
|
event.get_event_id_prefix(),
|
|
event.get_author_prefix(),
|
|
start.elapsed()
|
|
);
|
|
event_write = true;
|
|
// send this out to all clients
|
|
bcast_tx.send(event.clone()).ok();
|
|
}
|
|
}
|
|
Err(err) => {
|
|
warn!("event insert failed: {:?}", err);
|
|
notice_tx
|
|
.try_send(
|
|
"relay experienced an error trying to publish the latest event"
|
|
.to_owned(),
|
|
)
|
|
.ok();
|
|
}
|
|
}
|
|
|
|
// use rate limit, if defined, and if an event was actually written.
|
|
if event_write {
|
|
if let Some(ref lim) = lim_opt {
|
|
if let Err(n) = lim.check() {
|
|
let wait_for = n.wait_time_from(clock.now());
|
|
// check if we have recently logged rate
|
|
// limits, but print out a message only once
|
|
// per second.
|
|
if most_recent_rate_limit.elapsed().as_secs() > 10 {
|
|
warn!(
|
|
"rate limit reached for event creation (sleep for {:?}) (suppressing future messages for 10 seconds)",
|
|
wait_for
|
|
);
|
|
// reset last rate limit message
|
|
most_recent_rate_limit = Instant::now();
|
|
}
|
|
// block event writes, allowing them to queue up
|
|
thread::sleep(wait_for);
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
info!("database connection closed");
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
/// Persist an event to the database, returning rows added.
|
|
pub fn write_event(conn: &mut PooledConnection, e: &Event) -> Result<usize> {
|
|
// start transaction
|
|
let tx = conn.transaction()?;
|
|
// get relevant fields from event and convert to blobs.
|
|
let id_blob = hex::decode(&e.id).ok();
|
|
let pubkey_blob = hex::decode(&e.pubkey).ok();
|
|
let event_str = serde_json::to_string(&e).ok();
|
|
// ignore if the event hash is a duplicate.
|
|
let ins_count = tx.execute(
|
|
"INSERT OR IGNORE INTO event (event_hash, created_at, kind, author, content, first_seen, hidden) VALUES (?1, ?2, ?3, ?4, ?5, strftime('%s','now'), FALSE);",
|
|
params![id_blob, e.created_at, e.kind, pubkey_blob, event_str]
|
|
)?;
|
|
if ins_count == 0 {
|
|
// if the event was a duplicate, no need to insert event or
|
|
// pubkey references.
|
|
return Ok(ins_count);
|
|
}
|
|
// remember primary key of the event most recently inserted.
|
|
let ev_id = tx.last_insert_rowid();
|
|
// add all tags to the tag table
|
|
for tag in e.tags.iter() {
|
|
// ensure we have 2 values.
|
|
if tag.len() >= 2 {
|
|
let tagname = &tag[0];
|
|
let tagval = &tag[1];
|
|
// if tagvalue is hex;
|
|
if is_hex(tagval) {
|
|
tx.execute(
|
|
"INSERT OR IGNORE INTO tag (event_id, name, value_hex) VALUES (?1, ?2, ?3)",
|
|
params![ev_id, &tagname, hex::decode(&tagval).ok()],
|
|
)?;
|
|
} else {
|
|
tx.execute(
|
|
"INSERT OR IGNORE INTO tag (event_id, name, value) VALUES (?1, ?2, ?3)",
|
|
params![ev_id, &tagname, &tagval],
|
|
)?;
|
|
}
|
|
}
|
|
}
|
|
// if this event is for a metadata update, hide every other kind=0
|
|
// event from the same author that was issued earlier than this.
|
|
if e.kind == 0 {
|
|
let update_count = tx.execute(
|
|
"UPDATE event SET hidden=TRUE WHERE id!=? AND kind=0 AND author=? AND created_at <= ? and hidden!=TRUE",
|
|
params![ev_id, hex::decode(&e.pubkey).ok(), e.created_at],
|
|
)?;
|
|
if update_count > 0 {
|
|
info!(
|
|
"hid {} older metadata events for author {:?}",
|
|
update_count,
|
|
e.get_author_prefix()
|
|
);
|
|
}
|
|
}
|
|
// if this event is for a contact update, hide every other kind=3
|
|
// event from the same author that was issued earlier than this.
|
|
if e.kind == 3 {
|
|
let update_count = tx.execute(
|
|
"UPDATE event SET hidden=TRUE WHERE id!=? AND kind=3 AND author=? AND created_at <= ? and hidden!=TRUE",
|
|
params![ev_id, hex::decode(&e.pubkey).ok(), e.created_at],
|
|
)?;
|
|
if update_count > 0 {
|
|
info!(
|
|
"hid {} older contact events for author {:?}",
|
|
update_count,
|
|
e.get_author_prefix()
|
|
);
|
|
}
|
|
}
|
|
tx.commit()?;
|
|
Ok(ins_count)
|
|
}
|
|
|
|
/// Serialized event associated with a specific subscription request.
|
|
#[derive(PartialEq, Debug, Clone)]
|
|
pub struct QueryResult {
|
|
/// Subscription identifier
|
|
pub sub_id: String,
|
|
/// Serialized event
|
|
pub event: String,
|
|
}
|
|
|
|
/// Produce a arbitrary list of '?' parameters.
|
|
fn repeat_vars(count: usize) -> String {
|
|
if count == 0 {
|
|
return "".to_owned();
|
|
}
|
|
let mut s = "?,".repeat(count);
|
|
// Remove trailing comma
|
|
s.pop();
|
|
s
|
|
}
|
|
|
|
/// Create a dynamic SQL query string and params from a subscription.
|
|
fn query_from_sub(sub: &Subscription) -> (String, Vec<Box<dyn ToSql>>) {
|
|
// build a dynamic SQL query. all user-input is either an integer
|
|
// (sqli-safe), or a string that is filtered to only contain
|
|
// hexadecimal characters. Strings that require escaping (tag
|
|
// names/values) use parameters.
|
|
let mut query =
|
|
"SELECT DISTINCT(e.content) FROM event e LEFT JOIN tag t ON e.id=t.event_id ".to_owned();
|
|
// parameters
|
|
let mut params: Vec<Box<dyn ToSql>> = vec![];
|
|
|
|
// for every filter in the subscription, generate a where clause
|
|
let mut filter_clauses: Vec<String> = Vec::new();
|
|
for f in sub.filters.iter() {
|
|
// individual filter components
|
|
let mut filter_components: Vec<String> = Vec::new();
|
|
// Query for "authors", allowing prefix matches
|
|
if let Some(authvec) = &f.authors {
|
|
// take each author and convert to a hexsearch
|
|
let mut auth_searches: Vec<String> = vec![];
|
|
for auth in authvec {
|
|
match hex_range(auth) {
|
|
Some(HexSearch::Exact(ex)) => {
|
|
auth_searches.push("author=?".to_owned());
|
|
params.push(Box::new(ex));
|
|
}
|
|
Some(HexSearch::Range(lower, upper)) => {
|
|
auth_searches.push("(author>? AND author<?)".to_owned());
|
|
params.push(Box::new(lower));
|
|
params.push(Box::new(upper));
|
|
}
|
|
Some(HexSearch::LowerOnly(lower)) => {
|
|
auth_searches.push("author>?".to_owned());
|
|
params.push(Box::new(lower));
|
|
}
|
|
None => {
|
|
info!("Could not parse hex range from author {:?}", auth);
|
|
}
|
|
}
|
|
}
|
|
let authors_clause = format!("({})", auth_searches.join(" OR "));
|
|
filter_components.push(authors_clause);
|
|
}
|
|
// Query for Kind
|
|
if let Some(ks) = &f.kinds {
|
|
// kind is number, no escaping needed
|
|
let str_kinds: Vec<String> = ks.iter().map(|x| x.to_string()).collect();
|
|
let kind_clause = format!("kind IN ({})", str_kinds.join(", "));
|
|
filter_components.push(kind_clause);
|
|
}
|
|
// Query for event, allowing prefix matches
|
|
if let Some(idvec) = &f.ids {
|
|
// take each author and convert to a hexsearch
|
|
let mut id_searches: Vec<String> = vec![];
|
|
for id in idvec {
|
|
match hex_range(id) {
|
|
Some(HexSearch::Exact(ex)) => {
|
|
id_searches.push("event_hash=?".to_owned());
|
|
params.push(Box::new(ex));
|
|
}
|
|
Some(HexSearch::Range(lower, upper)) => {
|
|
id_searches.push("(event_hash>? AND event_hash<?)".to_owned());
|
|
params.push(Box::new(lower));
|
|
params.push(Box::new(upper));
|
|
}
|
|
Some(HexSearch::LowerOnly(lower)) => {
|
|
id_searches.push("event_hash>?".to_owned());
|
|
params.push(Box::new(lower));
|
|
}
|
|
None => {
|
|
info!("Could not parse hex range from id {:?}", id);
|
|
}
|
|
}
|
|
}
|
|
let id_clause = format!("({})", id_searches.join(" OR "));
|
|
filter_components.push(id_clause);
|
|
}
|
|
// Query for tags
|
|
if let Some(map) = &f.tags {
|
|
for (key, val) in map.iter() {
|
|
let mut str_vals: Vec<Box<dyn ToSql>> = vec![];
|
|
let mut blob_vals: Vec<Box<dyn ToSql>> = vec![];
|
|
for v in val {
|
|
if is_hex(v) {
|
|
if let Ok(h) = hex::decode(&v) {
|
|
blob_vals.push(Box::new(h));
|
|
}
|
|
} else {
|
|
str_vals.push(Box::new(v.to_owned()));
|
|
}
|
|
}
|
|
// create clauses with "?" params for each tag value being searched
|
|
let str_clause = format!("value IN ({})", repeat_vars(str_vals.len()));
|
|
let blob_clause = format!("value_hex IN ({})", repeat_vars(blob_vals.len()));
|
|
let tag_clause = format!("(name=? AND ({} OR {}))", str_clause, blob_clause);
|
|
// add the tag name as the first parameter
|
|
params.push(Box::new(key.to_owned()));
|
|
// add all tag values that are plain strings as params
|
|
params.append(&mut str_vals);
|
|
// add all tag values that are blobs as params
|
|
params.append(&mut blob_vals);
|
|
filter_components.push(tag_clause);
|
|
}
|
|
}
|
|
// Query for timestamp
|
|
if f.since.is_some() {
|
|
let created_clause = format!("created_at > {}", f.since.unwrap());
|
|
filter_components.push(created_clause);
|
|
}
|
|
// Query for timestamp
|
|
if f.until.is_some() {
|
|
let until_clause = format!("created_at < {}", f.until.unwrap());
|
|
filter_components.push(until_clause);
|
|
}
|
|
|
|
// combine all clauses, and add to filter_clauses
|
|
if !filter_components.is_empty() {
|
|
let mut fc = "( ".to_owned();
|
|
fc.push_str(&filter_components.join(" AND "));
|
|
fc.push_str(" )");
|
|
filter_clauses.push(fc);
|
|
}
|
|
}
|
|
|
|
// never display hidden events
|
|
query.push_str(" WHERE hidden!=TRUE ");
|
|
|
|
// combine all filters with OR clauses, if any exist
|
|
if !filter_clauses.is_empty() {
|
|
query.push_str(" AND (");
|
|
query.push_str(&filter_clauses.join(" OR "));
|
|
query.push_str(") ");
|
|
}
|
|
// add order clause
|
|
query.push_str(" ORDER BY created_at ASC");
|
|
debug!("query string: {}", query);
|
|
(query, params)
|
|
}
|
|
|
|
/// Perform a database query using a subscription.
|
|
///
|
|
/// The [`Subscription`] is converted into a SQL query. Each result
|
|
/// is published on the `query_tx` channel as it is returned. If a
|
|
/// message becomes available on the `abandon_query_rx` channel, the
|
|
/// query is immediately aborted.
|
|
pub async fn db_query(
|
|
sub: Subscription,
|
|
conn: PooledConnection,
|
|
query_tx: tokio::sync::mpsc::Sender<QueryResult>,
|
|
mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>,
|
|
) {
|
|
task::spawn_blocking(move || {
|
|
debug!("going to query for: {:?}", sub);
|
|
let mut row_count: usize = 0;
|
|
let start = Instant::now();
|
|
// generate SQL query
|
|
let (q, p) = query_from_sub(&sub);
|
|
// execute the query. Don't cache, since queries vary so much.
|
|
let mut stmt = conn.prepare(&q)?;
|
|
let mut event_rows = stmt.query(rusqlite::params_from_iter(p))?;
|
|
while let Some(row) = event_rows.next()? {
|
|
// check if this is still active (we could do this every N rows)
|
|
if abandon_query_rx.try_recv().is_ok() {
|
|
debug!("query aborted");
|
|
return Ok(());
|
|
}
|
|
row_count += 1;
|
|
let event_json = row.get(0)?;
|
|
query_tx
|
|
.blocking_send(QueryResult {
|
|
sub_id: sub.get_id(),
|
|
event: event_json,
|
|
})
|
|
.ok();
|
|
}
|
|
debug!(
|
|
"query completed ({} rows) in {:?}",
|
|
row_count,
|
|
start.elapsed()
|
|
);
|
|
let ok: Result<()> = Ok(());
|
|
ok
|
|
});
|
|
}
|