use axum::Router;
use axum::extract::State;
use axum::http::StatusCode;
use axum::routing::get;
use serde::{Deserialize, Serialize};
use time::format_description::well_known::Rfc3339;
use uuid::Uuid;
use crate::api::contenido;
use crate::api::extract::Query;
use crate::api::response::ApiResponse;
use crate::db::entities::{Ejecucion, Estado, FiltroDeEjecuciones};
use crate::error::AppError;
use crate::state::AppState;
pub const LIMITE_POR_DEFECTO: i64 = 50;
pub const LIMITE_MAXIMO: i64 = 200;
pub fn router() -> Router<AppState> {
Router::new().route("/api/v1/executions", get(listado))
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FiltrosDelCable {
pub job_id: Option<String>,
pub status: Option<String>,
pub limit: Option<i64>,
pub offset: Option<i64>,
}
impl FiltrosDelCable {
fn validar(self) -> Result<FiltroDeEjecuciones, AppError> {
let job_id = match self.job_id.as_deref() {
Some(valor) => Some(Uuid::parse_str(valor).map_err(|_| {
AppError::InvalidInput(
"el campo `job_id` de la consulta no tiene la forma de un UUID".to_owned(),
)
})?),
None => None,
};
let status = match self.status.as_deref() {
Some(valor) => Some(Estado::desde_texto(valor).ok_or_else(|| {
AppError::InvalidInput(format!(
"el campo `status` de la consulta no es uno de los cuatro estados admitidos: {}",
estados_admitidos()
))
})?),
None => None,
};
let limite = match self.limit {
Some(valor) if !(1..=LIMITE_MAXIMO).contains(&valor) => {
return Err(AppError::InvalidInput(format!(
"el campo `limit` de la consulta tiene que ser un número entre 1 y {LIMITE_MAXIMO}"
)));
}
Some(valor) => valor,
None => LIMITE_POR_DEFECTO,
};
let desplazamiento = match self.offset {
Some(valor) if valor < 0 => {
return Err(AppError::InvalidInput(
"el campo `offset` de la consulta no puede ser negativo".to_owned(),
));
}
Some(valor) => valor,
None => 0,
};
Ok(FiltroDeEjecuciones {
job_id,
status,
limite,
desplazamiento,
})
}
}
fn estados_admitidos() -> String {
[
Estado::Pending,
Estado::Running,
Estado::Success,
Estado::Failed,
]
.map(Estado::texto)
.join(", ")
}
#[derive(Debug, Serialize)]
pub struct EjecucionDelCable {
pub id: String,
pub job_id: String,
pub status: String,
pub snapshot_id: Option<String>,
pub bytes_processed: i64,
pub bytes_uploaded: i64,
pub error_message: Option<String>,
pub started_at: String,
pub finished_at: Option<String>,
}
impl EjecucionDelCable {
fn desde(fila: &Ejecucion) -> Result<Self, AppError> {
Ok(Self {
id: fila.id.hyphenated().to_string(),
job_id: fila.job_id.hyphenated().to_string(),
status: fila.status.texto().to_owned(),
snapshot_id: fila.snapshot_id.clone(),
bytes_processed: fila.bytes_processed,
bytes_uploaded: fila.bytes_uploaded,
error_message: fila.error_message.clone(),
started_at: formatear(fila.started_at)?,
finished_at: fila.finished_at.map(formatear).transpose()?,
})
}
}
fn formatear(fecha: time::OffsetDateTime) -> Result<String, AppError> {
fecha.format(&Rfc3339).map_err(|error| {
AppError::Other(format!("la fecha de la fila no se pudo formatear: {error}"))
})
}
pub async fn listado(
State(estado): State<AppState>,
Query(filtros): Query<FiltrosDelCable>,
) -> Result<ApiResponse, AppError> {
let filtro = filtros.validar()?;
let filas = estado.db.ejecuciones.listar_filtrado(&filtro).await?;
let vistas: Vec<EjecucionDelCable> = filas
.iter()
.map(EjecucionDelCable::desde)
.collect::<Result<_, _>>()?;
Ok(ApiResponse::new(
StatusCode::OK,
"ejecuciones listadas",
contenido(&vistas)?,
))
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use axum::http::{StatusCode, header};
use time::format_description::well_known::Rfc3339;
use time::macros::datetime;
use time::{Duration, OffsetDateTime};
use uuid::Uuid;
use crate::api::api_router;
use crate::db::entities::{
AltaDeEjecucion, AltaDeRepositorio, AltaDeTarea, AltaDeTarget, Estado, TipoDeTarget,
};
use crate::secret::SecretString;
use crate::state::{AppState, EstadoDeBase};
use crate::test_support::{
DirectorioDePrueba, base_con_clave, body_json, body_text, config_con_dsn, estado_de_prueba,
estado_desarrollo, get,
};
async fn sembrar_tarea(estado: &AppState) -> Uuid {
let target = estado
.db
.targets
.alta(AltaDeTarget {
name: "target-de-ejecuciones".to_owned(),
target_type: TipoDeTarget::Directory,
container_name: None,
db_user: None,
db_password: None,
db_name: None,
host_path: Some("/srv/datos".to_owned()),
db_path: None,
})
.await
.expect("el target de prueba se da de alta");
let repositorio = estado
.db
.repositorios
.alta(AltaDeRepositorio {
name: "repositorio-de-ejecuciones".to_owned(),
endpoint: "https://s3.example.test/ejecuciones".to_owned(),
bucket: "cubo-de-ejecuciones".to_owned(),
region: "eu-west-1".to_owned(),
access_key_id: "AKIAIOSFODNN7EXAMPLE".to_owned(),
secret_access_key: SecretString::new(
"credencial-de-prueba-0000000000000001".to_owned(),
),
passphrase: SecretString::new("passphrase-de-prueba-00000000000001".to_owned()),
})
.await
.expect("el repositorio de prueba se da de alta");
let tarea = estado
.db
.tareas
.alta(AltaDeTarea {
target_id: target.id,
repository_id: repositorio.id,
cron_expression: "0 3 * * *".to_owned(),
keep_daily: 7,
keep_weekly: 4,
keep_monthly: 6,
enabled: true,
})
.await
.expect("la tarea de prueba se da de alta");
tarea.id
}
async fn sembrar_ejecucion(
estado: &AppState,
job_id: Uuid,
estado_de_la_fila: Estado,
iniciada: OffsetDateTime,
procesados: i64,
) -> Uuid {
let fila = estado
.db
.ejecuciones
.alta(AltaDeEjecucion {
job_id,
status: estado_de_la_fila,
snapshot_id: Some("snapshot-de-prueba-00000000000000000001".to_owned()),
bytes_processed: procesados,
bytes_uploaded: procesados / 2,
error_message: None,
started_at: iniciada,
finished_at: Some(iniciada + Duration::seconds(30)),
})
.await
.expect("la ejecución de prueba se da de alta");
fila.id
}
async fn estado_con_base() -> (AppState, DirectorioDePrueba) {
let (directorio, pool, clave) = base_con_clave().await;
let estado = estado_de_prueba(
config_con_dsn(&directorio.dsn()),
EstadoDeBase::nuevo(pool, Arc::clone(&clave)),
);
(estado, directorio)
}
#[tokio::test]
async fn el_listado_sin_filtros_devuelve_las_ejecuciones_con_sus_metricas() {
let (estado, _directorio) = estado_con_base().await;
let tarea = sembrar_tarea(&estado).await;
let base = datetime!(2026-10-25 03:00:00 UTC);
for (desplazamiento, procesados) in [(0_i64, 1000_i64), (10, 2000), (20, 3000)].into_iter()
{
sembrar_ejecucion(
&estado,
tarea,
Estado::Success,
base + Duration::seconds(desplazamiento),
procesados,
)
.await;
}
let respuesta = get(api_router(estado), "/api/v1/executions").await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::OK,
"el listado contesta {status}: {cuerpo}"
);
let filas = cuerpo["data"]
.as_array()
.expect("el listado tiene que llegar como un array");
assert_eq!(filas.len(), 3, "sin filtros salen las tres filas");
for clave in [
"bytes_processed",
"bytes_uploaded",
"started_at",
"finished_at",
] {
assert!(
filas[0].get(clave).is_some(),
"la fila del listado tiene que traer la métrica `{clave}`: {}",
filas[0]
);
}
assert_eq!(
filas[0]["bytes_processed"], 3000,
"el orden es descendente por la fecha de inicio y la más reciente es la primera"
);
assert_eq!(
filas[2]["bytes_processed"], 1000,
"la más antigua tiene que quedar la última"
);
}
#[tokio::test]
async fn los_filtros_por_job_id_y_por_status_acotan_el_listado() {
let (estado, _directorio) = estado_con_base().await;
let tarea_a = sembrar_tarea(&estado).await;
let tarea_b = sembrar_tarea(&estado).await;
let base = datetime!(2026-10-25 03:00:00 UTC);
sembrar_ejecucion(&estado, tarea_a, Estado::Success, base, 1000).await;
sembrar_ejecucion(
&estado,
tarea_a,
Estado::Failed,
base + Duration::seconds(10),
2000,
)
.await;
sembrar_ejecucion(
&estado,
tarea_b,
Estado::Success,
base + Duration::seconds(20),
3000,
)
.await;
let respuesta = get(
api_router(estado.clone()),
&format!("/api/v1/executions?job_id={tarea_a}"),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::OK,
"el filtro por tarea contesta {status}"
);
let filas = cuerpo["data"].as_array().expect("el listado es un array");
assert_eq!(filas.len(), 2, "la tarea A tiene dos ejecuciones");
for fila in filas {
assert_eq!(
fila["job_id"],
tarea_a.hyphenated().to_string(),
"el filtro por tarea no puede devolver filas de otra"
);
}
let respuesta = get(
api_router(estado.clone()),
"/api/v1/executions?status=failed",
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::OK,
"el filtro por estado contesta {status}"
);
let filas = cuerpo["data"].as_array().expect("el listado es un array");
assert_eq!(filas.len(), 1, "solo una ejecución falló");
assert_eq!(filas[0]["status"], "failed", "y es la fallida");
let respuesta = get(
api_router(estado),
&format!("/api/v1/executions?job_id={tarea_a}&status=success"),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::OK,
"los dos filtros juntos contestan {status}"
);
let filas = cuerpo["data"].as_array().expect("el listado es un array");
assert_eq!(filas.len(), 1, "de la tarea A solo una terminó bien");
assert_eq!(filas[0]["status"], "success", "y es la satisfactoria");
}
#[tokio::test]
async fn el_limite_acota_y_el_desplazamiento_avanza() {
let (estado, _directorio) = estado_con_base().await;
let tarea = sembrar_tarea(&estado).await;
let base = datetime!(2026-10-25 03:00:00 UTC);
for (desplazamiento, procesados) in [(0_i64, 1000_i64), (10, 2000), (20, 3000)].into_iter()
{
sembrar_ejecucion(
&estado,
tarea,
Estado::Success,
base + Duration::seconds(desplazamiento),
procesados,
)
.await;
}
let primera =
body_json(get(api_router(estado.clone()), "/api/v1/executions?limit=1").await).await;
let filas = primera["data"].as_array().expect("el listado es un array");
assert_eq!(filas.len(), 1, "`limit=1` tiene que devolver una sola fila");
let identificador_primero = filas[0]["id"]
.as_str()
.expect("la fila trae su identificador")
.to_owned();
let segunda = body_json(
get(
api_router(estado.clone()),
"/api/v1/executions?limit=1&offset=1",
)
.await,
)
.await;
let filas = segunda["data"].as_array().expect("el listado es un array");
assert_eq!(filas.len(), 1, "`limit=1&offset=1` devuelve una sola fila");
let identificador_segundo = filas[0]["id"]
.as_str()
.expect("la fila trae su identificador")
.to_owned();
assert_ne!(
identificador_primero, identificador_segundo,
"el desplazamiento tiene que avanzar, no repetir la primera página"
);
let dos = body_json(get(api_router(estado), "/api/v1/executions?limit=2").await).await;
let filas = dos["data"].as_array().expect("el listado es un array");
assert_eq!(filas.len(), 2, "`limit=2` tiene que devolver dos filas");
}
#[tokio::test]
async fn un_limite_fuera_de_rango_o_no_numerico_contesta_400_sin_recortar() {
let (estado, _directorio) = estado_con_base().await;
let router = api_router(estado);
for ruta in [
"/api/v1/executions?limit=201",
"/api/v1/executions?limit=0",
"/api/v1/executions?limit=-1",
"/api/v1/executions?limit=abc",
"/api/v1/executions?offset=-1",
] {
let respuesta = get(router.clone(), ruta).await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::BAD_REQUEST,
"`{ruta}` tiene que contestar 400 y contestó {status}: {cuerpo}"
);
assert!(
cuerpo["data"].is_null(),
"un rechazo no puede traer listado: {cuerpo}"
);
}
let respuesta = get(router, "/api/v1/executions?limit=201").await;
let cuerpo = body_json(respuesta).await;
let mensaje = cuerpo["message"]
.as_str()
.expect("el rechazo trae mensaje")
.to_owned();
assert!(
mensaje.contains("limit"),
"el mensaje tiene que nombrar el campo: {mensaje}"
);
assert!(
!mensaje.contains("201"),
"el mensaje no repite el valor que llegó: {mensaje}"
);
}
#[tokio::test]
async fn dos_llamadas_seguidas_dan_el_mismo_orden_con_filas_empatadas() {
let (estado, _directorio) = estado_con_base().await;
let tarea = sembrar_tarea(&estado).await;
let empate = datetime!(2026-10-25 03:00:00 UTC);
sembrar_ejecucion(&estado, tarea, Estado::Success, empate, 1000).await;
sembrar_ejecucion(&estado, tarea, Estado::Failed, empate, 2000).await;
let mut secuencias = Vec::new();
for _ in 0..2 {
let cuerpo =
body_json(get(api_router(estado.clone()), "/api/v1/executions").await).await;
let filas = cuerpo["data"].as_array().expect("el listado es un array");
assert_eq!(filas.len(), 2, "las dos filas empatadas tienen que salir");
secuencias.push(
filas
.iter()
.map(|fila| {
fila["id"]
.as_str()
.expect("la fila trae su identificador")
.to_owned()
})
.collect::<Vec<_>>(),
);
}
assert_eq!(
secuencias[0], secuencias[1],
"el orden tiene que ser estable entre dos llamadas seguidas"
);
let mut ordenada = secuencias[0].clone();
ordenada.sort();
assert_eq!(
secuencias[0], ordenada,
"con la misma fecha, el desempate es el identificador"
);
}
#[tokio::test]
async fn las_fechas_de_una_fila_de_ejecucion_son_cadenas_rfc3339() {
let (estado, _directorio) = estado_con_base().await;
let tarea = sembrar_tarea(&estado).await;
let inicio = datetime!(2026-10-25 03:00:00 UTC);
sembrar_ejecucion(&estado, tarea, Estado::Success, inicio, 1000).await;
let cuerpo = body_json(get(api_router(estado), "/api/v1/executions").await).await;
let fila = &cuerpo["data"][0];
let empezada = fila["started_at"]
.as_str()
.unwrap_or_else(|| panic!("`started_at` tiene que ser una cadena: {fila}"));
let terminada = fila["finished_at"]
.as_str()
.unwrap_or_else(|| panic!("`finished_at` tiene que ser una cadena: {fila}"));
assert_eq!(
empezada, "2026-10-25T03:00:00Z",
"el formato del cable es RFC3339"
);
assert_eq!(
terminada, "2026-10-25T03:00:30Z",
"y el fin, treinta segundos después"
);
assert!(
!empezada.contains('.'),
"la fracción solo se escribe si los nanosegundos no son cero: {empezada}"
);
let instante = OffsetDateTime::parse(empezada, &Rfc3339)
.expect("la fecha del cable tiene que releerse como RFC3339");
assert_eq!(
instante, inicio,
"el texto tiene que significar el instante que se sembró"
);
}
#[tokio::test]
async fn un_parametro_de_consulta_mal_formado_contesta_400_con_la_envoltura() {
let respuesta = get(
api_router(estado_desarrollo().await),
"/api/v1/executions?limit=abc",
)
.await;
let status = respuesta.status();
let tipo = respuesta
.headers()
.get(header::CONTENT_TYPE)
.and_then(|valor| valor.to_str().ok())
.map(ToOwned::to_owned);
let texto = body_text(respuesta).await;
assert_eq!(
status,
StatusCode::BAD_REQUEST,
"el rechazo es un 400: {texto}"
);
assert!(
tipo.as_deref()
.is_some_and(|tipo| tipo.starts_with("application/json")),
"el rechazo tiene que salir por la envoltura, no en texto plano: {tipo:?}"
);
assert!(
!texto.contains("invalid digit"),
"el texto del intérprete no puede salir por el cable: {texto}"
);
assert!(
!texto.contains("abc"),
"el valor del parámetro no puede volver en el mensaje: {texto}"
);
let cuerpo: serde_json::Value =
serde_json::from_str(&texto).expect("el rechazo tiene que traer JSON en el cuerpo");
assert_eq!(
cuerpo["status"], 400,
"y el estado del cuerpo es el del 400"
);
assert!(
cuerpo["message"]
.as_str()
.is_some_and(|mensaje| !mensaje.is_empty()),
"el rechazo trae un mensaje: {texto}"
);
}
#[tokio::test]
async fn un_parametro_de_consulta_desconocido_se_rechaza() {
let respuesta = get(
api_router(estado_desarrollo().await),
"/api/v1/executions?limits=5",
)
.await;
let status = respuesta.status();
let texto = body_text(respuesta).await;
assert_eq!(
status,
StatusCode::BAD_REQUEST,
"la errata es un 400: {texto}"
);
assert!(
!texto.contains("unknown field"),
"el texto del intérprete no puede salir por el cable: {texto}"
);
let cuerpo: serde_json::Value =
serde_json::from_str(&texto).expect("el rechazo tiene que traer JSON en el cuerpo");
assert_eq!(
cuerpo["status"], 400,
"y el estado del cuerpo es el del 400"
);
assert!(
cuerpo["message"]
.as_str()
.is_some_and(|mensaje| !mensaje.is_empty()),
"el rechazo trae un mensaje: {texto}"
);
}
}