Hexor/OutFleet
1
0
Fork 0
mirror of https://github.com/house-of-vanity/OutFleet.git synced 2025-10-26 10:09:08 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Added usermanagement in TG admin

Browse Source
This commit is contained in:
AB from home.homenet
2025-10-24 18:11:34 +03:00
parent c6892b1a73
commit 78bf75b24e
89 changed files with 4389 additions and 2419 deletions
Download Patch File Download Diff File Expand all files Collapse all files

377
src/services/tasks.rs
View File

@@ -1,18 +1,21 @@
use anyhow::Result;
use tokio_cron_scheduler::{JobScheduler, Job};
use tracing::{info, error, warn, debug};
use crate::database::DatabaseManager;
use crate::database::repository::{ServerRepository, ServerInboundRepository, InboundTemplateRepository, InboundUsersRepository, CertificateRepository, UserRepository};
use crate::database::entities::inbound_users;
use crate::services::XrayService;
use crate::database::repository::{
CertificateRepository, InboundTemplateRepository, InboundUsersRepository,
ServerInboundRepository, ServerRepository, UserRepository,
};
use crate::database::DatabaseManager;
use crate::services::events::SyncEvent;
use sea_orm::{EntityTrait, ColumnTrait, QueryFilter, RelationTrait, JoinType};
use uuid::Uuid;
use crate::services::XrayService;
use anyhow::Result;
use chrono::{DateTime, Utc};
use sea_orm::{ColumnTrait, EntityTrait, JoinType, QueryFilter, RelationTrait};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use chrono::{DateTime, Utc};
use serde::{Serialize, Deserialize};
use tokio_cron_scheduler::{Job, JobScheduler};
use tracing::{debug, error, info, warn};
use uuid::Uuid;
pub struct TaskScheduler {
scheduler: JobScheduler,
@@ -47,7 +50,10 @@ impl TaskScheduler {
pub async fn new() -> Result<Self> {
let scheduler = JobScheduler::new().await?;
let task_status = Arc::new(RwLock::new(HashMap::new()));
Ok(Self { scheduler, task_status })
Ok(Self {
scheduler,
task_status,
})
}
/// Get current status of all tasks
@@ -56,15 +62,20 @@ impl TaskScheduler {
}
/// Start event-driven sync handler
pub async fn start_event_handler(db: DatabaseManager, mut event_receiver: tokio::sync::broadcast::Receiver<SyncEvent>) {
pub async fn start_event_handler(
db: DatabaseManager,
mut event_receiver: tokio::sync::broadcast::Receiver<SyncEvent>,
) {
let xray_service = XrayService::new();
tokio::spawn(async move {
while let Ok(event) = event_receiver.recv().await {
match event {
SyncEvent::InboundChanged(server_id) | SyncEvent::UserAccessChanged(server_id) => {
if let Err(e) = sync_single_server_by_id(&xray_service, &db, server_id).await {
SyncEvent::InboundChanged(server_id)
| SyncEvent::UserAccessChanged(server_id) => {
if let Err(e) =
sync_single_server_by_id(&xray_service, &db, server_id).await
{
error!("Failed to sync server {} from event: {}", server_id, e);
}
}
@@ -74,34 +85,36 @@ impl TaskScheduler {
}
pub async fn start(&mut self, db: DatabaseManager, xray_service: XrayService) -> Result<()> {
// Initialize task status
{
let mut status = self.task_status.write().unwrap();
status.insert("xray_sync".to_string(), TaskStatus {
name: "Xray Synchronization".to_string(),
description: "Synchronizes database state with xray servers".to_string(),
schedule: "0 * * * * * (every minute)".to_string(),
status: TaskState::Idle,
last_run: None,
next_run: Some(Utc::now() + chrono::Duration::minutes(1)),
total_runs: 0,
success_count: 0,
error_count: 0,
last_error: None,
last_duration_ms: None,
});
status.insert(
"xray_sync".to_string(),
TaskStatus {
name: "Xray Synchronization".to_string(),
description: "Synchronizes database state with xray servers".to_string(),
schedule: "0 * * * * * (every minute)".to_string(),
status: TaskState::Idle,
last_run: None,
next_run: Some(Utc::now() + chrono::Duration::minutes(1)),
total_runs: 0,
success_count: 0,
error_count: 0,
last_error: None,
last_duration_ms: None,
},
);
}
// Run initial sync in background to avoid blocking startup
let db_initial = db.clone();
let xray_service_initial = xray_service.clone();
let task_status_initial = self.task_status.clone();
tokio::spawn(async move {
info!("Starting initial xray sync in background...");
let start_time = Utc::now();
// Update status to running
{
let mut status = task_status_initial.write().unwrap();
@@ -111,7 +124,7 @@ impl TaskScheduler {
task.total_runs += 1;
}
}
match sync_xray_state(db_initial, xray_service_initial).await {
Ok(_) => {
let duration = (Utc::now() - start_time).num_milliseconds() as u64;
@@ -123,7 +136,7 @@ impl TaskScheduler {
task.last_error = None;
}
info!("Initial xray sync completed successfully in {}ms", duration);
},
}
Err(e) => {
let duration = (Utc::now() - start_time).num_milliseconds() as u64;
let mut status = task_status_initial.write().unwrap();
@@ -137,20 +150,20 @@ impl TaskScheduler {
}
}
});
// Add synchronization task that runs every minute
let db_clone = db.clone();
let xray_service_clone = xray_service.clone();
let task_status_clone = self.task_status.clone();
let sync_job = Job::new_async("0 */5 * * * *", move |_uuid, _l| {
let db = db_clone.clone();
let xray_service = xray_service_clone.clone();
let task_status = task_status_clone.clone();
Box::pin(async move {
let start_time = Utc::now();
// Update status to running
{
let mut status = task_status.write().unwrap();
@@ -161,7 +174,7 @@ impl TaskScheduler {
task.next_run = Some(start_time + chrono::Duration::minutes(1));
}
}
match sync_xray_state(db, xray_service).await {
Ok(_) => {
let duration = (Utc::now() - start_time).num_milliseconds() as u64;
@@ -172,7 +185,7 @@ impl TaskScheduler {
task.last_duration_ms = Some(duration);
task.last_error = None;
}
},
}
Err(e) => {
let duration = (Utc::now() - start_time).num_milliseconds() as u64;
let mut status = task_status.write().unwrap();
@@ -187,38 +200,42 @@ impl TaskScheduler {
}
})
})?;
self.scheduler.add(sync_job).await?;
// Add certificate renewal task that runs once a day at 2 AM
let db_clone_cert = db.clone();
let task_status_cert = self.task_status.clone();
// Initialize certificate renewal task status
{
let mut status = self.task_status.write().unwrap();
status.insert("cert_renewal".to_string(), TaskStatus {
name: "Certificate Renewal".to_string(),
description: "Renews Let's Encrypt certificates that expire within 15 days".to_string(),
schedule: "0 0 2 * * * (daily at 2 AM)".to_string(),
status: TaskState::Idle,
last_run: None,
next_run: Some(Utc::now() + chrono::Duration::days(1)),
total_runs: 0,
success_count: 0,
error_count: 0,
last_error: None,
last_duration_ms: None,
});
status.insert(
"cert_renewal".to_string(),
TaskStatus {
name: "Certificate Renewal".to_string(),
description: "Renews Let's Encrypt certificates that expire within 15 days"
.to_string(),
schedule: "0 0 2 * * * (daily at 2 AM)".to_string(),
status: TaskState::Idle,
last_run: None,
next_run: Some(Utc::now() + chrono::Duration::days(1)),
total_runs: 0,
success_count: 0,
error_count: 0,
last_error: None,
last_duration_ms: None,
},
);
}
let cert_renewal_job = Job::new_async("0 0 2 * * *", move |_uuid, _l| {
let db = db_clone_cert.clone();
let task_status = task_status_cert.clone();
Box::pin(async move {
let start_time = Utc::now();
// Update task status to running
{
let mut status = task_status.write().unwrap();
@@ -228,7 +245,7 @@ impl TaskScheduler {
task.total_runs += 1;
}
}
match check_and_renew_certificates(&db).await {
Ok(_) => {
let duration = (Utc::now() - start_time).num_milliseconds() as u64;
@@ -239,7 +256,7 @@ impl TaskScheduler {
task.last_duration_ms = Some(duration);
task.last_error = None;
}
},
}
Err(e) => {
let duration = (Utc::now() - start_time).num_milliseconds() as u64;
let mut status = task_status.write().unwrap();
@@ -254,9 +271,9 @@ impl TaskScheduler {
}
})
})?;
self.scheduler.add(cert_renewal_job).await?;
// Also run certificate check on startup
info!("Running initial certificate renewal check...");
tokio::spawn(async move {
@@ -264,7 +281,7 @@ impl TaskScheduler {
error!("Initial certificate renewal check failed: {}", e);
}
});
self.scheduler.start().await?;
Ok(())
}
@@ -281,7 +298,12 @@ impl TaskScheduler {
}
}
fn update_task_status_with_error(&self, task_id: &str, error: String, duration_ms: Option<u64>) {
fn update_task_status_with_error(
&self,
task_id: &str,
error: String,
duration_ms: Option<u64>,
) {
let mut status = self.task_status.write().unwrap();
if let Some(task) = status.get_mut(task_id) {
task.status = TaskState::Error;
@@ -301,11 +323,10 @@ impl TaskScheduler {
/// Synchronize xray server state with database state
async fn sync_xray_state(db: DatabaseManager, xray_service: XrayService) -> Result<()> {
let server_repo = ServerRepository::new(db.connection().clone());
let inbound_repo = ServerInboundRepository::new(db.connection().clone());
let template_repo = InboundTemplateRepository::new(db.connection().clone());
// Get all servers from database
let servers = match server_repo.find_all().await {
Ok(servers) => servers,
@@ -314,50 +335,50 @@ async fn sync_xray_state(db: DatabaseManager, xray_service: XrayService) -> Resu
return Err(e.into());
}
};
for server in servers {
let endpoint = server.get_grpc_endpoint();
// Test connection first
match xray_service.test_connection(server.id, &endpoint).await {
Ok(false) => {
warn!("Cannot connect to server {} at {}, skipping", server.name, endpoint);
warn!(
"Cannot connect to server {} at {}, skipping",
server.name, endpoint
);
continue;
},
}
Err(e) => {
error!("Error testing connection to server {}: {}", server.name, e);
continue;
}
_ => {}
}
// Get desired inbounds from database
let desired_inbounds = match get_desired_inbounds_from_db(&db, &server, &inbound_repo, &template_repo).await {
Ok(inbounds) => inbounds,
Err(e) => {
error!("Failed to get desired inbounds for server {}: {}", server.name, e);
continue;
}
};
let desired_inbounds =
match get_desired_inbounds_from_db(&db, &server, &inbound_repo, &template_repo).await {
Ok(inbounds) => inbounds,
Err(e) => {
error!(
"Failed to get desired inbounds for server {}: {}",
server.name, e
);
continue;
}
};
// Synchronize inbounds
if let Err(e) = sync_server_inbounds(
&xray_service,
server.id,
&endpoint,
&desired_inbounds
).await {
if let Err(e) =
sync_server_inbounds(&xray_service, server.id, &endpoint, &desired_inbounds).await
{
error!("Failed to sync inbounds for server {}: {}", server.name, e);
}
}
Ok(())
}
/// Get desired inbounds configuration from database
async fn get_desired_inbounds_from_db(
db: &DatabaseManager,
@@ -365,38 +386,47 @@ async fn get_desired_inbounds_from_db(
inbound_repo: &ServerInboundRepository,
template_repo: &InboundTemplateRepository,
) -> Result<HashMap<String, DesiredInbound>> {
// Get all inbounds for this server
let inbounds = inbound_repo.find_by_server_id(server.id).await?;
let mut desired_inbounds = HashMap::new();
for inbound in inbounds {
// Get template for this inbound
let template = match template_repo.find_by_id(inbound.template_id).await? {
Some(template) => template,
None => {
warn!("Template {} not found for inbound {}, skipping", inbound.template_id, inbound.tag);
warn!(
"Template {} not found for inbound {}, skipping",
inbound.template_id, inbound.tag
);
continue;
}
};
// Get users for this inbound
let users = get_users_for_inbound(db, inbound.id).await?;
// Get port from template or override
let port = inbound.port_override.unwrap_or(template.default_port);
// Get certificate if specified
let (cert_pem, key_pem) = if let Some(cert_id) = inbound.certificate_id {
match load_certificate_from_db(db, inbound.certificate_id).await {
Ok((cert, key)) => {
info!("Loaded certificate {} for inbound {}, has_cert={}, has_key={}",
cert_id, inbound.tag, cert.is_some(), key.is_some());
info!(
"Loaded certificate {} for inbound {}, has_cert={}, has_key={}",
cert_id,
inbound.tag,
cert.is_some(),
key.is_some()
);
(cert, key)
},
}
Err(e) => {
warn!("Failed to load certificate {} for inbound {}: {}", cert_id, inbound.tag, e);
warn!(
"Failed to load certificate {} for inbound {}: {}",
cert_id, inbound.tag, e
);
(None, None)
}
}
@@ -404,7 +434,7 @@ async fn get_desired_inbounds_from_db(
debug!("No certificate configured for inbound {}", inbound.tag);
(None, None)
};
let desired_inbound = DesiredInbound {
tag: inbound.tag.clone(),
port,
@@ -415,22 +445,24 @@ async fn get_desired_inbounds_from_db(
cert_pem,
key_pem,
};
desired_inbounds.insert(inbound.tag.clone(), desired_inbound);
}
Ok(desired_inbounds)
}
/// Get users for specific inbound from database
async fn get_users_for_inbound(db: &DatabaseManager, inbound_id: Uuid) -> Result<Vec<XrayUser>> {
let inbound_users_repo = InboundUsersRepository::new(db.connection().clone());
let inbound_users = inbound_users_repo.find_active_by_inbound_id(inbound_id).await?;
let inbound_users = inbound_users_repo
.find_active_by_inbound_id(inbound_id)
.await?;
// Get user details to generate emails
let user_repo = UserRepository::new(db.connection().clone());
let mut users: Vec<XrayUser> = Vec::new();
for inbound_user in inbound_users {
if let Some(user) = user_repo.find_by_id(inbound_user.user_id).await? {
@@ -442,23 +474,24 @@ async fn get_users_for_inbound(db: &DatabaseManager, inbound_id: Uuid) -> Result
});
}
}
Ok(users)
}
/// Load certificate from database
async fn load_certificate_from_db(db: &DatabaseManager, cert_id: Option<Uuid>) -> Result<(Option<String>, Option<String>)> {
async fn load_certificate_from_db(
db: &DatabaseManager,
cert_id: Option<Uuid>,
) -> Result<(Option<String>, Option<String>)> {
let cert_id = match cert_id {
Some(id) => id,
None => return Ok((None, None)),
};
let cert_repo = CertificateRepository::new(db.connection().clone());
match cert_repo.find_by_id(cert_id).await? {
Some(cert) => {
Ok((Some(cert.certificate_pem()), Some(cert.private_key_pem())))
},
Some(cert) => Ok((Some(cert.certificate_pem()), Some(cert.private_key_pem()))),
None => {
warn!("Certificate {} not found", cert_id);
Ok((None, None))
@@ -474,7 +507,9 @@ async fn sync_server_inbounds(
desired_inbounds: &HashMap<String, DesiredInbound>,
) -> Result<()> {
// Use optimized batch sync with single client
xray_service.sync_server_inbounds_optimized(server_id, endpoint, desired_inbounds).await
xray_service
.sync_server_inbounds_optimized(server_id, endpoint, desired_inbounds)
.await
}
/// Sync a single server by ID (for event-driven sync)
@@ -486,7 +521,7 @@ async fn sync_single_server_by_id(
let server_repo = ServerRepository::new(db.connection().clone());
let inbound_repo = ServerInboundRepository::new(db.connection().clone());
let template_repo = InboundTemplateRepository::new(db.connection().clone());
// Get server
let server = match server_repo.find_by_id(server_id).await? {
Some(server) => server,
@@ -495,22 +530,22 @@ async fn sync_single_server_by_id(
return Ok(());
}
};
// For now, sync all servers (can add active/inactive flag later)
// Get desired inbounds from database
let desired_inbounds = get_desired_inbounds_from_db(db, &server, &inbound_repo, &template_repo).await?;
let desired_inbounds =
get_desired_inbounds_from_db(db, &server, &inbound_repo, &template_repo).await?;
// Build endpoint
let endpoint = server.get_grpc_endpoint();
// Sync server
sync_server_inbounds(xray_service, server_id, &endpoint, &desired_inbounds).await?;
Ok(())
}
/// Represents desired inbound configuration from database
#[derive(Debug, Clone)]
pub struct DesiredInbound {
@@ -534,73 +569,79 @@ pub struct XrayUser {
/// Check and renew certificates that expire within 15 days
async fn check_and_renew_certificates(db: &DatabaseManager) -> Result<()> {
use crate::services::certificates::CertificateService;
use crate::database::repository::DnsProviderRepository;
use crate::services::certificates::CertificateService;
info!("Starting certificate renewal check...");
let cert_repo = CertificateRepository::new(db.connection().clone());
let dns_repo = DnsProviderRepository::new(db.connection().clone());
let cert_service = CertificateService::with_db(db.connection().clone());
// Get all certificates
let certificates = cert_repo.find_all().await?;
let mut renewed_count = 0;
let mut checked_count = 0;
for cert in certificates {
// Only check Let's Encrypt certificates with auto_renew enabled
if cert.cert_type != "letsencrypt" || !cert.auto_renew {
continue;
}
checked_count += 1;
// Check if certificate expires within 15 days
if cert.expires_soon(15) {
info!(
"Certificate '{}' (ID: {}) expires at {} - renewing...",
"Certificate '{}' (ID: {}) expires at {} - renewing...",
cert.name, cert.id, cert.expires_at
);
// Find the DNS provider used for this certificate
// For now, we'll use the first active Cloudflare provider
// In production, you might want to store the provider ID with the certificate
let providers = dns_repo.find_active_by_type("cloudflare").await?;
if providers.is_empty() {
error!(
"Cannot renew certificate '{}': No active Cloudflare DNS provider found",
"Cannot renew certificate '{}': No active Cloudflare DNS provider found",
cert.name
);
continue;
}
let dns_provider = &providers[0];
// Need to get the ACME email - for now using a default
// In production, this should be stored with the certificate
let acme_email = "admin@example.com"; // TODO: Store this with certificate
// Attempt to renew the certificate
match cert_service.generate_letsencrypt_certificate(
&cert.domain,
dns_provider.id,
acme_email,
false, // Use production Let's Encrypt
).await {
match cert_service
.generate_letsencrypt_certificate(
&cert.domain,
dns_provider.id,
acme_email,
false, // Use production Let's Encrypt
)
.await
{
Ok((new_cert_pem, new_key_pem)) => {
// Update the certificate in database
match cert_repo.update_certificate_data(
cert.id,
&new_cert_pem,
&new_key_pem,
chrono::Utc::now() + chrono::Duration::days(90), // Let's Encrypt certs are valid for 90 days
).await {
match cert_repo
.update_certificate_data(
cert.id,
&new_cert_pem,
&new_key_pem,
chrono::Utc::now() + chrono::Duration::days(90), // Let's Encrypt certs are valid for 90 days
)
.await
{
Ok(_) => {
info!("Successfully renewed certificate '{}'", cert.name);
renewed_count += 1;
// Trigger sync for all servers using this certificate
// This will be done via the event system
if let Err(e) = trigger_cert_renewal_sync(db, cert.id).await {
@@ -608,7 +649,10 @@ async fn check_and_renew_certificates(db: &DatabaseManager) -> Result<()> {
}
}
Err(e) => {
error!("Failed to save renewed certificate '{}' to database: {}", cert.name, e);
error!(
"Failed to save renewed certificate '{}' to database: {}",
cert.name, e
);
}
}
}
@@ -618,17 +662,17 @@ async fn check_and_renew_certificates(db: &DatabaseManager) -> Result<()> {
}
} else {
debug!(
"Certificate '{}' expires at {} - no renewal needed yet",
"Certificate '{}' expires at {} - no renewal needed yet",
cert.name, cert.expires_at
);
}
}
info!(
"Certificate renewal check completed: checked {}, renewed {}",
"Certificate renewal check completed: checked {}, renewed {}",
checked_count, renewed_count
);
Ok(())
}
@@ -636,23 +680,26 @@ async fn check_and_renew_certificates(db: &DatabaseManager) -> Result<()> {
async fn trigger_cert_renewal_sync(db: &DatabaseManager, cert_id: Uuid) -> Result<()> {
use crate::services::events::send_sync_event;
use crate::services::events::SyncEvent;
let inbound_repo = ServerInboundRepository::new(db.connection().clone());
// Find all server inbounds that use this certificate
let inbounds = inbound_repo.find_by_certificate_id(cert_id).await?;
// Collect unique server IDs
let mut server_ids = std::collections::HashSet::new();
for inbound in inbounds {
server_ids.insert(inbound.server_id);
}
// Trigger sync for each server
for server_id in server_ids {
info!("Triggering sync for server {} after certificate renewal", server_id);
info!(
"Triggering sync for server {} after certificate renewal",
server_id
);
send_sync_event(SyncEvent::InboundChanged(server_id));
}
Ok(())
}
}