more mess

bin/nanocld/specs/swagger.yaml

@@ -1298,13 +1298,13 @@ paths:
     put:
       tags:
       - Resources
-      summary: Patch a resource (update its version and/or spec) and create a new history
-      description: Patch a resource (update its version and/or spec) and create a new history
+      summary: Create a new resource spec and add history entry
+      description: Create a new resource spec and add history entry
       operationId: put_resource
       parameters:
       - name: name
         in: path
-        description: The resource name to patch
+        description: Name of the resource
         required: true
         schema:
           type: string
@@ -1316,13 +1316,13 @@ paths:
         required: true
       responses:
         '200':
-          description: The patched resource
+          description: Resource updated
           content:
             application/json:
               schema:
                 $ref: '#/components/schemas/Resource'
         '404':
-          description: Resource is not existing
+          description: Resource does not exit
           content:
             application/json:
               schema:

bin/nanocld/src/models/system.rs

@@ -9,7 +9,7 @@ use futures_util::{SinkExt, StreamExt};
 use nanocl_error::io::{IoResult, FromIo, IoError};
 use nanocl_stubs::{
   config::DaemonConfig,
-  system::{Event, EventPartial},
+  system::{Event, EventPartial, NativeEventAction, EventActor, EventKind},
 };
 
 use crate::{version, utils, repositories::generic::*};
@@ -206,4 +206,26 @@ impl SystemState {
   pub fn subscribe_raw(&self) -> IoResult<RawEventClient> {
     self.event_manager.raw.subscribe()
   }
+
+  pub fn emit_normal_native_action<A>(
+    &self,
+    actor: &A,
+    action: NativeEventAction,
+  ) where
+    A: Into<EventActor> + Clone,
+  {
+    let actor = actor.clone().into();
+    let event = EventPartial {
+      reporting_controller: "nanocl.io/core".to_owned(),
+      reporting_node: self.config.hostname.clone(),
+      kind: EventKind::Normal,
+      action: action.to_string(),
+      related: None,
+      reason: "state_sync".to_owned(),
+      note: None,
+      metadata: None,
+      actor: Some(actor),
+    };
+    self.spawn_emit_event(event);
+  }
 }

bin/nanocld/src/models/vm.rs

@@ -1,4 +1,5 @@
 use diesel::prelude::*;
+use nanocl_stubs::vm_spec::VmSpecPartial;
 
 use crate::schema::vms;
 
@@ -39,3 +40,9 @@ pub struct VmUpdateDb {
   /// The spec key reference
   pub spec_key: Option<uuid::Uuid>,
 }
+
+pub struct VmObjCreateIn {
+  pub namespace: String,
+  pub spec: VmSpecPartial,
+  pub version: String,
+}

bin/nanocld/src/objects/cargo.rs

@@ -1,9 +1,16 @@
-use nanocl_error::http::HttpResult;
-use nanocl_stubs::cargo::{Cargo, CargoDeleteQuery};
+use futures_util::{stream::FuturesUnordered, StreamExt};
+use bollard_next::container::RemoveContainerOptions;
+
+use nanocl_error::http::{HttpResult, HttpError};
+use nanocl_stubs::{
+  cargo::{Cargo, CargoDeleteQuery},
+  cargo_spec::ReplicationMode,
+};
 
 use crate::{
   utils,
-  models::{CargoDb, SystemState, CargoObjCreateIn},
+  repositories::generic::*,
+  models::{CargoDb, SystemState, CargoObjCreateIn, ProcessDb, SpecDb},
 };
 
 use super::generic::*;
@@ -16,9 +23,32 @@ impl ObjCreate for CargoDb {
     obj: &Self::ObjCreateIn,
     state: &SystemState,
   ) -> HttpResult<Self::ObjCreateOut> {
-    let cargo =
-      utils::cargo::create(&obj.namespace, &obj.spec, &obj.version, state)
-        .await?;
+    let cargo = CargoDb::create_from_spec(
+      &obj.namespace,
+      &obj.spec,
+      &obj.version,
+      &state.pool,
+    )
+    .await?;
+    let number = if let Some(mode) = &cargo.spec.replication {
+      match mode {
+        ReplicationMode::Static(replication_static) => {
+          replication_static.number
+        }
+        ReplicationMode::Auto => 1,
+        ReplicationMode::Unique => 1,
+        ReplicationMode::UniqueByNode => 1,
+        _ => 1,
+      }
+    } else {
+      1
+    };
+    if let Err(err) =
+      utils::cargo::create_instances(&cargo, number, state).await
+    {
+      CargoDb::del_by_pk(&cargo.spec.cargo_key, &state.pool).await?;
+      return Err(err);
+    }
     Ok(cargo)
   }
 }
@@ -32,7 +62,29 @@ impl ObjDelByPk for CargoDb {
     opts: &Self::ObjDelOpts,
     state: &SystemState,
   ) -> HttpResult<Self::ObjDelOut> {
-    let cargo = utils::cargo::delete_by_key(key, opts.force, state).await?;
+    let cargo = CargoDb::transform_read_by_pk(key, &state.pool).await?;
+    let processes =
+      ProcessDb::read_by_kind_key(&cargo.spec.cargo_key, &state.pool).await?;
+    processes
+      .into_iter()
+      .map(|process| async move {
+        utils::process::remove(
+          &process.key,
+          Some(RemoveContainerOptions {
+            force: opts.force.unwrap_or(false),
+            ..Default::default()
+          }),
+          state,
+        )
+        .await
+      })
+      .collect::<FuturesUnordered<_>>()
+      .collect::<Vec<Result<(), HttpError>>>()
+      .await
+      .into_iter()
+      .collect::<Result<Vec<_>, _>>()?;
+    CargoDb::del_by_pk(key, &state.pool).await?;
+    SpecDb::del_by_kind_key(key, &state.pool).await?;
     Ok(cargo)
   }
 }

bin/nanocld/src/objects/generic/create.rs

@@ -1,8 +1,6 @@
 use nanocl_error::http::HttpResult;
 use nanocl_stubs::system::{EventActor, NativeEventAction};
 
-use crate::utils;
-
 use crate::models::SystemState;
 
 /// A Create trait for all objects in Nanocl
@@ -28,11 +26,7 @@ pub trait ObjCreate {
     Self::ObjCreateOut: Into<EventActor> + Clone,
   {
     let obj = Self::fn_create_obj(obj, state).await?;
-    utils::event_emitter::emit_normal_native_action(
-      &obj,
-      NativeEventAction::Create,
-      state,
-    );
+    state.emit_normal_native_action(&obj, NativeEventAction::Create);
     Ok(obj)
   }
 }

bin/nanocld/src/objects/generic/delete.rs

@@ -1,8 +1,6 @@
 use nanocl_error::http::HttpResult;
 use nanocl_stubs::system::{EventActor, NativeEventAction};
 
-use crate::utils;
-
 use crate::models::SystemState;
 
 pub trait ObjDelByPk {
@@ -24,11 +22,7 @@ pub trait ObjDelByPk {
     Self::ObjDelOut: Into<EventActor> + Clone,
   {
     let obj = Self::fn_del_obj_by_pk(key, opts, state).await?;
-    utils::event_emitter::emit_normal_native_action(
-      &obj,
-      NativeEventAction::Delete,
-      state,
-    );
+    state.emit_normal_native_action(&obj, NativeEventAction::Delete);
     Ok(obj)
   }
 }

bin/nanocld/src/objects/generic/patch.rs

@@ -1,8 +1,6 @@
 use nanocl_error::http::HttpResult;
 use nanocl_stubs::system::{EventActor, NativeEventAction};
 
-use crate::utils;
-
 use crate::models::SystemState;
 
 pub trait ObjPatchByPk {
@@ -24,11 +22,7 @@ pub trait ObjPatchByPk {
     Self::ObjPatchOut: Into<EventActor> + Clone,
   {
     let obj = Self::fn_patch_obj_by_pk(pk, obj, state).await?;
-    utils::event_emitter::emit_normal_native_action(
-      &obj,
-      NativeEventAction::Patch,
-      state,
-    );
+    state.emit_normal_native_action(&obj, NativeEventAction::Update);
     Ok(obj)
   }
 }

bin/nanocld/src/objects/job.rs

@@ -0,0 +1,91 @@
+use bollard_next::container::RemoveContainerOptions;
+use futures_util::{StreamExt, stream::FuturesUnordered};
+
+use nanocl_error::http::{HttpResult, HttpError};
+use nanocl_stubs::job::{Job, JobPartial};
+
+use crate::{
+  utils,
+  repositories::generic::*,
+  models::{JobDb, ProcessDb},
+};
+
+use super::generic::*;
+
+impl ObjCreate for JobDb {
+  type ObjCreateIn = JobPartial;
+  type ObjCreateOut = Job;
+
+  async fn fn_create_obj(
+    obj: &Self::ObjCreateIn,
+    state: &crate::models::SystemState,
+  ) -> HttpResult<Self::ObjCreateOut> {
+    let db_model = JobDb::try_from_partial(obj)?;
+    let job = JobDb::create_from(db_model, &state.pool)
+      .await?
+      .to_spec(obj);
+    job
+      .containers
+      .iter()
+      .map(|container| {
+        let job_name = job.name.clone();
+        async move {
+          let mut container = container.clone();
+          let mut labels = container.labels.clone().unwrap_or_default();
+          labels.insert("io.nanocl.j".to_owned(), job_name.clone());
+          container.labels = Some(labels);
+          let short_id = utils::key::generate_short_id(6);
+          let name = format!("{job_name}-{short_id}.j");
+          utils::process::create(&name, "job", &job_name, container, state)
+            .await?;
+          Ok::<_, HttpError>(())
+        }
+      })
+      .collect::<FuturesUnordered<_>>()
+      .collect::<Vec<Result<(), HttpError>>>()
+      .await
+      .into_iter()
+      .collect::<Result<Vec<_>, _>>()?;
+    if let Some(schedule) = &job.schedule {
+      utils::job::add_cron_rule(&job, schedule, state).await?;
+    }
+    Ok(job)
+  }
+}
+
+impl ObjDelByPk for JobDb {
+  type ObjDelOpts = ();
+  type ObjDelOut = Job;
+
+  async fn fn_del_obj_by_pk(
+    pk: &str,
+    _opts: &Self::ObjDelOpts,
+    state: &crate::models::SystemState,
+  ) -> HttpResult<Self::ObjDelOut> {
+    let job = JobDb::read_by_pk(pk, &state.pool).await?.try_to_spec()?;
+    let processes = ProcessDb::read_by_kind_key(pk, &state.pool).await?;
+    processes
+      .into_iter()
+      .map(|process| async move {
+        utils::process::remove(
+          &process.key,
+          Some(RemoveContainerOptions {
+            force: true,
+            ..Default::default()
+          }),
+          state,
+        )
+        .await
+      })
+      .collect::<FuturesUnordered<_>>()
+      .collect::<Vec<Result<(), HttpError>>>()
+      .await
+      .into_iter()
+      .collect::<Result<Vec<_>, _>>()?;
+    JobDb::del_by_pk(&job.name, &state.pool).await?;
+    if job.schedule.is_some() {
+      utils::job::remove_cron_rule(&job, state).await?;
+    }
+    Ok(job)
+  }
+}

bin/nanocld/src/objects/mod.rs

@@ -1,4 +1,6 @@
+mod job;
 mod cargo;
+mod vm;
 mod secret;
 mod resource;
 

bin/nanocld/src/objects/resource.rs

@@ -26,7 +26,7 @@ impl ObjCreate for ResourceDb {
         &obj.name
       )));
     }
-    let obj = utils::resource::hook_create_resource(obj, &state.pool).await?;
+    let obj = utils::resource::hook_create(obj, &state.pool).await?;
     let resource = ResourceDb::create_from_spec(&obj, &state.pool).await?;
     Ok(resource)
   }
@@ -42,8 +42,7 @@ impl ObjDelByPk for ResourceDb {
     state: &SystemState,
   ) -> HttpResult<Self::ObjDelOut> {
     let resource = ResourceDb::transform_read_by_pk(key, &state.pool).await?;
-    if let Err(err) =
-      utils::resource::hook_delete_resource(&resource, &state.pool).await
+    if let Err(err) = utils::resource::hook_delete(&resource, &state.pool).await
     {
       log::warn!("{err}");
     }