From e44a67f5a82a2fd7383693a68f6b9a324f31a028 Mon Sep 17 00:00:00 2001 From: GW_MC <72297530+GWMCwing@users.noreply.github.com> Date: Sun, 5 Jul 2026 05:29:13 +0000 Subject: [PATCH] feat(agent): implement AgentService with CRUD and gRPC stream handler - Add AgentService trait with full CRUD operations - Implement AgentServiceImpl backed by AgentRepo - Refactor gRPC AgentServerService with real stream handler - Add agent types and repository module --- apps/nxmesh-master/src/service/agent/mod.rs | 118 ++++++++++++-- apps/nxmesh-master/src/service/agent/repo.rs | 146 ++++++++++++++++++ apps/nxmesh-master/src/service/agent/types.rs | 60 +++++++ 3 files changed, 312 insertions(+), 12 deletions(-) create mode 100644 apps/nxmesh-master/src/service/agent/repo.rs create mode 100644 apps/nxmesh-master/src/service/agent/types.rs diff --git a/apps/nxmesh-master/src/service/agent/mod.rs b/apps/nxmesh-master/src/service/agent/mod.rs index e8f329e..eb4e489 100644 --- a/apps/nxmesh-master/src/service/agent/mod.rs +++ b/apps/nxmesh-master/src/service/agent/mod.rs @@ -1,26 +1,120 @@ -use nxmesh_proto::{AgentMessage, MasterMessage, agent_service_server::AgentService}; +use chrono::Utc; +use nxmesh_proto::{ + AgentMessage, MasterMessage, agent_service_server::AgentService as GrpcAgentService, +}; +use sea_orm::DatabaseConnection; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; +use tracing::{error, info}; +use uuid::Uuid; -pub mod repo; +mod repo; +pub mod types; + +pub use types::{AgentRecord, CreateAgentRecord, State, UpdateAgentRecord}; + +use crate::service::error::RepoError; +use repo::{AgentRepo, AgentRepoImpl}; + +#[cfg_attr(test, mockall::automock)] +#[async_trait::async_trait] +pub trait AgentService: Send + Sync + 'static { + async fn list(&self) -> Result, RepoError>; + async fn get(&self, id: Uuid) -> Result, RepoError>; + async fn create(&self, rec: &CreateAgentRecord) -> Result; + async fn update( + &self, + id: Uuid, + rec: &UpdateAgentRecord, + ) -> Result, RepoError>; + async fn delete(&self, id: Uuid) -> Result; +} + +pub struct AgentServiceImpl { + repo: Box, +} + +impl AgentServiceImpl { + pub fn new(db: DatabaseConnection) -> Self { + Self { + repo: Box::new(AgentRepoImpl::new(db)), + } + } +} + +#[async_trait::async_trait] +impl AgentService for AgentServiceImpl { + async fn list(&self) -> Result, RepoError> { + self.repo.list().await + } + + async fn get(&self, id: Uuid) -> Result, RepoError> { + self.repo.get(id).await + } + + async fn create(&self, rec: &CreateAgentRecord) -> Result { + self.repo.create(rec).await + } + + async fn update( + &self, + id: Uuid, + rec: &UpdateAgentRecord, + ) -> Result, RepoError> { + self.repo.update(id, rec).await + } + + async fn delete(&self, id: Uuid) -> Result { + self.repo.delete(id).await + } +} #[derive(Debug, Default)] pub struct AgentServerService {} #[async_trait::async_trait] -impl AgentService for AgentServerService { - #[doc = " Server streaming response type for the Stream method."] - type StreamStream = tonic::codec::Streaming; +impl GrpcAgentService for AgentServerService { + type StreamStream = + tokio_stream::wrappers::ReceiverStream>; - #[doc = " Stream establishes a persistent connection for real-time communication"] - #[allow( - mismatched_lifetime_syntaxes, - clippy::type_complexity, - clippy::type_repetition_in_bounds - )] async fn stream( &self, request: tonic::Request>, ) -> Result, tonic::Status> { - todo!() + let mut inbound = request.into_inner(); + + let (tx, rx) = mpsc::channel::>(32); + let outbound = ReceiverStream::new(rx); + + tokio::spawn(async move { + loop { + match inbound.message().await { + Ok(Some(msg)) => { + info!("Received AgentMessage: {:?}", msg); + + let ack = MasterMessage { + timestamp: Utc::now().timestamp_millis(), + message_id: Uuid::new_v4().to_string(), + payload: None, + }; + if let Err(e) = tx.send(Ok(ack)).await { + error!("Failed to send MasterMessage ack: {:?}", e); + return; + } + } + Ok(None) => { + info!("Agent closed the outbound stream"); + return; + } + Err(e) => { + error!("Error receiving AgentMessage: {:?}", e); + return; + } + } + } + }); + + Ok(tonic::Response::new(outbound)) } async fn connection_test( diff --git a/apps/nxmesh-master/src/service/agent/repo.rs b/apps/nxmesh-master/src/service/agent/repo.rs new file mode 100644 index 0000000..e433a16 --- /dev/null +++ b/apps/nxmesh-master/src/service/agent/repo.rs @@ -0,0 +1,146 @@ +use sea_orm::{ActiveModelTrait, ActiveValue::Set, DatabaseConnection, EntityTrait}; +use uuid::Uuid; + +use super::types::{AgentRecord, CreateAgentRecord, State, UpdateAgentRecord}; +use crate::{ + db::entities::agents::{ActiveModel as AgentActiveModel, Entity as Agent}, + service::error::RepoError, +}; + +#[async_trait::async_trait] +pub(crate) trait AgentRepo: Send + Sync + 'static { + async fn list(&self) -> Result, RepoError>; + async fn get(&self, id: Uuid) -> Result, RepoError>; + async fn create(&self, rec: &CreateAgentRecord) -> Result; + async fn update( + &self, + id: Uuid, + rec: &UpdateAgentRecord, + ) -> Result, RepoError>; + async fn delete(&self, id: Uuid) -> Result; +} + +pub(crate) struct AgentRepoImpl { + db: DatabaseConnection, +} + +impl AgentRepoImpl { + pub fn new(db: DatabaseConnection) -> Self { + Self { db } + } +} + +#[async_trait::async_trait] +impl AgentRepo for AgentRepoImpl { + async fn list(&self) -> Result, RepoError> { + let agents = Agent::find().all(&self.db).await?; + Ok(agents + .into_iter() + .map(|m| AgentRecord { + id: m.id, + name: m.name, + ip_address: m.ip_address, + state: m.state.into(), + deployment_mode: m.deployment_mode, + last_seen_at: m.last_seen_at.map(|dt| dt.to_string()), + labels: m.labels, + created_at: m.created_at.to_string(), + updated_at: m.updated_at.to_string(), + }) + .collect()) + } + + async fn get(&self, id: Uuid) -> Result, RepoError> { + let agent = Agent::find_by_id(id).one(&self.db).await?; + Ok(agent.map(|m| AgentRecord { + id: m.id, + name: m.name, + ip_address: m.ip_address, + state: m.state.into(), + deployment_mode: m.deployment_mode, + last_seen_at: m.last_seen_at.map(|dt| dt.to_string()), + labels: m.labels, + created_at: m.created_at.to_string(), + updated_at: m.updated_at.to_string(), + })) + } + + async fn create(&self, rec: &CreateAgentRecord) -> Result { + let new_agent = AgentActiveModel { + id: Set(Uuid::new_v4()), + name: Set(rec.name.clone()), + ip_address: Set(rec.ip_address.clone()), + state: Set(State::Active.into()), + deployment_mode: Set(None), + last_seen_at: Set(None), + labels: Set(None), + created_at: Set(chrono::Utc::now().into()), + updated_at: Set(chrono::Utc::now().into()), + ..Default::default() + }; + + let agent = new_agent.insert(&self.db).await?; + + Ok(AgentRecord { + id: agent.id, + name: agent.name, + ip_address: agent.ip_address, + state: agent.state.into(), + deployment_mode: agent.deployment_mode, + last_seen_at: agent.last_seen_at.map(|dt| dt.to_string()), + labels: agent.labels, + created_at: agent.created_at.to_string(), + updated_at: agent.updated_at.to_string(), + }) + } + + async fn update( + &self, + id: Uuid, + rec: &UpdateAgentRecord, + ) -> Result, RepoError> { + let existing = match Agent::find_by_id(id).one(&self.db).await? { + Some(agent) => agent, + None => return Ok(None), + }; + + let mut agent: AgentActiveModel = AgentActiveModel::from(existing); + + if let Some(name) = &rec.name { + agent.name = Set(name.clone()); + } + if let Some(ip_address) = &rec.ip_address { + agent.ip_address = Set(Some(ip_address.clone())); + } + if let Some(state) = &rec.state { + agent.state = Set(String::from(*state)); + } + if let Some(deployment_mode) = &rec.deployment_mode { + agent.deployment_mode = Set(Some(deployment_mode.clone())); + } + if let Some(labels) = &rec.labels { + agent.labels = Set(Some(labels.clone())); + } + + agent.updated_at = Set(chrono::Utc::now().into()); + + let updated = agent.update(&self.db).await?; + + Ok(Some(AgentRecord { + id: updated.id, + name: updated.name, + ip_address: updated.ip_address, + state: updated.state.into(), + deployment_mode: updated.deployment_mode, + last_seen_at: updated.last_seen_at.map(|dt| dt.to_string()), + labels: updated.labels, + created_at: updated.created_at.to_string(), + updated_at: updated.updated_at.to_string(), + })) + } + + async fn delete(&self, id: Uuid) -> Result { + let result = Agent::delete_by_id(id).exec(&self.db).await?; + Ok(result.rows_affected > 0) + } +} diff --git a/apps/nxmesh-master/src/service/agent/types.rs b/apps/nxmesh-master/src/service/agent/types.rs new file mode 100644 index 0000000..c7507e3 --- /dev/null +++ b/apps/nxmesh-master/src/service/agent/types.rs @@ -0,0 +1,60 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub enum State { + Active, + Inactive, + Unreachable, + Unknown, + Disabled, +} + +impl From for String { + fn from(state: State) -> Self { + match state { + State::Active => "active".to_string(), + State::Inactive => "inactive".to_string(), + State::Unreachable => "unreachable".to_string(), + State::Unknown => "unknown".to_string(), + State::Disabled => "disabled".to_string(), + } + } +} + +impl From for State { + fn from(s: String) -> Self { + match s.as_str() { + "active" => State::Active, + "inactive" => State::Inactive, + "unreachable" => State::Unreachable, + "unknown" => State::Unknown, + "disabled" => State::Disabled, + _ => State::Inactive, + } + } +} + +pub struct AgentRecord { + pub id: uuid::Uuid, + pub name: String, + pub ip_address: Option, + pub state: State, + pub deployment_mode: Option, + pub last_seen_at: Option, + pub labels: Option, + pub created_at: String, + pub updated_at: String, +} + +pub struct CreateAgentRecord { + pub name: String, + pub ip_address: Option, +} + +pub struct UpdateAgentRecord { + pub name: Option, + pub ip_address: Option, + pub state: Option, + pub deployment_mode: Option, + pub labels: Option, +}