[WIP] Query history from history server

sable_history/Cargo.toml

@@ -11,6 +11,7 @@ built = { version = "0.5", features = [ "git2" ] }
 sable_network = { path = "../sable_network" }
 sable_server = { path = "../sable_server" }
 
+futures = "0.3"
 tokio = { version = "1.14", features = [ "full" ] }
 serde = { version = "1", features = [ "derive" ] }
 serde_with = "1.11"
@@ -25,4 +26,4 @@ chrono = "0.4"
 uuid = { version = "1.9.1", features = ["v7", "fast-rng", "serde"] }
 
 diesel = { version = "2.2", features = [ "postgres", "chrono", "uuid" ] }
-diesel-async = { version = "0.5", features = [ "postgres" ] }
+diesel-async = { version = "0.5", features = [ "postgres" ] }

sable_history/src/lib.rs

@@ -1,3 +1,5 @@
+mod pg_history_service;
+pub use pg_history_service::PgHistoryService;
 mod server;
 pub use server::*;
 

sable_history/src/pg_history_service.rs

@@ -0,0 +1,85 @@
+use std::borrow::BorrowMut;
+use std::collections::HashMap;
+
+use diesel::dsl::sql;
+use diesel::prelude::*;
+use diesel_async::{AsyncConnection, AsyncPgConnection};
+use futures::stream::StreamExt;
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use sable_network::prelude::*;
+
+use crate::schema::channels;
+
+/// Implementation of [`HistoryService`] backed PostgreSQL
+pub struct PgHistoryService<'a> {
+    database_connection: &'a Mutex<AsyncPgConnection>,
+}
+
+impl<'a> PgHistoryService<'a> {
+    pub fn new(database_connection: &'a Mutex<AsyncPgConnection>) -> Self {
+        Self {
+            database_connection,
+        }
+    }
+}
+
+impl<'a> HistoryService for PgHistoryService<'a> {
+    async fn list_targets(
+        &self,
+        user: UserId,
+        after_ts: Option<i64>,
+        before_ts: Option<i64>,
+        limit: Option<usize>,
+    ) -> HashMap<TargetId, i64> {
+        // TODO: access control
+        match diesel_async::RunQueryDsl::load_stream(
+            channels::dsl::channels.select((
+                channels::dsl::id,
+                sql::<diesel::pg::sql_types::Uuid>(
+                    "SELECT MAX(id) FROM messages WHERE target_channel=channels.id",
+                ),
+            )),
+            &mut *self.database_connection.lock().await,
+        )
+        .await
+        {
+            Err(e) => {
+                tracing::error!("Could not get history channels: {e}");
+                return HashMap::new();
+            }
+            Ok(rows) => {
+                rows.map(|row| row.expect("Could not deserialize row"))
+                    .map(
+                        |(channel_id, max_message_id): (i64, Uuid)| -> (TargetId, i64) {
+                            let (seconds, _) = max_message_id
+                                .get_timestamp()
+                                .expect("messages.id is not a UUID7")
+                                .to_unix();
+                            (
+                                TargetId::Channel(ChannelId::from(Snowflake::from(
+                                    u64::try_from(channel_id).expect("channel id is negative"),
+                                ))),
+                                seconds
+                                    .try_into()
+                                    .expect("message's UNIX timestamp is negative"),
+                            )
+                        },
+                    )
+                    .collect()
+                    .await
+            }
+        }
+    }
+
+    async fn get_entries(
+        &self,
+        user: UserId,
+        target: TargetId,
+        request: HistoryRequest,
+    ) -> Result<impl IntoIterator<Item = HistoryLogEntry>, HistoryError> {
+        todo!();
+        Ok(vec![])
+    }
+}

sable_history/src/server/mod.rs

@@ -5,11 +5,13 @@ use sable_network::prelude::*;
 use sable_server::ServerType;
 use serde::Deserialize;
 use tokio::sync::{mpsc::UnboundedReceiver, Mutex};
+use tracing::instrument;
 
 use std::sync::Arc;
 
 use diesel_async::{AsyncConnection, AsyncPgConnection};
 
+mod sync;
 mod update_handler;
 
 #[derive(Debug, Clone, Deserialize)]
@@ -20,7 +22,7 @@ pub struct HistoryServerConfig {
 pub struct HistoryServer {
     node: Arc<NetworkNode>,
     history_receiver: Mutex<UnboundedReceiver<sable_network::rpc::NetworkHistoryUpdate>>,
-    database_connection: Mutex<AsyncPgConnection>,
+    database_connection: Mutex<AsyncPgConnection>, // TODO: use a connection pool
 }
 
 impl ServerType for HistoryServer {
@@ -69,6 +71,14 @@ impl ServerType for HistoryServer {
                 {
                     let Some(update) = update else { break; };
 
+                    if let NetworkStateChange::NewServer(new_server) = &update.change
+                    {
+                        if new_server.server == self.node.id()
+                        {
+                            self.burst_to_network().await;
+                        }
+                    }
+
                     if let Err(error) = self.handle_history_update(update).await {
                         tracing::error!(?error, "Error return handling history update");
                     }
@@ -94,10 +104,56 @@ impl ServerType for HistoryServer {
         unimplemented!("history servers can't hot-upgrade");
     }
 
-    fn handle_remote_command(
+    #[instrument(skip_all)]
+    async fn handle_remote_command(
         &self,
-        _request: sable_network::rpc::RemoteServerRequestType,
+        req: sable_network::rpc::RemoteServerRequestType,
     ) -> sable_network::rpc::RemoteServerResponse {
-        todo!()
+        tracing::debug!(?req, "Got remote request");
+
+        use crate::server::rpc::RemoteServerRequestType::*;
+        use sable_network::rpc::RemoteServerResponse;
+
+        match req {
+            History(req) => {
+                use crate::server::rpc::RemoteHistoryServerRequestType::*;
+                use crate::server::rpc::RemoteHistoryServerResponse::*;
+
+                let history_service = crate::PgHistoryService::new(&self.database_connection);
+                match req {
+                    ListTargets {
+                        user,
+                        after_ts,
+                        before_ts,
+                        limit,
+                    } => TargetList(
+                        history_service
+                            .list_targets(user, after_ts, before_ts, limit)
+                            .await
+                            .into_iter()
+                            .collect(),
+                    ),
+                    GetEntries {
+                        user,
+                        target,
+                        request,
+                    } => Entries(
+                        history_service
+                            .get_entries(user, target, request)
+                            .await
+                            .map(|entries| entries.into_iter().collect()),
+                    ),
+                }
+                .into()
+            }
+            Services(_) => {
+                tracing::warn!(?req, "Got unsupported request (services)");
+                RemoteServerResponse::NotSupported
+            }
+            Ping => {
+                tracing::warn!(?req, "Got unsupported request (ping)");
+                RemoteServerResponse::NotSupported
+            }
+        }
     }
 }

sable_history/src/server/sync.rs

@@ -0,0 +1,12 @@
+use sable_network::network::wrapper::ObjectWrapper;
+
+use super::*;
+use crate::server::event::IntroduceHistoryServer;
+
+impl HistoryServer {
+    pub(super) async fn burst_to_network(&self) {
+        // Set ourselves as the active history node
+        self.node
+            .submit_event(self.node.id(), IntroduceHistoryServer {});
+    }
+}

sable_history/src/server/update_handler.rs

@@ -34,6 +34,7 @@ impl HistoryServer {
             | NetworkStateChange::ServerQuit(_)
             | NetworkStateChange::NewAuditLogEntry(_)
             | NetworkStateChange::UserLoginChange(_)
+            | NetworkStateChange::HistoryServerUpdate(_)
             | NetworkStateChange::ServicesUpdate(_)
             | NetworkStateChange::EventComplete(_) => Ok(()),
         }

sable_ircd/src/capability/account_tag.rs

@@ -27,6 +27,7 @@ fn account_for_tag(update: &NetworkStateChange, net: &Network) -> Option<String>
         NetworkStateChange::NewServer(_) => None,
         NetworkStateChange::ServerQuit(_) => None,
         NetworkStateChange::NewAuditLogEntry(_) => None,
+        NetworkStateChange::HistoryServerUpdate(_) => None,
         NetworkStateChange::ServicesUpdate(_) => None,
         NetworkStateChange::EventComplete(_) => None,
     }?;

sable_ircd/src/command/handlers/chathistory.rs

@@ -136,7 +136,7 @@ async fn handle_chathistory(
                 }
             };
 
-            let history_service = LocalHistoryService::new(server.node());
+            let history_service = server.node().history_service();
             match history_service
                 .get_entries(source.id(), target_id, request)
                 .await
@@ -160,7 +160,7 @@ async fn list_targets<'a>(
     to_ts: Option<i64>,
     limit: Option<usize>,
 ) {
-    let history_service = LocalHistoryService::new(server.node());
+    let history_service = server.node().history_service();
 
     let found_targets = history_service
         .list_targets(source.id(), to_ts, from_ts, limit)

sable_ircd/src/command/handlers/register.rs

@@ -35,7 +35,7 @@ async fn do_register_user(
     _email: &str,
     password: &str,
 ) -> CommandResult {
-    let Some(services_name) = network.current_services_name() else {
+    let Some(services_name) = network.current_services_server_name() else {
         response_to.send(message::Fail::new(
             "REGISTER",
             "TEMPORARILY_UNAVAILABLE",

sable_ircd/src/command/plumbing/argument_wrappers.rs

@@ -18,7 +18,7 @@ impl<'a> AmbientArgument<'a> for ServicesTarget<'a> {
         Ok(Self {
             name: ctx
                 .network()
-                .current_services_name()
+                .current_services_server_name()
                 .ok_or(CommandError::ServicesNotAvailable)?,
             server: ctx.server(),
         })

sable_ircd/src/messages/send_history.rs

@@ -50,6 +50,7 @@ impl SendHistoryItem<NetworkHistoryUpdate> for ClientServer {
             | NetworkStateChange::ServerQuit(_)
             | NetworkStateChange::NewAuditLogEntry(_)
             | NetworkStateChange::UserLoginChange(_)
+            | NetworkStateChange::HistoryServerUpdate(_)
             | NetworkStateChange::ServicesUpdate(_)
             | NetworkStateChange::EventComplete(_) => Ok(()),
         }
@@ -86,6 +87,7 @@ impl SendHistoryItem<HistoryLogEntry> for ClientServer {
             | NetworkStateChange::ServerQuit(_)
             | NetworkStateChange::NewAuditLogEntry(_)
             | NetworkStateChange::UserLoginChange(_)
+            | NetworkStateChange::HistoryServerUpdate(_)
             | NetworkStateChange::ServicesUpdate(_)
             | NetworkStateChange::EventComplete(_) => Ok(()),
         }

sable_ircd/src/server/server_type.rs

@@ -1,9 +1,11 @@
-use super::*;
-use crate::connection_collection::ConnectionCollectionState;
 use anyhow::Context;
+use tracing::instrument;
+
 use client_listener::SavedListenerCollection;
 use sable_server::ServerSaveError;
 
+use super::*;
+use crate::connection_collection::ConnectionCollectionState;
 use crate::monitor::MonitorSet;
 
 /// Saved state of a [`ClientServer`] for later resumption
@@ -168,7 +170,8 @@ impl sable_server::ServerType for ClientServer {
         }
     }
 
-    fn handle_remote_command(&self, cmd: RemoteServerRequestType) -> RemoteServerResponse {
+    #[instrument(skip_all)]
+    async fn handle_remote_command(&self, cmd: RemoteServerRequestType) -> RemoteServerResponse {
         match cmd {
             RemoteServerRequestType::Ping => RemoteServerResponse::Success,
             _ => RemoteServerResponse::NotSupported,

sable_network/Cargo.toml

@@ -21,6 +21,7 @@ pretty_assertions = "1.4"
 [dependencies]
 sable_macros = { path = "../sable_macros" }
 
+async-trait = "0.1.83"
 tracing = "0.1"
 thiserror = "1"
 serde_json = "1"
@@ -42,7 +43,7 @@ rand = "0.8"
 arrayvec = { version = "0.7", features = [ "serde" ] }
 hashers = "1"
 serde_with = "1.11"
-parking_lot = { version = "0.12", features = [ "serde" ] }
+parking_lot = { version = "0.12.2", features = [ "serde", "arc_lock" ] }
 wildmatch = "2.1"
 concurrent_log = { version = "0.2.4", features = [ "serde" ] }
 ipnet = { version = "2", features = [ "serde" ] }