feat: with compose

Cargo.lock

@@ -938,9 +938,9 @@ checksum = "1171693293099992e19cddea4e8b849964e9846f4acee11b3948bcc337be8776"
 
 [[package]]
 name = "libredox"
-version = "0.1.9"
+version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "391290121bad3d37fbddad76d8f5d1c1c314cfc646d143d7e07a3086ddff0ce3"
+checksum = "4488594b9328dee448adb906d8b126d9b7deb7cf5c22161ee591610bb1be83c0"
 dependencies = [
  "bitflags",
  "libc",
@@ -1349,9 +1349,9 @@ dependencies = [
 
 [[package]]
 name = "redox_users"
-version = "0.5.2"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4e608c6638b9c18977b00b475ac1f28d14e84b27d8d42f70e0bf1e3dec127ac"
+checksum = "dd6f9d3d47bdd2ad6945c5015a226ec6155d0bcdfd8f7cd29f86b71f8de99d2b"
 dependencies = [
  "getrandom 0.2.16",
  "libredox",

crates/norun-grpc-interface/src/grpc/norun.v1.rs

@@ -35,9 +35,37 @@ pub struct Project {
     #[prost(string, tag = "2")]
     pub image: ::prost::alloc::string::String,
     #[prost(string, tag = "3")]
+    #[prost(uint32, optional, tag="2")]
+    pub port: ::core::option::Option<u32>,
+    #[prost(oneof="project::ProjectType", tags="3, 4")]
+    pub project_type: ::core::option::Option<project::ProjectType>,
+}
+/// Nested message and enum types in `Project`.
+pub mod project {
+    #[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Oneof)]
+    pub enum ProjectType {
+        #[prost(message, tag="3")]
+        Container(super::Container),
+        #[prost(message, tag="4")]
+        Compose(super::Compose),
+    }
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Container {
+    #[prost(string, tag="1")]
+    pub image: ::prost::alloc::string::String,
+    #[prost(string, tag="2")]
     pub version: ::prost::alloc::string::String,
     #[prost(uint32, optional, tag = "4")]
     pub port: ::core::option::Option<u32>,
 }
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Compose {
+    #[prost(map="string, bytes", tag="1")]
+    pub files: ::std::collections::HashMap<::prost::alloc::string::String, ::prost::alloc::vec::Vec<u8>>,
+}
 include!("norun.v1.tonic.rs");
 // @@protoc_insertion_point(module)

crates/norun-grpc-interface/src/grpc/norun.v1.tonic.rs

@@ -4,7 +4,6 @@ pub mod registry_service_client {
     #![allow(unused_variables, dead_code, missing_docs, clippy::let_unit_value)]
     use tonic::codegen::*;
     use tonic::codegen::http::Uri;
-    ///
     #[derive(Debug, Clone)]
     pub struct RegistryServiceClient<T> {
         inner: tonic::client::Grpc<T>,
@@ -85,7 +84,6 @@ pub mod registry_service_client {
             self.inner = self.inner.max_encoding_message_size(limit);
             self
         }
-        ///
         pub async fn publish(
             &mut self,
             request: impl tonic::IntoRequest<super::PublishRequest>,
@@ -111,7 +109,6 @@ pub mod registry_service_client {
                 .insert(GrpcMethod::new("norun.v1.RegistryService", "Publish"));
             self.inner.unary(req, path, codec).await
         }
-        ///
         pub async fn get_topic(
             &mut self,
             request: impl tonic::IntoRequest<super::GetTopicRequest>,
@@ -146,12 +143,10 @@ pub mod registry_service_server {
     /// Generated trait containing gRPC methods that should be implemented for use with RegistryServiceServer.
     #[async_trait]
     pub trait RegistryService: Send + Sync + 'static {
-        ///
         async fn publish(
             &self,
             request: tonic::Request<super::PublishRequest>,
         ) -> std::result::Result<tonic::Response<super::PublishResponse>, tonic::Status>;
-        ///
         async fn get_topic(
             &self,
             request: tonic::Request<super::GetTopicRequest>,
@@ -160,7 +155,6 @@ pub mod registry_service_server {
             tonic::Status,
         >;
     }
-    ///
     #[derive(Debug)]
     pub struct RegistryServiceServer<T: RegistryService> {
         inner: _Inner<T>,

crates/norun/src/cli.rs

@@ -1,10 +1,14 @@
 use clap::{Parser, Subcommand};
 
 use crate::{
-    cli::{publish::PublishCommand, serve::ServeCommand, subscribe::SubscribeCommand},
+    cli::{
+        node::NodeCommand, publish::PublishCommand, serve::ServeCommand,
+        subscribe::SubscribeCommand,
+    },
     state::ClientState,
 };
 
+mod node;
 mod publish;
 mod serve;
 mod subscribe;
@@ -25,6 +29,7 @@ struct Cli {
 
 #[derive(Subcommand)]
 enum CliSubcommands {
+    Node(NodeCommand),
     Subscribe(SubscribeCommand),
     Publish(PublishCommand),
     Serve(ServeCommand),
@@ -36,6 +41,7 @@ pub async fn execute() -> anyhow::Result<()> {
     let state = ClientState::new(&cmd.server_url);
 
     match cmd.subcommands {
+        CliSubcommands::Node(cmd) => cmd.execute(&state).await,
         CliSubcommands::Publish(cmd) => cmd.execute(&state).await,
         CliSubcommands::Serve(cmd) => cmd.execute().await,
         CliSubcommands::Subscribe(cmd) => cmd.execute(&state).await,

crates/norun/src/cli/node.rs

@@ -0,0 +1,22 @@
+use crate::{cli::node::start::StartCommand, state::ClientState};
+
+mod start;
+
+#[derive(clap::Parser)]
+pub struct NodeCommand {
+    #[clap(subcommand)]
+    commands: NodeCommands,
+}
+
+#[derive(clap::Subcommand)]
+pub enum NodeCommands {
+    Start(StartCommand),
+}
+
+impl NodeCommand {
+    pub async fn execute(&self, state: &ClientState) -> anyhow::Result<()> {
+        match &self.commands {
+            NodeCommands::Start(start_command) => start_command.execute(state).await,
+        }
+    }
+}

crates/norun/src/cli/node/start.rs

@@ -0,0 +1,15 @@
+use crate::{node::services::node_service::NodeServiceState, state::ClientState};
+
+#[derive(clap::Parser)]
+pub struct StartCommand {}
+
+impl StartCommand {
+    pub async fn execute(&self, state: &ClientState) -> anyhow::Result<()> {
+        notmad::Mad::builder()
+            .add(state.node_service())
+            .run()
+            .await?;
+
+        Ok(())
+    }
+}

crates/norun/src/cli/subscribe.rs

@@ -1,3 +1,5 @@
+use norun_grpc_interface::project::ProjectType;
+
 use crate::{
     container_runtime::ContainerRuntimeState, grpc_client::GrpcClientState, models::port::Port,
     state::ClientState,
@@ -19,16 +21,26 @@ impl SubscribeCommand {
         for project in projects.projects {
             println!("project: {project:?}");
 
-            runtime
-                .ensure_running(
-                    &project.name,
-                    &format!("{}:{}", project.image, project.version),
-                    vec![Port {
-                        host_port: 38080,
-                        container_port: 80,
-                    }],
-                )
-                .await?;
+            match project.project_type {
+                Some(project_type) => match project_type {
+                    ProjectType::Container(container) => {
+                        runtime
+                            .ensure_running(
+                                &project.name,
+                                &format!("{}:{}", container.image, container.version),
+                                vec![Port {
+                                    host_port: 38080,
+                                    container_port: 80,
+                                }],
+                            )
+                            .await?;
+                    }
+                    ProjectType::Compose(compose) => {
+                        // Allocate a local project
+                    }
+                },
+                None => todo!(),
+            }
         }
 
         Ok(())

crates/norun/src/grpc_client.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use norun_grpc_interface::{
     GetTopicRequest, Projects, PublishRequest, registry_service_client::RegistryServiceClient,
 };
@@ -70,9 +72,39 @@ impl From<ProjectFile> for norun_grpc_interface::Project {
     fn from(value: ProjectFile) -> Self {
         Self {
             name: value.project.name,
-            image: value.container.image,
-            version: value.container.version,
             port: value.expose.and_then(|e| e.port),
+            project_type: Some({
+                match (value.container, value.compose) {
+                    (None, None) => panic!("either a container or compose is required"),
+                    (Some(_), Some(_)) => {
+                        panic!("either a container or compose is required, but not both")
+                    }
+                    (Some(container), None) => {
+                        norun_grpc_interface::project::ProjectType::Container(
+                            norun_grpc_interface::Container {
+                                image: container.image,
+                                version: container.version,
+                            },
+                        )
+                    }
+                    (None, Some(compose)) => norun_grpc_interface::project::ProjectType::Compose(
+                        norun_grpc_interface::Compose {
+                            // TODO: dirty hack to get files out for compose, it should instead
+                            // be transformed into an intermediary format, or simply extraced from the <From>
+                            files: compose
+                                .include
+                                .into_iter()
+                                .map(|i| {
+                                    (
+                                        i.to_string_lossy().to_string(),
+                                        std::fs::read(&i).expect("to be able to read include file"),
+                                    )
+                                })
+                                .collect::<HashMap<_, _>>(),
+                        },
+                    ),
+                }
+            }),
         }
     }
 }

crates/norun/src/main.rs

@@ -14,6 +14,8 @@ mod grpc_server;
 
 mod container_runtime;
 
+mod node;
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt()

crates/norun/src/node.rs

@@ -0,0 +1,3 @@
+pub mod config;
+
+pub mod services;

crates/norun/src/node/config.rs

@@ -0,0 +1,15 @@
+use std::collections::BTreeMap;
+
+use serde::Deserialize;
+
+#[derive(Debug, Deserialize)]
+pub struct Config {
+    node_name: Option<String>,
+
+    subscriptions: BTreeMap<String, Subscription>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct Subscription {
+    enabled: Option<bool>,
+}

crates/norun/src/node/services.rs

@@ -0,0 +1,3 @@
+pub mod config_service;
+pub mod data_store;
+pub mod node_service;

crates/norun/src/node/services/config_service.rs

@@ -0,0 +1,31 @@
+use anyhow::Context;
+
+use crate::{node::config::Config, state::ClientState};
+
+pub struct ConfigService {}
+
+impl ConfigService {
+    pub async fn get_config(&self) -> anyhow::Result<Config> {
+        let config_file_path = dirs::config_dir()
+            .context("failed to get config dir")?
+            .join("norun")
+            .join("node")
+            .join("config.toml");
+
+        let config_file_content = tokio::fs::read_to_string(&config_file_path).await?;
+
+        let config: Config = toml::from_str(&config_file_content)?;
+
+        Ok(config)
+    }
+}
+
+pub trait ConfigServiceState {
+    fn config_service(&self) -> ConfigService;
+}
+
+impl ConfigServiceState for ClientState {
+    fn config_service(&self) -> ConfigService {
+        ConfigService {}
+    }
+}

crates/norun/src/node/services/data_store.rs

@@ -0,0 +1,69 @@
+use std::sync::Arc;
+
+use notmad::{Component, MadError};
+use tokio::sync::Mutex;
+use tokio_util::sync::CancellationToken;
+
+use crate::state::ClientState;
+
+#[derive(Clone)]
+pub struct DataStore {
+    sender: tokio::sync::mpsc::Sender<()>,
+    receiver: Arc<Mutex<tokio::sync::mpsc::Receiver<()>>>,
+}
+
+impl DataStore {
+    pub async fn execute(&self, cancellation_token: CancellationToken) -> anyhow::Result<()> {
+        let mut rec = self.receiver.lock().await;
+
+        loop {
+            let msg = tokio::select! {
+                _ = cancellation_token.cancelled() => {
+                    return Ok(())
+                },
+                item = rec.recv() => {
+                    match item {
+                        Some(item) => item,
+                        None => return Ok(()),
+                    }
+                }
+            };
+
+            tracing::debug!("handling item");
+        }
+    }
+
+    pub async fn publish(&self) -> anyhow::Result<()> {
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl Component for DataStore {
+    fn name(&self) -> Option<String> {
+        Some("norun/node/data-store".into())
+    }
+
+    async fn run(&self, cancellation_token: CancellationToken) -> Result<(), MadError> {
+        self.execute(cancellation_token)
+            .await
+            .map_err(notmad::MadError::Inner)?;
+
+        Ok(())
+    }
+}
+
+pub trait DataStoreState {
+    fn data_store(&self) -> DataStore;
+}
+
+impl DataStoreState for ClientState {
+    fn data_store(&self) -> DataStore {
+        let (sender, receiver) = tokio::sync::mpsc::channel(100);
+
+        DataStore {
+            sender,
+            receiver: Arc::new(Mutex::new(receiver)),
+        }
+    }
+}

crates/norun/src/node/services/node_service.rs

@@ -0,0 +1,35 @@
+use notmad::{Component, MadError};
+use tokio_util::sync::CancellationToken;
+
+use crate::state::ClientState;
+
+pub struct NodeService {}
+
+#[async_trait::async_trait]
+impl Component for NodeService {
+    fn name(&self) -> Option<String> {
+        Some("norun/node".into())
+    }
+
+    async fn setup(&self) -> Result<(), MadError> {
+        tracing::info!("starting norun/node!");
+
+        Ok(())
+    }
+
+    async fn run(&self, cancellation_token: CancellationToken) -> Result<(), MadError> {
+        cancellation_token.cancelled().await;
+
+        Ok(())
+    }
+}
+
+pub trait NodeServiceState {
+    fn node_service(&self) -> NodeService;
+}
+
+impl NodeServiceState for ClientState {
+    fn node_service(&self) -> NodeService {
+        NodeService {}
+    }
+}

crates/norun/src/project_file.rs

@@ -1,4 +1,4 @@
-use std::path::Path;
+use std::path::{Path, PathBuf};
 
 use anyhow::Context;
 use serde::Deserialize;
@@ -8,8 +8,9 @@ const NORUN_PROJECT_FILE_NAME: &str = "norun.toml";
 #[derive(Clone, Debug, Deserialize, PartialEq)]
 pub struct ProjectFile {
     pub project: ProjectDecl,
-    pub container: ContainerDecl,
+    pub container: Option<ContainerDecl>,
     pub expose: Option<ExposeDecl>,
+    pub compose: Option<ComposeDecl>,
 }
 
 #[derive(Clone, Debug, Deserialize, PartialEq)]
@@ -23,6 +24,11 @@ pub struct ContainerDecl {
     pub version: String,
 }
 
+#[derive(Clone, Debug, Deserialize, PartialEq)]
+pub struct ComposeDecl {
+    pub include: Vec<PathBuf>,
+}
+
 #[derive(Clone, Debug, Deserialize, PartialEq)]
 pub struct ExposeDecl {
     pub port: Option<u32>,
@@ -72,10 +78,11 @@ port = 8080
             project: ProjectDecl {
                 name: "hello-world".into(),
             },
-            container: ContainerDecl {
+            container: Some(ContainerDecl {
                 image: "kasperhermansen/hello-world".into(),
                 version: "latest".into(),
-            },
+            }),
+            compose: None,
             expose: Some(ExposeDecl { port: Some(8080) }),
         };
 

interface/proto/norun/v1/registry.proto

@@ -27,9 +27,19 @@ message Projects {
 
 message Project {
   string name = 1;
+  optional uint32 port = 2;
 
-  string image = 2;
-  string version = 3;
-
-  optional uint32 port = 4;
+  oneof project_type {
+    Container container = 3;
+    Compose compose = 4;
+  }
+}
+
+message Container {
+  string image = 1;
+  string version = 2;
+}
+
+message Compose {
+  map<string, bytes> files = 1;
 }

mise.toml

@@ -0,0 +1,11 @@
+[tasks."node"]
+env = { RUST_LOG = "norun=trace,notmad=debug,info" }
+run = "cargo run -p norun -- node"
+
+[tasks."test"]
+alias = ["t"]
+run = "cargo nextest run"
+
+[tasks."generate"]
+alias = ["g", "gen"]
+run = "buf generate"