initial commit

.gitignore

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

Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "multibridge"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+futures = "0.3"
+tokio = { version = "1.28", features = ["rt-multi-thread", "macros", "time", "sync"] }
+async-trait = "0.1"
+env_logger = "0.10"
+log = "0.4"
+
+irc = "0.15"
+matrix-sdk = "0.6"
+serenity = "0.11"

src/bridge_irc/mod.rs

@@ -0,0 +1,67 @@
+use async_trait::async_trait;
+use futures::StreamExt;
+use tokio::select;
+use irc::{client::Client, proto::Command};
+use crate::linkmap::Linkmap;
+use crate::message::Id;
+use crate::{message::{Message, Sender, Receiver}, supervisor::{Task, TaskResult}};
+use irc::proto::Message as IrcMessage;
+
+pub use irc::client::prelude::Config as IrcConfig;
+
+pub struct IrcTask {
+	config: IrcConfig,
+	channels: Linkmap<String>,
+}
+
+impl IrcTask {
+	pub fn new(config: IrcConfig, channels: Linkmap<String>) -> Self {
+		Self {
+			config, channels
+		}
+	}
+
+	async fn handle_bridge_msg(&self, id: Id, client: &mut Client, msg: Message) -> TaskResult {
+		for channel in self.channels.get_channels(&msg.link) {
+			if msg.origin.0 == id && &msg.origin.1 == channel {
+				continue
+			}
+			client.send_privmsg(channel, format!("<{}> {}", msg.author, msg.content))?;
+		}
+		Ok(())
+	}
+
+	async fn handle_irc_msg(&self, id: Id, tx: &mut Sender, msg: IrcMessage) -> TaskResult {
+		if let Command::PRIVMSG(channel, message) = &msg.command {
+			let Some(link) = self.channels.get_link(channel) else { return Ok(()) };
+			let Some(author) = msg.source_nickname() else { return Ok(()) };
+			tx.send(Message {
+				origin: (id, channel.to_owned()),
+				link: link.clone(),
+				author: author.to_owned(),
+				content: message.clone(),
+			})?;
+		}
+		Ok(())
+	}
+}
+
+#[async_trait]
+impl Task for IrcTask {
+	async fn start(&self, id: Id, mut tx: Sender, mut rx: Receiver) -> TaskResult {
+		let mut client = Client::from_config(self.config.clone()).await?;
+		client.identify()?;
+
+		let mut stream = client.stream()?;
+
+		loop {
+			select! {
+				bridge_msg = rx.recv() => self.handle_bridge_msg(id, &mut client, bridge_msg?).await?,
+				irc_msg = stream.next() => match irc_msg {
+					Some(msg) => self.handle_irc_msg(id, &mut tx, msg?).await?,
+					None => return Err("Lost connection to IRC server".into()),
+				}
+			};
+		}
+	}
+}

src/bridge_matrix/mod.rs

@@ -0,0 +1,102 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use futures::StreamExt;
+use matrix_sdk::Client;
+use matrix_sdk::config::SyncSettings;
+use matrix_sdk::event_handler::Ctx;
+use matrix_sdk::room::Room;
+use matrix_sdk::ruma::events::room::message::{SyncRoomMessageEvent, RoomMessageEventContent};
+use matrix_sdk::ruma::{RoomId, OwnedUserId};
+use tokio::select;
+use crate::linkmap::Linkmap;
+use crate::message::{Id, Message};
+use crate::{message::{Sender, Receiver}, supervisor::{Task, TaskResult}};
+
+pub struct MatrixTask {
+	rooms: Arc<Linkmap<String>>,
+	user: OwnedUserId,
+	passwd: String,
+}
+
+impl MatrixTask {
+	pub fn new(user: OwnedUserId, passwd: String, rooms: Linkmap<String>) -> Self {
+		Self {
+			user,
+			passwd,
+			rooms: Arc::new(rooms)
+		}
+	}
+}
+
+struct Context {
+	id: Id,
+	tx: Sender,
+	rooms: Arc<Linkmap<String>>,
+}
+
+#[async_trait]
+impl Task for MatrixTask {
+	async fn start(&self, id: Id, tx: Sender, mut rx: Receiver) -> TaskResult {
+		let client = Client::builder()
+			.server_name(&self.user.server_name())
+			.build().await?;
+		client.login_username(&self.user, &self.passwd)
+			.send()
+			.await?;
+
+		client.add_event_handler_context(Arc::new(Context {
+			id,
+			tx,
+			rooms: self.rooms.clone()
+		}));
+		client.add_event_handler(handle_matrix_msg);
+
+		let mut sync_stream = Box::pin(client.sync_stream(SyncSettings::default()).await);
+
+		loop {
+			select! {
+				response = sync_stream.next() => match response {
+					Some(Ok(_)) => (),
+					Some(Err(e)) => return Err(e.into()),
+					None => return Err("Lost connection to Matrix server".into())
+				},
+				message = rx.recv() => match message {
+					Ok(msg) => handle_bridge_msg(id, &client, &self.rooms, msg).await?,
+					Err(e) => return Err(e.into())
+				}
+			}
+		}
+	}
+}
+
+async fn handle_matrix_msg(ev: SyncRoomMessageEvent, client: Client, room: Room, ctx: Ctx<Arc<Context>>) -> TaskResult {
+	if Some(ev.sender()) == client.user_id() { return Ok(()) }
+	let SyncRoomMessageEvent::Original(ev) = ev else { return Ok(()) };
+	let Ok(Some(sender)) = room.get_member(&ev.sender).await else { return Ok(()) };
+	let Some(link) = ctx.rooms.get_link(room.room_id().as_str()) else { return Ok(()) };
+	let body = ev.content.body();
+
+	ctx.tx.send(Message {
+		origin: (ctx.id, room.room_id().to_string()),
+		link: link.clone(),
+		author: sender.name().to_owned(),
+		content: body.to_owned(),
+	})?;
+	Ok(())
+}
+
+async fn handle_bridge_msg(id: Id, client: &Client, links: &Linkmap<String>, msg: Message) -> TaskResult {
+	let content = RoomMessageEventContent::text_plain(
+		format!("<{}> {}", msg.author, msg.content)
+	);
+	for room in links.get_channels(&msg.link) {
+		if msg.origin.0 == id && &msg.origin.1 == room {
+			continue;
+		}
+		let Ok(room) = <&RoomId>::try_from(room.as_str()) else { continue };
+		let Some(Room::Joined(room)) = client.get_room(room) else { continue };
+		room.send(content.clone(), None).await?;
+	}
+	Ok(())
+}

src/linkmap.rs

@@ -0,0 +1,54 @@
+use std::{collections::HashMap, borrow::Borrow};
+
+use crate::message::Link;
+use std::hash::Hash;
+
+pub struct Linkmap<T: Hash + Eq + Clone> {
+	to: HashMap<T, Link>,
+	from: HashMap<Link, Vec<T>>,
+}
+
+impl <T: Hash + Eq + Clone> Linkmap<T> {
+	pub fn new() -> Self {
+		Self {
+			to: HashMap::new(),
+			from: HashMap::new(),
+		}
+	}
+
+	pub fn with_capacity(capacity: usize) -> Self {
+		Self {
+			to: HashMap::with_capacity(capacity),
+			from: HashMap::with_capacity(capacity),
+		}
+	}
+
+	pub fn get_link<K>(&self, channel: &K) -> Option<&Link>
+	where T: Borrow<K>, K: Hash + Eq + ?Sized {
+		self.to.get(channel)
+	}
+
+	pub fn get_channels(&self, link: &Link) -> &[T] {
+		self.from.get(link).map(|a| a.as_slice()).unwrap_or(&[])
+	}
+
+	pub fn insert(&mut self, link: Link, channel: T) {
+		self.to.insert(channel.clone(), link.clone());
+		self.from.entry(link)
+			.or_insert(Vec::with_capacity(1))
+			.push(channel.clone());
+	}
+}
+
+impl <T: Hash + Eq + Clone> FromIterator<(T, Link)> for Linkmap<T> {
+	fn from_iter<I: IntoIterator<Item=(T, Link)>>(iter: I) -> Self {
+		let to: HashMap<T, Link> = iter.into_iter().collect();
+		let mut from = HashMap::with_capacity(to.len());
+		for (k, v) in &to {
+			from.entry(v.clone())
+				.or_insert(Vec::with_capacity(1))
+				.push(k.clone());
+		}
+		Self { to, from }
+	}
+}

src/main.rs

@@ -0,0 +1,24 @@
+use bridge_matrix::MatrixTask;
+use linkmap::Linkmap;
+use matrix_sdk::ruma::user_id;
+use supervisor::Task;
+use bridge_irc::{IrcConfig, IrcTask};
+
+use crate::supervisor::run_tasks;
+
+mod supervisor;
+mod message;
+mod linkmap;
+
+mod bridge_irc;
+mod bridge_matrix;
+
+#[tokio::main]
+async fn main() {
+	env_logger::init();
+
+	let tasks: Vec<Box<dyn Task>> = vec![
+		// tasks withheld for privacy
+	];
+	run_tasks(tasks).await;
+}

src/message.rs

@@ -0,0 +1,30 @@
+pub type Sender = tokio::sync::broadcast::Sender<Message>;
+pub type Receiver = tokio::sync::broadcast::Receiver<Message>;
+
+#[derive(Clone, Debug, Eq, PartialEq, Hash)]
+pub struct Link(String);
+
+impl <S> From<S> for Link
+where S: Into<String> {
+    fn from(value: S) -> Self {
+        Self(value.into())
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct Id(usize);
+
+impl Id {
+	pub const fn new(n: usize) -> Self {
+		Self(n)
+	}
+}
+
+#[derive(Clone, Debug)]
+pub struct Message {
+	pub origin: (Id, String),
+	pub link: Link,
+	pub author: String,
+	pub content: String,
+}
+

src/supervisor.rs

@@ -0,0 +1,59 @@
+use std::{any::Any, time::Duration};
+use std::error::Error;
+use std::panic::AssertUnwindSafe;
+use futures::FutureExt;
+
+use async_trait::async_trait;
+use futures::StreamExt;
+use futures::stream::FuturesUnordered;
+use log::{warn, info, error};
+
+use crate::message::{Sender, Receiver, Id};
+
+pub type TaskResult = Result<(), Box<dyn Error>>;
+
+#[async_trait]
+pub trait Task {
+	async fn start(&self, origin: Id, tx: Sender, rc: Receiver) -> TaskResult;
+	fn restart_timeout(&self) -> Option<Duration> {
+		Some(Duration::from_secs(15))
+	}
+}
+
+enum ExitStatus {
+	Success,
+	Error(Box<dyn Error>),
+	Panic(Box<dyn Any + 'static>),
+}
+
+async fn start_task(id: usize, task: &dyn Task, tx: Sender, timeout: Duration) -> (usize, ExitStatus) {
+	tokio::time::sleep(timeout).await;
+	let rx = tx.subscribe();
+	let future = AssertUnwindSafe(task.start(Id::new(id), tx, rx)).catch_unwind();
+	let result = match future.await {
+		Ok(Ok(_)) => ExitStatus::Success,
+		Ok(Err(e)) => ExitStatus::Error(e),
+		Err(e) => ExitStatus::Panic(e),
+	};
+	(id, result)
+}
+
+pub async fn run_tasks(tasks: Vec<Box<dyn Task>>) {
+	let mut futures = FuturesUnordered::new();
+	let (tx, _) = tokio::sync::broadcast::channel(64);
+	for (id, task) in tasks.iter().enumerate() {
+		futures.push(start_task(id, task.as_ref(), tx.clone(), Duration::ZERO));
+	}
+	while let Some((id, result)) = futures.next().await {
+		let task = &tasks[id];
+		match &result {
+			ExitStatus::Success => warn!("task {id:?} exited successfully"),
+			ExitStatus::Error(e) => warn!("task {id:?}: exited with error: {e}"),
+			ExitStatus::Panic(_) => error!("task {id:?}: panicked"),
+		}
+		if let Some(dur) = task.restart_timeout() {
+			info!("task {id:?}: retrying in {dur:?}");
+			futures.push(start_task(id, task.as_ref(), tx.clone(), dur));
+		}
+	}
+}