i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 01:10:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(gateway): unify background task runtime and storage

Browse Source
This commit is contained in:
fawney19
2026-05-09 21:19:29 +08:00
parent 4a64d078f3
commit 209322b499
64 changed files with 4602 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
crates/aether-data/src/backend/mysql.rs
View File

@@ -13,6 +13,9 @@ use crate::repository::auth_modules::{
AuthModuleReadRepository, AuthModuleWriteRepository, MysqlAuthModuleReadRepository,
MysqlAuthModuleRepository,
};
use crate::repository::background_tasks::{
BackgroundTaskReadRepository, BackgroundTaskWriteRepository, MysqlBackgroundTaskRepository,
};
use crate::repository::billing::{BillingReadRepository, MysqlBillingReadRepository};
use crate::repository::candidate_selection::{
MinimalCandidateSelectionReadRepository, MysqlMinimalCandidateSelectionReadRepository,
@@ -123,6 +126,14 @@ impl MysqlBackend {
Arc::new(MysqlBillingReadRepository::new(self.pool_clone()))
}
pub fn background_task_read_repository(&self) -> Arc<dyn BackgroundTaskReadRepository> {
Arc::new(MysqlBackgroundTaskRepository::new(self.pool_clone()))
}
pub fn background_task_write_repository(&self) -> Arc<dyn BackgroundTaskWriteRepository> {
Arc::new(MysqlBackgroundTaskRepository::new(self.pool_clone()))
}
pub fn request_candidate_read_repository(&self) -> Arc<dyn RequestCandidateReadRepository> {
Arc::new(MysqlRequestCandidateRepository::new(self.pool_clone()))
}

11
crates/aether-data/src/backend/postgres.rs
View File

@@ -15,6 +15,9 @@ use crate::repository::auth_modules::{
AuthModuleReadRepository, AuthModuleWriteRepository, SqlxAuthModuleReadRepository,
SqlxAuthModuleRepository,
};
use crate::repository::background_tasks::{
BackgroundTaskReadRepository, BackgroundTaskWriteRepository, SqlxBackgroundTaskRepository,
};
use crate::repository::billing::{BillingReadRepository, SqlxBillingReadRepository};
use crate::repository::candidate_selection::{
MinimalCandidateSelectionReadRepository, SqlxMinimalCandidateSelectionReadRepository,
@@ -118,6 +121,14 @@ impl PostgresBackend {
Arc::new(SqlxBillingReadRepository::new(self.pool_clone()))
}
pub fn background_task_read_repository(&self) -> Arc<dyn BackgroundTaskReadRepository> {
Arc::new(SqlxBackgroundTaskRepository::new(self.pool_clone()))
}
pub fn background_task_write_repository(&self) -> Arc<dyn BackgroundTaskWriteRepository> {
Arc::new(SqlxBackgroundTaskRepository::new(self.pool_clone()))
}
pub fn minimal_candidate_selection_read_repository(
&self,
) -> Arc<dyn MinimalCandidateSelectionReadRepository> {

12
crates/aether-data/src/backend/read.rs
View File

@@ -6,6 +6,7 @@ use crate::repository::announcements::AnnouncementReadRepository;
use crate::repository::audit::AuditLogReadRepository;
use crate::repository::auth::AuthApiKeyReadRepository;
use crate::repository::auth_modules::AuthModuleReadRepository;
use crate::repository::background_tasks::BackgroundTaskReadRepository;
use crate::repository::billing::BillingReadRepository;
use crate::repository::candidate_selection::MinimalCandidateSelectionReadRepository;
use crate::repository::candidates::RequestCandidateReadRepository;
@@ -27,6 +28,7 @@ pub struct DataReadRepositories {
audit_logs: Option<Arc<dyn AuditLogReadRepository>>,
auth_api_keys: Option<Arc<dyn AuthApiKeyReadRepository>>,
auth_modules: Option<Arc<dyn AuthModuleReadRepository>>,
background_tasks: Option<Arc<dyn BackgroundTaskReadRepository>>,
billing: Option<Arc<dyn BillingReadRepository>>,
gemini_file_mappings: Option<Arc<dyn GeminiFileMappingReadRepository>>,
global_models: Option<Arc<dyn GlobalModelReadRepository>>,
@@ -50,6 +52,7 @@ impl fmt::Debug for DataReadRepositories {
.field("has_announcements", &self.announcements.is_some())
.field("has_audit_logs", &self.audit_logs.is_some())
.field("has_auth_modules", &self.auth_modules.is_some())
.field("has_background_tasks", &self.background_tasks.is_some())
.field("has_billing", &self.billing.is_some())
.field(
"has_gemini_file_mappings",
@@ -97,6 +100,10 @@ impl DataReadRepositories {
.map(PostgresBackend::auth_module_read_repository)
.or_else(|| mysql.map(MysqlBackend::auth_module_read_repository))
.or_else(|| sqlite.map(SqliteBackend::auth_module_read_repository)),
background_tasks: postgres
.map(PostgresBackend::background_task_read_repository)
.or_else(|| mysql.map(MysqlBackend::background_task_read_repository))
.or_else(|| sqlite.map(SqliteBackend::background_task_read_repository)),
billing: postgres
.map(PostgresBackend::billing_read_repository)
.or_else(|| mysql.map(MysqlBackend::billing_read_repository))
@@ -177,6 +184,10 @@ impl DataReadRepositories {
self.auth_modules.clone()
}
pub fn background_tasks(&self) -> Option<Arc<dyn BackgroundTaskReadRepository>> {
self.background_tasks.clone()
}
pub fn billing(&self) -> Option<Arc<dyn BillingReadRepository>> {
self.billing.clone()
}
@@ -240,6 +251,7 @@ impl DataReadRepositories {
|| self.announcements.is_some()
|| self.audit_logs.is_some()
|| self.auth_modules.is_some()
|| self.background_tasks.is_some()
|| self.billing.is_some()
|| self.gemini_file_mappings.is_some()
|| self.global_models.is_some()

11
crates/aether-data/src/backend/sqlite.rs
View File

@@ -13,6 +13,9 @@ use crate::repository::auth_modules::{
AuthModuleReadRepository, AuthModuleWriteRepository, SqliteAuthModuleReadRepository,
SqliteAuthModuleRepository,
};
use crate::repository::background_tasks::{
BackgroundTaskReadRepository, BackgroundTaskWriteRepository, SqliteBackgroundTaskRepository,
};
use crate::repository::billing::{BillingReadRepository, SqliteBillingReadRepository};
use crate::repository::candidate_selection::{
MinimalCandidateSelectionReadRepository, SqliteMinimalCandidateSelectionReadRepository,
@@ -124,6 +127,14 @@ impl SqliteBackend {
Arc::new(SqliteBillingReadRepository::new(self.pool_clone()))
}
pub fn background_task_read_repository(&self) -> Arc<dyn BackgroundTaskReadRepository> {
Arc::new(SqliteBackgroundTaskRepository::new(self.pool_clone()))
}
pub fn background_task_write_repository(&self) -> Arc<dyn BackgroundTaskWriteRepository> {
Arc::new(SqliteBackgroundTaskRepository::new(self.pool_clone()))
}
pub fn request_candidate_read_repository(&self) -> Arc<dyn RequestCandidateReadRepository> {
Arc::new(SqliteRequestCandidateRepository::new(self.pool_clone()))
}

12
crates/aether-data/src/backend/write.rs
View File

@@ -5,6 +5,7 @@ use super::{MysqlBackend, PostgresBackend, SqliteBackend};
use crate::repository::announcements::AnnouncementWriteRepository;
use crate::repository::auth::AuthApiKeyWriteRepository;
use crate::repository::auth_modules::AuthModuleWriteRepository;
use crate::repository::background_tasks::BackgroundTaskWriteRepository;
use crate::repository::candidates::RequestCandidateWriteRepository;
use crate::repository::gemini_file_mappings::GeminiFileMappingWriteRepository;
use crate::repository::global_models::GlobalModelWriteRepository;
@@ -23,6 +24,7 @@ pub struct DataWriteRepositories {
announcements: Option<Arc<dyn AnnouncementWriteRepository>>,
auth_api_keys: Option<Arc<dyn AuthApiKeyWriteRepository>>,
auth_modules: Option<Arc<dyn AuthModuleWriteRepository>>,
background_tasks: Option<Arc<dyn BackgroundTaskWriteRepository>>,
request_candidates: Option<Arc<dyn RequestCandidateWriteRepository>>,
gemini_file_mappings: Option<Arc<dyn GeminiFileMappingWriteRepository>>,
global_models: Option<Arc<dyn GlobalModelWriteRepository>>,
@@ -43,6 +45,7 @@ impl fmt::Debug for DataWriteRepositories {
.field("has_announcements", &self.announcements.is_some())
.field("has_auth_api_keys", &self.auth_api_keys.is_some())
.field("has_auth_modules", &self.auth_modules.is_some())
.field("has_background_tasks", &self.background_tasks.is_some())
.field("has_request_candidates", &self.request_candidates.is_some())
.field(
"has_gemini_file_mappings",
@@ -81,6 +84,10 @@ impl DataWriteRepositories {
.map(PostgresBackend::auth_module_write_repository)
.or_else(|| mysql.map(MysqlBackend::auth_module_write_repository))
.or_else(|| sqlite.map(SqliteBackend::auth_module_write_repository)),
background_tasks: postgres
.map(PostgresBackend::background_task_write_repository)
.or_else(|| mysql.map(MysqlBackend::background_task_write_repository))
.or_else(|| sqlite.map(SqliteBackend::background_task_write_repository)),
request_candidates: postgres
.map(PostgresBackend::request_candidate_write_repository)
.or_else(|| mysql.map(MysqlBackend::request_candidate_write_repository))
@@ -149,6 +156,10 @@ impl DataWriteRepositories {
self.auth_modules.clone()
}
pub fn background_tasks(&self) -> Option<Arc<dyn BackgroundTaskWriteRepository>> {
self.background_tasks.clone()
}
pub fn usage(&self) -> Option<Arc<dyn UsageWriteRepository>> {
self.usage.clone()
}
@@ -201,6 +212,7 @@ impl DataWriteRepositories {
self.announcements.is_some()
|| self.auth_api_keys.is_some()
|| self.auth_modules.is_some()
|| self.background_tasks.is_some()
|| self.request_candidates.is_some()
|| self.gemini_file_mappings.is_some()
|| self.global_models.is_some()

2
crates/aether-data/src/lifecycle/bootstrap/postgres.rs
View File

@@ -7,7 +7,7 @@ use tracing::info;
// Generated by build.rs from schema/bootstrap/postgres.
pub(crate) static EMPTY_DATABASE_SNAPSHOT_SQL: &str =
include_str!(concat!(env!("OUT_DIR"), "/empty_database_snapshot.sql"));
pub(crate) const EMPTY_DATABASE_SNAPSHOT_CUTOFF_VERSION: i64 = 20260508000000;
pub(crate) const EMPTY_DATABASE_SNAPSHOT_CUTOFF_VERSION: i64 = 20260509000000;
const PUBLIC_BASE_TABLE_COUNT_SQL: &str = r#"
SELECT COUNT(*)::BIGINT

16
crates/aether-data/src/lifecycle/migrate/tests.rs
View File

@@ -294,6 +294,7 @@ fn empty_database_snapshot_covers_current_cutoff_versions() {
20260507000000,
20260507120000,
20260508000000,
20260509000000,
]
);
}
@@ -513,11 +514,21 @@ fn mysql_and_sqlite_migrations_include_enabled_incrementals() {
assert_eq!(
mysql_versions,
vec![20260403000000, 20260507120000, 20260508000000]
vec![
20260403000000,
20260507120000,
20260508000000,
20260509000000
]
);
assert_eq!(
sqlite_versions,
vec![20260403000000, 20260507120000, 20260508000000]
vec![
20260403000000,
20260507120000,
20260508000000,
20260509000000
]
);
}
@@ -1022,6 +1033,7 @@ fn pending_migrations_from_applied_skips_versions_already_applied() {
20260507000000,
20260507120000,
20260508000000,
20260509000000,
]
);
}

218
crates/aether-data/src/repository/background_tasks/memory.rs Normal file
View File

@@ -0,0 +1,218 @@
use std::collections::{BTreeMap, BTreeSet};
use std::sync::RwLock;
use async_trait::async_trait;
use super::{
BackgroundTaskListQuery, BackgroundTaskReadRepository, BackgroundTaskStatus,
BackgroundTaskSummary, BackgroundTaskWriteRepository, StoredBackgroundTaskEvent,
StoredBackgroundTaskRun, StoredBackgroundTaskRunPage, UpsertBackgroundTaskEvent,
UpsertBackgroundTaskRun,
};
use crate::DataLayerError;
#[derive(Debug, Default)]
struct InMemoryBackgroundTaskIndex {
runs: BTreeMap<String, StoredBackgroundTaskRun>,
events_by_run: BTreeMap<String, Vec<StoredBackgroundTaskEvent>>,
}
#[derive(Debug, Default)]
pub struct InMemoryBackgroundTaskRepository {
index: RwLock<InMemoryBackgroundTaskIndex>,
}
impl InMemoryBackgroundTaskRepository {
fn matches_filter(run: &StoredBackgroundTaskRun, query: &BackgroundTaskListQuery) -> bool {
if let Some(kind) = query.kind {
if run.kind != kind {
return false;
}
}
if let Some(status) = query.status {
if run.status != status {
return false;
}
}
if let Some(trigger) = query.trigger.as_deref() {
if run.trigger != trigger {
return false;
}
}
if let Some(task_key_substring) = query.task_key_substring.as_deref() {
let needle = task_key_substring.to_ascii_lowercase();
if !run.task_key.to_ascii_lowercase().contains(&needle) {
return false;
}
}
true
}
pub fn seed_runs<I>(runs: I) -> Self
where
I: IntoIterator<Item = StoredBackgroundTaskRun>,
{
let mut index = InMemoryBackgroundTaskIndex::default();
for run in runs {
index.runs.insert(run.id.clone(), run);
}
Self {
index: RwLock::new(index),
}
}
}
#[async_trait]
impl BackgroundTaskReadRepository for InMemoryBackgroundTaskRepository {
async fn find_run(
&self,
run_id: &str,
) -> Result<Option<StoredBackgroundTaskRun>, DataLayerError> {
Ok(self
.index
.read()
.expect("background task repository lock")
.runs
.get(run_id)
.cloned())
}
async fn list_runs(
&self,
query: &BackgroundTaskListQuery,
) -> Result<StoredBackgroundTaskRunPage, DataLayerError> {
let mut items = self
.index
.read()
.expect("background task repository lock")
.runs
.values()
.filter(|run| Self::matches_filter(run, query))
.cloned()
.collect::<Vec<_>>();
items.sort_by(|left, right| {
right
.created_at_unix_secs
.cmp(&left.created_at_unix_secs)
.then_with(|| right.updated_at_unix_secs.cmp(&left.updated_at_unix_secs))
});
let total = items.len();
let limit = query.limit.max(1);
let items = items
.into_iter()
.skip(query.offset)
.take(limit)
.collect::<Vec<_>>();
Ok(StoredBackgroundTaskRunPage { items, total })
}
async fn list_events(
&self,
run_id: &str,
offset: usize,
limit: usize,
) -> Result<Vec<StoredBackgroundTaskEvent>, DataLayerError> {
let Some(events) = self
.index
.read()
.expect("background task repository lock")
.events_by_run
.get(run_id)
.cloned()
else {
return Ok(Vec::new());
};
let limit = limit.max(1);
Ok(events.into_iter().skip(offset).take(limit).collect())
}
async fn summarize_runs(&self) -> Result<BackgroundTaskSummary, DataLayerError> {
let runs = self
.index
.read()
.expect("background task repository lock")
.runs
.values()
.cloned()
.collect::<Vec<_>>();
let mut by_status = BTreeMap::new();
let mut by_kind = BTreeMap::new();
let mut running_count = 0_u64;
for run in runs {
*by_status
.entry(run.status.as_database().to_string())
.or_insert(0) += 1;
*by_kind
.entry(run.kind.as_database().to_string())
.or_insert(0) += 1;
if run.status == BackgroundTaskStatus::Running {
running_count += 1;
}
}
let total = by_status.values().copied().sum();
Ok(BackgroundTaskSummary {
total,
running_count,
by_status,
by_kind,
})
}
}
#[async_trait]
impl BackgroundTaskWriteRepository for InMemoryBackgroundTaskRepository {
async fn upsert_run(
&self,
run: UpsertBackgroundTaskRun,
) -> Result<StoredBackgroundTaskRun, DataLayerError> {
run.validate()?;
let stored = run.into_stored();
self.index
.write()
.expect("background task repository lock")
.runs
.insert(stored.id.clone(), stored.clone());
Ok(stored)
}
async fn request_cancel(
&self,
run_id: &str,
updated_at_unix_secs: u64,
) -> Result<bool, DataLayerError> {
let mut guard = self.index.write().expect("background task repository lock");
let Some(run) = guard.runs.get_mut(run_id) else {
return Ok(false);
};
run.cancel_requested = true;
run.updated_at_unix_secs = updated_at_unix_secs;
Ok(true)
}
async fn upsert_event(
&self,
event: UpsertBackgroundTaskEvent,
) -> Result<StoredBackgroundTaskEvent, DataLayerError> {
event.validate()?;
let stored = event.into_stored();
let mut guard = self.index.write().expect("background task repository lock");
let entries = guard
.events_by_run
.entry(stored.run_id.clone())
.or_default();
if let Some(position) = entries.iter().position(|value| value.id == stored.id) {
entries[position] = stored.clone();
} else {
entries.push(stored.clone());
}
let mut seen = BTreeSet::new();
entries.retain(|entry| seen.insert(entry.id.clone()));
entries.sort_by(|left, right| {
left.created_at_unix_secs
.cmp(&right.created_at_unix_secs)
.then_with(|| left.id.cmp(&right.id))
});
Ok(stored)
}
}

17
crates/aether-data/src/repository/background_tasks/mod.rs Normal file
View File

@@ -0,0 +1,17 @@
mod memory;
mod mysql;
mod postgres;
mod sqlite;
#[allow(unused_imports)]
pub(crate) use aether_data_contracts::repository::background_tasks::{
BackgroundTaskKind, BackgroundTaskListQuery, BackgroundTaskReadRepository,
BackgroundTaskRepository, BackgroundTaskStatus, BackgroundTaskSummary,
BackgroundTaskWriteRepository, StoredBackgroundTaskEvent, StoredBackgroundTaskRun,
StoredBackgroundTaskRunPage, UpsertBackgroundTaskEvent, UpsertBackgroundTaskRun,
};
pub use memory::InMemoryBackgroundTaskRepository;
pub use mysql::MysqlBackgroundTaskRepository;
pub use postgres::SqlxBackgroundTaskRepository;
pub use sqlite::SqliteBackgroundTaskRepository;

449
crates/aether-data/src/repository/background_tasks/mysql.rs Normal file
View File

@@ -0,0 +1,449 @@
use async_trait::async_trait;
use sqlx::{mysql::MySqlRow, MySql, QueryBuilder, Row};
use super::{
BackgroundTaskKind, BackgroundTaskListQuery, BackgroundTaskReadRepository,
BackgroundTaskStatus, BackgroundTaskSummary, BackgroundTaskWriteRepository,
StoredBackgroundTaskEvent, StoredBackgroundTaskRun, StoredBackgroundTaskRunPage,
UpsertBackgroundTaskEvent, UpsertBackgroundTaskRun,
};
use crate::driver::mysql::MysqlPool;
use crate::error::SqlResultExt;
use crate::DataLayerError;
const RUN_COLUMNS: &str = r#"
SELECT
id,
task_key,
kind,
`trigger`,
status,
attempt,
max_attempts,
owner_instance,
progress_percent,
progress_message,
payload_json,
result_json,
error_message,
cancel_requested,
created_by,
created_at_unix_secs,
started_at_unix_secs,
finished_at_unix_secs,
updated_at_unix_secs
FROM background_task_runs
"#;
const EVENT_COLUMNS: &str = r#"
SELECT
id,
run_id,
event_type,
message,
payload_json,
created_at_unix_secs
FROM background_task_events
"#;
#[derive(Debug, Clone)]
pub struct MysqlBackgroundTaskRepository {
pool: MysqlPool,
}
impl MysqlBackgroundTaskRepository {
pub fn new(pool: MysqlPool) -> Self {
Self { pool }
}
fn apply_run_filter(builder: &mut QueryBuilder<'_, MySql>, query: &BackgroundTaskListQuery) {
let mut has_where = false;
if let Some(kind) = query.kind {
if !has_where {
builder.push(" WHERE ");
has_where = true;
} else {
builder.push(" AND ");
}
builder.push("kind = ").push_bind(kind.as_database());
}
if let Some(status) = query.status {
if !has_where {
builder.push(" WHERE ");
has_where = true;
} else {
builder.push(" AND ");
}
builder.push("status = ").push_bind(status.as_database());
}
if let Some(trigger) = query.trigger.as_deref() {
if !has_where {
builder.push(" WHERE ");
has_where = true;
} else {
builder.push(" AND ");
}
builder.push("`trigger` = ").push_bind(trigger.to_string());
}
if let Some(task_key_substring) = query.task_key_substring.as_deref() {
if !has_where {
builder.push(" WHERE ");
} else {
builder.push(" AND ");
}
builder.push("LOWER(task_key) LIKE ").push_bind(format!(
"%{}%",
task_key_substring.trim().to_ascii_lowercase()
));
}
}
}
#[async_trait]
impl BackgroundTaskReadRepository for MysqlBackgroundTaskRepository {
async fn find_run(
&self,
run_id: &str,
) -> Result<Option<StoredBackgroundTaskRun>, DataLayerError> {
let row = sqlx::query(&format!("{RUN_COLUMNS} WHERE id = ? LIMIT 1"))
.bind(run_id)
.fetch_optional(&self.pool)
.await
.map_sql_err()?;
row.as_ref().map(map_run_row).transpose()
}
async fn list_runs(
&self,
query: &BackgroundTaskListQuery,
) -> Result<StoredBackgroundTaskRunPage, DataLayerError> {
let limit = query.limit.max(1);
let mut count_builder =
QueryBuilder::<MySql>::new("SELECT COUNT(id) AS total FROM background_task_runs");
Self::apply_run_filter(&mut count_builder, query);
let total = count_builder
.build_query_scalar::<i64>()
.fetch_one(&self.pool)
.await
.map_sql_err()?;
let mut builder = QueryBuilder::<MySql>::new(RUN_COLUMNS);
Self::apply_run_filter(&mut builder, query);
builder
.push(" ORDER BY created_at_unix_secs DESC, updated_at_unix_secs DESC")
.push(" LIMIT ")
.push_bind(i64_from_usize(limit, "run limit")?)
.push(" OFFSET ")
.push_bind(i64_from_usize(query.offset, "run offset")?);
let rows = builder.build().fetch_all(&self.pool).await.map_sql_err()?;
let items = rows
.iter()
.map(map_run_row)
.collect::<Result<Vec<_>, _>>()?;
Ok(StoredBackgroundTaskRunPage {
items,
total: usize::try_from(total).unwrap_or_default(),
})
}
async fn list_events(
&self,
run_id: &str,
offset: usize,
limit: usize,
) -> Result<Vec<StoredBackgroundTaskEvent>, DataLayerError> {
let limit = limit.max(1);
let rows = sqlx::query(&format!(
"{EVENT_COLUMNS} WHERE run_id = ? ORDER BY created_at_unix_secs ASC, id ASC LIMIT ? OFFSET ?"
))
.bind(run_id)
.bind(i64_from_usize(limit, "event limit")?)
.bind(i64_from_usize(offset, "event offset")?)
.fetch_all(&self.pool)
.await
.map_sql_err()?;
rows.iter().map(map_event_row).collect()
}
async fn summarize_runs(&self) -> Result<BackgroundTaskSummary, DataLayerError> {
let total = sqlx::query_scalar::<_, i64>("SELECT COUNT(id) FROM background_task_runs")
.fetch_one(&self.pool)
.await
.map_sql_err()?;
let running_count = sqlx::query_scalar::<_, i64>(
"SELECT COUNT(id) FROM background_task_runs WHERE status = 'running'",
)
.fetch_one(&self.pool)
.await
.map_sql_err()?;
let status_rows = sqlx::query(
"SELECT status, COUNT(id) AS total FROM background_task_runs GROUP BY status",
)
.fetch_all(&self.pool)
.await
.map_sql_err()?;
let kind_rows =
sqlx::query("SELECT kind, COUNT(id) AS total FROM background_task_runs GROUP BY kind")
.fetch_all(&self.pool)
.await
.map_sql_err()?;
let mut by_status = std::collections::BTreeMap::new();
for row in status_rows {
let key: String = row.try_get("status").map_sql_err()?;
let count: i64 = row.try_get("total").map_sql_err()?;
by_status.insert(key, u64::try_from(count).unwrap_or_default());
}
let mut by_kind = std::collections::BTreeMap::new();
for row in kind_rows {
let key: String = row.try_get("kind").map_sql_err()?;
let count: i64 = row.try_get("total").map_sql_err()?;
by_kind.insert(key, u64::try_from(count).unwrap_or_default());
}
Ok(BackgroundTaskSummary {
total: u64::try_from(total).unwrap_or_default(),
running_count: u64::try_from(running_count).unwrap_or_default(),
by_status,
by_kind,
})
}
}
#[async_trait]
impl BackgroundTaskWriteRepository for MysqlBackgroundTaskRepository {
async fn upsert_run(
&self,
run: UpsertBackgroundTaskRun,
) -> Result<StoredBackgroundTaskRun, DataLayerError> {
run.validate()?;
sqlx::query(
r#"
INSERT INTO background_task_runs (
id,
task_key,
kind,
`trigger`,
status,
attempt,
max_attempts,
owner_instance,
progress_percent,
progress_message,
payload_json,
result_json,
error_message,
cancel_requested,
created_by,
created_at_unix_secs,
started_at_unix_secs,
finished_at_unix_secs,
updated_at_unix_secs
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE
task_key = VALUES(task_key),
kind = VALUES(kind),
`trigger` = VALUES(`trigger`),
status = VALUES(status),
attempt = VALUES(attempt),
max_attempts = VALUES(max_attempts),
owner_instance = VALUES(owner_instance),
progress_percent = VALUES(progress_percent),
progress_message = VALUES(progress_message),
payload_json = VALUES(payload_json),
result_json = VALUES(result_json),
error_message = VALUES(error_message),
cancel_requested = VALUES(cancel_requested),
created_by = VALUES(created_by),
created_at_unix_secs = VALUES(created_at_unix_secs),
started_at_unix_secs = VALUES(started_at_unix_secs),
finished_at_unix_secs = VALUES(finished_at_unix_secs),
updated_at_unix_secs = VALUES(updated_at_unix_secs)
"#,
)
.bind(&run.id)
.bind(&run.task_key)
.bind(run.kind.as_database())
.bind(&run.trigger)
.bind(run.status.as_database())
.bind(i64::from(run.attempt))
.bind(i64::from(run.max_attempts))
.bind(run.owner_instance.as_deref())
.bind(i32::from(run.progress_percent))
.bind(run.progress_message.as_deref())
.bind(json_to_string(&run.payload_json, "payload_json")?)
.bind(json_to_string(&run.result_json, "result_json")?)
.bind(run.error_message.as_deref())
.bind(run.cancel_requested)
.bind(run.created_by.as_deref())
.bind(u64_to_i64(
run.created_at_unix_secs,
"created_at_unix_secs",
)?)
.bind(run.started_at_unix_secs.map(|value| value as i64))
.bind(run.finished_at_unix_secs.map(|value| value as i64))
.bind(u64_to_i64(
run.updated_at_unix_secs,
"updated_at_unix_secs",
)?)
.execute(&self.pool)
.await
.map_sql_err()?;
self.find_run(&run.id).await?.ok_or_else(|| {
DataLayerError::UnexpectedValue("background task run missing after upsert".to_string())
})
}
async fn request_cancel(
&self,
run_id: &str,
updated_at_unix_secs: u64,
) -> Result<bool, DataLayerError> {
let affected = sqlx::query(
"UPDATE background_task_runs SET cancel_requested = TRUE, updated_at_unix_secs = ? WHERE id = ?",
)
.bind(u64_to_i64(updated_at_unix_secs, "updated_at_unix_secs")?)
.bind(run_id)
.execute(&self.pool)
.await
.map_sql_err()?
.rows_affected();
Ok(affected > 0)
}
async fn upsert_event(
&self,
event: UpsertBackgroundTaskEvent,
) -> Result<StoredBackgroundTaskEvent, DataLayerError> {
event.validate()?;
sqlx::query(
r#"
INSERT INTO background_task_events (
id, run_id, event_type, message, payload_json, created_at_unix_secs
) VALUES (?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
run_id = VALUES(run_id),
event_type = VALUES(event_type),
message = VALUES(message),
payload_json = VALUES(payload_json),
created_at_unix_secs = VALUES(created_at_unix_secs)
"#,
)
.bind(&event.id)
.bind(&event.run_id)
.bind(&event.event_type)
.bind(&event.message)
.bind(json_to_string(&event.payload_json, "payload_json")?)
.bind(u64_to_i64(
event.created_at_unix_secs,
"created_at_unix_secs",
)?)
.execute(&self.pool)
.await
.map_sql_err()?;
let row = sqlx::query(&format!("{EVENT_COLUMNS} WHERE id = ? LIMIT 1"))
.bind(&event.id)
.fetch_one(&self.pool)
.await
.map_sql_err()?;
map_event_row(&row)
}
}
fn map_run_row(row: &MySqlRow) -> Result<StoredBackgroundTaskRun, DataLayerError> {
let kind: String = row.try_get("kind").map_sql_err()?;
let status: String = row.try_get("status").map_sql_err()?;
let attempt: i64 = row.try_get("attempt").map_sql_err()?;
let max_attempts: i64 = row.try_get("max_attempts").map_sql_err()?;
let progress_percent: i32 = row.try_get("progress_percent").map_sql_err()?;
let created_at_unix_secs: i64 = row.try_get("created_at_unix_secs").map_sql_err()?;
let started_at_unix_secs: Option<i64> = row.try_get("started_at_unix_secs").map_sql_err()?;
let finished_at_unix_secs: Option<i64> = row.try_get("finished_at_unix_secs").map_sql_err()?;
let updated_at_unix_secs: i64 = row.try_get("updated_at_unix_secs").map_sql_err()?;
Ok(StoredBackgroundTaskRun {
id: row.try_get("id").map_sql_err()?,
task_key: row.try_get("task_key").map_sql_err()?,
kind: BackgroundTaskKind::from_database(&kind)?,
trigger: row.try_get("trigger").map_sql_err()?,
status: BackgroundTaskStatus::from_database(&status)?,
attempt: u32::try_from(attempt).unwrap_or_default(),
max_attempts: u32::try_from(max_attempts).unwrap_or_default(),
owner_instance: row.try_get("owner_instance").map_sql_err()?,
progress_percent: u16::try_from(progress_percent).unwrap_or_default(),
progress_message: row.try_get("progress_message").map_sql_err()?,
payload_json: parse_optional_json(
row.try_get("payload_json").ok().flatten(),
"payload_json",
)?,
result_json: parse_optional_json(row.try_get("result_json").ok().flatten(), "result_json")?,
error_message: row.try_get("error_message").map_sql_err()?,
cancel_requested: row.try_get("cancel_requested").map_sql_err()?,
created_by: row.try_get("created_by").map_sql_err()?,
created_at_unix_secs: u64::try_from(created_at_unix_secs).unwrap_or_default(),
started_at_unix_secs: started_at_unix_secs.and_then(|value| u64::try_from(value).ok()),
finished_at_unix_secs: finished_at_unix_secs.and_then(|value| u64::try_from(value).ok()),
updated_at_unix_secs: u64::try_from(updated_at_unix_secs).unwrap_or_default(),
})
}
fn map_event_row(row: &MySqlRow) -> Result<StoredBackgroundTaskEvent, DataLayerError> {
let created_at_unix_secs: i64 = row.try_get("created_at_unix_secs").map_sql_err()?;
Ok(StoredBackgroundTaskEvent {
id: row.try_get("id").map_sql_err()?,
run_id: row.try_get("run_id").map_sql_err()?,
event_type: row.try_get("event_type").map_sql_err()?,
message: row.try_get("message").map_sql_err()?,
payload_json: parse_optional_json(
row.try_get("payload_json").ok().flatten(),
"payload_json",
)?,
created_at_unix_secs: u64::try_from(created_at_unix_secs).unwrap_or_default(),
})
}
fn i64_from_usize(value: usize, label: &str) -> Result<i64, DataLayerError> {
i64::try_from(value).map_err(|_| {
DataLayerError::UnexpectedValue(format!("background task {label} overflow: {value}"))
})
}
fn u64_to_i64(value: u64, label: &str) -> Result<i64, DataLayerError> {
i64::try_from(value).map_err(|_| {
DataLayerError::UnexpectedValue(format!("background task {label} overflow: {value}"))
})
}
fn json_to_string(
value: &Option<serde_json::Value>,
field_name: &str,
) -> Result<Option<String>, DataLayerError> {
value
.as_ref()
.map(|value| {
serde_json::to_string(value).map_err(|err| {
DataLayerError::UnexpectedValue(format!(
"background task {field_name} is unserializable: {err}"
))
})
})
.transpose()
}
fn parse_optional_json(
value: Option<String>,
field_name: &str,
) -> Result<Option<serde_json::Value>, DataLayerError> {
value
.filter(|value| !value.trim().is_empty())
.map(|value| {
serde_json::from_str(&value).map_err(|err| {
DataLayerError::UnexpectedValue(format!(
"background task {field_name} contains invalid JSON: {err}"
))
})
})
.transpose()
}

426
crates/aether-data/src/repository/background_tasks/postgres.rs Normal file
View File

@@ -0,0 +1,426 @@
use async_trait::async_trait;
use sqlx::{postgres::PgRow, PgPool, Postgres, QueryBuilder, Row};
use super::{
BackgroundTaskKind, BackgroundTaskListQuery, BackgroundTaskReadRepository,
BackgroundTaskStatus, BackgroundTaskSummary, BackgroundTaskWriteRepository,
StoredBackgroundTaskEvent, StoredBackgroundTaskRun, StoredBackgroundTaskRunPage,
UpsertBackgroundTaskEvent, UpsertBackgroundTaskRun,
};
use crate::error::SqlxResultExt;
use crate::DataLayerError;
const RUN_COLUMNS: &str = r#"
SELECT
id,
task_key,
kind,
"trigger",
status,
attempt,
max_attempts,
owner_instance,
progress_percent,
progress_message,
payload_json,
result_json,
error_message,
cancel_requested,
created_by,
created_at_unix_secs,
started_at_unix_secs,
finished_at_unix_secs,
updated_at_unix_secs
FROM background_task_runs
"#;
const EVENT_COLUMNS: &str = r#"
SELECT
id,
run_id,
event_type,
message,
payload_json,
created_at_unix_secs
FROM background_task_events
"#;
#[derive(Debug, Clone)]
pub struct SqlxBackgroundTaskRepository {
pool: PgPool,
}
impl SqlxBackgroundTaskRepository {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
fn apply_run_filter(
builder: &mut QueryBuilder<'_, Postgres>,
query: &BackgroundTaskListQuery,
include_where: bool,
) {
let mut has_where = include_where;
let mut push_where = |builder: &mut QueryBuilder<'_, Postgres>| {
if has_where {
builder.push(" AND ");
} else {
builder.push(" WHERE ");
has_where = true;
}
};
if let Some(kind) = query.kind {
push_where(builder);
builder.push("kind = ").push_bind(kind.as_database());
}
if let Some(status) = query.status {
push_where(builder);
builder.push("status = ").push_bind(status.as_database());
}
if let Some(trigger) = query.trigger.as_deref() {
push_where(builder);
builder
.push("\"trigger\" = ")
.push_bind(trigger.to_string());
}
if let Some(task_key_substring) = query.task_key_substring.as_deref() {
push_where(builder);
builder
.push("task_key ILIKE ")
.push_bind(format!("%{}%", task_key_substring.trim()));
}
}
}
#[async_trait]
impl BackgroundTaskReadRepository for SqlxBackgroundTaskRepository {
async fn find_run(
&self,
run_id: &str,
) -> Result<Option<StoredBackgroundTaskRun>, DataLayerError> {
let row = sqlx::query(&format!("{RUN_COLUMNS} WHERE id = $1 LIMIT 1"))
.bind(run_id)
.fetch_optional(&self.pool)
.await
.map_postgres_err()?;
row.as_ref().map(map_run_row).transpose()
}
async fn list_runs(
&self,
query: &BackgroundTaskListQuery,
) -> Result<StoredBackgroundTaskRunPage, DataLayerError> {
let limit = query.limit.max(1);
let mut count_builder =
QueryBuilder::<Postgres>::new("SELECT COUNT(id) AS total FROM background_task_runs");
Self::apply_run_filter(&mut count_builder, query, false);
let total = count_builder
.build_query_scalar::<i64>()
.fetch_one(&self.pool)
.await
.map_postgres_err()?;
let mut builder = QueryBuilder::<Postgres>::new(RUN_COLUMNS);
Self::apply_run_filter(&mut builder, query, false);
builder
.push(" ORDER BY created_at_unix_secs DESC, updated_at_unix_secs DESC")
.push(" LIMIT ")
.push_bind(i64_from_usize(limit, "background task run limit")?)
.push(" OFFSET ")
.push_bind(i64_from_usize(query.offset, "background task run offset")?);
let rows = builder
.build()
.fetch_all(&self.pool)
.await
.map_postgres_err()?;
let items = rows
.iter()
.map(map_run_row)
.collect::<Result<Vec<_>, _>>()?;
Ok(StoredBackgroundTaskRunPage {
items,
total: usize::try_from(total).unwrap_or_default(),
})
}
async fn list_events(
&self,
run_id: &str,
offset: usize,
limit: usize,
) -> Result<Vec<StoredBackgroundTaskEvent>, DataLayerError> {
let limit = limit.max(1);
let rows = sqlx::query(&format!(
"{EVENT_COLUMNS} WHERE run_id = $1 ORDER BY created_at_unix_secs ASC, id ASC LIMIT $2 OFFSET $3"
))
.bind(run_id)
.bind(i64_from_usize(limit, "background task event limit")?)
.bind(i64_from_usize(offset, "background task event offset")?)
.fetch_all(&self.pool)
.await
.map_postgres_err()?;
rows.iter().map(map_event_row).collect()
}
async fn summarize_runs(&self) -> Result<BackgroundTaskSummary, DataLayerError> {
let total = sqlx::query_scalar::<_, i64>("SELECT COUNT(id) FROM background_task_runs")
.fetch_one(&self.pool)
.await
.map_postgres_err()?;
let running_count = sqlx::query_scalar::<_, i64>(
"SELECT COUNT(id) FROM background_task_runs WHERE status = 'running'",
)
.fetch_one(&self.pool)
.await
.map_postgres_err()?;
let status_rows = sqlx::query(
"SELECT status, COUNT(id) AS total FROM background_task_runs GROUP BY status",
)
.fetch_all(&self.pool)
.await
.map_postgres_err()?;
let kind_rows =
sqlx::query("SELECT kind, COUNT(id) AS total FROM background_task_runs GROUP BY kind")
.fetch_all(&self.pool)
.await
.map_postgres_err()?;
let mut by_status = std::collections::BTreeMap::new();
for row in status_rows {
let key: String = row.try_get("status").map_postgres_err()?;
let count: i64 = row.try_get("total").map_postgres_err()?;
by_status.insert(key, u64::try_from(count).unwrap_or_default());
}
let mut by_kind = std::collections::BTreeMap::new();
for row in kind_rows {
let key: String = row.try_get("kind").map_postgres_err()?;
let count: i64 = row.try_get("total").map_postgres_err()?;
by_kind.insert(key, u64::try_from(count).unwrap_or_default());
}
Ok(BackgroundTaskSummary {
total: u64::try_from(total).unwrap_or_default(),
running_count: u64::try_from(running_count).unwrap_or_default(),
by_status,
by_kind,
})
}
}
#[async_trait]
impl BackgroundTaskWriteRepository for SqlxBackgroundTaskRepository {
async fn upsert_run(
&self,
run: UpsertBackgroundTaskRun,
) -> Result<StoredBackgroundTaskRun, DataLayerError> {
run.validate()?;
sqlx::query(
r#"
INSERT INTO background_task_runs (
id,
task_key,
kind,
"trigger",
status,
attempt,
max_attempts,
owner_instance,
progress_percent,
progress_message,
payload_json,
result_json,
error_message,
cancel_requested,
created_by,
created_at_unix_secs,
started_at_unix_secs,
finished_at_unix_secs,
updated_at_unix_secs
) VALUES (
$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19
)
ON CONFLICT(id) DO UPDATE SET
task_key = EXCLUDED.task_key,
kind = EXCLUDED.kind,
"trigger" = EXCLUDED."trigger",
status = EXCLUDED.status,
attempt = EXCLUDED.attempt,
max_attempts = EXCLUDED.max_attempts,
owner_instance = EXCLUDED.owner_instance,
progress_percent = EXCLUDED.progress_percent,
progress_message = EXCLUDED.progress_message,
payload_json = EXCLUDED.payload_json,
result_json = EXCLUDED.result_json,
error_message = EXCLUDED.error_message,
cancel_requested = EXCLUDED.cancel_requested,
created_by = EXCLUDED.created_by,
created_at_unix_secs = EXCLUDED.created_at_unix_secs,
started_at_unix_secs = EXCLUDED.started_at_unix_secs,
finished_at_unix_secs = EXCLUDED.finished_at_unix_secs,
updated_at_unix_secs = EXCLUDED.updated_at_unix_secs
"#,
)
.bind(&run.id)
.bind(&run.task_key)
.bind(run.kind.as_database())
.bind(&run.trigger)
.bind(run.status.as_database())
.bind(u32_to_i32(run.attempt, "attempt")?)
.bind(u32_to_i32(run.max_attempts, "max_attempts")?)
.bind(run.owner_instance.as_deref())
.bind(i32::from(run.progress_percent))
.bind(run.progress_message.as_deref())
.bind(run.payload_json.clone())
.bind(run.result_json.clone())
.bind(run.error_message.as_deref())
.bind(run.cancel_requested)
.bind(run.created_by.as_deref())
.bind(u64_to_i64(
run.created_at_unix_secs,
"created_at_unix_secs",
)?)
.bind(run.started_at_unix_secs.map(|value| value as i64))
.bind(run.finished_at_unix_secs.map(|value| value as i64))
.bind(u64_to_i64(
run.updated_at_unix_secs,
"updated_at_unix_secs",
)?)
.execute(&self.pool)
.await
.map_postgres_err()?;
self.find_run(&run.id).await?.ok_or_else(|| {
DataLayerError::UnexpectedValue("background task run missing after upsert".to_string())
})
}
async fn request_cancel(
&self,
run_id: &str,
updated_at_unix_secs: u64,
) -> Result<bool, DataLayerError> {
let affected = sqlx::query(
"UPDATE background_task_runs SET cancel_requested = TRUE, updated_at_unix_secs = $2 WHERE id = $1",
)
.bind(run_id)
.bind(u64_to_i64(updated_at_unix_secs, "updated_at_unix_secs")?)
.execute(&self.pool)
.await
.map_postgres_err()?
.rows_affected();
Ok(affected > 0)
}
async fn upsert_event(
&self,
event: UpsertBackgroundTaskEvent,
) -> Result<StoredBackgroundTaskEvent, DataLayerError> {
event.validate()?;
sqlx::query(
r#"
INSERT INTO background_task_events (
id,
run_id,
event_type,
message,
payload_json,
created_at_unix_secs
) VALUES ($1,$2,$3,$4,$5,$6)
ON CONFLICT(id) DO UPDATE SET
run_id = EXCLUDED.run_id,
event_type = EXCLUDED.event_type,
message = EXCLUDED.message,
payload_json = EXCLUDED.payload_json,
created_at_unix_secs = EXCLUDED.created_at_unix_secs
"#,
)
.bind(&event.id)
.bind(&event.run_id)
.bind(&event.event_type)
.bind(&event.message)
.bind(event.payload_json.clone())
.bind(u64_to_i64(
event.created_at_unix_secs,
"created_at_unix_secs",
)?)
.execute(&self.pool)
.await
.map_postgres_err()?;
let row = sqlx::query(&format!("{EVENT_COLUMNS} WHERE id = $1 LIMIT 1"))
.bind(&event.id)
.fetch_one(&self.pool)
.await
.map_postgres_err()?;
map_event_row(&row)
}
}
fn map_run_row(row: &PgRow) -> Result<StoredBackgroundTaskRun, DataLayerError> {
let kind: String = row.try_get("kind").map_postgres_err()?;
let status: String = row.try_get("status").map_postgres_err()?;
let attempt: i32 = row.try_get("attempt").map_postgres_err()?;
let max_attempts: i32 = row.try_get("max_attempts").map_postgres_err()?;
let progress_percent: i32 = row.try_get("progress_percent").map_postgres_err()?;
let created_at_unix_secs: i64 = row.try_get("created_at_unix_secs").map_postgres_err()?;
let started_at_unix_secs: Option<i64> =
row.try_get("started_at_unix_secs").map_postgres_err()?;
let finished_at_unix_secs: Option<i64> =
row.try_get("finished_at_unix_secs").map_postgres_err()?;
let updated_at_unix_secs: i64 = row.try_get("updated_at_unix_secs").map_postgres_err()?;
Ok(StoredBackgroundTaskRun {
id: row.try_get("id").map_postgres_err()?,
task_key: row.try_get("task_key").map_postgres_err()?,
kind: BackgroundTaskKind::from_database(&kind)?,
trigger: row.try_get("trigger").map_postgres_err()?,
status: BackgroundTaskStatus::from_database(&status)?,
attempt: u32::try_from(attempt).unwrap_or_default(),
max_attempts: u32::try_from(max_attempts).unwrap_or_default(),
owner_instance: row.try_get("owner_instance").map_postgres_err()?,
progress_percent: u16::try_from(progress_percent).unwrap_or_default(),
progress_message: row.try_get("progress_message").map_postgres_err()?,
payload_json: row.try_get("payload_json").map_postgres_err()?,
result_json: row.try_get("result_json").map_postgres_err()?,
error_message: row.try_get("error_message").map_postgres_err()?,
cancel_requested: row.try_get("cancel_requested").map_postgres_err()?,
created_by: row.try_get("created_by").map_postgres_err()?,
created_at_unix_secs: u64::try_from(created_at_unix_secs).unwrap_or_default(),
started_at_unix_secs: started_at_unix_secs.and_then(|value| u64::try_from(value).ok()),
finished_at_unix_secs: finished_at_unix_secs.and_then(|value| u64::try_from(value).ok()),
updated_at_unix_secs: u64::try_from(updated_at_unix_secs).unwrap_or_default(),
})
}
fn map_event_row(row: &PgRow) -> Result<StoredBackgroundTaskEvent, DataLayerError> {
let created_at_unix_secs: i64 = row.try_get("created_at_unix_secs").map_postgres_err()?;
Ok(StoredBackgroundTaskEvent {
id: row.try_get("id").map_postgres_err()?,
run_id: row.try_get("run_id").map_postgres_err()?,
event_type: row.try_get("event_type").map_postgres_err()?,
message: row.try_get("message").map_postgres_err()?,
payload_json: row.try_get("payload_json").map_postgres_err()?,
created_at_unix_secs: u64::try_from(created_at_unix_secs).unwrap_or_default(),
})
}
fn i64_from_usize(value: usize, label: &str) -> Result<i64, DataLayerError> {
i64::try_from(value).map_err(|_| {
DataLayerError::UnexpectedValue(format!("background task {label} overflow: {value}"))
})
}
fn u64_to_i64(value: u64, label: &str) -> Result<i64, DataLayerError> {
i64::try_from(value).map_err(|_| {
DataLayerError::UnexpectedValue(format!("background task {label} overflow: {value}"))
})
}
fn u32_to_i32(value: u32, label: &str) -> Result<i32, DataLayerError> {
i32::try_from(value).map_err(|_| {
DataLayerError::UnexpectedValue(format!("background task {label} overflow: {value}"))
})
}

430
crates/aether-data/src/repository/background_tasks/sqlite.rs Normal file
View File

@@ -0,0 +1,430 @@
use async_trait::async_trait;
use sqlx::{sqlite::SqliteRow, QueryBuilder, Row, Sqlite};
use super::{
BackgroundTaskKind, BackgroundTaskListQuery, BackgroundTaskReadRepository,
BackgroundTaskStatus, BackgroundTaskSummary, BackgroundTaskWriteRepository,
StoredBackgroundTaskEvent, StoredBackgroundTaskRun, StoredBackgroundTaskRunPage,
UpsertBackgroundTaskEvent, UpsertBackgroundTaskRun,
};
use crate::driver::sqlite::SqlitePool;
use crate::error::SqlResultExt;
use crate::DataLayerError;
const RUN_COLUMNS: &str = r#"
SELECT
id,
task_key,
kind,
"trigger",
status,
attempt,
max_attempts,
owner_instance,
progress_percent,
progress_message,
payload_json,
result_json,
error_message,
cancel_requested,
created_by,
created_at_unix_secs,
started_at_unix_secs,
finished_at_unix_secs,
updated_at_unix_secs
FROM background_task_runs
"#;
const EVENT_COLUMNS: &str = r#"
SELECT
id,
run_id,
event_type,
message,
payload_json,
created_at_unix_secs
FROM background_task_events
"#;
#[derive(Debug, Clone)]
pub struct SqliteBackgroundTaskRepository {
pool: SqlitePool,
}
impl SqliteBackgroundTaskRepository {
pub fn new(pool: SqlitePool) -> Self {
Self { pool }
}
fn apply_run_filter(builder: &mut QueryBuilder<'_, Sqlite>, query: &BackgroundTaskListQuery) {
let mut has_where = false;
if let Some(kind) = query.kind {
if !has_where {
builder.push(" WHERE ");
has_where = true;
} else {
builder.push(" AND ");
}
builder.push("kind = ").push_bind(kind.as_database());
}
if let Some(status) = query.status {
if !has_where {
builder.push(" WHERE ");
has_where = true;
} else {
builder.push(" AND ");
}
builder.push("status = ").push_bind(status.as_database());
}
if let Some(trigger) = query.trigger.as_deref() {
if !has_where {
builder.push(" WHERE ");
has_where = true;
} else {
builder.push(" AND ");
}
builder
.push("\"trigger\" = ")
.push_bind(trigger.to_string());
}
if let Some(task_key_substring) = query.task_key_substring.as_deref() {
if !has_where {
builder.push(" WHERE ");
} else {
builder.push(" AND ");
}
builder.push("LOWER(task_key) LIKE ").push_bind(format!(
"%{}%",
task_key_substring.trim().to_ascii_lowercase()
));
}
}
}
#[async_trait]
impl BackgroundTaskReadRepository for SqliteBackgroundTaskRepository {
async fn find_run(
&self,
run_id: &str,
) -> Result<Option<StoredBackgroundTaskRun>, DataLayerError> {
let row = sqlx::query(&format!("{RUN_COLUMNS} WHERE id = ? LIMIT 1"))
.bind(run_id)
.fetch_optional(&self.pool)
.await
.map_sql_err()?;
row.as_ref().map(map_run_row).transpose()
}
async fn list_runs(
&self,
query: &BackgroundTaskListQuery,
) -> Result<StoredBackgroundTaskRunPage, DataLayerError> {
let limit = query.limit.max(1);
let mut count_builder =
QueryBuilder::<Sqlite>::new("SELECT COUNT(id) AS total FROM background_task_runs");
Self::apply_run_filter(&mut count_builder, query);
let total = count_builder
.build_query_scalar::<i64>()
.fetch_one(&self.pool)
.await
.map_sql_err()?;
let mut builder = QueryBuilder::<Sqlite>::new(RUN_COLUMNS);
Self::apply_run_filter(&mut builder, query);
builder
.push(" ORDER BY created_at_unix_secs DESC, updated_at_unix_secs DESC")
.push(" LIMIT ")
.push_bind(i64_from_usize(limit, "run limit")?)
.push(" OFFSET ")
.push_bind(i64_from_usize(query.offset, "run offset")?);
let rows = builder.build().fetch_all(&self.pool).await.map_sql_err()?;
let items = rows
.iter()
.map(map_run_row)
.collect::<Result<Vec<_>, _>>()?;
Ok(StoredBackgroundTaskRunPage {
items,
total: usize::try_from(total).unwrap_or_default(),
})
}
async fn list_events(
&self,
run_id: &str,
offset: usize,
limit: usize,
) -> Result<Vec<StoredBackgroundTaskEvent>, DataLayerError> {
let limit = limit.max(1);
let rows = sqlx::query(&format!(
"{EVENT_COLUMNS} WHERE run_id = ? ORDER BY created_at_unix_secs ASC, id ASC LIMIT ? OFFSET ?"
))
.bind(run_id)
.bind(i64_from_usize(limit, "event limit")?)
.bind(i64_from_usize(offset, "event offset")?)
.fetch_all(&self.pool)
.await
.map_sql_err()?;
rows.iter().map(map_event_row).collect()
}
async fn summarize_runs(&self) -> Result<BackgroundTaskSummary, DataLayerError> {
let total = sqlx::query_scalar::<_, i64>("SELECT COUNT(id) FROM background_task_runs")
.fetch_one(&self.pool)
.await
.map_sql_err()?;
let running_count = sqlx::query_scalar::<_, i64>(
"SELECT COUNT(id) FROM background_task_runs WHERE status = 'running'",
)
.fetch_one(&self.pool)
.await
.map_sql_err()?;
let status_rows = sqlx::query(
"SELECT status, COUNT(id) AS total FROM background_task_runs GROUP BY status",
)
.fetch_all(&self.pool)
.await
.map_sql_err()?;
let kind_rows =
sqlx::query("SELECT kind, COUNT(id) AS total FROM background_task_runs GROUP BY kind")
.fetch_all(&self.pool)
.await
.map_sql_err()?;
let mut by_status = std::collections::BTreeMap::new();
for row in status_rows {
let key: String = row.try_get("status").map_sql_err()?;
let count: i64 = row.try_get("total").map_sql_err()?;
by_status.insert(key, u64::try_from(count).unwrap_or_default());
}
let mut by_kind = std::collections::BTreeMap::new();
for row in kind_rows {
let key: String = row.try_get("kind").map_sql_err()?;
let count: i64 = row.try_get("total").map_sql_err()?;
by_kind.insert(key, u64::try_from(count).unwrap_or_default());
}
Ok(BackgroundTaskSummary {
total: u64::try_from(total).unwrap_or_default(),
running_count: u64::try_from(running_count).unwrap_or_default(),
by_status,
by_kind,
})
}
}
#[async_trait]
impl BackgroundTaskWriteRepository for SqliteBackgroundTaskRepository {
async fn upsert_run(
&self,
run: UpsertBackgroundTaskRun,
) -> Result<StoredBackgroundTaskRun, DataLayerError> {
run.validate()?;
sqlx::query(
r#"
INSERT INTO background_task_runs (
id,
task_key,
kind,
"trigger",
status,
attempt,
max_attempts,
owner_instance,
progress_percent,
progress_message,
payload_json,
result_json,
error_message,
cancel_requested,
created_by,
created_at_unix_secs,
started_at_unix_secs,
finished_at_unix_secs,
updated_at_unix_secs
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(id) DO UPDATE SET
task_key = excluded.task_key,
kind = excluded.kind,
"trigger" = excluded."trigger",
status = excluded.status,
attempt = excluded.attempt,
max_attempts = excluded.max_attempts,
owner_instance = excluded.owner_instance,
progress_percent = excluded.progress_percent,
progress_message = excluded.progress_message,
payload_json = excluded.payload_json,
result_json = excluded.result_json,
error_message = excluded.error_message,
cancel_requested = excluded.cancel_requested,
created_by = excluded.created_by,
created_at_unix_secs = excluded.created_at_unix_secs,
started_at_unix_secs = excluded.started_at_unix_secs,
finished_at_unix_secs = excluded.finished_at_unix_secs,
updated_at_unix_secs = excluded.updated_at_unix_secs
"#,
)
.bind(&run.id)
.bind(&run.task_key)
.bind(run.kind.as_database())
.bind(&run.trigger)
.bind(run.status.as_database())
.bind(i64::from(run.attempt))
.bind(i64::from(run.max_attempts))
.bind(run.owner_instance.as_deref())
.bind(i32::from(run.progress_percent))
.bind(run.progress_message.as_deref())
.bind(run.payload_json.as_ref().map(serde_json::Value::to_string))
.bind(run.result_json.as_ref().map(serde_json::Value::to_string))
.bind(run.error_message.as_deref())
.bind(run.cancel_requested)
.bind(run.created_by.as_deref())
.bind(u64_to_i64(
run.created_at_unix_secs,
"created_at_unix_secs",
)?)
.bind(run.started_at_unix_secs.map(|value| value as i64))
.bind(run.finished_at_unix_secs.map(|value| value as i64))
.bind(u64_to_i64(
run.updated_at_unix_secs,
"updated_at_unix_secs",
)?)
.execute(&self.pool)
.await
.map_sql_err()?;
self.find_run(&run.id).await?.ok_or_else(|| {
DataLayerError::UnexpectedValue("background task run missing after upsert".to_string())
})
}
async fn request_cancel(
&self,
run_id: &str,
updated_at_unix_secs: u64,
) -> Result<bool, DataLayerError> {
let affected = sqlx::query(
"UPDATE background_task_runs SET cancel_requested = 1, updated_at_unix_secs = ? WHERE id = ?",
)
.bind(u64_to_i64(updated_at_unix_secs, "updated_at_unix_secs")?)
.bind(run_id)
.execute(&self.pool)
.await
.map_sql_err()?
.rows_affected();
Ok(affected > 0)
}
async fn upsert_event(
&self,
event: UpsertBackgroundTaskEvent,
) -> Result<StoredBackgroundTaskEvent, DataLayerError> {
event.validate()?;
sqlx::query(
r#"
INSERT INTO background_task_events (
id, run_id, event_type, message, payload_json, created_at_unix_secs
) VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
run_id = excluded.run_id,
event_type = excluded.event_type,
message = excluded.message,
payload_json = excluded.payload_json,
created_at_unix_secs = excluded.created_at_unix_secs
"#,
)
.bind(&event.id)
.bind(&event.run_id)
.bind(&event.event_type)
.bind(&event.message)
.bind(
event
.payload_json
.as_ref()
.map(serde_json::Value::to_string),
)
.bind(u64_to_i64(
event.created_at_unix_secs,
"created_at_unix_secs",
)?)
.execute(&self.pool)
.await
.map_sql_err()?;
let row = sqlx::query(&format!("{EVENT_COLUMNS} WHERE id = ? LIMIT 1"))
.bind(&event.id)
.fetch_one(&self.pool)
.await
.map_sql_err()?;
map_event_row(&row)
}
}
fn map_run_row(row: &SqliteRow) -> Result<StoredBackgroundTaskRun, DataLayerError> {
let kind: String = row.try_get("kind").map_sql_err()?;
let status: String = row.try_get("status").map_sql_err()?;
let attempt: i64 = row.try_get("attempt").map_sql_err()?;
let max_attempts: i64 = row.try_get("max_attempts").map_sql_err()?;
let progress_percent: i32 = row.try_get("progress_percent").map_sql_err()?;
let created_at_unix_secs: i64 = row.try_get("created_at_unix_secs").map_sql_err()?;
let started_at_unix_secs: Option<i64> = row.try_get("started_at_unix_secs").map_sql_err()?;
let finished_at_unix_secs: Option<i64> = row.try_get("finished_at_unix_secs").map_sql_err()?;
let updated_at_unix_secs: i64 = row.try_get("updated_at_unix_secs").map_sql_err()?;
Ok(StoredBackgroundTaskRun {
id: row.try_get("id").map_sql_err()?,
task_key: row.try_get("task_key").map_sql_err()?,
kind: BackgroundTaskKind::from_database(&kind)?,
trigger: row.try_get("trigger").map_sql_err()?,
status: BackgroundTaskStatus::from_database(&status)?,
attempt: u32::try_from(attempt).unwrap_or_default(),
max_attempts: u32::try_from(max_attempts).unwrap_or_default(),
owner_instance: row.try_get("owner_instance").map_sql_err()?,
progress_percent: u16::try_from(progress_percent).unwrap_or_default(),
progress_message: row.try_get("progress_message").map_sql_err()?,
payload_json: parse_optional_json(row.try_get("payload_json").map_sql_err()?)?,
result_json: parse_optional_json(row.try_get("result_json").map_sql_err()?)?,
error_message: row.try_get("error_message").map_sql_err()?,
cancel_requested: row.try_get("cancel_requested").map_sql_err()?,
created_by: row.try_get("created_by").map_sql_err()?,
created_at_unix_secs: u64::try_from(created_at_unix_secs).unwrap_or_default(),
started_at_unix_secs: started_at_unix_secs.and_then(|value| u64::try_from(value).ok()),
finished_at_unix_secs: finished_at_unix_secs.and_then(|value| u64::try_from(value).ok()),
updated_at_unix_secs: u64::try_from(updated_at_unix_secs).unwrap_or_default(),
})
}
fn map_event_row(row: &SqliteRow) -> Result<StoredBackgroundTaskEvent, DataLayerError> {
let created_at_unix_secs: i64 = row.try_get("created_at_unix_secs").map_sql_err()?;
Ok(StoredBackgroundTaskEvent {
id: row.try_get("id").map_sql_err()?,
run_id: row.try_get("run_id").map_sql_err()?,
event_type: row.try_get("event_type").map_sql_err()?,
message: row.try_get("message").map_sql_err()?,
payload_json: parse_optional_json(row.try_get("payload_json").map_sql_err()?)?,
created_at_unix_secs: u64::try_from(created_at_unix_secs).unwrap_or_default(),
})
}
fn parse_optional_json(value: Option<String>) -> Result<Option<serde_json::Value>, DataLayerError> {
value
.map(|raw| {
serde_json::from_str::<serde_json::Value>(&raw).map_err(|err| {
DataLayerError::UnexpectedValue(format!(
"invalid background task json payload: {err}"
))
})
})
.transpose()
}
fn i64_from_usize(value: usize, label: &str) -> Result<i64, DataLayerError> {
i64::try_from(value).map_err(|_| {
DataLayerError::UnexpectedValue(format!("background task {label} overflow: {value}"))
})
}
fn u64_to_i64(value: u64, label: &str) -> Result<i64, DataLayerError> {
i64::try_from(value).map_err(|_| {
DataLayerError::UnexpectedValue(format!("background task {label} overflow: {value}"))
})
}

1
crates/aether-data/src/repository/mod.rs
View File

@@ -8,6 +8,7 @@ pub mod announcements;
pub mod audit;
pub mod auth;
pub mod auth_modules;
pub mod background_tasks;
pub mod billing;
pub mod candidate_selection;
pub mod candidates;