feat: with compose

NETWORKING.md

@@ -0,0 +1,12 @@
+# Networking
+
+We've got multiple hosts potentially in play, and multiple containers that wants
+to expose themselves on the primary host, but only 1 process can take a tcp port
+on the same host at a time.
+
+1. The primary instance, will bind to common tcp ports as required (80, 443) can
+   be changed.
+   1. TODO: TLS termination
+1. Will send traffic to peers hosting their versions of a url.
+1. On the peers, each container will bind to a port (TBD), maybe we will do some
+   long running connections instead. SOCK protocol

README.md

@@ -1,7 +1,11 @@
 # Norun
 
+## Server
+
+Many servers
+
 ```bash
-norun subscribe apps  
+norun subscribe apps
 ```
 
 ```bash  
@@ -18,6 +22,7 @@ name = "hello-world"
 [container]
 image = "kasperhermansen/hello-world"
 version = "latest" # default
+replicas = 3
 
 [expose]
 port = 8080

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

@@ -3,24 +3,69 @@
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct PublishRequest {
-    #[prost(message, optional, tag="1")]
+    #[prost(message, optional, tag = "1")]
     pub project: ::core::option::Option<Project>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
-pub struct PublishResponse {
+pub struct PublishResponse {}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct GetTopicRequest {
+    #[prost(string, tag = "1")]
+    pub topic: ::prost::alloc::string::String,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct GetTopicResponse {
+    #[prost(message, optional, tag = "1")]
+    pub projects: ::core::option::Option<Projects>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Projects {
+    #[prost(message, repeated, tag = "1")]
+    pub projects: ::prost::alloc::vec::Vec<Project>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Project {
-    #[prost(string, tag="1")]
+    #[prost(string, tag = "1")]
     pub name: ::prost::alloc::string::String,
-    #[prost(string, tag="2")]
+    #[prost(string, tag = "2")]
     pub image: ::prost::alloc::string::String,
-    #[prost(string, tag="3")]
+    #[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")]
+    #[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)
+// @@protoc_insertion_point(module)

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

@@ -109,6 +109,31 @@ 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>,
+        ) -> std::result::Result<
+            tonic::Response<super::GetTopicResponse>,
+            tonic::Status,
+        > {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| {
+                    tonic::Status::new(
+                        tonic::Code::Unknown,
+                        format!("Service was not ready: {}", e.into()),
+                    )
+                })?;
+            let codec = tonic::codec::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static(
+                "/norun.v1.RegistryService/GetTopic",
+            );
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("norun.v1.RegistryService", "GetTopic"));
+            self.inner.unary(req, path, codec).await
+        }
     }
 }
 /// Generated server implementations.
@@ -122,6 +147,13 @@ pub mod registry_service_server {
             &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>,
+        ) -> std::result::Result<
+            tonic::Response<super::GetTopicResponse>,
+            tonic::Status,
+        >;
     }
     #[derive(Debug)]
     pub struct RegistryServiceServer<T: RegistryService> {
@@ -248,6 +280,52 @@ pub mod registry_service_server {
                     };
                     Box::pin(fut)
                 }
+                "/norun.v1.RegistryService/GetTopic" => {
+                    #[allow(non_camel_case_types)]
+                    struct GetTopicSvc<T: RegistryService>(pub Arc<T>);
+                    impl<
+                        T: RegistryService,
+                    > tonic::server::UnaryService<super::GetTopicRequest>
+                    for GetTopicSvc<T> {
+                        type Response = super::GetTopicResponse;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::GetTopicRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as RegistryService>::get_topic(&inner, request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let inner = inner.0;
+                        let method = GetTopicSvc(inner);
+                        let codec = tonic::codec::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 _ => {
                     Box::pin(async move {
                         Ok(

crates/norun/Cargo.toml

@@ -21,6 +21,11 @@ tonic = { workspace = true }
 tokio-util = "0.7.15"
 async-trait = "0.1.88"
 notmad = "0.7.2"
+bollard = "0.19.1"
+futures-util = "0.3.31"
+dirs = "6.0.0"
+uuid = { version = "1.17.0", features = ["serde", "v4"] }
+ron = "0.10.1"
 
 [dev-dependencies]
 pretty_assertions = "1.4.1"

crates/norun/src/cli.rs

@@ -1,12 +1,17 @@
 use clap::{Parser, Subcommand};
 
 use crate::{
-    cli::{publish::PublishCommand, serve::ServeCommand},
+    cli::{
+        node::NodeCommand, publish::PublishCommand, serve::ServeCommand,
+        subscribe::SubscribeCommand,
+    },
     state::ClientState,
 };
 
+mod node;
 mod publish;
 mod serve;
+mod subscribe;
 
 #[derive(Parser)]
 #[command(author, version, about)]
@@ -24,6 +29,8 @@ struct Cli {
 
 #[derive(Subcommand)]
 enum CliSubcommands {
+    Node(NodeCommand),
+    Subscribe(SubscribeCommand),
     Publish(PublishCommand),
     Serve(ServeCommand),
 }
@@ -34,7 +41,9 @@ 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/publish.rs

@@ -1,12 +1,9 @@
 use std::path::PathBuf;
 
-use crate::{grpc_client::GrpcClientState, models::ProjectTag, project_file, state::ClientState};
+use crate::{grpc_client::GrpcClientState, project_file, state::ClientState};
 
 #[derive(clap::Parser, Debug)]
 pub struct PublishCommand {
-    #[arg(value_parser = clap::value_parser!(ProjectTag))]
-    project_tag: ProjectTag,
-
     #[arg(long = "project-path", default_value = ".")]
     project_path: PathBuf,
 }

crates/norun/src/cli/subscribe.rs

@@ -0,0 +1,48 @@
+use norun_grpc_interface::project::ProjectType;
+
+use crate::{
+    container_runtime::ContainerRuntimeState, grpc_client::GrpcClientState, models::port::Port,
+    state::ClientState,
+};
+
+#[derive(clap::Parser)]
+pub struct SubscribeCommand {
+    #[arg(long)]
+    topic: String,
+}
+
+impl SubscribeCommand {
+    pub async fn execute(&self, state: &ClientState) -> anyhow::Result<()> {
+        let projects = state.grpc_client().subscribe(&self.topic).await?;
+
+        let runtime = state.container_runtime();
+
+        println!("printing found projects (len={})", projects.projects.len());
+        for project in projects.projects {
+            println!("project: {project:?}");
+
+            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/container_runtime.rs

@@ -0,0 +1,105 @@
+use std::{collections::HashMap, sync::LazyLock};
+
+use bollard::{
+    Docker,
+    query_parameters::{
+        CreateContainerOptionsBuilder, CreateImageOptionsBuilder, ListContainersOptionsBuilder,
+        StartContainerOptionsBuilder,
+    },
+    secret::{ContainerCreateBody, HostConfig, PortBinding},
+};
+use futures_util::TryStreamExt;
+
+use crate::{models::port::Port, state::ClientState};
+
+#[derive(Clone)]
+pub struct ContainerRuntime {
+    client: Docker,
+}
+
+impl ContainerRuntime {
+    #[tracing::instrument(skip(self), level = "trace")]
+    pub async fn ensure_running(
+        &self,
+        name: &str,
+        image: &str,
+        ports: Vec<Port>,
+    ) -> anyhow::Result<()> {
+        tracing::debug!("ensuring that image is running");
+
+        let containers = self
+            .client
+            .list_containers(Some(
+                ListContainersOptionsBuilder::default()
+                    .all(true)
+                    .filters(&HashMap::from([(
+                        "name".to_string(),
+                        vec![name.to_string()],
+                    )]))
+                    .build(),
+            ))
+            .await?;
+        if !containers.is_empty() {
+            // Reconcile difference
+            return Ok(());
+        }
+
+        let _ = self
+            .client
+            .create_image(
+                Some(CreateImageOptionsBuilder::new().from_image(image).build()),
+                None,
+                None,
+            )
+            .try_collect::<Vec<_>>()
+            .await?;
+
+        let ports: HashMap<_, _> = ports
+            .iter()
+            .map(|p| {
+                (
+                    format!("{}/tcp", p.container_port),
+                    Some(vec![PortBinding {
+                        host_ip: Some("0.0.0.0".into()),
+                        host_port: Some(p.host_port.to_string()),
+                    }]),
+                )
+            })
+            .collect();
+        self.client
+            .create_container(
+                Some(CreateContainerOptionsBuilder::new().name(name).build()),
+                ContainerCreateBody {
+                    image: Some(image.into()),
+                    host_config: Some(HostConfig {
+                        port_bindings: Some(ports),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+            )
+            .await?;
+
+        self.client
+            .start_container(name, Some(StartContainerOptionsBuilder::default().build()))
+            .await?;
+
+        Ok(())
+    }
+}
+
+pub trait ContainerRuntimeState {
+    fn container_runtime(&self) -> ContainerRuntime;
+}
+
+impl ContainerRuntimeState for ClientState {
+    fn container_runtime(&self) -> ContainerRuntime {
+        static CLIENT: LazyLock<bollard::Docker> = LazyLock::new(|| {
+            Docker::connect_with_defaults().expect("to be able to connect to a docker daemon")
+        });
+
+        ContainerRuntime {
+            client: CLIENT.clone(),
+        }
+    }
+}

crates/norun/src/grpc_client.rs

@@ -1,4 +1,8 @@
-use norun_grpc_interface::{PublishRequest, registry_service_client::RegistryServiceClient};
+use std::collections::HashMap;
+
+use norun_grpc_interface::{
+    GetTopicRequest, Projects, PublishRequest, registry_service_client::RegistryServiceClient,
+};
 use tokio::sync::OnceCell;
 use tonic::transport::Channel;
 
@@ -26,6 +30,23 @@ impl GrpcClient {
         Ok(())
     }
 
+    pub async fn subscribe(&self, topic: &str) -> anyhow::Result<Projects> {
+        tracing::trace!("calling subscribe via. grpc on registry");
+
+        let mut registry_client = self.get_registry_client().await?;
+
+        let res = registry_client
+            .get_topic(GetTopicRequest {
+                topic: topic.to_string(),
+            })
+            .await?;
+
+        let res = res.into_inner();
+
+        res.projects
+            .ok_or_else(|| anyhow::anyhow!("failed to get projects from server"))
+    }
+
     async fn get_registry_client(&self) -> anyhow::Result<RegistryServiceClient<Channel>> {
         let client = self
             .registry_client
@@ -51,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/grpc_server/registry.rs

@@ -1,6 +1,9 @@
 use norun_grpc_interface::{registry_service_server::RegistryService, *};
 
-use crate::{server::services::registry::RegistryServiceState, state::ServerState};
+use crate::{
+    server::services::registry::{self, RegistryServiceState},
+    state::ServerState,
+};
 
 pub struct GrpcRegistryService {
     pub state: ServerState,
@@ -8,7 +11,7 @@ pub struct GrpcRegistryService {
 
 #[async_trait::async_trait]
 impl RegistryService for GrpcRegistryService {
-    #[tracing::instrument(skip(self), level = "debug")]
+    #[tracing::instrument(skip(self), level = "trace")]
     async fn publish(
         &self,
         request: tonic::Request<PublishRequest>,
@@ -30,4 +33,34 @@ impl RegistryService for GrpcRegistryService {
 
         Ok(tonic::Response::new(PublishResponse {}))
     }
+
+    #[tracing::instrument(skip(self), level = "trace")]
+    async fn get_topic(
+        &self,
+        request: tonic::Request<GetTopicRequest>,
+    ) -> std::result::Result<tonic::Response<GetTopicResponse>, tonic::Status> {
+        tracing::debug!("subscribe called");
+
+        let req = request.into_inner();
+
+        let projects = self
+            .state
+            .registry_service()
+            .get_topic(&req.topic)
+            .await
+            .inspect_err(|e| tracing::warn!("failed to subscribe on topic: {}", e))
+            .map_err(|e| tonic::Status::internal(e.to_string()))?;
+
+        Ok(tonic::Response::new(GetTopicResponse {
+            projects: Some(projects.into()),
+        }))
+    }
+}
+
+impl From<registry::Projects> for Projects {
+    fn from(value: registry::Projects) -> Self {
+        Self {
+            projects: value.projects.into_iter().collect(),
+        }
+    }
 }

crates/norun/src/main.rs

@@ -7,9 +7,15 @@ mod state;
 
 mod server;
 
+mod services;
+
 mod grpc_client;
 mod grpc_server;
 
+mod container_runtime;
+
+mod node;
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt()

crates/norun/src/models.rs

@@ -1,2 +1,24 @@
 pub mod project_tag;
 pub use project_tag::*;
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+pub mod port {
+    #[derive(Clone, Debug, PartialEq)]
+    pub struct Port {
+        pub host_port: usize,
+        pub container_port: usize,
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
+pub struct Project {
+    pub id: Uuid,
+    pub spec: ProjectSpec,
+}
+
+#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
+pub enum ProjectSpec {
+    Compose {},
+    Container {},
+}

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) }),
         };
 

crates/norun/src/server/services/registry.rs

@@ -1,7 +1,4 @@
-use std::{
+use std::sync::{Arc, LazyLock};
-    collections::BTreeMap,
-    sync::{Arc, LazyLock},
-};
 
 use norun_grpc_interface::Project;
 use tokio::sync::Mutex;
@@ -25,6 +22,25 @@ impl RegistryService {
 
         Ok(())
     }
+
+    pub async fn get_topic(&self, topic: &str) -> anyhow::Result<Projects> {
+        tracing::debug!("get projects for topic");
+
+        let projects = {
+            let store = self.store.lock().await;
+            store
+                .iter()
+                .filter(|i| i.name == topic)
+                .cloned()
+                .collect::<Vec<_>>()
+        };
+
+        Ok(Projects { projects })
+    }
+}
+
+pub struct Projects {
+    pub projects: Vec<Project>,
 }
 
 pub trait RegistryServiceState {

crates/norun/src/services.rs

@@ -0,0 +1 @@
+pub mod project_registry;

crates/norun/src/services/project_registry.rs

@@ -0,0 +1,104 @@
+use std::path::PathBuf;
+
+use anyhow::Context;
+use tokio::io::AsyncWriteExt;
+
+use crate::{models::Project, state::ClientState};
+
+pub struct ProjectRegistry {
+    state_dir: PathBuf,
+}
+
+impl ProjectRegistry {
+    pub async fn get_project(&self, project: &Project) -> anyhow::Result<Option<Project>> {
+        let project_dir = self.project_file(project)?;
+
+        if !project_dir.exists() {
+            return Ok(None);
+        }
+
+        let project_content = tokio::fs::read_to_string(&project_dir)
+            .await
+            .context("failed to read ron file")?;
+
+        let project: Project = ron::from_str(&project_content)
+            .context(format!("failed to read: '{}'", project_dir.display()))?;
+
+        Ok(Some(project))
+    }
+
+    pub async fn create_project(&self, project: &Project) -> anyhow::Result<()> {
+        match self.get_project(project).await {
+            Ok(_) => anyhow::bail!("project already exists"),
+            Err(_) => {
+                // continue
+            }
+        }
+
+        let project_file_path = self.project_file(project)?;
+        if let Some(project_file) = project_file_path.parent() {
+            tokio::fs::create_dir_all(project_file)
+                .await
+                .context("create ron project dir")?;
+        }
+
+        let mut project_file = tokio::fs::File::create_new(&project_file_path)
+            .await
+            .context("create project file")?;
+
+        let project_content = ron::to_string(project)?;
+
+        project_file
+            .write_all(&project_content.as_bytes())
+            .await
+            .context("write project file")?;
+
+        Ok(())
+    }
+
+    pub async fn update_project(&self, project: &Project) -> anyhow::Result<()> {
+        let project_file_path = self.project_file(project)?;
+        if let Some(project_file) = project_file_path.parent() {
+            tokio::fs::create_dir_all(project_file)
+                .await
+                .context("update ron project dir")?;
+        }
+
+        let mut project_file = tokio::fs::File::create_new(&project_file_path)
+            .await
+            .context("update project file")?;
+
+        let project_content = ron::to_string(project)?;
+
+        project_file
+            .write_all(project_content.as_bytes())
+            .await
+            .context("update project file")?;
+
+        Ok(())
+    }
+
+    fn project_file(&self, project: &Project) -> anyhow::Result<PathBuf> {
+        let project_dir = self
+            .state_dir
+            .join(project.id.to_string())
+            .join("project.ron");
+
+        Ok(project_dir)
+    }
+}
+
+pub trait ProjectRegistryState {
+    fn project_registry(&self) -> ProjectRegistry;
+}
+
+impl ProjectRegistryState for ClientState {
+    fn project_registry(&self) -> ProjectRegistry {
+        ProjectRegistry {
+            state_dir: dirs::state_dir()
+                .expect("to be able to find state")
+                .join("norun")
+                .join("projects"),
+        }
+    }
+}

examples/hello-world/norun.toml

@@ -2,7 +2,7 @@
 name = "hello-world"
 
 [container]
-image = "kasperhermansen/hello-world"
+image = "library/hello-world"
 version = "latest"
 
 [expose]

examples/nginx/norun.toml

@@ -0,0 +1,9 @@
+[project]
+name = "nginx"
+
+[container]
+image = "library/nginx"
+version = "latest"
+
+[expose]
+port = 80

interface/proto/norun/v1/registry.proto

@@ -2,20 +2,44 @@ syntax = "proto3";
 
 package norun.v1;
 
-service RegistryService {
-  rpc Publish(PublishRequest) returns (PublishResponse) {}
-}
 
 message PublishRequest {
   Project project = 1;
 }
 message PublishResponse {}
 
+message GetTopicRequest {
+  string topic = 1;
+}
+message GetTopicResponse {
+  Projects projects = 1;
+}
+
+
+service RegistryService {
+  rpc Publish(PublishRequest) returns (PublishResponse) {}
+  rpc GetTopic(GetTopicRequest) returns (GetTopicResponse) {}
+}
+
+message Projects {
+  repeated Project projects = 1;
+}
+
 message Project {
   string name = 1;
+  optional uint32 port = 2;
 
-  string image = 2;
+  oneof project_type {
-  string version = 3;
+    Container container = 3;
-
+    Compose compose = 4;
-  optional uint32 port = 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"