Implement core

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "hermit"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+tokio = { version = '*', features = [ "full" ] }
+futures = '*'
+reqwest = '*'
+serde = { version = '*', features = [ "derive" ] }
+serde_json = '*'

flake.lock

@@ -0,0 +1,74 @@
+{
+  "nodes": {
+    "flake-utils": {
+      "locked": {
+        "lastModified": 1652776076,
+        "narHash": "sha256-gzTw/v1vj4dOVbpBSJX4J0DwUR6LIyXo7/SuuTJp1kM=",
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "rev": "04c1b180862888302ddfb2e3ad9eaa63afc60cf8",
+        "type": "github"
+      },
+      "original": {
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "type": "github"
+      }
+    },
+    "naersk": {
+      "inputs": {
+        "nixpkgs": "nixpkgs"
+      },
+      "locked": {
+        "lastModified": 1653413650,
+        "narHash": "sha256-wojDHjb+eU80MPH+3HQaK0liUy8EgR95rvmCl24i58Y=",
+        "owner": "nix-community",
+        "repo": "naersk",
+        "rev": "69daaceebe12c070cd5ae69ba38f277bbf033695",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-community",
+        "repo": "naersk",
+        "type": "github"
+      }
+    },
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1653738054,
+        "narHash": "sha256-IaR8iLN4Ms3f5EjU1CJkXSc49ZzyS5qv03DtVAti6/s=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "17b62c338f2a0862a58bb6951556beecd98ccda9",
+        "type": "github"
+      },
+      "original": {
+        "id": "nixpkgs",
+        "type": "indirect"
+      }
+    },
+    "nixpkgs_2": {
+      "locked": {
+        "lastModified": 1653738054,
+        "narHash": "sha256-IaR8iLN4Ms3f5EjU1CJkXSc49ZzyS5qv03DtVAti6/s=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "17b62c338f2a0862a58bb6951556beecd98ccda9",
+        "type": "github"
+      },
+      "original": {
+        "id": "nixpkgs",
+        "type": "indirect"
+      }
+    },
+    "root": {
+      "inputs": {
+        "flake-utils": "flake-utils",
+        "naersk": "naersk",
+        "nixpkgs": "nixpkgs_2"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

flake.nix

@@ -0,0 +1,33 @@
+{
+  inputs = {
+    flake-utils.url = "github:numtide/flake-utils";
+    naersk.url = "github:nix-community/naersk";
+  };
+
+  outputs = { self, nixpkgs, flake-utils, naersk }:
+    flake-utils.lib.eachDefaultSystem (
+      system: let
+        pkgs = nixpkgs.legacyPackages."${system}";
+        naersk-lib = naersk.lib."${system}";
+      in
+        rec {
+          # `nix build`
+          packages."hermit" = naersk-lib.buildPackage {
+            pname = "hermit";
+            root = ./.;
+          };
+          defaultPackage = packages."hermit";
+
+          # `nix run`
+          apps."hermit"= flake-utils.lib.mkApp {
+            drv = packages."hermit";
+          };
+          defaultApp = apps."hermit";
+
+          # `nix develop`
+          devShell = pkgs.mkShell {
+            nativeBuildInputs = with pkgs; [ rustc cargo openssl pkgconfig ];
+          };
+        }
+    );
+}

src/main.rs

@@ -0,0 +1,565 @@
+use futures::prelude::*;
+
+#[tokio::main]
+async fn main() {
+    todo!()
+}
+
+mod task {
+
+    //! Async tasks, communicating with each other across threads through generic
+    //! streams and sinks.
+
+	use std::pin::Pin;
+    use std::sync::Arc;
+	use futures::prelude::*;
+	use crate::{flow::Flow, Activity, ctrl::Message, rule::Rule};
+
+    /// Perform a [`Task`].
+    pub fn run (task: impl Task) {
+        tokio::spawn(task.run());
+    }
+
+	/// A computation running indefinitely on a separate thread.
+    pub trait Task {
+        
+        /// The future representing this computation.
+		type Future: Future<Output = ()> + Send + 'static;
+		
+		/// Execute the task.
+		fn run (self) -> Self::Future;
+
+    }
+	
+    /// API request event processing.
+	pub struct Api <F, A, C, P> {
+    	/// Input stream of API request events from the frontend endpoints.
+    	pub fe_rx: F,
+    	/// Input stream of API request events from the ActivityPub
+    	/// endpoints.
+    	pub ap_rx: A,
+    	/// Output stream to the [`Ctrl`] task.
+    	pub ctrl_tx: C,
+    	/// Output stream to the [Activity processor pipeline][Process].
+    	pub pipe_tx: P,
+	}
+
+	/// Processes CLI commands and sends them to either the [`Auto`] task (which
+	/// takes care of scheduling automated maintenance tasks) or the [`Ctrl`] task,
+	/// which propagates control messages through the system, like live config
+	/// updates or shutdown messages for example.
+	pub struct Ipc <A, C> {
+    	/// Output stream to the [`Auto`] task.
+		pub auto_tx: A,
+		/// Output stream to the [`Ctrl`] task.
+		pub ctrl_tx: C,
+	}
+
+	/// Delivers control messages to other running tasks.
+	pub struct Ctrl <A, I, S> {
+    	/// Message stream from the [`Api`] task.
+		pub api_rx: A,
+		/// Message stream from the [`Ipc`] task.
+		pub ipc_rx: I,
+		/// Fan-out to all running tasks that are subscribed to [control messages][Ctrl].
+		pub tx: S,
+	}
+
+	/// Performs automated maintenance tasks.
+	pub struct Auto <E, C> {
+    	/// Receiver for manual job triggers received from the [`Ipc`] task.
+		pub ipc_rx: E,
+		/// Receiver for [control messages][Ctrl].
+		pub ctrl_rx: C,
+	}
+
+	pub struct Process <D, C, F> {
+    	pub data_rx: D,
+		pub ctrl_rx: C,
+		pub fe_tx: F,
+	}
+
+	impl<D, C, F> Task for Process<D, C, F>
+	where
+		D: Stream<Item = Flow<serde_json::Value>> + Unpin + Send + 'static,
+		C: Stream<Item = Message> + Unpin + Send + 'static,
+		Arc<F>: Sink<Activity> + Send + Sync + 'static,
+		F: Send + Sync + 'static,
+	{
+        type Future = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
+
+        fn run (self) -> Self::Future {
+
+            let Self { mut data_rx, mut ctrl_rx, fe_tx } = self;
+
+        	Box::pin(async move {
+				
+				let mut config = crate::conf::Config::new("localhost");
+				let ctx = crate::Context {};
+                let sink = Arc::new(fe_tx);
+				
+				loop {
+					tokio::select! {
+						// Await control commands from `Ctrl`.
+						Some (message) = ctrl_rx.next() => match message {
+    						// Live config reloading.
+    						Message::Reconfigure (c) => c(&mut config),
+    						// Graceful termination command from `Ctrl`.
+    						Message::Terminate => break,
+						},
+						// Listen for incoming activities.
+						Some (data) = data_rx.next() => {
+
+    						// Dereferencing and other unfucking.
+    						let d    = ctx.dereferencer();
+    						let data = match data.apply(|j| d.dereference(j)).await {
+        						Ok (data) => data,
+        						_ => continue,
+    						};
+
+    						// Run both incoming and outgoing activities through the filtering system.
+    						let action = |act| config.rules.iter().try_fold(act, |a, r| r.apply(a));
+							let data = match data.map(action).to_option() {
+								// Activity survived the filtering process, bind it to `data`.
+								Some (data) => data,
+								// Activity got filtered out, move on.
+								None => continue,
+							};
+
+							// Perform each activity in the context of the instance.
+							let c    = ctx.clone();
+							let copy = data.clone();
+							let data = match data.apply(|a| a.perform(c)).await {
+    							// Everything went ok, we can continue, bind `copy` to `data`.
+    							Ok (_) => copy,
+    							// Something went wrong while performing the activity,
+    							// report error and move on.
+    							_ => continue,
+							};
+
+							// If incoming: push a notification to the frontend.
+							let incoming = {
+    							let s = sink.clone();
+    							let c = config.notify.clone();
+    							move |a: Activity| a.notify(c, s)
+							};
+							// If outgoing: deliver the activity to its targets.
+							let outgoing = {
+    							let s = ctx.signer();
+    							move |a: Activity| a.deliver(s)
+							};
+
+							match data.pick(incoming, outgoing).await {
+								Ok (_) => println!("Yay"),
+								_ => println!("Boo"),
+							};
+
+						},
+					}
+				}
+			})
+        }
+
+	}
+
+}
+
+pub mod flow {
+
+	//! Functional control flow based on the source and destination
+	//! of a message flowing through the system.
+
+	use std::future::Future;
+
+	/// A wrapper type that annotates a message with the flow it is
+	/// supposed to take, without allowing that flow to be inspected
+	/// or modified.
+	#[derive(Clone)]
+    pub struct Flow <T> {
+		flow: Direction,
+		data: T,
+    }
+
+    #[derive(Clone, Copy)]
+    enum Direction {
+        Incoming,
+        Outgoing,
+    }
+
+	impl<T> Flow<T> {
+    	
+    	#[allow(non_snake_case)]
+    	/// Make the data take the "incoming" flow.
+    	pub fn Incoming (data: T) -> Flow<T> {
+			Flow { data, flow: Direction::Incoming }
+		}
+
+		#[allow(non_snake_case)]
+		/// Make the data take the "outbound" flow.
+		pub fn Outgoing (data: T) -> Flow<T> {
+			Flow { data, flow: Direction::Outgoing }
+		}
+
+		/// Apply a function `f` to the value inside, without disturbing
+		/// the flow direction.
+		pub async fn apply <F, A, U, E> (self, f: F) -> Result<Flow<U>, E>
+		where
+    		A: Future<Output = Result<U, E>>,
+		    F: FnOnce (T) -> A,
+		{
+    		let Flow { data, flow } = self;
+    		Ok (Flow {
+        		data: f(data).await?,
+        		flow,
+    		})
+		}
+
+		/// If the message is taking the incoming flow, apply `f`, if it is taking the
+		/// outgoing flow, apply `g`.
+		pub async fn pick <F, G, A, B, U, E> (self, f: F, g: G) -> Result<Flow<U>, E>
+		where
+			A: Future<Output = Result<U, E>>,
+			B: Future<Output = Result<U, E>>,
+			F: FnOnce (T) -> A,
+			G: FnOnce (T) -> B,
+    	{
+			match self.flow {
+				Direction::Incoming => self.apply(f).await,
+				Direction::Outgoing => self.apply(g).await,
+			}
+    	}
+
+		/// Map over the contained value.
+    	pub fn map <F, U> (self, f: F) -> Flow<U>
+    	where
+        	F: FnOnce (T) -> U,
+        {
+			Flow {
+    			data: f(self.data),
+    			flow: self.flow,
+			}
+        }
+		
+	}
+
+	impl<T> Flow<Option<T>> {
+    	/// Swap the containers.
+		pub fn to_option (self) -> Option<Flow<T>> {
+    		let Flow { flow, data } = self;
+    		data.map(|data| Flow {
+        		flow,
+        		data,
+    		})
+		}
+	}
+
+	impl<T, E> Flow<Result<T, E>> {
+    	/// Swap the containers.
+		pub fn to_result (self) -> Result<Flow<T>, E> {
+    		let Flow { flow, data } = self;
+    		data.map(|data| Flow {
+        		flow,
+        		data,
+    		})
+		}
+	}
+
+}
+
+/// Control messages.
+pub mod ctrl {
+
+    use std::sync::Arc;
+
+    use crate::conf::Config;
+
+    #[derive(Clone)]
+	pub enum Message {
+    	/// Modify the existing configuration of each task.
+		Reconfigure (Arc<Box<dyn Fn (&mut Config) + Send + Sync>>),
+		Terminate,
+	}
+
+}
+
+/// Configuration.
+pub mod conf {
+
+	use crate::rule::Rule;
+
+	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<Box<dyn Rule + Send + Sync>>,
+		/// Notification predicate.
+		pub notify: Notify,
+	}
+
+	impl Config {
+    	/// Create a new default config.
+		pub fn new (hostname: impl ToString) -> Config {
+    		let (notify, rules) = def();
+            Config {
+				host: hostname.to_string(),
+				port: 6969,
+				notify,
+				rules,
+            }
+		}
+	}
+
+	#[derive(Clone, Copy)]
+	pub struct Notify {
+		pub post_liked: bool,
+		pub post_shared: bool,
+		pub follow_requested: bool,
+		pub new_follower: bool,
+	}
+
+	impl Default for Notify {
+        fn default () -> Self {
+            Notify {
+                post_liked: true,
+                post_shared: true,
+                follow_requested: true,
+                new_follower: true,
+            }
+        }
+    }
+
+    /// Shortcut for creating a default instance
+	fn def <T> () -> T where T: Default { T::default() }
+
+}
+
+#[derive(Clone)]
+pub struct Context {}
+
+impl Context {
+	pub fn dereferencer (&self) -> Dereferencer {
+		Dereferencer { web: reqwest::Client::new() }
+	}
+
+	pub fn signer (&self) -> &(dyn sign::Sign + Send + Sync) {
+		todo!()
+	}
+}
+
+pub struct Dereferencer {
+    web: reqwest::Client
+}
+
+impl Dereferencer {
+	pub async fn dereference (&self, a: serde_json::Value) -> Result<Activity> {
+		todo!()
+	}
+}
+
+#[derive(Debug)]
+pub enum Error {}
+
+pub type Result <T, E = Error> = std::result::Result<T, E>;
+
+#[derive(Clone)]
+pub enum Activity {
+	Create (act::Create),
+	Follow (act::Follow),
+}
+
+impl Activity {
+	pub async fn perform (self, ctx: Context) -> Result<()> {
+		todo!()
+	}
+
+	pub async fn notify <S> (self, cfg: conf::Notify, sink: S) -> Result<()>
+	where
+    	S: Sink<Activity>,
+	{
+		todo!()
+	}
+
+	pub async fn deliver <S> (self, signer: &S) -> Result<()>
+	where
+    	S: sign::Sign + ?Sized,
+	{
+		todo!()
+	}
+}
+
+pub mod act {
+
+    #[derive(Clone)]
+	pub enum Create {
+		Note {},
+	}
+
+    #[derive(Clone)]
+   	pub enum Follow {
+		Actor {},
+   	}
+}
+
+pub mod sign {
+
+    //! Request signing.
+
+    use reqwest::Request;
+    use crate::Result;
+
+	pub trait Sign {
+		fn sign (&self, req: &mut Request) -> Result<()>;
+	}
+}
+
+pub mod rule {
+    
+	//! User-defined activity transformation rules.
+	//!
+	//! Every [`Rule`] is a function `fn (Activity) -> Option<Activity>`.
+
+	use super::Activity;
+
+	/// Transforms an [`Activity`].
+	///
+	/// ```
+	/// use hermit::{ Activity, rule::{ Filter, Rule, keep } };
+	///
+	/// // Fails to compile if the given parameter is not a `Rule`
+	/// fn is_rule <R: Rule> (x: R) -> R { x }
+	///
+	/// // Closures of `Activity -> Activity` or
+	/// // `Activity -> Option<Activity>` can be used.
+	/// let closure = is_rule(|a: Activity| Some(a));
+	///
+	/// // `hermit::rule::Filter` implements `Rule`. This one will
+	/// // filter every activity.
+	/// let filter = is_rule(Filter (|_| true))
+	///
+	/// // `hermit::rule::keep` is a function pointer, and they
+	/// // always implement the `Fn*` traits.
+	/// let function = is_rule(keep);
+	///
+	/// // Rules can be combined using the `then` operator, in which
+	/// // case they will be applied in sequence.
+	/// let combined = is_rule(closure.then(filter).then(keep));
+	///
+	/// // Check if it works! Due to `filter`, any input this combined
+	/// // rule is applied to will be dropped.
+	/// let result = combined.apply(todo!());
+	/// assert!(result.is_none())
+	/// ```
+	pub trait Rule {
+    	
+    	/// Apply the rule to the [`Activity`].
+    	///
+    	/// If this function returns `None`, the activity is dropped and will
+    	/// not be processed further. This allows rules to function both as
+    	/// transformations and as filters.
+    	fn apply (&self, act: Activity) -> Option<Activity>;
+
+		/// Sequence `next` after `self` in a lazy way.
+    	fn then <R> (self, next: R) -> Then<Self, R>
+    	where
+        	Self: Sized,
+        	R: Rule,
+    	{
+			Then (self, next)
+    	}
+
+		/// Apply `self` only if `pred` holds.
+    	fn only_if <P> (self, pred: P) -> Cond<P, Self>
+    	where
+        	Self: Sized,
+        	P: Fn (&Activity) -> bool,
+    	{
+			Cond { rule: self, pred }
+    	}
+
+	}
+
+	impl<F, O> Rule for F
+	where
+    	O: Into<Option<Activity>>,
+    	F: Fn (Activity) -> O,
+	{
+		fn apply (&self, act: Activity) -> Option<Activity> {
+    		self(act).into()
+		}
+	}
+
+ 	// Primitives
+
+	/// Always keep passed activities.
+	pub fn keep (a: Activity) -> Option<Activity> { Some (a) }
+
+	/// Always drop passed activities.
+	pub fn drop (_: Activity) -> Option<Activity> { None }
+
+	/// A simple filtering rule that drops the activity if it matches the predicate `P`.
+    pub struct Filter <P> (pub P)
+    where
+        P: Fn (&Activity) -> bool;
+
+    impl<P> Rule for Filter<P>
+    where
+        P: Fn (&Activity) -> bool
+    {
+        fn apply (&self, act: Activity) -> Option<Activity> {
+            let Self (f) = self;
+            if f(&act) {
+				None
+            } else {
+				Some (act)
+            }
+        }
+    }
+
+	// Combinators
+
+	/// Sequence two rules.
+	///
+	/// `B` will only be applied if `A` returns [`Some`], otherwise it
+	/// short-circuits.
+	pub struct Then <A, B> (A, B);
+
+	impl<A, B> Rule for Then<A, B>
+	where
+    	A: Rule,
+    	B: Rule,
+	{
+        fn apply (&self, act: Activity) -> Option<Activity> {
+            let Self (a, b) = self;
+            a.apply(act).and_then(|act| {
+                b.apply(act)
+            })
+        }
+    }
+
+	/// Apply a rule conditionally.
+	///
+	/// If the predicate `P` returns `true`, apply `R`. Otherwise, return the
+	/// activity unmodified.
+    pub struct Cond <P, R> {
+		pred: P,
+		rule: R,
+    }
+
+	impl<P, R> Rule for Cond<P, R>
+	where
+    	P: Fn (&Activity) -> bool,
+    	R: Rule,
+	{
+        fn apply (&self, act: Activity) -> Option<Activity> {
+            let Self { pred, rule } = self;
+            if pred(&act) {
+                rule.apply(act)
+            } else {
+				Some (act)
+            }
+        }
+    }
+
+}