just a in-between state

4 files changed, 253 insertions, 84 deletions

diff --git a/client-native-rift/src/crypto.rs b/client-native-rift/src/crypto.rs
index 3f00c67..9bd8908 100644
--- a/client-native-rift/src/crypto.rs
+++ b/client-native-rift/src/crypto.rs
@@ -2,17 +2,20 @@ use aes_gcm::{
     aead::{generic_array::sequence::GenericSequence, Aead},
     Aes256Gcm, KeyInit, Nonce,
 };
+use log::info;
 
 pub struct Key(Aes256Gcm);
 
 impl Key {
-    pub fn derive(secret: String) -> Self {
+    pub fn derive(secret: &str) -> Self {
+        info!("running key generation... this might take some™ time");
         let salt = base64::decode("thisisagoodsaltAAAAAAA==").unwrap();
-        let mut key = [0u8; 256];
+        let mut key = [0u8; 32];
         fastpbkdf2::pbkdf2_hmac_sha256(secret.as_bytes(), salt.as_slice(), 250000, &mut key);
 
         let key = Aes256Gcm::new_from_slice(key.as_slice()).unwrap();
 
+        info!("done");
         Self(key)
     }
     pub fn encrypt(&self, s: &str) -> String {
diff --git a/client-native-rift/src/main.rs b/client-native-rift/src/main.rs
index d9935e8..fbc8387 100644
--- a/client-native-rift/src/main.rs
+++ b/client-native-rift/src/main.rs
@@ -1,18 +1,30 @@
 #![feature(async_closure)]
+#![feature(box_syntax)]
 
-use std::{sync::Arc, time::Duration};
-
+use crate::protocol::{RTCSessionDescriptionInit, RelayMessageWrapper, ServerboundPacket};
+use clap::{Parser, Subcommand};
+use crypto::Key;
+use futures_util::{SinkExt, StreamExt};
+use log::{debug, info};
+use protocol::{ClientboundPacket, RelayMessage};
 use signaling::signaling_connect;
+use std::{
+    collections::HashMap,
+    sync::{Arc, Weak},
+    time::Duration,
+};
+use tokio::sync::{mpsc::UnboundedSender, Mutex, RwLock};
 use webrtc::{
     api::{
         interceptor_registry::register_default_interceptors, media_engine::MediaEngine, APIBuilder,
+        API,
     },
     data_channel::data_channel_message::DataChannelMessage,
     ice_transport::ice_server::RTCIceServer,
     interceptor::registry::Registry,
     peer_connection::{
-        configuration::RTCConfiguration, math_rand_alpha,
-        peer_connection_state::RTCPeerConnectionState,
+        configuration::RTCConfiguration, peer_connection_state::RTCPeerConnectionState,
+        RTCPeerConnection,
     },
 };
 
@@ -21,6 +33,7 @@ pub mod protocol;
 pub mod signaling;
 
 fn main() {
+    env_logger::init_from_env("LOG");
     tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .build()
@@ -28,8 +41,26 @@ fn main() {
         .block_on(run())
 }
 
+#[derive(Parser)]
+struct Args {
+    #[clap(long, default_value = "meet.metamuffin.org")]
+    signaling_host: String,
+    #[clap(short, long)]
+    secret: String,
+    #[clap(subcommand)]
+    action: Action,
+}
+#[derive(Subcommand)]
+enum Action {
+    Send {},
+    Receive {},
+}
+
 async fn run() {
-    let (send, recv) = signaling_connect("meet.metamuffin.org", "hunter2").await;
+    let args = Args::parse();
+    let (sender, mut recv) = signaling_connect(&args.signaling_host, &args.secret).await;
+
+    let key = crypto::Key::derive(&args.secret);
 
     let mut media_engine = MediaEngine::default();
     media_engine.register_default_codecs().unwrap();
@@ -40,48 +71,28 @@ async fn run() {
         .with_interceptor_registry(registry)
         .build();
 
-    let config = RTCConfiguration {
-        ice_servers: vec![RTCIceServer {
-            urls: vec!["stun:metamuffin.org:16900".to_owned()],
-            ..Default::default()
-        }],
-        ..Default::default()
-    };
+    let state = Arc::new(State {
+        peers: Default::default(),
+        key,
+        api,
+        my_id: RwLock::new(None),
+        sender,
+        args,
+    });
 
-    let peer_connection = Arc::new(api.new_peer_connection(config).await.unwrap());
+    {
+        let state = state.clone();
+        tokio::spawn(async move {
+            debug!("receiving packets now");
+            while let Some(packet) = recv.recv().await {
+                debug!("{packet:?}");
+                let state = state.clone();
+                state.on_message(packet).await
+            }
+        });
+    }
 
-    let data_channel = peer_connection
-        .create_data_channel("data", None)
-        .await
-        .unwrap();
-
-    let (done_tx, mut done_rx) = tokio::sync::mpsc::channel::<()>(1);
-
-    peer_connection
-        .on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
-            println!("conn state: {s}");
-            Box::pin(async {})
-        }))
-        .await;
-
-    let d_label = data_channel.label().to_owned();
-    data_channel
-        .on_message(Box::new(move |msg: DataChannelMessage| {
-            let msg_str = String::from_utf8(msg.data.to_vec()).unwrap();
-            println!("Message from DataChannel '{}': '{}'", d_label, msg_str);
-            Box::pin(async {})
-        }))
-        .await;
-
-    let offer = peer_connection.create_offer(None).await.unwrap();
-    peer_connection
-        .set_local_description(offer.clone())
-        .await
-        .unwrap();
-
-    println!("{offer:?}");
-
-    tokio::time::sleep(Duration::from_secs(5)).await;
+    tokio::time::sleep(Duration::from_secs(10000)).await;
 
     // // Wait for the answer to be pasted
     // let line = signal::must_read_stdin().unwrap();
@@ -91,3 +102,146 @@ async fn run() {
     // // Apply the answer as the remote description
     // peer_connection.set_remote_description(answer).await?;
 }
+
+pub struct State {
+    args: Args,
+    api: API,
+    key: Key,
+    my_id: RwLock<Option<usize>>,
+    sender: UnboundedSender<ServerboundPacket>,
+    peers: RwLock<HashMap<usize, Arc<Peer>>>,
+}
+impl State {
+    pub async fn my_id(&self) -> usize {
+        self.my_id.read().await.expect("not initialized yet")
+    }
+
+    pub async fn on_message(&self, packet: ClientboundPacket) {
+        match packet {
+            protocol::ClientboundPacket::Init {
+                your_id,
+                version: _,
+            } => {
+                *self.my_id.write().await = Some(your_id);
+            }
+            protocol::ClientboundPacket::ClientJoin { id } => {
+                if id == self.my_id().await {
+                    // we joined - YAY!
+                } else {
+                    if let Action::Send { .. } = &self.args.action {}
+                }
+            }
+            protocol::ClientboundPacket::ClientLeave { id: _ } => {}
+            protocol::ClientboundPacket::Message { sender, message } => {
+                let message = self.key.decrypt(&message);
+                let p = serde_json::from_str::<RelayMessageWrapper>(&message).unwrap();
+                self.on_relay(sender, p.inner).await;
+            }
+        }
+    }
+
+    pub async fn on_relay(&self, sender: usize, p: RelayMessage) {}
+
+    pub async fn send_relay(&self, receipient: usize, inner: RelayMessage) {
+        self.sender
+            .send(ServerboundPacket::Relay {
+                recipient: Some(0),
+                message: self.key.encrypt(
+                    &serde_json::to_string(&RelayMessageWrapper {
+                        sender: self.my_id.read().await.expect("not ready to relay yet.."),
+                        inner,
+                    })
+                    .unwrap(),
+                ),
+            })
+            .unwrap()
+    }
+}
+
+pub struct Peer {
+    state: Arc<State>, // maybe use Weak later
+    peer_connection: RTCPeerConnection,
+    id: usize,
+}
+
+impl Peer {
+    pub async fn create(state: Arc<State>, id: usize) -> Arc<Self> {
+        let config = RTCConfiguration {
+            ice_servers: vec![RTCIceServer {
+                urls: vec!["stun:metamuffin.org:16900".to_owned()],
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+
+        let peer_connection = state.api.new_peer_connection(config).await.unwrap();
+
+        let peer = Arc::new(Self {
+            peer_connection,
+            id,
+            state: state.clone(),
+        });
+        peer.peer_connection
+            .on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
+                println!("conn state: {s}");
+                Box::pin(async {})
+            }))
+            .await;
+
+        {
+            let peer2 = peer.clone();
+            peer.peer_connection
+                .on_negotiation_needed(box move || {
+                    let peer = peer2.clone();
+                    Box::pin(async { peer.on_negotiation_needed().await })
+                })
+                .await;
+        }
+        peer
+    }
+
+    pub async fn send_relay(&self, inner: RelayMessage) {
+        self.state.send_relay(self.id, inner).await
+    }
+
+    pub async fn start_transfer(&self) {
+        info!("starting data channel");
+        let data_channel = self
+            .peer_connection
+            .create_data_channel("file-transfer", None)
+            .await
+            .unwrap();
+
+        data_channel
+            .on_message(Box::new(move |msg: DataChannelMessage| {
+                let msg_str = String::from_utf8(msg.data.to_vec()).unwrap();
+                println!("message! '{}'", msg_str);
+                Box::pin(async {})
+            }))
+            .await;
+    }
+
+    pub async fn on_relay(&self, p: RelayMessage) {
+        match p {
+            protocol::RelayMessage::Offer(o) => todo!(),
+            protocol::RelayMessage::Answer(a) => todo!(),
+            protocol::RelayMessage::IceCandidate(c) => {
+                self.peer_connection.add_ice_candidate(c).await.unwrap();
+            }
+        }
+    }
+
+    pub async fn on_negotiation_needed(self: Arc<Self>) {
+        info!("({}) negotiation needed", self.id);
+        let offer = self.peer_connection.create_offer(None).await.unwrap();
+        self.peer_connection
+            .set_local_description(offer.clone())
+            .await
+            .unwrap();
+        self.send_relay(protocol::RelayMessage::Offer(RTCSessionDescriptionInit {
+            sdp: offer.sdp,
+            ty: offer.sdp_type,
+        }))
+        .await
+    }
+}
diff --git a/client-native-rift/src/protocol.rs b/client-native-rift/src/protocol.rs
index b3719d0..431dc42 100644
--- a/client-native-rift/src/protocol.rs
+++ b/client-native-rift/src/protocol.rs
@@ -1,5 +1,7 @@
 use serde::{Deserialize, Serialize};
-use webrtc::peer_connection::sdp::sdp_type::RTCSdpType;
+use webrtc::{
+    ice_transport::ice_candidate::RTCIceCandidateInit, peer_connection::sdp::sdp_type::RTCSdpType,
+};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
@@ -22,8 +24,8 @@ pub enum ServerboundPacket {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RelayMessageWrapper {
-    sender: usize, // redundant, but ensures the server didnt cheat
-    inner: RelayMessage,
+    pub sender: usize, // redundant, but ensures the server didnt cheat
+    pub inner: RelayMessage,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -39,12 +41,3 @@ pub struct RTCSessionDescriptionInit {
     pub ty: RTCSdpType,
     pub sdp: String,
 }
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct RTCIceCandidateInit {
-    pub candidate: String,
-    pub sdp_m_line_index: Option<usize>,
-    pub sdp_mid: Option<String>,
-    pub username_fragment: Option<String>,
-}
diff --git a/client-native-rift/src/signaling.rs b/client-native-rift/src/signaling.rs
index c61b982..2ac3edc 100644
--- a/client-native-rift/src/signaling.rs
+++ b/client-native-rift/src/signaling.rs
@@ -1,40 +1,59 @@
 use crate::protocol::ClientboundPacket;
 use crate::{crypto::hash, protocol::ServerboundPacket};
 use futures_util::{SinkExt, StreamExt};
+use log::{debug, info};
+use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
 use tokio_tungstenite::tungstenite::{self, Message};
 
 pub async fn signaling_connect(
     host: &str,
     secret: &str,
 ) -> (
-    impl SinkExt<ServerboundPacket>,
-    impl StreamExt<Item = Option<ClientboundPacket>>,
+    UnboundedSender<ServerboundPacket>,
+    UnboundedReceiver<ClientboundPacket>,
 ) {
-    let (conn, _) = tokio_tungstenite::connect_async(
-        url::Url::parse(&format!("wss://{host}/signaling/{}", hash(secret))).unwrap(),
-    )
-    .await
-    .unwrap();
+    let uri = format!("wss://{host}/signaling/{}", hash(secret));
+    info!("connecting to signaling server at {uri:?}");
+    let (conn, _) = tokio_tungstenite::connect_async(url::Url::parse(&uri).unwrap())
+        .await
+        .unwrap();
+    info!("connection established");
 
-    let (tx, rx) = conn.split();
-    let prx = rx.map(|mesg| {
-        let mesg = mesg.unwrap();
-        match mesg {
-            tungstenite::Message::Text(t) => {
-                let p: ClientboundPacket = serde_json::from_str(t.as_str()).unwrap();
-                Some(p)
-            }
-            tungstenite::Message::Close(_) => {
-                eprintln!("ws closed :(");
-                None
+    let (mut tx, rx) = conn.split();
+
+    let (in_tx, in_rx) = unbounded_channel();
+    let (out_tx, mut out_rx) = unbounded_channel();
+
+    tokio::spawn(async {
+        rx.for_each(move |mesg| {
+            info!("packet in");
+            let mesg = mesg.unwrap();
+            match mesg {
+                tungstenite::Message::Text(t) => {
+                    let p: ClientboundPacket = serde_json::from_str(t.as_str()).unwrap();
+                    debug!("<-  {p:?}");
+                    in_tx.send(p).unwrap()
+                }
+                tungstenite::Message::Close(_) => {
+                    eprintln!("ws closed :(");
+                    unreachable!();
+                }
+                _ => (),
             }
-            _ => None,
-        }
+            Box::pin(async { () })
+        })
+        .await;
     });
-    let ptx = tx.with(async move |p| {
-        Ok::<_, tokio_tungstenite::tungstenite::error::Error>(Message::Text(
-            serde_json::to_string::<ServerboundPacket>(&p).unwrap(),
-        ))
+    tokio::spawn(async move {
+        while let Some(p) = out_rx.recv().await {
+            debug!(" ->  {p:?}");
+            tx.send(Message::Text(
+                serde_json::to_string::<ServerboundPacket>(&p).unwrap(),
+            ))
+            .await
+            .unwrap()
+        }
     });
-    (ptx, prx)
+
+    (out_tx, in_rx)
 }