From cc6a39303593cee9d7681002bbff7584ed99e010 Mon Sep 17 00:00:00 2001 From: Riley Apeldoorn Date: Mon, 13 Jun 2022 23:08:11 +0200 Subject: [PATCH] Database abstractions --- Cargo.lock | 403 ++++++++++++++++++++++++++++++++++++++++++++++- Cargo.toml | 10 +- src/main.rs | 444 +++++++++++++++++++++++++++++++++++++++++++++++----- 3 files changed, 813 insertions(+), 44 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 74aef71..fc15a1a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,17 @@ # It is not intended for manual editing. version = 3 +[[package]] +name = "ahash" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fcb51a0695d8f838b1ee009b3fbf66bda078cd64590202a864a8f3e8c4315c47" +dependencies = [ + "getrandom", + "once_cell", + "version_check", +] + [[package]] name = "async-trait" version = "0.1.56" @@ -13,6 +24,15 @@ dependencies = [ "syn", ] +[[package]] +name = "atoi" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "616896e05fc0e2649463a93a15183c6a16bf03413a7af88ef1285ddedfa9cda5" +dependencies = [ + "num-traits", +] + [[package]] name = "autocfg" version = "1.1.0" @@ -143,6 +163,41 @@ dependencies = [ "libc", ] +[[package]] +name = "crc" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "49fc9a695bca7f35f5f4c15cddc84415f66a74ea78eef08e90c5024f2b540e23" +dependencies = [ + "crc-catalog", +] + +[[package]] +name = "crc-catalog" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccaeedb56da03b09f598226e25e80088cb4cd25f316e6e4df7d695f0feeb1403" + +[[package]] +name = "crossbeam-queue" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f25d8400f4a7a5778f0e4e52384a48cbd9b5c495d110786187fc750075277a2" +dependencies = [ + "cfg-if", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf124c720b7686e3c2663cf54062ab0f68a88af2fb6a030e87e30bf721fcb38" +dependencies = [ + "cfg-if", + "lazy_static", +] + [[package]] name = "crypto-common" version = "0.1.3" @@ -161,8 +216,41 @@ checksum = "f2fb860ca6fafa5552fb6d0e816a69c8e49f0908bf524e30a90d97c85892d506" dependencies = [ "block-buffer", "crypto-common", + "subtle", ] +[[package]] +name = "dirs" +version = "4.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca3aa72a6f96ea37bbc5aa912f6788242832f75369bdfdadcb0e38423f100059" +dependencies = [ + "dirs-sys", +] + +[[package]] +name = "dirs-sys" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b1d1d91c932ef41c0f2663aa8b0ca0342d444d842c06914aa0a7e352d0bada6" +dependencies = [ + "libc", + "redox_users", + "winapi", +] + +[[package]] +name = "dotenv" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77c90badedccf4105eca100756a0b1289e191f6fcbdadd3cee1d2f614f97da8f" + +[[package]] +name = "either" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e78d4f1cc4ae33bbfc157ed5d5a5ef3bc29227303d595861deb238fcec4e9457" + [[package]] name = "encoding_rs" version = "0.8.31" @@ -172,6 +260,12 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "event-listener" +version = "2.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77f3309417938f28bf8228fcff79a4a37103981e3e186d2ccd19c74b38f4eb71" + [[package]] name = "fastrand" version = "1.7.0" @@ -254,6 +348,17 @@ dependencies = [ "futures-util", ] +[[package]] +name = "futures-intrusive" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "62007592ac46aa7c2b6416f7deb9a8a8f63a01e0f1d6e1787d5630170db2b63e" +dependencies = [ + "futures-core", + "lock_api", + "parking_lot 0.11.2", +] + [[package]] name = "futures-io" version = "0.3.21" @@ -346,6 +451,27 @@ name = "hashbrown" version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ab5ef0d4909ef3724cc8cce6ccc8572c5c817592e9285f5464f8e86f8bd3726e" +dependencies = [ + "ahash", +] + +[[package]] +name = "hashlink" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7249a3129cbc1ffccd74857f81464a323a152173cdb134e0fd81bc803b29facf" +dependencies = [ + "hashbrown", +] + +[[package]] +name = "heck" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2540771e65fc8cb83cd6e8a237f70c319bd5c29f78ed1084ba5d50eeac86f7f9" +dependencies = [ + "unicode-segmentation", +] [[package]] name = "hermit" @@ -356,7 +482,9 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sqlx", "tokio", + "url", ] [[package]] @@ -368,6 +496,30 @@ dependencies = [ "libc", ] +[[package]] +name = "hex" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" + +[[package]] +name = "hkdf" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "791a029f6b9fc27657f6f188ec6e5e43f6911f6f878e0dc5501396e09809d437" +dependencies = [ + "hmac", +] + +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest", +] + [[package]] name = "http" version = "0.2.8" @@ -481,6 +633,15 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "879d54834c8c76457ef4293a689b2a8c59b076067ad77b15efafbb05f92a592b" +[[package]] +name = "itertools" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9a9d19fa1e79b6215ff29b9d6880b706147f16e9b1dbb1e4e5947b5b02bc5e3" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.2" @@ -539,6 +700,15 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "73cbba799671b762df5a175adf59ce145165747bb891505c43d09aefbbf38beb" +[[package]] +name = "md-5" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "658646b21e0b72f7866c7038ab086d3d5e1cd6271f060fd37defb241949d0582" +dependencies = [ + "digest", +] + [[package]] name = "memchr" version = "2.5.0" @@ -551,6 +721,12 @@ version = "0.3.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2a60c7ce501c71e03a9c9c0d35b861413ae925bd979cc7a4e30d060069aaac8d" +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "mio" version = "0.8.3" @@ -581,6 +757,25 @@ dependencies = [ "tempfile", ] +[[package]] +name = "nom" +version = "7.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a8903e5a29a317527874d0402f867152a3d21c908bb0b933e416c65e301d4c36" +dependencies = [ + "memchr", + "minimal-lexical", +] + +[[package]] +name = "num-traits" +version = "0.2.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "578ede34cf02f8924ab9447f50c28075b4d3e5b269972345e7e0372b38c6cdcd" +dependencies = [ + "autocfg", +] + [[package]] name = "num_cpus" version = "1.13.1" @@ -642,6 +837,17 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "parking_lot" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d17b78036a60663b797adeaee46f5c9dfebb86948d1255007a1d6be0271ff99" +dependencies = [ + "instant", + "lock_api", + "parking_lot_core 0.8.5", +] + [[package]] name = "parking_lot" version = "0.12.1" @@ -649,7 +855,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f" dependencies = [ "lock_api", - "parking_lot_core", + "parking_lot_core 0.9.3", +] + +[[package]] +name = "parking_lot_core" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d76e8e1493bcac0d2766c42737f34458f1c8c50c0d23bcb24ea953affb273216" +dependencies = [ + "cfg-if", + "instant", + "libc", + "redox_syscall", + "smallvec", + "winapi", ] [[package]] @@ -665,6 +885,12 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "paste" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c520e05135d6e763148b6426a837e239041653ba7becd2e538c076c738025fc" + [[package]] name = "percent-encoding" version = "2.1.0" @@ -772,6 +998,17 @@ dependencies = [ "bitflags", ] +[[package]] +name = "redox_users" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b033d837a7cf162d7993aded9304e30a83213c648b6e389db233191f891e5c2b" +dependencies = [ + "getrandom", + "redox_syscall", + "thiserror", +] + [[package]] name = "remove_dir_all" version = "0.5.3" @@ -916,6 +1153,17 @@ dependencies = [ "digest", ] +[[package]] +name = "sha2" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "55deaec60f81eefe3cce0dc50bda92d6d8e88f2a27df7c5033b42afeb1ed2676" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "signal-hook-registry" version = "1.4.0" @@ -947,6 +1195,123 @@ dependencies = [ "winapi", ] +[[package]] +name = "sqlformat" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4b7922be017ee70900be125523f38bdd644f4f06a1b16e8fa5a8ee8c34bffd4" +dependencies = [ + "itertools", + "nom", + "unicode_categories", +] + +[[package]] +name = "sqlx" +version = "0.5.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "551873805652ba0d912fec5bbb0f8b4cdd96baf8e2ebf5970e5671092966019b" +dependencies = [ + "sqlx-core", + "sqlx-macros", +] + +[[package]] +name = "sqlx-core" +version = "0.5.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e48c61941ccf5ddcada342cd59e3e5173b007c509e1e8e990dafc830294d9dc5" +dependencies = [ + "ahash", + "atoi", + "base64", + "bitflags", + "byteorder", + "bytes", + "crc", + "crossbeam-queue", + "dirs", + "either", + "event-listener", + "futures-channel", + "futures-core", + "futures-intrusive", + "futures-util", + "hashlink", + "hex", + "hkdf", + "hmac", + "indexmap", + "itoa", + "libc", + "log", + "md-5", + "memchr", + "once_cell", + "paste", + "percent-encoding", + "rand", + "serde", + "serde_json", + "sha-1", + "sha2", + "smallvec", + "sqlformat", + "sqlx-rt", + "stringprep", + "thiserror", + "tokio-stream", + "url", + "whoami", +] + +[[package]] +name = "sqlx-macros" +version = "0.5.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc0fba2b0cae21fc00fe6046f8baa4c7fcb49e379f0f592b04696607f69ed2e1" +dependencies = [ + "dotenv", + "either", + "heck", + "once_cell", + "proc-macro2", + "quote", + "sha2", + "sqlx-core", + "sqlx-rt", + "syn", + "url", +] + +[[package]] +name = "sqlx-rt" +version = "0.5.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4db708cd3e459078f85f39f96a00960bd841f66ee2a669e90bf36907f5a79aae" +dependencies = [ + "native-tls", + "once_cell", + "tokio", + "tokio-native-tls", +] + +[[package]] +name = "stringprep" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ee348cb74b87454fff4b551cbf727025810a004f88aeacae7f85b87f4e9a1c1" +dependencies = [ + "unicode-bidi", + "unicode-normalization", +] + +[[package]] +name = "subtle" +version = "2.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6bdef32e8150c2a081110b42772ffe7d7c9032b606bc226c8260fd97e0976601" + [[package]] name = "syn" version = "1.0.96" @@ -1025,7 +1390,7 @@ dependencies = [ "mio", "num_cpus", "once_cell", - "parking_lot", + "parking_lot 0.12.1", "pin-project-lite", "signal-hook-registry", "socket2", @@ -1054,6 +1419,17 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df54d54117d6fdc4e4fea40fe1e4e566b3505700e148a6827e59b34b0d2600d9" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-tungstenite" version = "0.17.1" @@ -1201,6 +1577,18 @@ dependencies = [ "tinyvec", ] +[[package]] +name = "unicode-segmentation" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e8820f5d777f6224dc4be3632222971ac30164d4a258d595640799554ebfd99" + +[[package]] +name = "unicode_categories" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39ec24b3121d976906ece63c9daad25b85969647682eee313cb5779fdd69e14e" + [[package]] name = "url" version = "2.2.2" @@ -1211,6 +1599,7 @@ dependencies = [ "idna", "matches", "percent-encoding", + "serde", ] [[package]] @@ -1329,6 +1718,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "whoami" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "524b58fa5a20a2fb3014dd6358b70e6579692a56ef6fce928834e488f42f65e8" +dependencies = [ + "wasm-bindgen", + "web-sys", +] + [[package]] name = "winapi" version = "0.3.9" diff --git a/Cargo.toml b/Cargo.toml index 3bf425a..2e340a0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,9 +6,11 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -tokio = { version = '*', features = [ "full" ] } +tokio = { version = '*', features = [ "full" ] } futures = '*' -reqwest = '*' -serde = { version = '*', features = [ "derive" ] } +reqwest = { version = '*', features = [ "json" ] } +serde = { version = '*', features = [ "derive" ] } serde_json = '*' -axum = { version = '*', features = [ "ws", "serde_json" ] } +axum = { version = '*', features = [ "ws", "serde_json" ] } +url = { version = '*', features = [ "serde" ] } +sqlx = { version = '*', features = [ "postgres", "runtime-tokio-native-tls" ] } diff --git a/src/main.rs b/src/main.rs index 4a2d4ad..a6fff94 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,11 +1,19 @@ -use std::sync::Arc; +pub use id::Id; +use serde_json::{from_value, Value}; use futures::prelude::*; -use reqwest::{IntoUrl, Response}; +use sign::Sign; +use conf::Config; #[tokio::main] -async fn main() { - todo!() +async fn main () { + + let cfg = Config::new("hmt.riley.lgbt"); + let ctx = Context { + config: cfg, + signer: todo!(), + }; + } mod task { @@ -14,13 +22,17 @@ mod task { //! streams and sinks. use std::pin::Pin; - use std::sync::Arc; use futures::prelude::*; - use crate::{flow::Flow, Activity, ctrl::Message, rule::Rule}; + use serde_json::Value; + use crate::{sign::Sign, flow::Flow, Activity, ctrl::Message, Context}; /// Perform a [`Task`]. - pub fn run (task: impl Task) { - tokio::spawn(task.run()); + pub fn run (ctx: &Context, task: impl Task) + where + S: Sign + Clone + Send + Sync + 'static + { + let ctx = ctx.clone(); + tokio::spawn(task.run(ctx)); } /// A computation running indefinitely on a separate thread. @@ -30,7 +42,9 @@ mod task { type Future: Future + Send + 'static; /// Execute the task. - fn run (self) -> Self::Future; + fn run (self, ctx: Context) -> Self::Future + where + S: Sign + Clone + Send + Sync + 'static; } @@ -83,19 +97,21 @@ mod task { impl Task for Process where - D: Stream> + Unpin + Send + 'static, + D: Stream> + Unpin + Send + 'static, C: Stream + Unpin + Send + 'static, { type Future = Pin + Send + 'static>>; - fn run (self) -> Self::Future { + fn run (self, ctx: Context) -> Self::Future + where + S: Sign + Clone + Send + Sync + 'static + { let Self { mut data_rx, mut ctrl_rx } = self; Box::pin(async move { let mut config = crate::conf::Config::new("localhost"); - let ctx = crate::Context {}; loop { tokio::select! { @@ -284,6 +300,7 @@ pub mod ctrl { pub enum Message { /// Modify the existing configuration of each task. Reconfigure (Arc>), + /// Shut down everything. Terminate, } @@ -292,16 +309,18 @@ pub mod ctrl { /// Configuration. pub mod conf { - use crate::rule::Rule; + use std::sync::Arc; + use crate::rule::Rule; + #[derive(Clone)] pub struct Config { /// The domain of the instance. pub host: String, /// The port to host the instance on. Defaults to `6969`. pub port: u16, /// Filtering rules applied to each activity. - pub rules: Vec>, - /// Notification predicate. + pub rules: Vec>>, + /// Notification configuration. pub notify: Notify, } @@ -343,22 +362,68 @@ pub mod conf { } #[derive(Clone)] -pub struct Context {} +pub struct Context { + config: Config, + signer: S, + client: db::Client, +} -impl Context { - - pub fn dereferencer (&self) -> Dereferencer { - Dereferencer { web: reqwest::Client::new() } +impl Context { + + /// Attempt an action within the context of the database. + pub async fn with_db <'a, F, O, T> (&'a mut self, f: F) -> Result + where + F: FnOnce (&'a mut db::Client) -> O, + O: Future> + 'a, + { + f(&mut self.client).await + } + + /// Get all actors on the instance. + pub fn actors (&self) -> impl Iterator + '_ { + None.into_iter() + } + + /// Get a dereferencer. + pub fn dereferencer (&self) -> Dereferencer + where + S: Sign + Clone + { + Dereferencer { + web: reqwest::Client::new(), + signer: self.signer.clone(), + db: self.client.clone(), + } } - pub fn signer (&self) -> &(dyn sign::Sign + Send + Sync) { - todo!() + /// Access the inner [`Sign`] provider. + pub fn signer (&self) -> &S { + &self.signer } + /// Access a notifier that delivers notifications to their intended targets. pub fn notifier (&self) -> Notifier { todo!() } + /// Conjure an activity "from thin air" as though it were posted through a client. + pub (crate) async fn conjure (&self, act: impl Into) -> Result<()> { + let act = act.into(); + todo!() + } + +} + +pub trait IntoUrl { + fn into_url (self) -> Option; +} + +impl IntoUrl for T where T: ToString { + fn into_url (self) -> Option { + self.to_string() + .parse() + .ok() + } } pub struct Notifier { @@ -366,47 +431,168 @@ pub struct Notifier { socket: Box + Send + Sync + Unpin>, } -pub struct Dereferencer { +/// A type that provides dereferencing facilities for [`Activity`] data. +pub struct Dereferencer { web: reqwest::Client, + db: db::Client, + signer: S, } -impl Dereferencer { +impl Dereferencer +where + S: Sign +{ /// Perform the dereferencing. - pub async fn dereference (&self, json: serde_json::Value) -> Result { - todo!() + pub async fn dereference (&self, json: Value) -> Result { + match json["type"].as_str() { + Some ("Create") => self.deref_create(json).await.map(Activity::from), + _ => todo!() + } + } + + fn db_client (&self) -> &db::Client { + &self.db + } + + fn web_client (&self) -> &reqwest::Client { + &self.web + } + + /// Fetch a JSON value. + pub async fn fetch (&self, url: impl IntoUrl) -> Result { + + let client = self.web_client(); + + let url = match url.into_url() { + Some (url) => url, + None => todo!(), + }; + + let req = { + let mut r = client.get(url).build()?; + self.signer.sign(&mut r)?; + r + }; + + let value = client + .execute(req) + .await? + .json() + .await?; + + Ok (value) + + } + + /// Attempt to dereference to a [`Create`](ap::Create) activity. + async fn deref_create (&self, json: Value) -> Result { + let json = if let Value::String (url) = json { + self.fetch(url).await? + } else { json }; + + match json["object"]["type"].as_str() { + Some ("Note" | "Article") => todo!(), //Ok (act::Create::Note { id }), + _ => return Err (todo!()), + } } } #[derive(Debug)] pub enum Error { Http (reqwest::Error), + Json (serde_json::Error), + Sqlx (sqlx::Error), +} + +impl From for Error { + fn from (e: sqlx::Error) -> Self { Error::Sqlx (e) } } impl From for Error { fn from (e: reqwest::Error) -> Self { Error::Http (e) } } +impl From for Error { + fn from (e: serde_json::Error) -> Self { Error::Json (e) } +} + fn err (e: impl Into) -> Error { e.into() } pub type Result = std::result::Result; +#[derive(Clone)] +pub struct Actor { + id: Id, + is_locked: bool, +} + #[derive(Clone)] pub enum Activity { - Create (act::Create), - Follow (act::Follow), + Create (ap::Create), + Follow (ap::Follow), + Accept (ap::Accept), } impl Activity { - pub async fn perform (self, ctx: Context) -> Result<()> { - todo!() + pub async fn perform (self, mut ctx: Context) -> Result<()> + where + S: sign::Sign + { + use ap::*; + + match self { + Activity::Follow (Follow::Actor { id, actor, object, .. }) => { + + // Find the actor this activity refers to. If it's not a local + // actor, we don't care. + let x = ctx.actors().find(|a| object.id == a.id); + match x { + + // Unlocked account + Some (a) if !a.is_locked => { + + // Prepare the operation. + let op = db::ops::Following { + from: actor.id.clone(), + to: object.id.clone(), + id: id.clone(), + }; + + // Use the database connection to perform an action. + ctx.with_db(|db| db.insert(op)).await?; + + // Reply with an `Accept` activity if the account is not + // locked, so the remote knows it's ok to follow this actor + // immediately. + ctx.conjure(Accept::Follow { + object: Follow::Actor { + id: id.clone(), + object, + actor, + }, + actor: a, + id, + }).await + + }, + + _ => todo!(), + } + }, + _ => todo!(), + } } /// Send a notification to the given [`Sink`]. pub async fn notify (self, notifier: Notifier) -> Result<()> { let Notifier { config, mut socket } = notifier; match &self { - Activity::Follow (..) if config.new_follower => socket.send(self).await.map_err(err), - // Otherwise, do nothing + // Only notify if the config value is set to `true`. + Activity::Follow (..) if config.new_follower => + socket.send(self) + .map_err(err) + .await, + // In all other cases, do nothing _ => Ok (()) } } @@ -464,17 +650,184 @@ impl Activity { } } -pub mod act { +pub mod db { + use crate::{Id, Result}; + use futures::prelude::*; + use sqlx::{Executor, pool::PoolConnection}; + + /// `const ()` but in Rust + fn void (_: T) -> () { () } + + type Database = sqlx::Postgres; + + pub struct Config {} + + /// A database client. + #[derive(Clone)] + pub struct Client { + /// The internal connection pool. + pool: sqlx::Pool, + } + + impl Client { + + pub async fn new (_: Config) -> Result { + todo!() + } + + /// Fetch the data mapped to the given `key` from the database. + pub async fn get (&self, key: T::Key) -> Result> + where + T: Get, + { + self.with_conn(|c| T::get(key, c)) + .await + } + + /// Perfom an insertion on the database. + pub async fn insert (&mut self, data: T) -> Result<()> + where + T: Insert, + { + self.with_conn(|c| data.set(c)) + .await + .map(void) + } + + /// Delete something from the database. + pub async fn delete (&mut self, key: T::Key) -> Result<()> + where + T: Delete, + { + self.with_conn(|c| T::del(key, c)) + .await + } + + /// Handles the getting-a-connection logic. + async fn with_conn (&self, f: F) -> Result + where + F: FnOnce (&mut PoolConnection) -> O, + O: Future>, + { + use crate::err; + + self.pool + .acquire() + .map_err(err) + .and_then(|mut c| { + f(&mut c) + }) + .await + } + + } + + pub trait Object: Sized { + type Key: Eq; + fn key (&self) -> &Self::Key; + } + + pub trait Insert: Object { + type Future: Future>; + fn set <'e, E> (self, exec: E) -> Self::Future + where + E: Executor<'e>; + } + + pub trait Delete: Object { + type Future: Future>; + fn del <'e, E> (key: Self::Key, exec: E) -> Self::Future where E: Executor<'e>; + } + + pub trait Get: Object { + type Future: Future>>; + fn get <'e, E> (key: Self::Key, exec: E) -> Self::Future where E: Executor<'e>; + } + + pub mod ops { + + //! Database operations (queries and updates). + + use super::*; + + pub struct Following { + pub from: Id, + pub to: Id, + pub id: Id, + } + + impl Object for Following { + type Key = Id; + fn key (&self) -> &Self::Key { &self.id } + } + + impl Insert for Following { + type Future = future::BoxFuture<'static, Result>; + fn set <'e, E> (self, exec: E) -> Self::Future + where + E: Executor<'e> + { + todo!() + } + } + } +} + +mod id { + use serde::{ Deserialize, Serialize }; + + #[derive(PartialEq, Eq, Clone, Serialize, Deserialize)] + pub struct Id (reqwest::Url); + + impl crate::IntoUrl for Id { + fn into_url (self) -> Option { Some (self.0) } + } + +} + +pub mod ap { + + //! ActivityPub types and utilities. + + use crate::{ Id, Activity, Actor }; #[derive(Clone)] pub enum Create { - Note {}, + Note { + id: Id, + }, } + impl From for Activity { + fn from (a: Create) -> Self { Self::Create (a) } + } + #[derive(Clone)] pub enum Follow { - Actor {}, + Actor { + id: Id, + actor: Actor, + object: Actor, + }, } + + impl From for Activity { + fn from (a: Follow) -> Self { Self::Follow (a) } + } + + #[derive(Clone)] + pub enum Accept { + Follow { + id: Id, + actor: Actor, + object: Follow, + } + } + + impl From for Activity { + fn from (a: Accept) -> Self { Self::Accept (a) } + } + } pub mod sign { @@ -558,7 +911,7 @@ pub mod rule { impl Rule for F where O: Into>, - F: Fn (Activity) -> O, + F: Fn (Activity) -> O + Clone, { fn apply (&self, act: Activity) -> Option { self(act).into() @@ -574,13 +927,14 @@ pub mod rule { pub fn drop (_: Activity) -> Option { None } /// A simple filtering rule that drops the activity if it matches the predicate `P`. + #[derive(Clone)] pub struct Filter

(pub P) where P: Fn (&Activity) -> bool; impl

Rule for Filter

where - P: Fn (&Activity) -> bool + P: Fn (&Activity) -> bool + Clone, { fn apply (&self, act: Activity) -> Option { let Self (f) = self; @@ -598,6 +952,7 @@ pub mod rule { /// /// `B` will only be applied if `A` returns [`Some`], otherwise it /// short-circuits. + #[derive(Clone)] pub struct Then (A, B); impl Rule for Then @@ -617,6 +972,7 @@ pub mod rule { /// /// If the predicate `P` returns `true`, apply `R`. Otherwise, return the /// activity unmodified. + #[derive(Clone)] pub struct Cond { pred: P, rule: R, @@ -624,7 +980,7 @@ pub mod rule { impl Rule for Cond where - P: Fn (&Activity) -> bool, + P: Fn (&Activity) -> bool + Clone, R: Rule, { fn apply (&self, act: Activity) -> Option { @@ -637,4 +993,16 @@ pub mod rule { } } + /// Execute a command and drop if nonzero exit code or empty stdout. + /// If the exit code is zero, stdout will be deserialized to an + /// [`Activity`]. + #[derive(Clone)] + pub struct Exec (std::path::PathBuf); + + impl Exec { + pub fn new (path: impl AsRef) -> Option { + todo!() + } + } + }