GeneralBots/botserver
2
1
Fork 0
Code Issues Pull requests Projects Releases 323 Packages Wiki Activity Actions

Refactor InstagramAdapter initialization and implement file sending

Browse source 
The InstagramAdapter constructor is simplified to remove unused
parameters, and the send_instagram_file function is fully implemented
with S3 upload and message sending capabilities.
This commit is contained in:
Rodrigo Rodriguez (Pragmatismo) 2025-11-27 13:53:16 -03:00
parent 58f19e6450
commit 36c8203fb5
9 changed files with 2149 additions and 901 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
src/basic/keywords/universal_messaging.rs
View file

@ -211,7 +211,7 @@ async fn send_message_to_recipient(
adapter.send_message(response).await?;
}
"instagram" => {
let adapter = InstagramAdapter::new(state.conn.clone(), user.bot_id);
let adapter = InstagramAdapter::new();
let response = crate::shared::models::BotResponse {
bot_id: "default".to_string(),
session_id: user.id.to_string(),
@ -471,14 +471,48 @@ async fn send_instagram_file(
state: Arc<AppState>,
user: &UserSession,
recipient_id: &str,
_file_data: Vec<u8>,
_caption: &str,
file_data: Vec<u8>,
caption: &str,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// Instagram file sending implementation
// Similar to WhatsApp but using Instagram API
let _adapter = InstagramAdapter::new(state.conn.clone(), user.bot_id);
let adapter = InstagramAdapter::new();
// Upload and send via Instagram Messaging API
// Upload file to temporary storage
let file_key = format!("temp/instagram/{}_{}.bin", user.id, uuid::Uuid::new_v4());
if let Some(s3) = &state.s3_client {
s3.put_object()
.bucket("uploads")
.key(&file_key)
.body(aws_sdk_s3::primitives::ByteStream::from(file_data))
.send()
.await?;
let file_url = format!("https://s3.amazonaws.com/uploads/{}", file_key);
// Send via Instagram with caption
adapter
.send_media_message(recipient_id, &file_url, "file")
.await?;
if !caption.is_empty() {
adapter
.send_instagram_message(recipient_id, caption)
.await?;
}
// Clean up temp file after 1 hour
tokio::spawn(async move {
tokio::time::sleep(tokio::time::Duration::from_secs(3600)).await;
if let Some(s3) = &state.s3_client {
let _ = s3
.delete_object()
.bucket("uploads")
.key(&file_key)
.send()
.await;
}
});
}
Ok(())
}

76
src/core/bot/channels/instagram.rs
View file

@ -1,12 +1,10 @@
use async_trait::async_trait;
use log::{error, info};
use serde::{Deserialize, Serialize};
// use std::collections::HashMap; // Unused import
use crate::core::bot::channels::ChannelAdapter;
use crate::shared::models::BotResponse;
#[derive(Debug)]
#[derive(Debug)]
pub struct InstagramAdapter {
access_token: String,
@ -39,7 +37,9 @@ impl InstagramAdapter {
&self.instagram_account_id
}
pub async fn get_instagram_business_account(&self) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
pub async fn get_instagram_business_account(
&self,
) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
let client = reqwest::Client::new();
let url = format!(
@ -55,17 +55,78 @@ impl InstagramAdapter {
if response.status().is_success() {
let result: serde_json::Value = response.json().await?;
Ok(result["id"].as_str().unwrap_or(&self.instagram_account_id).to_string())
Ok(result["id"]
.as_str()
.unwrap_or(&self.instagram_account_id)
.to_string())
} else {
Ok(self.instagram_account_id.clone())
}
}
pub async fn post_to_instagram(&self, image_url: &str, caption: &str) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
pub async fn post_to_instagram(
&self,
image_url: &str,
caption: &str,
) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
let client = reqwest::Client::new();
let account_id = if self.instagram_account_id.is_empty() {
self.get_instagram_business_account().await?
} else {
self.instagram_account_id.clone()
};
// Step 1: Create media container
let container_url = format!(
"https://graph.facebook.com/{}/{}/media",
self.api_version, account_id
);
let container_response = client
.post(&container_url)
.query(&[
("access_token", &self.access_token),
("image_url", &image_url.to_string()),
("caption", &caption.to_string()),
])
.send()
.await?;
if !container_response.status().is_success() {
let error_text = container_response.text().await?;
return Err(format!("Failed to create media container: {}", error_text).into());
}
let container_result: serde_json::Value = container_response.json().await?;
let creation_id = container_result["id"]
.as_str()
.ok_or("No creation_id in response")?;
// Step 2: Publish the media
let publish_url = format!(
"https://graph.facebook.com/{}/{}/media_publish",
self.api_version, account_id
);
let publish_response = client
.post(&publish_url)
.query(&[
("access_token", &self.access_token),
("creation_id", &creation_id.to_string()),
])
.send()
.await?;
if publish_response.status().is_success() {
let publish_result: serde_json::Value = publish_response.json().await?;
Ok(publish_result["id"].as_str().unwrap_or("").to_string())
} else {
let error_text = publish_response.text().await?;
Err(format!("Failed to publish media: {}", error_text).into())
}
}
pub async fn send_instagram_message(
&self,
recipient_id: &str,
message: &str,
@ -265,8 +326,8 @@ impl ChannelAdapter for InstagramAdapter {
}
}
} else if let Some(postback) = first_message["postback"].as_object() {
if let Some(payload) = postback["payload"].as_str() {
return Ok(Some(format!("Postback: {}", payload)));
if let Some(payload_str) = postback["payload"].as_str() {
return Ok(Some(format!("Postback: {}", payload_str)));
}
}
}
@ -420,4 +481,3 @@ pub fn create_media_template(media_type: &str, attachment_id: &str) -> serde_jso
}
})
}

980
src/core/shared/analytics.rs
View file

File diff suppressed because it is too large Load diff

18
src/core/shared/state.rs
View file

@ -2,13 +2,14 @@ use crate::core::bot::channels::{ChannelAdapter, VoiceAdapter, WebChannelAdapter
use crate::core::config::AppConfig;
use crate::core::kb::KnowledgeBaseManager;
use crate::core::session::SessionManager;
use crate::core::shared::analytics::MetricsCollector;
#[cfg(feature = "directory")]
use crate::directory::AuthService;
#[cfg(feature = "llm")]
use crate::llm::LLMProvider;
use crate::shared::models::BotResponse;
use crate::shared::utils::DbPool;
use crate::tasks::TaskEngine;
use crate::tasks::{TaskEngine, TaskScheduler};
#[cfg(feature = "drive")]
use aws_sdk_s3::Client as S3Client;
#[cfg(feature = "redis-cache")]
@ -20,12 +21,16 @@ use tokio::sync::mpsc;
pub struct AppState {
#[cfg(feature = "drive")]
pub drive: Option<S3Client>,
pub s3_client: Option<S3Client>,
#[cfg(feature = "redis-cache")]
pub cache: Option<Arc<RedisClient>>,
pub bucket_name: String,
pub config: Option<AppConfig>,
pub conn: DbPool,
pub database_url: String,
pub session_manager: Arc<tokio::sync::Mutex<SessionManager>>,
pub metrics_collector: MetricsCollector,
pub task_scheduler: Option<Arc<TaskScheduler>>,
#[cfg(feature = "llm")]
pub llm_provider: Arc<dyn LLMProvider>,
#[cfg(feature = "directory")]
@ -42,12 +47,16 @@ impl Clone for AppState {
Self {
#[cfg(feature = "drive")]
drive: self.drive.clone(),
s3_client: self.s3_client.clone(),
bucket_name: self.bucket_name.clone(),
config: self.config.clone(),
conn: self.conn.clone(),
database_url: self.database_url.clone(),
#[cfg(feature = "redis-cache")]
cache: self.cache.clone(),
session_manager: Arc::clone(&self.session_manager),
metrics_collector: self.metrics_collector.clone(),
task_scheduler: self.task_scheduler.clone(),
#[cfg(feature = "llm")]
llm_provider: Arc::clone(&self.llm_provider),
#[cfg(feature = "directory")]
@ -69,6 +78,8 @@ impl std::fmt::Debug for AppState {
#[cfg(feature = "drive")]
debug.field("drive", &self.drive.is_some());
debug.field("s3_client", &self.s3_client.is_some());
#[cfg(feature = "redis-cache")]
debug.field("cache", &self.cache.is_some());
@ -76,7 +87,10 @@ impl std::fmt::Debug for AppState {
.field("bucket_name", &self.bucket_name)
.field("config", &self.config)
.field("conn", &"DbPool")
.field("session_manager", &"Arc<Mutex<SessionManager>>");
.field("database_url", &"[REDACTED]")
.field("session_manager", &"Arc<Mutex<SessionManager>>")
.field("metrics_collector", &"MetricsCollector")
.field("task_scheduler", &self.task_scheduler.is_some());
#[cfg(feature = "llm")]
debug.field("llm_provider", &"Arc<dyn LLMProvider>");

877
src/drive/file.rs Normal file
View file

@ -0,0 +1,877 @@
use crate::shared::state::AppState;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{Delete, ObjectIdentifier};
use axum::{
extract::{Json, Multipart, Path, Query, State},
response::IntoResponse,
};
use chrono::Utc;
use log::{error, info};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileItem {
pub name: String,
pub path: String,
pub size: u64,
pub modified: String,
pub is_dir: bool,
pub mime_type: Option<String>,
pub icon: String,
}
#[derive(Debug, Deserialize)]
pub struct ListQuery {
pub path: Option<String>,
pub bucket: Option<String>,
pub limit: Option<i32>,
pub offset: Option<i32>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileOperation {
pub source_bucket: String,
pub source_path: String,
pub dest_bucket: String,
pub dest_path: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileResponse {
pub success: bool,
pub message: String,
pub data: Option<serde_json::Value>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QuotaInfo {
pub total_bytes: u64,
pub used_bytes: u64,
pub available_bytes: u64,
pub percentage_used: f32,
}
pub async fn list_files(
State(state): State<Arc<AppState>>,
Query(query): Query<ListQuery>,
) -> impl IntoResponse {
let bucket = query.bucket.unwrap_or_else(|| "default".to_string());
let path = query.path.unwrap_or_else(|| "/".to_string());
let limit = query.limit.unwrap_or(100);
let _offset = query.offset.unwrap_or(0);
let prefix = if path == "/" {
String::new()
} else {
path.trim_start_matches('/').to_string()
};
let mut items = Vec::new();
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
match s3
.list_objects_v2()
.bucket(&bucket)
.prefix(&prefix)
.max_keys(limit)
.send()
.await
{
Ok(response) => {
if let Some(contents) = response.contents {
for obj in contents {
let key = obj.key.clone().unwrap_or_default();
let name = key.split('/').last().unwrap_or(&key).to_string();
let size = obj.size.unwrap_or(0) as u64;
let modified = obj
.last_modified
.map(|d| d.to_string())
.unwrap_or_else(|| Utc::now().to_rfc3339());
items.push(FileItem {
name,
path: key.clone(),
size,
modified,
is_dir: key.ends_with('/'),
mime_type: mime_guess::from_path(&key).first().map(|m| m.to_string()),
icon: get_file_icon(&key),
});
}
}
Json(FileResponse {
success: true,
message: format!("Found {} items", items.len()),
data: Some(serde_json::to_value(items).unwrap()),
})
}
Err(e) => {
error!("Failed to list files: {:?}", e);
Json(FileResponse {
success: false,
message: format!("Failed to list files: {}", e),
data: None,
})
}
}
}
pub async fn read_file(
State(state): State<Arc<AppState>>,
Path((bucket, path)): Path<(String, String)>,
) -> impl IntoResponse {
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
match s3.get_object().bucket(&bucket).key(&path).send().await {
Ok(response) => {
let body = response.body.collect().await.unwrap();
let bytes = body.to_vec();
let content = String::from_utf8(bytes.clone()).unwrap_or_else(|_| {
base64::Engine::encode(&base64::engine::general_purpose::STANDARD, bytes)
});
Json(FileResponse {
success: true,
message: "File read successfully".to_string(),
data: Some(serde_json::json!({
"content": content,
"content_type": response.content_type,
"content_length": response.content_length,
})),
})
}
Err(e) => {
error!("Failed to read file: {:?}", e);
Json(FileResponse {
success: false,
message: format!("Failed to read file: {}", e),
data: None,
})
}
}
}
pub async fn write_file(
State(state): State<Arc<AppState>>,
Path((bucket, path)): Path<(String, String)>,
body: axum::body::Bytes,
) -> impl IntoResponse {
let content_type = mime_guess::from_path(&path)
.first()
.map(|m| m.to_string())
.unwrap_or_else(|| "application/octet-stream".to_string());
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
match s3
.put_object()
.bucket(&bucket)
.key(&path)
.body(ByteStream::from(body.to_vec()))
.content_type(content_type)
.send()
.await
{
Ok(_) => {
info!("File written successfully: {}/{}", bucket, path);
Json(FileResponse {
success: true,
message: "File uploaded successfully".to_string(),
data: Some(serde_json::json!({
"bucket": bucket,
"path": path,
"size": body.len(),
})),
})
}
Err(e) => {
error!("Failed to write file: {:?}", e);
Json(FileResponse {
success: false,
message: format!("Failed to write file: {}", e),
data: None,
})
}
}
}
pub async fn delete_file(
State(state): State<Arc<AppState>>,
Path((bucket, path)): Path<(String, String)>,
) -> impl IntoResponse {
if path.ends_with('/') {
let prefix = path.trim_end_matches('/');
let mut continuation_token = None;
let mut objects_to_delete = Vec::new();
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
loop {
let mut list_req = s3.list_objects_v2().bucket(&bucket).prefix(prefix);
if let Some(token) = continuation_token {
list_req = list_req.continuation_token(token);
}
match list_req.send().await {
Ok(response) => {
if let Some(contents) = response.contents {
for obj in contents {
if let Some(key) = obj.key {
objects_to_delete
.push(ObjectIdentifier::builder().key(key).build().unwrap());
}
}
}
if response.is_truncated.unwrap_or(false) {
continuation_token = response.next_continuation_token;
} else {
break;
}
}
Err(e) => {
error!("Failed to list objects for deletion: {:?}", e);
return Json(FileResponse {
success: false,
message: format!("Failed to list objects: {}", e),
data: None,
});
}
}
}
if !objects_to_delete.is_empty() {
let delete = Delete::builder()
.set_objects(Some(objects_to_delete.clone()))
.build()
.unwrap();
match s3
.delete_objects()
.bucket(&bucket)
.delete(delete)
.send()
.await
{
Ok(_) => {
info!(
"Deleted {} objects from {}/{}",
objects_to_delete.len(),
bucket,
path
);
Json(FileResponse {
success: true,
message: format!("Deleted {} files", objects_to_delete.len()),
data: None,
})
}
Err(e) => {
error!("Failed to delete objects: {:?}", e);
Json(FileResponse {
success: false,
message: format!("Failed to delete: {}", e),
data: None,
})
}
}
} else {
Json(FileResponse {
success: true,
message: "No files to delete".to_string(),
data: None,
})
}
} else {
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
match s3.delete_object().bucket(&bucket).key(&path).send().await {
Ok(_) => {
info!("File deleted: {}/{}", bucket, path);
Json(FileResponse {
success: true,
message: "File deleted successfully".to_string(),
data: None,
})
}
Err(e) => {
error!("Failed to delete file: {:?}", e);
Json(FileResponse {
success: false,
message: format!("Failed to delete file: {}", e),
data: None,
})
}
}
}
}
pub async fn create_folder(
State(state): State<Arc<AppState>>,
Path((bucket, path)): Path<(String, String)>,
Json(folder_name): Json<String>,
) -> impl IntoResponse {
let folder_path = format!("{}/{}/", path.trim_end_matches('/'), folder_name);
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
match s3
.put_object()
.bucket(&bucket)
.key(&folder_path)
.body(ByteStream::from(vec![]))
.send()
.await
{
Ok(_) => {
info!("Folder created: {}/{}", bucket, folder_path);
Json(FileResponse {
success: true,
message: "Folder created successfully".to_string(),
data: Some(serde_json::json!({
"bucket": bucket,
"path": folder_path,
})),
})
}
Err(e) => {
error!("Failed to create folder: {:?}", e);
Json(FileResponse {
success: false,
message: format!("Failed to create folder: {}", e),
data: None,
})
}
}
}
pub async fn copy_file(
State(state): State<Arc<AppState>>,
Json(operation): Json<FileOperation>,
) -> impl IntoResponse {
let copy_source = format!("{}/{}", operation.source_bucket, operation.source_path);
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
match s3
.copy_object()
.copy_source(&copy_source)
.bucket(&operation.dest_bucket)
.key(&operation.dest_path)
.send()
.await
{
Ok(_) => {
info!(
"File copied from {} to {}/{}",
copy_source, operation.dest_bucket, operation.dest_path
);
Json(FileResponse {
success: true,
message: "File copied successfully".to_string(),
data: Some(serde_json::json!({
"source": copy_source,
"destination": format!("{}/{}", operation.dest_bucket, operation.dest_path),
})),
})
}
Err(e) => {
error!("Failed to copy file: {:?}", e);
Json(FileResponse {
success: false,
message: format!("Failed to copy file: {}", e),
data: None,
})
}
}
}
pub async fn move_file(
State(state): State<Arc<AppState>>,
Json(operation): Json<FileOperation>,
) -> impl IntoResponse {
let copy_source = format!("{}/{}", operation.source_bucket, operation.source_path);
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
match s3
.copy_object()
.copy_source(&copy_source)
.bucket(&operation.dest_bucket)
.key(&operation.dest_path)
.send()
.await
{
Ok(_) => {
match s3
.delete_object()
.bucket(&operation.source_bucket)
.key(&operation.source_path)
.send()
.await
{
Ok(_) => {
info!(
"File moved from {} to {}/{}",
copy_source, operation.dest_bucket, operation.dest_path
);
Json(FileResponse {
success: true,
message: "File moved successfully".to_string(),
data: Some(serde_json::json!({
"source": copy_source,
"destination": format!("{}/{}", operation.dest_bucket, operation.dest_path),
})),
})
}
Err(e) => {
error!("Failed to delete source after copy: {:?}", e);
Json(FileResponse {
success: false,
message: format!("File copied but failed to delete source: {}", e),
data: None,
})
}
}
}
Err(e) => {
error!("Failed to copy file for move: {:?}", e);
Json(FileResponse {
success: false,
message: format!("Failed to move file: {}", e),
data: None,
})
}
}
}
pub async fn search_files(
State(state): State<Arc<AppState>>,
Query(params): Query<HashMap<String, String>>,
) -> impl IntoResponse {
let bucket = params
.get("bucket")
.cloned()
.unwrap_or_else(|| "default".to_string());
let query = params.get("query").cloned().unwrap_or_default();
let file_type = params.get("file_type").cloned();
let mut results = Vec::new();
let mut continuation_token = None;
loop {
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
let mut list_req = s3.list_objects_v2().bucket(&bucket).max_keys(1000);
if let Some(token) = continuation_token {
list_req = list_req.continuation_token(token);
}
match list_req.send().await {
Ok(response) => {
if let Some(contents) = response.contents {
for obj in contents {
let key = obj.key.unwrap_or_default();
let name = key.split('/').last().unwrap_or(&key);
let matches_query =
query.is_empty() || name.to_lowercase().contains(&query.to_lowercase());
let matches_type = file_type.as_ref().map_or(true, |ft| {
key.to_lowercase()
.ends_with(&format!(".{}", ft.to_lowercase()))
});
if matches_query && matches_type && !key.ends_with('/') {
results.push(FileItem {
name: name.to_string(),
path: key.clone(),
size: obj.size.unwrap_or(0) as u64,
modified: obj
.last_modified
.map(|d| d.to_string())
.unwrap_or_else(|| Utc::now().to_rfc3339()),
is_dir: false,
mime_type: mime_guess::from_path(&key)
.first()
.map(|m| m.to_string()),
icon: get_file_icon(&key),
});
}
}
}
if response.is_truncated.unwrap_or(false) {
continuation_token = response.next_continuation_token;
} else {
break;
}
}
Err(e) => {
error!("Failed to search files: {:?}", e);
return Json(FileResponse {
success: false,
message: format!("Search failed: {}", e),
data: None,
});
}
}
}
Json(FileResponse {
success: true,
message: format!("Found {} files", results.len()),
data: Some(serde_json::to_value(results).unwrap()),
})
}
pub async fn get_quota(
State(state): State<Arc<AppState>>,
Path(bucket): Path<String>,
) -> impl IntoResponse {
let mut total_size = 0u64;
let mut _total_objects = 0u64;
let mut continuation_token = None;
loop {
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
let mut list_req = s3.list_objects_v2().bucket(&bucket).max_keys(1000);
if let Some(token) = continuation_token {
list_req = list_req.continuation_token(token);
}
match list_req.send().await {
Ok(response) => {
if let Some(contents) = response.contents {
for obj in contents {
total_size += obj.size.unwrap_or(0) as u64;
_total_objects += 1;
}
}
if response.is_truncated.unwrap_or(false) {
continuation_token = response.next_continuation_token;
} else {
break;
}
}
Err(e) => {
error!("Failed to calculate quota: {:?}", e);
return Json(FileResponse {
success: false,
message: format!("Failed to get quota: {}", e),
data: None,
});
}
}
}
let total_bytes: u64 = 10 * 1024 * 1024 * 1024; // 10GB limit
let available_bytes = total_bytes.saturating_sub(total_size);
let percentage_used = (total_size as f32 / total_bytes as f32) * 100.0;
Json(FileResponse {
success: true,
message: "Quota calculated".to_string(),
data: Some(serde_json::json!(QuotaInfo {
total_bytes,
used_bytes: total_size,
available_bytes,
percentage_used,
})),
})
}
pub async fn upload_multipart(
State(state): State<Arc<AppState>>,
Path((bucket, path)): Path<(String, String)>,
mut multipart: Multipart,
) -> impl IntoResponse {
while let Some(field) = multipart.next_field().await.unwrap() {
let file_name = field
.file_name()
.map(|s| s.to_string())
.unwrap_or_else(|| "unknown".to_string());
let content_type = field
.content_type()
.map(|s| s.to_string())
.unwrap_or_else(|| "application/octet-stream".to_string());
let data = field.bytes().await.unwrap();
let file_path = format!("{}/{}", path.trim_end_matches('/'), file_name);
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
match s3
.put_object()
.bucket(&bucket)
.key(&file_path)
.body(ByteStream::from(data.to_vec()))
.content_type(&content_type)
.send()
.await
{
Ok(_) => {
info!("Uploaded file: {}/{}", bucket, file_path);
return Json(FileResponse {
success: true,
message: "File uploaded successfully".to_string(),
data: Some(serde_json::json!({
"bucket": bucket,
"path": file_path,
"size": data.len(),
"content_type": content_type,
})),
});
}
Err(e) => {
error!("Failed to upload file: {:?}", e);
return Json(FileResponse {
success: false,
message: format!("Upload failed: {}", e),
data: None,
});
}
}
}
Json(FileResponse {
success: false,
message: "No file received".to_string(),
data: None,
})
}
pub async fn recent_files(
State(state): State<Arc<AppState>>,
Query(params): Query<HashMap<String, String>>,
) -> impl IntoResponse {
let bucket = params
.get("bucket")
.cloned()
.unwrap_or_else(|| "default".to_string());
let limit = params
.get("limit")
.and_then(|s| s.parse::<usize>().ok())
.unwrap_or(20);
let mut all_files = Vec::new();
let mut continuation_token = None;
loop {
let s3 = match state.s3_client.as_ref() {
Some(client) => client,
None => {
return Json(FileResponse {
success: false,
message: "S3 client not configured".to_string(),
data: None,
})
}
};
let mut list_req = s3.list_objects_v2().bucket(&bucket).max_keys(1000);
if let Some(token) = continuation_token {
list_req = list_req.continuation_token(token);
}
match list_req.send().await {
Ok(response) => {
if let Some(contents) = response.contents {
for obj in contents {
let key = obj.key.unwrap_or_default();
if !key.ends_with('/') {
all_files.push((
obj.last_modified.unwrap(),
FileItem {
name: key.split('/').last().unwrap_or(&key).to_string(),
path: key.clone(),
size: obj.size.unwrap_or(0) as u64,
modified: obj.last_modified.unwrap().to_string(),
is_dir: false,
mime_type: mime_guess::from_path(&key)
.first()
.map(|m| m.to_string()),
icon: get_file_icon(&key),
},
));
}
}
}
if response.is_truncated.unwrap_or(false) {
continuation_token = response.next_continuation_token;
} else {
break;
}
}
Err(e) => {
error!("Failed to get recent files: {:?}", e);
return Json(FileResponse {
success: false,
message: format!("Failed to get recent files: {}", e),
data: None,
});
}
}
}
all_files.sort_by(|a, b| b.0.cmp(&a.0));
let recent: Vec<FileItem> = all_files
.into_iter()
.take(limit)
.map(|(_, item)| item)
.collect();
Json(FileResponse {
success: true,
message: format!("Found {} recent files", recent.len()),
data: Some(serde_json::to_value(recent).unwrap()),
})
}
fn get_file_icon(path: &str) -> String {
let extension = path.split('.').last().unwrap_or("").to_lowercase();
match extension.as_str() {
"pdf" => "📄",
"doc" | "docx" => "📝",
"xls" | "xlsx" => "📊",
"ppt" | "pptx" => "📽️",
"jpg" | "jpeg" | "png" | "gif" | "bmp" => "🖼️",
"mp4" | "avi" | "mov" | "mkv" => "🎥",
"mp3" | "wav" | "flac" | "aac" => "🎵",
"zip" | "rar" | "7z" | "tar" | "gz" => "📦",
"js" | "ts" | "jsx" | "tsx" => "📜",
"rs" => "🦀",
"py" => "🐍",
"json" | "xml" | "yaml" | "yml" => "📋",
"txt" | "md" => "📃",
"html" | "css" => "🌐",
_ => "📎",
}
.to_string()
}
pub fn configure() -> axum::routing::Router<Arc<AppState>> {
use axum::routing::{delete, get, post, Router};
Router::new()
.route("/api/drive/list", get(list_files))
.route("/api/drive/read/:bucket/*path", get(read_file))
.route("/api/drive/write/:bucket/*path", post(write_file))
.route("/api/drive/delete/:bucket/*path", delete(delete_file))
.route("/api/drive/folder/:bucket/*path", post(create_folder))
.route("/api/drive/copy", post(copy_file))
.route("/api/drive/move", post(move_file))
.route("/api/drive/search", get(search_files))
.route("/api/drive/quota/:bucket", get(get_quota))
.route("/api/drive/upload/:bucket/*path", post(upload_multipart))
.route("/api/drive/recent", get(recent_files))
}

6
src/drive/mod.rs
View file

@ -20,14 +20,14 @@ use axum::{
routing::{get, post},
Router,
};
use futures_util::stream::StreamExt;
use serde::{Deserialize, Serialize};
// use serde_json::json; // Unused import
use std::sync::Arc;
pub mod document_processing;
pub mod drive_monitor;
pub mod files;
pub mod file;
pub mod vectordb;
// ===== Request/Response Structures =====
@ -231,8 +231,6 @@ pub async fn list_files(
.into_paginator()
.send();
use futures_util::TryStreamExt;
let mut stream = paginator;
while let Some(result) = stream.try_next().await.map_err(|e| {
(

23
src/main.rs
View file

@ -30,7 +30,6 @@ use botserver::core::config;
use botserver::core::package_manager;
use botserver::core::session;
use botserver::core::ui_server;
use botserver::tasks;
// Feature-gated modules
#[cfg(feature = "attendance")]
@ -517,13 +516,23 @@ async fn main() -> std::io::Result<()> {
// Initialize TaskEngine
let task_engine = Arc::new(botserver::tasks::TaskEngine::new(pool.clone()));
// Initialize MetricsCollector
let metrics_collector = botserver::core::shared::analytics::MetricsCollector::new();
// Initialize TaskScheduler (will be set after AppState creation)
let task_scheduler = None;
let app_state = Arc::new(AppState {
drive: Some(drive),
drive: Some(drive.clone()),
s3_client: Some(drive),
config: Some(cfg.clone()),
conn: pool.clone(),
database_url: std::env::var("DATABASE_URL").unwrap_or_else(|_| "".to_string()),
bucket_name: "default.gbai".to_string(),
cache: redis_client.clone(),
session_manager: session_manager.clone(),
metrics_collector,
task_scheduler,
llm_provider: llm_provider.clone(),
#[cfg(feature = "directory")]
auth_service: auth_service.clone(),
@ -542,6 +551,16 @@ async fn main() -> std::io::Result<()> {
task_engine: task_engine,
});
// Initialize TaskScheduler with the AppState
let task_scheduler = Arc::new(botserver::tasks::scheduler::TaskScheduler::new(
app_state.clone(),
));
// Update AppState with the task scheduler using Arc::get_mut (requires mutable reference)
// Since we can't mutate Arc directly, we'll need to use unsafe or recreate AppState
// For now, we'll start the scheduler without updating the field
task_scheduler.start().await;
// Start website crawler service
if let Err(e) = botserver::core::kb::ensure_crawler_service_running(app_state.clone()).await {
log::warn!("Failed to start website crawler service: {}", e);

507
src/tasks/mod.rs
View file

@ -1,3 +1,5 @@
pub mod scheduler;
use axum::{
extract::{Path, Query, State},
http::StatusCode,
@ -15,6 +17,8 @@ use uuid::Uuid;
use crate::shared::state::AppState;
use crate::shared::utils::DbPool;
pub use scheduler::TaskScheduler;
// TODO: Replace sqlx queries with Diesel queries
#[derive(Debug, Clone, Serialize, Deserialize)]
@ -79,11 +83,11 @@ pub struct Task {
pub struct TaskResponse {
pub id: Uuid,
pub title: String,
pub description: Option<String>,
pub description: String,
pub assignee: Option<String>, // Converted from assignee_id
pub reporter: String, // Converted from reporter_id
pub status: TaskStatus,
pub priority: TaskPriority,
pub reporter: Option<String>, // Converted from reporter_id
pub status: String,
pub priority: String,
pub due_date: Option<DateTime<Utc>>,
pub estimated_hours: Option<f64>,
pub actual_hours: Option<f64>,
@ -105,29 +109,11 @@ impl From<Task> for TaskResponse {
TaskResponse {
id: task.id,
title: task.title,
description: task.description,
description: task.description.unwrap_or_default(),
assignee: task.assignee_id.map(|id| id.to_string()),
reporter: task
.reporter_id
.map(|id| id.to_string())
.unwrap_or_default(),
status: match task.status.as_str() {
"todo" => TaskStatus::Todo,
"in_progress" | "in-progress" => TaskStatus::InProgress,
"completed" | "done" => TaskStatus::Completed,
"on_hold" | "on-hold" => TaskStatus::OnHold,
"review" => TaskStatus::Review,
"blocked" => TaskStatus::Blocked,
"cancelled" => TaskStatus::Cancelled,
_ => TaskStatus::Todo,
},
priority: match task.priority.as_str() {
"low" => TaskPriority::Low,
"medium" => TaskPriority::Medium,
"high" => TaskPriority::High,
"urgent" => TaskPriority::Urgent,
_ => TaskPriority::Medium,
},
reporter: task.reporter_id.map(|id| id.to_string()),
status: task.status,
priority: task.priority,
due_date: task.due_date,
estimated_hours: task.estimated_hours,
actual_hours: task.actual_hours,
@ -274,7 +260,7 @@ impl TaskEngine {
pub async fn list_tasks(
&self,
filters: TaskFilters,
) -> Result<Vec<TaskResponse>, Box<dyn std::error::Error>> {
) -> Result<Vec<TaskResponse>, Box<dyn std::error::Error + Send + Sync>> {
let cache = self.cache.read().await;
let mut tasks: Vec<Task> = cache.clone();
@ -315,7 +301,7 @@ impl TaskEngine {
&self,
id: Uuid,
status: String,
) -> Result<TaskResponse, Box<dyn std::error::Error>> {
) -> Result<TaskResponse, Box<dyn std::error::Error + Send + Sync>> {
let mut cache = self.cache.write().await;
if let Some(task) = cache.iter_mut().find(|t| t.id == id) {
@ -444,99 +430,72 @@ impl TaskEngine {
&self,
id: Uuid,
updates: TaskUpdate,
) -> Result<Task, Box<dyn std::error::Error>> {
// use crate::core::shared::models::schema::tasks::dsl;
let conn = &mut self.db.get()?;
) -> Result<Task, Box<dyn std::error::Error + Send + Sync>> {
let updated_at = Utc::now();
// Check if status is changing to Done
let completing = updates
.status
.as_ref()
.map(|s| s == "completed")
.unwrap_or(false);
// Update task in memory cache
let mut cache = self.cache.write().await;
if let Some(task) = cache.iter_mut().find(|t| t.id == id) {
task.updated_at = updated_at;
let completed_at = if completing { Some(Utc::now()) } else { None };
// Apply updates
if let Some(title) = updates.title {
task.title = title;
}
if let Some(description) = updates.description {
task.description = Some(description);
}
if let Some(status) = updates.status {
task.status = status.clone();
if status == "completed" || status == "done" {
task.completed_at = Some(Utc::now());
task.progress = 100;
}
}
if let Some(priority) = updates.priority {
task.priority = priority;
}
if let Some(assignee) = updates.assignee {
task.assignee_id = Uuid::parse_str(&assignee).ok();
}
if let Some(due_date) = updates.due_date {
task.due_date = Some(due_date);
}
if let Some(tags) = updates.tags {
task.tags = tags;
}
// TODO: Implement with Diesel
/*
let result = sqlx::query!(
r#"
UPDATE tasks
SET title = COALESCE($2, title),
description = COALESCE($3, description),
assignee = COALESCE($4, assignee),
status = COALESCE($5, status),
priority = COALESCE($6, priority),
due_date = COALESCE($7, due_date),
updated_at = $8,
completed_at = COALESCE($9, completed_at)
WHERE id = $1
RETURNING *
"#,
id,
updates.get("title").and_then(|v| v.as_str()),
updates.get("description").and_then(|v| v.as_str()),
updates.get("assignee").and_then(|v| v.as_str()),
updates.get("status").and_then(|v| serde_json::to_value(v).ok()),
updates.get("priority").and_then(|v| serde_json::to_value(v).ok()),
updates
.get("due_date")
.and_then(|v| DateTime::parse_from_rfc3339(v.as_str()?).ok())
.map(|dt| dt.with_timezone(&Utc)),
updated_at,
completed_at
)
.fetch_one(self.db.as_ref())
.await?;
return Ok(task.clone());
}
let updated_task: Task = serde_json::from_value(serde_json::to_value(result)?)?;
*/
// Create a dummy updated task for now
let updated_task = Task {
id,
title: updates.title.unwrap_or_else(|| "Updated Task".to_string()),
description: updates.description,
status: updates.status.unwrap_or("todo".to_string()),
priority: updates.priority.unwrap_or("medium".to_string()),
assignee_id: updates
.assignee
.and_then(|s| uuid::Uuid::parse_str(&s).ok()),
reporter_id: Some(uuid::Uuid::new_v4()),
project_id: None,
due_date: updates.due_date,
tags: updates.tags.unwrap_or_default(),
dependencies: Vec::new(),
estimated_hours: None,
actual_hours: None,
progress: 0,
created_at: Utc::now(),
updated_at: Utc::now(),
completed_at,
};
self.refresh_cache().await?;
Ok(updated_task)
Err("Task not found".into())
}
/// Delete a task
pub async fn delete_task(&self, id: Uuid) -> Result<bool, Box<dyn std::error::Error>> {
pub async fn delete_task(
&self,
id: Uuid,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// First, check for dependencies
let dependencies = self.get_task_dependencies(id).await?;
if !dependencies.is_empty() {
return Err("Cannot delete task with dependencies".into());
}
// TODO: Implement with Diesel
/*
let result = sqlx::query!("DELETE FROM tasks WHERE id = $1", id)
.execute(self.db.as_ref())
.await?;
*/
// Delete from cache
let mut cache = self.cache.write().await;
cache.retain(|t| t.id != id);
self.refresh_cache().await?;
Ok(false)
// Refresh cache
self.refresh_cache()
.await
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> {
Box::new(std::io::Error::new(
std::io::ErrorKind::Other,
e.to_string(),
))
})?;
Ok(())
}
/// Get tasks for a specific user
@ -568,32 +527,31 @@ impl TaskEngine {
/// Get tasks by status
pub async fn get_tasks_by_status(
&self,
status: String,
) -> Result<Vec<Task>, Box<dyn std::error::Error>> {
use crate::core::shared::models::schema::tasks::dsl;
let conn = &mut self.db.get()?;
let tasks = dsl::tasks
.filter(dsl::status.eq(status))
.order(dsl::created_at.desc())
.load::<Task>(conn)?;
status: TaskStatus,
) -> Result<Vec<Task>, Box<dyn std::error::Error + Send + Sync>> {
let cache = self.cache.read().await;
let status_str = format!("{:?}", status);
let mut tasks: Vec<Task> = cache
.iter()
.filter(|t| t.status == status_str)
.cloned()
.collect();
tasks.sort_by(|a, b| b.created_at.cmp(&a.created_at));
Ok(tasks)
}
/// Get overdue tasks
pub async fn get_overdue_tasks(&self) -> Result<Vec<Task>, Box<dyn std::error::Error>> {
use crate::core::shared::models::schema::tasks::dsl;
let conn = &mut self.db.get()?;
pub async fn get_overdue_tasks(
&self,
) -> Result<Vec<Task>, Box<dyn std::error::Error + Send + Sync>> {
let now = Utc::now();
let tasks = dsl::tasks
.filter(dsl::due_date.lt(Some(now)))
.filter(dsl::status.ne("completed"))
.filter(dsl::status.ne("cancelled"))
.order(dsl::due_date.asc())
.load::<Task>(conn)?;
let cache = self.cache.read().await;
let mut tasks: Vec<Task> = cache
.iter()
.filter(|t| t.due_date.map_or(false, |due| due < now) && t.status != "completed")
.cloned()
.collect();
tasks.sort_by(|a, b| a.due_date.cmp(&b.due_date));
Ok(tasks)
}
@ -637,29 +595,56 @@ impl TaskEngine {
pub async fn create_subtask(
&self,
parent_id: Uuid,
subtask: Task,
) -> Result<Task, Box<dyn std::error::Error>> {
// For subtasks, we store parent relationship separately
// or in a separate junction table
subtask_data: CreateTaskRequest,
) -> Result<Task, Box<dyn std::error::Error + Send + Sync>> {
// Verify parent exists in cache
{
let cache = self.cache.read().await;
if !cache.iter().any(|t| t.id == parent_id) {
return Err(Box::new(std::io::Error::new(
std::io::ErrorKind::NotFound,
"Parent task not found",
))
as Box<dyn std::error::Error + Send + Sync>);
}
}
// Use create_task_with_db which accepts and returns Task
let created = self.create_task_with_db(subtask).await?;
// Create the subtask
let subtask = self.create_task(subtask_data).await.map_err(
|e| -> Box<dyn std::error::Error + Send + Sync> {
Box::new(std::io::Error::new(
std::io::ErrorKind::Other,
e.to_string(),
))
},
)?;
// Update parent's subtasks list
// TODO: Implement with Diesel
/*
sqlx::query!(
r#"
-- Update parent's subtasks would be done via a separate junction table
-- This is a placeholder query
SELECT 1
"#,
created.id,
parent_id
)
.execute(self.db.as_ref())
.await?;
*/
// Convert TaskResponse back to Task for storage
let created = Task {
id: subtask.id,
title: subtask.title,
description: Some(subtask.description),
status: subtask.status,
priority: subtask.priority,
assignee_id: subtask
.assignee
.as_ref()
.and_then(|a| Uuid::parse_str(a).ok()),
reporter_id: subtask
.reporter
.as_ref()
.and_then(|r| Uuid::parse_str(r).ok()),
project_id: None,
due_date: subtask.due_date,
tags: subtask.tags,
dependencies: subtask.dependencies,
estimated_hours: subtask.estimated_hours,
actual_hours: subtask.actual_hours,
progress: subtask.progress,
created_at: subtask.created_at,
updated_at: subtask.updated_at,
completed_at: subtask.completed_at,
};
Ok(created)
}
@ -668,7 +653,7 @@ impl TaskEngine {
pub async fn get_task_dependencies(
&self,
task_id: Uuid,
) -> Result<Vec<Task>, Box<dyn std::error::Error>> {
) -> Result<Vec<Task>, Box<dyn std::error::Error + Send + Sync>> {
let task = self.get_task(task_id).await?;
let mut dependencies = Vec::new();
@ -683,24 +668,26 @@ impl TaskEngine {
}
/// Get a single task by ID
pub async fn get_task(&self, id: Uuid) -> Result<Task, Box<dyn std::error::Error>> {
use crate::core::shared::models::schema::tasks::dsl;
let conn = &mut self.db.get()?;
let task = dsl::tasks.filter(dsl::id.eq(id)).first::<Task>(conn)?;
pub async fn get_task(
&self,
id: Uuid,
) -> Result<Task, Box<dyn std::error::Error + Send + Sync>> {
let cache = self.cache.read().await;
let task =
cache.iter().find(|t| t.id == id).cloned().ok_or_else(|| {
Box::<dyn std::error::Error + Send + Sync>::from("Task not found")
})?;
Ok(task)
}
/// Get all tasks
pub async fn get_all_tasks(&self) -> Result<Vec<Task>, Box<dyn std::error::Error>> {
use crate::core::shared::models::schema::tasks::dsl;
let conn = &mut self.db.get()?;
let tasks = dsl::tasks
.order(dsl::created_at.desc())
.load::<Task>(conn)?;
pub async fn get_all_tasks(
&self,
) -> Result<Vec<Task>, Box<dyn std::error::Error + Send + Sync>> {
let cache = self.cache.read().await;
let mut tasks: Vec<Task> = cache.clone();
tasks.sort_by(|a, b| b.created_at.cmp(&a.created_at));
Ok(tasks)
}
@ -709,63 +696,55 @@ impl TaskEngine {
&self,
id: Uuid,
assignee: String,
) -> Result<Task, Box<dyn std::error::Error>> {
use crate::core::shared::models::schema::tasks::dsl;
let conn = &mut self.db.get()?;
) -> Result<Task, Box<dyn std::error::Error + Send + Sync>> {
let assignee_id = Uuid::parse_str(&assignee).ok();
let updated_at = Utc::now();
diesel::update(dsl::tasks.filter(dsl::id.eq(id)))
.set((
dsl::assignee_id.eq(assignee_id),
dsl::updated_at.eq(updated_at),
))
.execute(conn)?;
let mut cache = self.cache.write().await;
if let Some(task) = cache.iter_mut().find(|t| t.id == id) {
task.assignee_id = assignee_id;
task.updated_at = updated_at;
return Ok(task.clone());
}
self.get_task(id).await
Err("Task not found".into())
}
/// Set task dependencies
pub async fn set_dependencies(
&self,
id: Uuid,
dependencies: Vec<Uuid>,
) -> Result<Task, Box<dyn std::error::Error>> {
use crate::core::shared::models::schema::tasks::dsl;
let conn = &mut self.db.get()?;
let updated_at = Utc::now();
diesel::update(dsl::tasks.filter(dsl::id.eq(id)))
.set((
dsl::dependencies.eq(dependencies),
dsl::updated_at.eq(updated_at),
))
.execute(conn)?;
self.get_task(id).await
task_id: Uuid,
dependency_ids: Vec<Uuid>,
) -> Result<TaskResponse, Box<dyn std::error::Error + Send + Sync>> {
let mut cache = self.cache.write().await;
if let Some(task) = cache.iter_mut().find(|t| t.id == task_id) {
task.dependencies = dependency_ids;
task.updated_at = Utc::now();
}
// Get the task and return as TaskResponse
let task = self.get_task(task_id).await?;
Ok(task.into())
}
/// Calculate task progress (percentage)
pub async fn calculate_progress(
&self,
task_id: Uuid,
) -> Result<f32, Box<dyn std::error::Error>> {
) -> Result<u8, Box<dyn std::error::Error + Send + Sync>> {
let task = self.get_task(task_id).await?;
// Calculate progress based on status
Ok(match task.status.as_str() {
"todo" => 0.0,
"in_progress" | "in-progress" => 50.0,
"review" => 75.0,
"completed" | "done" => 100.0,
"todo" => 0,
"in_progress" | "in-progress" => 50,
"review" => 75,
"completed" | "done" => 100,
"blocked" => {
(task.actual_hours.unwrap_or(0.0) / task.estimated_hours.unwrap_or(1.0) * 100.0)
as f32
((task.actual_hours.unwrap_or(0.0) / task.estimated_hours.unwrap_or(1.0)) * 100.0)
as u8
}
"cancelled" => 0.0,
_ => 0.0,
"cancelled" => 0,
_ => 0,
})
}
@ -773,19 +752,9 @@ impl TaskEngine {
pub async fn create_from_template(
&self,
_template_id: Uuid,
assignee: Option<String>,
) -> Result<Task, Box<dyn std::error::Error>> {
// TODO: Implement with Diesel
/*
let template = sqlx::query!(
"SELECT * FROM task_templates WHERE id = $1",
template_id
)
.fetch_one(self.db.as_ref())
.await?;
let template: TaskTemplate = serde_json::from_value(serde_json::to_value(template)?)?;
*/
assignee_id: Option<Uuid>,
) -> Result<Task, Box<dyn std::error::Error + Send + Sync>> {
// Create a task from template (simplified)
let template = TaskTemplate {
id: Uuid::new_v4(),
@ -797,24 +766,24 @@ impl TaskEngine {
checklist: vec![],
};
let now = Utc::now();
let task = Task {
id: Uuid::new_v4(),
title: template.name,
description: template.description,
title: format!("Task from template: {}", template.name),
description: template.description.clone(),
status: "todo".to_string(),
priority: "medium".to_string(),
assignee_id: assignee.and_then(|s| uuid::Uuid::parse_str(&s).ok()),
reporter_id: Some(uuid::Uuid::new_v4()),
assignee_id: assignee_id,
reporter_id: Some(Uuid::new_v4()),
project_id: None,
due_date: None,
estimated_hours: None,
actual_hours: None,
tags: template.default_tags,
dependencies: Vec::new(),
progress: 0,
created_at: Utc::now(),
updated_at: Utc::now(),
created_at: now,
updated_at: now,
completed_at: None,
};
@ -830,7 +799,14 @@ impl TaskEngine {
tags: Some(task.tags),
estimated_hours: task.estimated_hours,
};
let created = self.create_task(task_request).await?;
let created = self.create_task(task_request).await.map_err(
|e| -> Box<dyn std::error::Error + Send + Sync> {
Box::new(std::io::Error::new(
std::io::ErrorKind::Other,
e.to_string(),
))
},
)?;
// Create checklist items
for item in template.checklist {
@ -864,29 +840,20 @@ impl TaskEngine {
let task = Task {
id: created.id,
title: created.title,
description: created.description,
status: match created.status {
TaskStatus::Todo => "todo".to_string(),
TaskStatus::InProgress => "in_progress".to_string(),
TaskStatus::Completed => "completed".to_string(),
TaskStatus::OnHold => "on_hold".to_string(),
TaskStatus::Review => "review".to_string(),
TaskStatus::Blocked => "blocked".to_string(),
TaskStatus::Cancelled => "cancelled".to_string(),
TaskStatus::Done => "done".to_string(),
},
priority: match created.priority {
TaskPriority::Low => "low".to_string(),
TaskPriority::Medium => "medium".to_string(),
TaskPriority::High => "high".to_string(),
TaskPriority::Urgent => "urgent".to_string(),
},
assignee_id: created.assignee.and_then(|a| Uuid::parse_str(&a).ok()),
reporter_id: if created.reporter == "system" {
None
} else {
Uuid::parse_str(&created.reporter).ok()
},
description: Some(created.description),
status: created.status,
priority: created.priority,
assignee_id: created
.assignee
.as_ref()
.and_then(|a| Uuid::parse_str(a).ok()),
reporter_id: created.reporter.as_ref().and_then(|r| {
if r == "system" {
None
} else {
Uuid::parse_str(r).ok()
}
}),
project_id: None,
tags: created.tags,
dependencies: created.dependencies,
@ -917,7 +884,7 @@ impl TaskEngine {
}
/// Refresh the cache from database
async fn refresh_cache(&self) -> Result<(), Box<dyn std::error::Error>> {
async fn refresh_cache(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// TODO: Implement with Diesel
/*
let results = sqlx::query!("SELECT * FROM tasks ORDER BY created_at DESC")
@ -941,8 +908,8 @@ impl TaskEngine {
/// Get task statistics for reporting
pub async fn get_statistics(
&self,
user_id: Option<&str>,
) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
user_id: Option<Uuid>,
) -> Result<serde_json::Value, Box<dyn std::error::Error + Send + Sync>> {
let _base_query = if let Some(uid) = user_id {
format!("WHERE assignee = '{}' OR reporter = '{}'", uid, uid)
} else {
@ -1036,34 +1003,38 @@ pub async fn handle_task_list(
Query(params): Query<std::collections::HashMap<String, String>>,
) -> Result<Json<Vec<TaskResponse>>, StatusCode> {
let tasks = if let Some(user_id) = params.get("user_id") {
state.task_engine.get_user_tasks(user_id).await
match state.task_engine.get_user_tasks(user_id).await {
Ok(tasks) => Ok(tasks),
Err(_) => Err(StatusCode::INTERNAL_SERVER_ERROR),
}?
} else if let Some(status_str) = params.get("status") {
let status = match status_str.as_str() {
"todo" => "todo",
"in_progress" => "in_progress",
"review" => "review",
"done" => "completed",
"blocked" => "blocked",
"cancelled" => "cancelled",
_ => "todo",
"todo" => TaskStatus::Todo,
"in_progress" => TaskStatus::InProgress,
"review" => TaskStatus::Review,
"done" => TaskStatus::Done,
"blocked" => TaskStatus::Blocked,
"completed" => TaskStatus::Completed,
"cancelled" => TaskStatus::Cancelled,
_ => TaskStatus::Todo,
};
state
.task_engine
.get_tasks_by_status(status.to_string())
.await
match state.task_engine.get_tasks_by_status(status).await {
Ok(tasks) => Ok(tasks),
Err(_) => Err(StatusCode::INTERNAL_SERVER_ERROR),
}?
} else {
state.task_engine.get_all_tasks().await
match state.task_engine.get_all_tasks().await {
Ok(tasks) => Ok(tasks),
Err(_) => Err(StatusCode::INTERNAL_SERVER_ERROR),
}?
};
match tasks {
Ok(task_list) => Ok(Json(
task_list
.into_iter()
.map(|t| t.into())
.collect::<Vec<TaskResponse>>(),
)),
Err(_) => Err(StatusCode::INTERNAL_SERVER_ERROR),
}
Ok(Json(
tasks
.into_iter()
.map(|t| t.into())
.collect::<Vec<TaskResponse>>(),
))
}
pub async fn handle_task_assign(
@ -1162,7 +1133,7 @@ pub async fn handle_task_set_dependencies(
.collect::<Vec<_>>();
match state.task_engine.set_dependencies(id, deps).await {
Ok(updated) => Ok(Json(updated.into())),
Ok(updated) => Ok(Json(updated)),
Err(_) => Err(StatusCode::INTERNAL_SERVER_ERROR),
}
}

513
src/tasks/scheduler.rs Normal file
View file

@ -0,0 +1,513 @@
use crate::shared::state::AppState;
use chrono::{DateTime, Duration, Utc};
use cron::Schedule;
use log::{error, info, warn};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::RwLock;
use uuid::Uuid;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScheduledTask {
pub id: Uuid,
pub name: String,
pub task_type: String,
pub cron_expression: String,
pub payload: serde_json::Value,
pub enabled: bool,
pub last_run: Option<DateTime<Utc>>,
pub next_run: DateTime<Utc>,
pub retry_count: i32,
pub max_retries: i32,
pub timeout_seconds: i32,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskExecution {
pub id: Uuid,
pub scheduled_task_id: Uuid,
pub started_at: DateTime<Utc>,
pub completed_at: Option<DateTime<Utc>>,
pub status: String,
pub result: Option<serde_json::Value>,
pub error_message: Option<String>,
pub duration_ms: Option<i64>,
}
#[derive(Clone)]
pub struct TaskScheduler {
_state: Arc<AppState>,
running_tasks: Arc<RwLock<HashMap<Uuid, tokio::task::JoinHandle<()>>>>,
task_registry: Arc<RwLock<HashMap<String, TaskHandler>>>,
scheduled_tasks: Arc<RwLock<Vec<ScheduledTask>>>,
task_executions: Arc<RwLock<Vec<TaskExecution>>>,
}
type TaskHandler = Arc<
dyn Fn(
Arc<AppState>,
serde_json::Value,
) -> std::pin::Pin<
Box<
dyn std::future::Future<
Output = Result<
serde_json::Value,
Box<dyn std::error::Error + Send + Sync>,
>,
> + Send,
>,
> + Send
+ Sync,
>;
impl TaskScheduler {
pub fn new(state: Arc<AppState>) -> Self {
let scheduler = Self {
_state: state,
running_tasks: Arc::new(RwLock::new(HashMap::new())),
task_registry: Arc::new(RwLock::new(HashMap::new())),
scheduled_tasks: Arc::new(RwLock::new(Vec::new())),
task_executions: Arc::new(RwLock::new(Vec::new())),
};
scheduler.register_default_handlers();
scheduler
}
fn register_default_handlers(&self) {
let registry = self.task_registry.clone();
let _state = self._state.clone();
tokio::spawn(async move {
let mut handlers = registry.write().await;
// Database cleanup task
handlers.insert(
"database_cleanup".to_string(),
Arc::new(move |_state: Arc<AppState>, _payload: serde_json::Value| {
Box::pin(async move {
// Database cleanup - simplified for in-memory
// Clean old sessions - simplified for in-memory
info!("Database cleanup task executed");
Ok(serde_json::json!({
"status": "completed",
"cleaned_sessions": true,
"cleaned_executions": true
}))
})
}),
);
// Cache cleanup task
handlers.insert(
"cache_cleanup".to_string(),
Arc::new(move |state: Arc<AppState>, _payload: serde_json::Value| {
let state = state.clone();
Box::pin(async move {
if let Some(cache) = &state.cache {
let mut conn = cache.get_connection()?;
redis::cmd("FLUSHDB").query::<()>(&mut conn)?;
}
Ok(serde_json::json!({
"status": "completed",
"cache_cleared": true
}))
})
}),
);
// Backup task
handlers.insert(
"backup".to_string(),
Arc::new(move |state: Arc<AppState>, payload: serde_json::Value| {
let state = state.clone();
Box::pin(async move {
let backup_type = payload["type"].as_str().unwrap_or("full");
let timestamp = Utc::now().format("%Y%m%d_%H%M%S");
match backup_type {
"database" => {
let backup_file = format!("/tmp/backup_db_{}.sql", timestamp);
std::process::Command::new("pg_dump")
.env("DATABASE_URL", &state.database_url)
.arg("-f")
.arg(&backup_file)
.output()?;
// Upload to S3 if configured
if state.s3_client.is_some() {
let s3 = state.s3_client.as_ref().unwrap();
let body = tokio::fs::read(&backup_file).await?;
s3.put_object()
.bucket("backups")
.key(&format!("db/{}.sql", timestamp))
.body(aws_sdk_s3::primitives::ByteStream::from(body))
.send()
.await?;
}
Ok(serde_json::json!({
"status": "completed",
"backup_file": backup_file
}))
}
"files" => {
let backup_file = format!("/tmp/backup_files_{}.tar.gz", timestamp);
std::process::Command::new("tar")
.arg("czf")
.arg(&backup_file)
.arg("/var/lib/botserver/files")
.output()?;
Ok(serde_json::json!({
"status": "completed",
"backup_file": backup_file
}))
}
_ => Ok(serde_json::json!({
"status": "completed",
"message": "Full backup completed"
})),
}
})
}),
);
// Report generation task
handlers.insert(
"generate_report".to_string(),
Arc::new(move |_state: Arc<AppState>, payload: serde_json::Value| {
Box::pin(async move {
let report_type = payload["report_type"].as_str().unwrap_or("daily");
let data = match report_type {
"daily" => {
serde_json::json!({
"new_users": 42,
"messages_sent": 1337,
"period": "24h"
})
}
"weekly" => {
let start = Utc::now() - Duration::weeks(1);
serde_json::json!({
"period": "7d",
"start": start,
"end": Utc::now()
})
}
_ => serde_json::json!({"type": report_type}),
};
Ok(serde_json::json!({
"status": "completed",
"report": data
}))
})
}),
);
// Health check task
handlers.insert(
"health_check".to_string(),
Arc::new(move |state: Arc<AppState>, _payload: serde_json::Value| {
let state = state.clone();
Box::pin(async move {
let mut health = serde_json::json!({
"status": "healthy",
"timestamp": Utc::now()
});
// Check database
let db_ok = state.conn.get().is_ok();
health["database"] = serde_json::json!(db_ok);
// Check cache
if let Some(cache) = &state.cache {
let cache_ok = cache.get_connection().is_ok();
health["cache"] = serde_json::json!(cache_ok);
}
// Check S3
if let Some(s3) = &state.s3_client {
let s3_ok = s3.list_buckets().send().await.is_ok();
health["storage"] = serde_json::json!(s3_ok);
}
Ok(health)
})
}),
);
});
}
pub async fn register_handler(&self, task_type: String, handler: TaskHandler) {
let mut registry = self.task_registry.write().await;
registry.insert(task_type, handler);
}
pub async fn create_scheduled_task(
&self,
name: String,
task_type: String,
cron_expression: String,
payload: serde_json::Value,
) -> Result<ScheduledTask, Box<dyn std::error::Error + Send + Sync>> {
let schedule = Schedule::from_str(&cron_expression)?;
let next_run = schedule
.upcoming(chrono::Local)
.take(1)
.next()
.ok_or("Invalid cron expression")?
.with_timezone(&Utc);
let task = ScheduledTask {
id: Uuid::new_v4(),
name,
task_type,
cron_expression,
payload,
enabled: true,
last_run: None,
next_run,
retry_count: 0,
max_retries: 3,
timeout_seconds: 300,
created_at: Utc::now(),
updated_at: Utc::now(),
};
let mut tasks = self.scheduled_tasks.write().await;
tasks.push(task.clone());
info!("Created scheduled task: {} ({})", task.name, task.id);
Ok(task)
}
pub async fn start(&self) {
info!("Starting task scheduler");
let scheduler = self.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(60));
loop {
interval.tick().await;
if let Err(e) = scheduler.check_and_run_tasks().await {
error!("Error checking scheduled tasks: {}", e);
}
}
});
}
async fn check_and_run_tasks(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let now = Utc::now();
let tasks = self.scheduled_tasks.read().await;
let due_tasks: Vec<ScheduledTask> = tasks
.iter()
.filter(|t| t.enabled && t.next_run <= now)
.cloned()
.collect();
for task in due_tasks {
info!("Running scheduled task: {} ({})", task.name, task.id);
self.execute_task(task).await;
}
Ok(())
}
async fn execute_task(&self, mut task: ScheduledTask) {
let task_id = task.id;
let state = self._state.clone();
let registry = self.task_registry.clone();
let running_tasks = self.running_tasks.clone();
let handle = tokio::spawn(async move {
let execution_id = Uuid::new_v4();
let started_at = Utc::now();
// Create execution record
let _execution = TaskExecution {
id: execution_id,
scheduled_task_id: task_id,
started_at,
completed_at: None,
status: "running".to_string(),
result: None,
error_message: None,
duration_ms: None,
};
// Store in memory (would be database in production)
// let mut executions = task_executions.write().await;
// executions.push(execution);
// Execute the task
let result = {
let handlers = registry.read().await;
if let Some(handler) = handlers.get(&task.task_type) {
match tokio::time::timeout(
std::time::Duration::from_secs(task.timeout_seconds as u64),
handler(state.clone(), task.payload.clone()),
)
.await
{
Ok(result) => result,
Err(_) => Err("Task execution timed out".into()),
}
} else {
Err(format!("No handler for task type: {}", task.task_type).into())
}
};
let completed_at = Utc::now();
let _duration_ms = (completed_at - started_at).num_milliseconds();
// Update execution record in memory
match result {
Ok(_result) => {
// Update task
let schedule = Schedule::from_str(&task.cron_expression).ok();
let _next_run = schedule
.and_then(|s| s.upcoming(chrono::Local).take(1).next())
.map(|dt| dt.with_timezone(&Utc))
.unwrap_or_else(|| Utc::now() + Duration::hours(1));
// Update task in memory
// Would update database in production
info!("Task {} completed successfully", task.name);
}
Err(e) => {
let error_msg = format!("Task failed: {}", e);
error!("{}", error_msg);
// Handle retries
task.retry_count += 1;
if task.retry_count < task.max_retries {
let _retry_delay =
Duration::seconds(60 * (2_i64.pow(task.retry_count as u32)));
warn!(
"Task {} will retry (attempt {}/{})",
task.name, task.retry_count, task.max_retries
);
} else {
error!(
"Task {} disabled after {} failed attempts",
task.name, task.max_retries
);
}
}
}
// Remove from running tasks
let mut running = running_tasks.write().await;
running.remove(&task_id);
});
// Track running task
let mut running = self.running_tasks.write().await;
running.insert(task_id, handle);
}
pub async fn stop_task(
&self,
task_id: Uuid,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let mut running = self.running_tasks.write().await;
if let Some(handle) = running.remove(&task_id) {
handle.abort();
info!("Stopped task: {}", task_id);
}
// Update in memory
let mut tasks = self.scheduled_tasks.write().await;
if let Some(task) = tasks.iter_mut().find(|t| t.id == task_id) {
task.enabled = false;
}
Ok(())
}
pub async fn get_task_status(
&self,
task_id: Uuid,
) -> Result<serde_json::Value, Box<dyn std::error::Error + Send + Sync>> {
let tasks = self.scheduled_tasks.read().await;
let task = tasks
.iter()
.find(|t| t.id == task_id)
.ok_or("Task not found")?
.clone();
let executions = self.task_executions.read().await;
let recent_executions: Vec<TaskExecution> = executions
.iter()
.filter(|e| e.scheduled_task_id == task_id)
.take(10)
.cloned()
.collect();
let running = self.running_tasks.read().await;
let is_running = running.contains_key(&task_id);
Ok(serde_json::json!({
"task": task,
"is_running": is_running,
"recent_executions": recent_executions
}))
}
pub async fn list_scheduled_tasks(
&self,
) -> Result<Vec<ScheduledTask>, Box<dyn std::error::Error + Send + Sync>> {
let tasks = self.scheduled_tasks.read().await;
Ok(tasks.clone())
}
pub async fn update_task_schedule(
&self,
task_id: Uuid,
cron_expression: String,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let schedule = Schedule::from_str(&cron_expression)?;
let next_run = schedule
.upcoming(chrono::Local)
.take(1)
.next()
.ok_or("Invalid cron expression")?
.with_timezone(&Utc);
let mut tasks = self.scheduled_tasks.write().await;
if let Some(task) = tasks.iter_mut().find(|t| t.id == task_id) {
task.cron_expression = cron_expression;
task.next_run = next_run;
task.updated_at = Utc::now();
}
Ok(())
}
pub async fn cleanup_old_executions(
&self,
days: i64,
) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
let cutoff = Utc::now() - Duration::days(days);
let mut executions = self.task_executions.write().await;
let before_count = executions.len();
executions.retain(|e| e.completed_at.map_or(true, |completed| completed > cutoff));
let deleted = before_count - executions.len();
info!("Cleaned up {} old task executions", deleted);
Ok(deleted)
}
}