eba4fd1
use std::time::Instant;
use std::collections::HashMap;
use models::chat::messages;
use models::issue::issues;
use models::repo::repos;
use models::room::rooms;
use models::space::spaces;
use models::users::{user, user_sessions};
use sea_orm::{
ColumnTrait, ConnectionTrait, DbBackend, EntityTrait, PaginatorTrait, QueryFilter, Statement,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use crate::errors::AdminError;
use crate::services::logs::JobLogDto;
use crate::services::queues::QueueSnapshotDto;
use crate::services::service::AdminService;
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct ProcessInfoDto {
pub version: String,
pub uptime_secs: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct DependencyDto {
pub ok: bool,
pub latency_ms: Option<u64>,
pub detail: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct DependenciesDto {
pub database: DependencyDto,
pub redis: DependencyDto,
pub storage: DependencyDto,
pub search: DependencyDto,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct FeaturesDto {
pub search_enabled: bool,
pub import_enabled: bool,
pub email_enabled: bool,
pub email_queue_enabled: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct CountsDto {
pub users: i64,
pub active_users: i64,
pub spaces: i64,
pub active_spaces: i64,
pub repos: i64,
pub rooms: i64,
pub issues: i64,
pub messages: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct SystemStatusDto {
pub process: ProcessInfoDto,
pub dependencies: DependenciesDto,
pub features: FeaturesDto,
pub counts: CountsDto,
pub queues: Vec<QueueSnapshotDto>,
pub recent_failures: Vec<JobLogDto>,
pub generated_at: OffsetDateTime,
}
fn latency_ms(start: Instant) -> u64 {
start.elapsed().as_millis() as u64
}
fn info_value(info: &str, key: &str) -> Option<String> {
info.lines()
.find_map(|line| line.strip_prefix(key))
.map(|value| value.trim().to_string())
}
impl AdminService {
async fn database_health(&self) -> DependencyDto {
let start = Instant::now();
let result = self
.db()
.read()
.query_one_raw(Statement::from_string(DbBackend::Postgres, "SELECT 1"))
.await;
match result {
Ok(_) => DependencyDto {
ok: true,
latency_ms: Some(latency_ms(start)),
detail: None,
},
Err(error) => DependencyDto {
ok: false,
latency_ms: Some(latency_ms(start)),
detail: Some(error.to_string()),
},
}
}
async fn redis_health(&self) -> DependencyDto {
let start = Instant::now();
let mut conn = self.redis().connection_clone();
let (ok, info) = match &mut conn {
export::RedisConnection::Standalone(connection) => {
let ping = redis::cmd("PING")
.query_async::<String>(connection)
.await
.is_ok();
let info = redis::cmd("INFO")
.query_async::<String>(connection)
.await
.ok();
(ping, info)
}
export::RedisConnection::Cluster(connection) => {
let ping = redis::cmd("PING")
.query_async::<String>(connection)
.await
.is_ok();
let info = redis::cmd("INFO")
.query_async::<String>(connection)
.await
.ok();
(ping, info)
}
};
let detail = info.map(|info| {
let memory = info_value(&info, "used_memory_human:")
.unwrap_or_else(|| info_value(&info, "used_memory:").unwrap_or_default());
let clients = info_value(&info, "connected_clients:").unwrap_or_default();
let keys = info
.lines()
.find_map(|line| line.strip_prefix("db0:keys="))
.and_then(|value| value.split(',').next())
.unwrap_or_default()
.to_string();
format!("memory={memory} clients={clients} keys={keys}")
});
DependencyDto {
ok,
latency_ms: Some(latency_ms(start)),
detail,
}
}
async fn storage_health(&self) -> DependencyDto {
let start = Instant::now();
match self.storage().client().bucket_exists().await {
Ok(ok) => DependencyDto {
ok,
latency_ms: Some(latency_ms(start)),
detail: Some(format!("bucket={}", self.storage().client().bucket_name())),
},
Err(error) => DependencyDto {
ok: false,
latency_ms: Some(latency_ms(start)),
detail: Some(error.to_string()),
},
}
}
async fn search_health(&self) -> DependencyDto {
let config = &self.config().search;
if !config.enabled {
return DependencyDto {
ok: false,
latency_ms: None,
detail: Some("search is disabled".to_string()),
};
}
let start = Instant::now();
let url = format!("{}/health", config.url.trim_end_matches('/'));
let client = match reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(2))
.build()
{
Ok(client) => client,
Err(error) => {
return DependencyDto {
ok: false,
latency_ms: Some(latency_ms(start)),
detail: Some(error.to_string()),
};
}
};
match client.get(url).send().await {
Ok(response) => DependencyDto {
ok: response.status().is_success(),
latency_ms: Some(latency_ms(start)),
detail: Some(format!("status={}", response.status().as_u16())),
},
Err(error) => DependencyDto {
ok: false,
latency_ms: Some(latency_ms(start)),
detail: Some(error.to_string()),
},
}
}
async fn counts(&self) -> Result<CountsDto, AdminError> {
Ok(CountsDto {
users: user::Entity::find().count(self.db().read()).await? as i64,
active_users: user::Entity::find()
.filter(models::users::user::Column::IsActive.eq(true))
.count(self.db().read())
.await? as i64,
spaces: spaces::Entity::find().count(self.db().read()).await? as i64,
active_spaces: spaces::Entity::find()
.filter(models::space::spaces::Column::IsActive.eq(true))
.count(self.db().read())
.await? as i64,
repos: repos::Entity::find().count(self.db().read()).await? as i64,
rooms: rooms::Entity::find().count(self.db().read()).await? as i64,
issues: issues::Entity::find().count(self.db().read()).await? as i64,
messages: messages::Entity::find().count(self.db().read()).await? as i64,
})
}
pub async fn system_status(&self) -> Result<SystemStatusDto, AdminError> {
let (database, redis, storage, search, counts, queues) = tokio::join!(
self.database_health(),
self.redis_health(),
self.storage_health(),
self.search_health(),
self.counts(),
self.queue_snapshots(),
);
let counts = counts?;
let queues = queues?;
let recent_failures = self
.list_job_logs(crate::services::logs::ListJobLogsQuery {
source: None,
from: None,
to: None,
limit: Some(10),
offset: Some(0),
})
.await?
.items;
Ok(SystemStatusDto {
process: ProcessInfoDto {
version: env!("CARGO_PKG_VERSION").to_string(),
uptime_secs: self.uptime_secs(),
},
dependencies: DependenciesDto {
database,
redis,
storage,
search,
},
features: FeaturesDto {
search_enabled: self.config().search.enabled,
import_enabled: self.config().import.enabled,
email_enabled: self.config().email.enabled(),
email_queue_enabled: self.config().email_queue.enabled(),
},
counts,
queues,
recent_failures,
generated_at: OffsetDateTime::now_utc(),
})
}
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct HostEntryDto {
pub name: String,
pub enabled: bool,
pub bind_addr: Option<String>,
pub port: Option<u16>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminEndpointDto {
pub enabled: bool,
pub bind_addr: String,
pub port: u16,
pub cors_allow_origins: Vec<String>,
pub cors_allow_credentials: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct SearchConfigDto {
pub enabled: bool,
pub index_prefix: String,
pub max_file_bytes: usize,
pub batch_size: usize,
pub poll_interval_secs: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct ImportConfigDto {
pub enabled: bool,
pub allowed_hosts: Vec<String>,
pub allow_insecure_tls: bool,
pub max_concurrent_jobs: usize,
pub max_repo_bytes: u64,
pub max_attempts: i32,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct GitConfigDto {
pub repo_root: String,
pub objects_root: String,
pub max_blob_size: u64,
pub sync_topic: String,
pub sync_concurrency: usize,
pub base_url: String,
pub graphql_complexity_limit: u64,
pub graphql_introspection: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct StorageConfigDto {
pub bucket: Option<String>,
pub region: Option<String>,
pub path_style: Option<bool>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct DatabaseConfigDto {
pub endpoint: Option<String>,
pub host: Option<String>,
pub port: Option<u16>,
pub database: Option<String>,
pub ssl_mode: Option<String>,
pub max_connections: Option<u32>,
pub min_connections: Option<u32>,
pub replica_count: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct RedisConfigDto {
pub endpoint: String,
pub mode: String,
pub database: Option<u16>,
pub pool_size: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct SessionConfigDto {
pub mode: String,
pub cookie_name: Option<String>,
pub cookie_domain: Option<String>,
pub cookie_secure: Option<bool>,
pub cookie_same_site: String,
pub lifespan_secs: Option<i64>,
pub table_name: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct EmailConfigDto {
pub enabled: bool,
pub queue_enabled: bool,
pub smtp_host: String,
pub smtp_port: u16,
pub from_address: Option<String>,
pub from_name: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct SystemConfigDto {
pub version: String,
pub hosts: Vec<HostEntryDto>,
pub admin: AdminEndpointDto,
pub features: FeaturesDto,
pub search: SearchConfigDto,
pub import: ImportConfigDto,
pub git: GitConfigDto,
pub storage: StorageConfigDto,
pub database: DatabaseConfigDto,
pub redis: RedisConfigDto,
pub session: SessionConfigDto,
pub email: EmailConfigDto,
pub generated_at: OffsetDateTime,
}
fn redact_endpoint(value: &str) -> String {
let Some(scheme_end) = value.find("://") else {
return value.to_string();
};
let (scheme, rest) = value.split_at(scheme_end + 3);
let end = rest.find(['/', '?']).unwrap_or(rest.len());
let authority = &rest[..end];
let host = authority.rsplit('@').next().unwrap_or(authority);
format!("{scheme}{host}{}", &rest[end..])
}
impl AdminService {
pub async fn system_config(&self) -> Result<SystemConfigDto, AdminError> {
let config = self.config();
let hosts = vec![
HostEntryDto {
name: "apihost".to_string(),
enabled: config.apihost.enabled(),
bind_addr: config.apihost.bind_addr.clone(),
port: config.apihost.port,
},
HostEntryDto {
name: "githost".to_string(),
enabled: config.githost.enabled(),
bind_addr: config.githost.http_bind_addr.map(|value| value.to_string()),
port: config.githost.http_port,
},
];
Ok(SystemConfigDto {
version: env!("CARGO_PKG_VERSION").to_string(),
hosts,
admin: AdminEndpointDto {
enabled: config.admin.enabled(),
bind_addr: config.admin.bind_addr().to_string(),
port: config.admin.port(),
cors_allow_origins: config.admin.cors_allow_origins.clone(),
cors_allow_credentials: config.admin.cors_allow_credentials(),
},
features: FeaturesDto {
search_enabled: config.search.enabled,
import_enabled: config.import.enabled,
email_enabled: config.email.enabled(),
email_queue_enabled: config.email_queue.enabled(),
},
search: SearchConfigDto {
enabled: config.search.enabled,
index_prefix: config.search.index_prefix.clone(),
max_file_bytes: config.search.max_file_bytes,
batch_size: config.search.batch_size,
poll_interval_secs: config.search.poll_interval_secs,
},
import: ImportConfigDto {
enabled: config.import.enabled,
allowed_hosts: config.import.allowed_hosts.clone(),
allow_insecure_tls: config.import.allow_insecure_tls,
max_concurrent_jobs: config.import.max_concurrent_jobs,
max_repo_bytes: config.import.max_repo_bytes,
max_attempts: config.import.max_attempts,
},
git: GitConfigDto {
repo_root: config.git.repo_root.clone(),
objects_root: config.git.objects_root.clone(),
max_blob_size: config.git.max_blob_size,
sync_topic: config.git.sync.queue_topic.clone(),
sync_concurrency: config.git.sync.concurrency,
base_url: config.git.base_url.clone(),
graphql_complexity_limit: config.git.graphql_complexity_limit,
graphql_introspection: config.git.graphql_introspection,
},
storage: StorageConfigDto {
bucket: config.s3.bucket.clone(),
region: config.s3.region.clone(),
path_style: config.s3.path_style,
},
database: DatabaseConfigDto {
endpoint: config.database.url.as_deref().map(redact_endpoint),
host: config.database.host.clone(),
port: config.database.port,
database: config.database.database.clone(),
ssl_mode: config
.database
.ssl_mode
.as_ref()
.map(|mode| mode.as_query_value().to_string()),
max_connections: config.database.max_connections,
min_connections: config.database.min_connections,
replica_count: config.database.replicas.len(),
},
redis: RedisConfigDto {
endpoint: redact_endpoint(config.redis.url()),
mode: format!("{:?}", config.redis.mode()).to_ascii_lowercase(),
database: config.redis.database,
pool_size: config.redis.pool_size(),
},
session: SessionConfigDto {
mode: format!("{:?}", config.session.mode.unwrap_or_default()).to_ascii_lowercase(),
cookie_name: config.session.cookie_name.clone(),
cookie_domain: config.session.cookie_domain.clone(),
cookie_secure: config.session.cookie_secure,
cookie_same_site: format!(
"{:?}",
config.session.cookie_same_site.unwrap_or_default()
)
.to_ascii_lowercase(),
lifespan_secs: config.session.lifespan_secs,
table_name: config.session.table_name.clone(),
},
email: EmailConfigDto {
enabled: config.email.enabled(),
queue_enabled: config.email_queue.enabled(),
smtp_host: config.email.smtp_host().to_string(),
smtp_port: config.email.smtp_port(),
from_address: config.email.from_address().map(str::to_string),
from_name: config.email.from_name().map(str::to_string),
},
generated_at: OffsetDateTime::now_utc(),
})
}
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct TrendPointDto {
pub date: String,
pub users: i64,
pub spaces: i64,
pub repos: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct DashboardDto {
pub counts: CountsDto,
pub trend: Vec<TrendPointDto>,
pub active_sessions: i64,
pub restricted_users: i64,
pub recent_failures: Vec<JobLogDto>,
pub queues: Vec<QueueSnapshotDto>,
pub generated_at: OffsetDateTime,
}
fn day_key(value: OffsetDateTime) -> String {
let date = value.date();
format!(
"{:04}-{:02}-{:02}",
date.year(),
u8::from(date.month()),
date.day()
)
}
impl AdminService {
pub async fn dashboard(&self) -> Result<DashboardDto, AdminError> {
let (counts, queues) = tokio::join!(self.counts(), self.queue_snapshots());
let counts = counts?;
let queues = queues?;
let now = OffsetDateTime::now_utc();
let since = (now.date() - time::Duration::days(13))
.midnight()
.assume_utc();
let users = user::Entity::find()
.filter(user::Column::CreatedAt.gte(since))
.all(self.db().read())
.await?;
let spaces = spaces::Entity::find()
.filter(models::space::spaces::Column::CreatedAt.gte(since))
.all(self.db().read())
.await?;
let repos = repos::Entity::find()
.filter(repos::Column::CreatedAt.gte(since))
.all(self.db().read())
.await?;
let mut buckets: Vec<TrendPointDto> = (0..14)
.map(|offset| TrendPointDto {
date: day_key(since + time::Duration::days(offset)),
users: 0,
spaces: 0,
repos: 0,
})
.collect();
let mut index: HashMap<String, usize> = HashMap::new();
for (position, bucket) in buckets.iter().enumerate() {
index.insert(bucket.date.clone(), position);
}
for row in &users {
if let Some(position) = index.get(&day_key(row.created_at)) {
buckets[*position].users += 1;
}
}
for row in &spaces {
if let Some(position) = index.get(&day_key(row.created_at)) {
buckets[*position].spaces += 1;
}
}
for row in &repos {
if let Some(position) = index.get(&day_key(row.created_at)) {
buckets[*position].repos += 1;
}
}
let active_sessions = user_sessions::Entity::find()
.filter(user_sessions::Column::IsRevoked.eq(false))
.count(self.db().read())
.await? as i64;
let restricted_users = user::Entity::find()
.filter(user::Column::IsRestricted.eq(true))
.count(self.db().read())
.await? as i64;
let recent_failures = self
.list_job_logs(crate::services::logs::ListJobLogsQuery {
source: None,
from: None,
to: None,
limit: Some(8),
offset: Some(0),
})
.await?
.items;
Ok(DashboardDto {
counts,
trend: buckets,
active_sessions,
restricted_users,
recent_failures,
queues,
generated_at: now,
})
}
}