From 9aca4cadf03288015363c9d4d3ea1acb26867cc0 Mon Sep 17 00:00:00 2001 From: pavel Date: Tue, 17 Feb 2026 19:07:06 +0100 Subject: [PATCH] stuff --- frontend/vite.config.js | 2 +- src/config.rs | 5 + src/domain/agent/mod.rs | 33 ++++- src/domain/agent/tools.rs | 72 +++++++++ src/domain/auth.rs | 23 +++ src/domain/calendar/mod.rs | 292 +++++++++++++++++++++++++++++++++++++ src/domain/mod.rs | 1 + src/domain/tasks.rs | 3 + src/error.rs | 16 +- src/scheduler.rs | 12 +- src/server/chat.rs | 3 + src/server/mod.rs | 41 +++++- src/server/tasks.rs | 1 + 13 files changed, 483 insertions(+), 21 deletions(-) create mode 100644 src/domain/calendar/mod.rs diff --git a/frontend/vite.config.js b/frontend/vite.config.js index a71c6f4..a06ef36 100644 --- a/frontend/vite.config.js +++ b/frontend/vite.config.js @@ -4,7 +4,7 @@ export default defineConfig({ server: { proxy: { '/api': { - target: 'http://localhost:3000', + target: 'http://localhost:3001', changeOrigin: true, ws: true, } diff --git a/src/config.rs b/src/config.rs index b69a4c0..b81a727 100644 --- a/src/config.rs +++ b/src/config.rs @@ -15,6 +15,7 @@ pub struct Config { pub agent_max_turns: u32, pub agent_max_duration_secs: u64, pub vapid_private_key: String, + pub calendar_api_url: String, } impl Config { @@ -58,6 +59,9 @@ impl Config { let vapid_private_key = env::var("VAPID_PRIVATE_KEY") .map_err(|_| AppError::Config("VAPID_PRIVATE_KEY must be set".into()))?; + let calendar_api_url = + env::var("CALENDAR_API_URL").unwrap_or_else(|_| "http://localhost:8000".to_string()); + Ok(Config { database_url, port, @@ -71,6 +75,7 @@ impl Config { agent_max_turns, agent_max_duration_secs, vapid_private_key, + calendar_api_url, }) } } diff --git a/src/domain/agent/mod.rs b/src/domain/agent/mod.rs index 1b6dde8..95f1c5e 100644 --- a/src/domain/agent/mod.rs +++ b/src/domain/agent/mod.rs @@ -3,6 +3,7 @@ pub mod tools; use chrono::Utc; use sea_orm::DatabaseConnection; +use std::sync::Arc; use std::time::{Duration, Instant}; use self::api::{ChatRequest, ChatResponse, Message, Tool}; @@ -13,6 +14,8 @@ pub struct Agent { url: String, zen_api_key: Option, tavily_api_key: Option, + calendar_client: Arc, + pub user_sub: Option, pub messages: Vec, tools: Option>, logs: String, @@ -26,6 +29,8 @@ impl Agent { db: DatabaseConnection, zen_api_key: Option, tavily_api_key: Option, + calendar_client: Arc, + user_sub: Option, initial_message: String, ) -> AppResult { let intro = format!( @@ -50,13 +55,22 @@ impl Agent { }, ]; - Self::with_messages(db, zen_api_key, tavily_api_key, messages) + Self::with_messages( + db, + zen_api_key, + tavily_api_key, + calendar_client, + user_sub, + messages, + ) } pub fn with_messages( db: DatabaseConnection, zen_api_key: Option, tavily_api_key: Option, + calendar_client: Arc, + user_sub: Option, messages: Vec, ) -> AppResult { let tools = Some(tools::get_tools()); @@ -72,6 +86,8 @@ impl Agent { url: "https://opencode.ai/zen/v1/chat/completions".to_string(), zen_api_key, tavily_api_key, + calendar_client, + user_sub, messages, tools, logs: String::new(), @@ -166,12 +182,15 @@ impl Agent { for tool_call in tool_calls { self.log(&format!("Calling tool: {}", tool_call.function.name)); - let (tool_message, is_final, tool_answer) = - tools::handle_tool_call(tool_call, &self.tavily_api_key, &self.db) - .await - .map_err(|e| { - AppError::Internal(format!("Tool execution failed: {}", e)) - })?; + let (tool_message, is_final, tool_answer) = tools::handle_tool_call( + tool_call, + &self.tavily_api_key, + &self.db, + &self.calendar_client, + self.user_sub.as_deref(), + ) + .await + .map_err(|e| AppError::Internal(format!("Tool execution failed: {}", e)))?; if let Some(ans) = tool_answer { self.answer = Some(ans.clone()); diff --git a/src/domain/agent/tools.rs b/src/domain/agent/tools.rs index bbdff4c..2b257c0 100644 --- a/src/domain/agent/tools.rs +++ b/src/domain/agent/tools.rs @@ -3,6 +3,7 @@ use sea_orm::{ ColumnTrait, Condition, DatabaseConnection, EntityTrait, QueryFilter, QueryOrder, QuerySelect, }; use serde::Deserialize; +use std::sync::Arc; use uuid::Uuid; #[derive(Deserialize)] @@ -96,6 +97,47 @@ pub fn get_tools() -> Vec { }), }, }, + Tool { + tool_type: "function".to_string(), + function: FunctionDefinition { + name: "calendar_list_events".to_string(), + description: "List calendar events".to_string(), + parameters: serde_json::json!({ + "type": "object", + "properties": { + "upcoming": { + "type": "boolean", + "description": "If true, only upcoming events will be listed" + } + } + }), + }, + }, + Tool { + tool_type: "function".to_string(), + function: FunctionDefinition { + name: "calendar_create_event".to_string(), + description: "Create a new calendar event".to_string(), + parameters: serde_json::json!({ + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Name of the event" + }, + "from": { + "type": "string", + "description": "Start time in ISO 8601 format (e.g., 2023-10-27T10:00:00Z)" + }, + "to": { + "type": "string", + "description": "End time in ISO 8601 format (e.g., 2023-10-27T11:00:00Z)" + } + }, + "required": ["name", "from", "to"] + }), + }, + }, ] } @@ -103,6 +145,8 @@ pub async fn handle_tool_call( tool_call: &ToolCall, tavily_api_key: &Option, db: &DatabaseConnection, + calendar: &Arc, + user_sub: Option<&str>, ) -> Result<(Message, bool, Option), Box> { let mut answer = None; let name = &tool_call.function.name; @@ -198,6 +242,34 @@ pub async fn handle_tool_call( )); } (out, false) + } else if name == "calendar_list_events" { + let args: serde_json::Value = serde_json::from_str(&tool_call.function.arguments)?; + let upcoming = args["upcoming"].as_bool(); + match calendar + .list_events(user_sub.map(|s| s.to_string()), upcoming) + .await + { + Ok(events) => { + tracing::info!("{:#?}", events); + (serde_json::to_string(&events)?, false) + } + Err(e) => (format!("Error listing events: {}", e), false), + } + } else if name == "calendar_create_event" { + let args: serde_json::Value = serde_json::from_str(&tool_call.function.arguments)?; + let name_val = args["name"].as_str().unwrap_or_default(); + let from_val = args["from"].as_str().unwrap_or_default(); + let to_val = args["to"].as_str().unwrap_or_default(); + match calendar + .create_event(user_sub.map(|s| s.to_string()), name_val, from_val, to_val) + .await + { + Ok(event) => ( + format!("Event created: {}", serde_json::to_string(&event)?), + false, + ), + Err(e) => (format!("Error creating event: {}", e), false), + } } else { (format!("Error: Unknown tool {}", name), false) }; diff --git a/src/domain/auth.rs b/src/domain/auth.rs index c70e0c7..10a1255 100644 --- a/src/domain/auth.rs +++ b/src/domain/auth.rs @@ -190,4 +190,27 @@ impl Authenticator { Ok(res) } + + pub async fn client_credentials( + &self, + scope: &str, + ) -> Result> { + let params = [ + ("grant_type", "client_credentials"), + ("client_id", &self.client_id), + ("client_secret", &self.client_secret), + ("scope", scope), + ]; + + let res = self + .client + .post(&self.token_url) + .form(¶ms) + .send() + .await? + .json() + .await?; + + Ok(res) + } } diff --git a/src/domain/calendar/mod.rs b/src/domain/calendar/mod.rs new file mode 100644 index 0000000..d6176d1 --- /dev/null +++ b/src/domain/calendar/mod.rs @@ -0,0 +1,292 @@ +use crate::domain::auth::Authenticator; +use crate::error::{AppError, AppResult}; +use chrono::{DateTime, Duration, Utc}; +use reqwest::Client; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; +use tokio::sync::RwLock; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CalendarEvent { + pub id: Option, + pub name: String, + pub from: DateTime, + pub to: DateTime, + pub user_sub: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CreateEventRequest { + pub name: String, + pub from: String, + pub to: String, + pub user_sub: Option, +} +struct TokenState { + access_token: String, + expires_at: DateTime, +} + +pub struct CalendarClient { + base_url: String, + client: Client, + authenticator: Arc, + token_state: RwLock>, +} + +impl CalendarClient { + pub fn new(base_url: String, authenticator: Arc) -> Self { + Self { + base_url: base_url.trim_end_matches('/').to_string(), + client: Client::new(), + authenticator, + token_state: RwLock::new(None), + } + } + + async fn get_token(&self) -> AppResult { + { + let state = self.token_state.read().await; + if let Some(token) = &*state { + if token.expires_at > Utc::now() + Duration::seconds(30) { + tracing::debug!("Using cached Calendar API token"); + return Ok(token.access_token.clone()); + } + } + } + + let mut state = self.token_state.write().await; + // Double check after acquiring write lock + if let Some(token) = &*state { + if token.expires_at > Utc::now() + Duration::seconds(30) { + return Ok(token.access_token.clone()); + } + } + + tracing::info!("Refreshing Calendar API token via Client Credentials flow"); + let token_data = self + .authenticator + .client_credentials("profile") + .await + .map_err(|e| AppError::Internal(format!("Failed to get client credentials: {}", e)))?; + + let access_token = token_data["access_token"] + .as_str() + .ok_or_else(|| AppError::Internal("Missing access_token in response".into()))? + .to_string(); + + let expires_in = token_data["expires_in"].as_i64().unwrap_or(3600); + + let expires_at = Utc::now() + Duration::seconds(expires_in); + + *state = Some(TokenState { + access_token: access_token.clone(), + expires_at, + }); + + Ok(access_token) + } + + pub async fn list_events( + &self, + user_sub: Option, + upcoming: Option, + ) -> AppResult> { + let token = self.get_token().await?; + let mut url = format!("{}/service/v1/events", self.base_url); + let mut params = Vec::new(); + if let Some(uid) = &user_sub { + params.push(format!("user_sub={}", uid)); + } + if let Some(u) = upcoming { + params.push(format!("upcoming={}", u)); + } + + if !params.is_empty() { + url.push_str("?"); + url.push_str(¶ms.join("&")); + } + + tracing::info!(method = "GET", %url, "Sending Calendar API request"); + + let res = self + .client + .get(&url) + .bearer_auth(token) + .send() + .await + .map_err(AppError::Network)?; + + let status = res.status(); + tracing::info!(%status, %url, "Received Calendar API response"); + + if !status.is_success() { + let error_body = res.text().await.unwrap_or_default(); + tracing::error!(%status, %url, body = %error_body, "Calendar API request failed"); + return Err(AppError::Internal(format!( + "Failed to list events: {} - {}", + status, error_body + ))); + } + + let body = res + .json() + .await + .map_err(|e| AppError::Internal(e.to_string())); + tracing::info!(%status, %url, body = ?body, "Received Calendar API response"); + body + } + + pub async fn create_event( + &self, + user_sub: Option, + name: &str, + from: &str, + to: &str, + ) -> AppResult { + let token = self.get_token().await?; + let url = format!("{}/service/v1/events", self.base_url); + + let request = CreateEventRequest { + name: name.to_string(), + from: from.to_string(), + to: to.to_string(), + user_sub, + }; + + tracing::info!(method = "POST", %url, "Sending Calendar API request"); + + let res = self + .client + .post(&url) + .bearer_auth(token) + .json(&request) + .send() + .await + .map_err(AppError::Network)?; + + let status = res.status(); + tracing::info!(%status, %url, "Received Calendar API response"); + + if !status.is_success() { + let error_body = res.text().await.unwrap_or_default(); + tracing::error!(%status, %url, body = %error_body, "Calendar API request failed"); + return Err(AppError::Internal(format!( + "Failed to create event: {} - {}", + status, error_body + ))); + } + + res.json() + .await + .map_err(|e| AppError::Internal(e.to_string())) + } + + #[allow(dead_code)] + pub async fn get_event(&self, id: i32) -> AppResult { + let token = self.get_token().await?; + let url = format!("{}/service/v1/events/{}", self.base_url, id); + tracing::info!(method = "GET", %url, "Sending Calendar API request"); + + let res = self + .client + .get(&url) + .bearer_auth(token) + .send() + .await + .map_err(AppError::Network)?; + + let status = res.status(); + tracing::info!(%status, %url, "Received Calendar API response"); + + if !status.is_success() { + let error_body = res.text().await.unwrap_or_default(); + tracing::error!(%status, %url, body = %error_body, "Calendar API request failed"); + return Err(AppError::Internal(format!( + "Failed to get event: {} - {}", + status, error_body + ))); + } + + res.json() + .await + .map_err(|e| AppError::Internal(e.to_string())) + } + + #[allow(dead_code)] + pub async fn update_event( + &self, + id: i32, + user_sub: Option, + name: &str, + from: &str, + to: &str, + ) -> AppResult { + let token = self.get_token().await?; + let url = format!("{}/service/v1/events/{}", self.base_url, id); + + let request = CreateEventRequest { + name: name.to_string(), + from: from.to_string(), + to: to.to_string(), + user_sub, + }; + + tracing::info!(method = "PUT", %url, "Sending Calendar API request"); + + let res = self + .client + .put(&url) + .bearer_auth(token) + .json(&request) + .send() + .await + .map_err(AppError::Network)?; + + let status = res.status(); + tracing::info!(%status, %url, "Received Calendar API response"); + + if !status.is_success() { + let error_body = res.text().await.unwrap_or_default(); + tracing::error!(%status, %url, body = %error_body, "Calendar API request failed"); + return Err(AppError::Internal(format!( + "Failed to update event: {} - {}", + status, error_body + ))); + } + + res.json() + .await + .map_err(|e| AppError::Internal(e.to_string())) + } + + #[allow(dead_code)] + pub async fn delete_event(&self, id: i32) -> AppResult<()> { + let token = self.get_token().await?; + let url = format!("{}/service/v1/events/{}", self.base_url, id); + + tracing::info!(method = "DELETE", %url, "Sending Calendar API request"); + + let res = self + .client + .delete(&url) + .bearer_auth(token) + .send() + .await + .map_err(AppError::Network)?; + + let status = res.status(); + tracing::info!(%status, %url, "Received Calendar API response"); + + if !status.is_success() { + let error_body = res.text().await.unwrap_or_default(); + tracing::error!(%status, %url, body = %error_body, "Calendar API request failed"); + return Err(AppError::Internal(format!( + "Failed to delete event: {} - {}", + status, error_body + ))); + } + + Ok(()) + } +} diff --git a/src/domain/mod.rs b/src/domain/mod.rs index 70e887f..4a21816 100644 --- a/src/domain/mod.rs +++ b/src/domain/mod.rs @@ -1,4 +1,5 @@ pub mod agent; pub mod auth; +pub mod calendar; pub mod notifications; pub mod tasks; diff --git a/src/domain/tasks.rs b/src/domain/tasks.rs index 90c2286..2ea8b26 100644 --- a/src/domain/tasks.rs +++ b/src/domain/tasks.rs @@ -56,6 +56,7 @@ pub async fn execute_agent_run( db: &DatabaseConnection, _scheduler: &Arc, config: &Arc, + calendar_client: Arc, task_id: Uuid, goal: String, ) -> AppResult { @@ -92,6 +93,8 @@ pub async fn execute_agent_run( db.clone(), config.zen_api_key.clone(), config.tavily_api_key.clone(), + calendar_client.clone(), + None, goal.clone(), )?; diff --git a/src/error.rs b/src/error.rs index 01943d5..81194ce 100644 --- a/src/error.rs +++ b/src/error.rs @@ -32,16 +32,20 @@ pub enum AppError { impl IntoResponse for AppError { fn into_response(self) -> Response { - let (status, error_message) = match self { + let (status, error_message) = match &self { AppError::Database(err) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()), - AppError::Config(err) => (StatusCode::INTERNAL_SERVER_ERROR, err), - AppError::NotFound(err) => (StatusCode::NOT_FOUND, err), - AppError::Unauthorized(err) => (StatusCode::UNAUTHORIZED, err), - AppError::Internal(err) => (StatusCode::INTERNAL_SERVER_ERROR, err), + AppError::Config(err) => (StatusCode::INTERNAL_SERVER_ERROR, err.clone()), + AppError::NotFound(err) => (StatusCode::NOT_FOUND, err.clone()), + AppError::Unauthorized(err) => (StatusCode::UNAUTHORIZED, err.clone()), + AppError::Internal(err) => (StatusCode::INTERNAL_SERVER_ERROR, err.clone()), AppError::Network(err) => (StatusCode::BAD_GATEWAY, err.to_string()), - AppError::InvalidRequest(err) => (StatusCode::BAD_REQUEST, err), + AppError::InvalidRequest(err) => (StatusCode::BAD_REQUEST, err.clone()), }; + if status.is_server_error() || status.is_client_error() { + tracing::error!(%status, error = %self, "AppError converted to response"); + } + let body = Json(json!({ "error": error_message, })); diff --git a/src/scheduler.rs b/src/scheduler.rs index feac531..c98debd 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -13,6 +13,7 @@ pub struct Scheduler { db: DatabaseConnection, tasks_to_jobs: DashMap, config: Arc, + pub calendar_client: Arc, pub tx: tokio::sync::broadcast::Sender, pub push_sender: Arc, } @@ -21,6 +22,7 @@ impl Scheduler { pub async fn new( db: DatabaseConnection, config: Arc, + calendar_client: Arc, tx: tokio::sync::broadcast::Sender, push_sender: Arc, ) -> AppResult { @@ -36,6 +38,7 @@ impl Scheduler { db, tasks_to_jobs: DashMap::new(), config, + calendar_client, tx, push_sender, }) @@ -52,13 +55,17 @@ impl Scheduler { let tx = self.tx.clone(); let push_sender = self.push_sender.clone(); + let calendar_client = self.calendar_client.clone(); let job = Job::new_async(cron_expr, move |_uuid, _l| { let db = db.clone(); let config = config.clone(); let tx = tx.clone(); let push_sender = push_sender.clone(); + let calendar_client = calendar_client.clone(); Box::pin(async move { - if let Err(e) = Self::run_task(db, config, tx, push_sender, task_id).await { + if let Err(e) = + Self::run_task(db, config, calendar_client, tx, push_sender, task_id).await + { tracing::error!("Error in scheduled task {}: {}", task_id, e); } }) @@ -91,6 +98,7 @@ impl Scheduler { async fn run_task( db: DatabaseConnection, config: Arc, + calendar_client: Arc, tx: tokio::sync::broadcast::Sender, push_sender: Arc, task_id: Uuid, @@ -132,6 +140,8 @@ impl Scheduler { db.clone(), config.zen_api_key.clone(), config.tavily_api_key.clone(), + calendar_client.clone(), + None, task.goal.clone(), )?; diff --git a/src/server/chat.rs b/src/server/chat.rs index 5163275..d6b6bcd 100644 --- a/src/server/chat.rs +++ b/src/server/chat.rs @@ -17,6 +17,7 @@ pub struct ChatResult { pub async fn chat_handler( State(state): State>, + user: crate::server::auth::AuthenticatedUser, Json(payload): Json, ) -> Result, AppError> { let msg_count = payload.messages.len(); @@ -45,6 +46,8 @@ pub async fn chat_handler( state.db.clone(), state.config.zen_api_key.clone(), state.config.tavily_api_key.clone(), + state.calendar_client.clone(), + Some(user.0.sub), messages, )?; diff --git a/src/server/mod.rs b/src/server/mod.rs index 6a9d998..5160b4e 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -13,7 +13,6 @@ use sea_orm::{Database, DatabaseConnection, EntityTrait}; use std::sync::Arc; use tower_http::cors::{AllowOrigin, CorsLayer}; -use crate::entities::task::Entity as Task; use crate::scheduler::Scheduler; use crate::error::AppResult; @@ -25,6 +24,7 @@ pub struct AppState { pub config: Arc, pub verifier: Arc, pub authenticator: Arc, + pub calendar_client: Arc, pub tx: tokio::sync::broadcast::Sender, } @@ -39,13 +39,26 @@ pub async fn start(config: crate::config::Config) -> AppResult<()> { &config.vapid_private_key.clone(), )?); + let (verifier, authenticator) = setup_auth(&config).await?; + let calendar_client = Arc::new(crate::domain::calendar::CalendarClient::new( + config.calendar_api_url.clone(), + authenticator.clone(), + )); + let scheduler = Arc::new( - Scheduler::new(db.clone(), config.clone(), tx.clone(), push_sender.clone()) - .await - .map_err(|e| crate::error::AppError::Internal(e.to_string()))?, + Scheduler::new( + db.clone(), + config.clone(), + calendar_client.clone(), + tx.clone(), + push_sender.clone(), + ) + .await + .map_err(|e| crate::error::AppError::Internal(e.to_string()))?, ); // Load existing scheduled tasks + use crate::entities::task::Entity as Task; let existing_tasks = Task::find() .all(&db) .await @@ -56,14 +69,13 @@ pub async fn start(config: crate::config::Config) -> AppResult<()> { } } - let (verifier, authenticator) = setup_auth(&config).await?; - let state = Arc::new(AppState { db, scheduler, config: config.clone(), verifier, authenticator, + calendar_client, tx, }); @@ -136,6 +148,7 @@ fn build_app(state: Arc, config: &crate::config::Config) -> Router { .route("/api/notifications/vapid-key", get(notifications::push_handlers::get_vapid_key)) .route("/api/tasks/:id/subscription", get(notifications::push_handlers::get_subscription_status)) .route("/api/tasks/:id/subscribe", post(notifications::push_handlers::subscribe_task).delete(notifications::push_handlers::unsubscribe_task)) + .layer(axum::middleware::from_fn(log_error_responses)) .layer(cors) .layer(tower_http::set_header::SetResponseHeaderLayer::overriding( axum::http::header::CONTENT_SECURITY_POLICY, @@ -153,6 +166,22 @@ fn build_app(state: Arc, config: &crate::config::Config) -> Router { .with_state(state) } +async fn log_error_responses( + req: axum::extract::Request, + next: axum::middleware::Next, +) -> axum::response::Response { + let method = req.method().clone(); + let uri = req.uri().clone(); + let res = next.run(req).await; + let status = res.status(); + + if status.is_client_error() || status.is_server_error() { + tracing::error!(%method, %uri, %status, "Response error"); + } + + res +} + fn build_cors_layer(config: &crate::config::Config) -> CorsLayer { let allow_origin = if let Some(origins) = &config.cors_allowed_origins { let values: Vec = origins diff --git a/src/server/tasks.rs b/src/server/tasks.rs index 5c5da2b..fddc842 100644 --- a/src/server/tasks.rs +++ b/src/server/tasks.rs @@ -109,6 +109,7 @@ pub async fn rerun_task( &state.db, &state.scheduler, &state.config, + state.calendar_client.clone(), task.id, task.goal, )