#![allow(missing_docs, clippy::missing_docs_in_private_items)]
use crate::error::HandlerError;
use crate::model::{
CreateSubscription, HalLink, SubscriptionHal, SubscriptionLinks, SubscriptionPage,
SubscriptionPageLinks, SubscriptionWithBranch, UpdateSubscription,
};
use crate::repository::subscription::SubscriptionRepository;
use crate::state::AppState;
use axum::{
Json,
extract::{Path, Query, State},
};
use rovo::rovo;
use serde::Deserialize;
use tracing::info;
fn map_to_hal(sub_with_branch: SubscriptionWithBranch) -> SubscriptionHal {
let id = sub_with_branch.subscription.id;
SubscriptionHal {
subscription: sub_with_branch.subscription,
source_branch: sub_with_branch.source_branch,
links: SubscriptionLinks {
self_link: HalLink {
href: format!("/subscriptions/{}", id),
},
update: HalLink {
href: format!("/subscriptions/{}", id),
},
delete: HalLink {
href: format!("/subscriptions/{}", id),
},
},
}
}
#[allow(rustdoc::invalid_html_tags)]
#[rovo]
pub async fn create_subscription(
state: State<AppState>,
payload: Json<CreateSubscription>,
) -> Result<Json<SubscriptionHal>, HandlerError> {
create_subscription_inner(state, payload).await
}
async fn create_subscription_inner(
State(state): State<AppState>,
Json(payload): Json<CreateSubscription>,
) -> Result<Json<SubscriptionHal>, HandlerError> {
let sub_with_branch = state.repository.create(&payload).await?;
info!(
"Registered new subscription for branch ID {} (repo: {}, branch: {}): {:?}",
sub_with_branch.subscription.branch_id,
sub_with_branch.source_branch.repo_url,
sub_with_branch.source_branch.name,
sub_with_branch.subscription
);
Ok(Json(map_to_hal(sub_with_branch)))
}
#[derive(Debug, Deserialize, rovo::schemars::JsonSchema)]
pub struct ListSubscriptionsQuery {
pub limit: Option<usize>,
pub last_id: Option<i64>,
}
#[allow(rustdoc::invalid_html_tags)]
#[rovo]
pub async fn list_subscriptions(
state: State<AppState>,
query: Query<ListSubscriptionsQuery>,
) -> Result<Json<SubscriptionPage>, HandlerError> {
list_subscriptions_inner(state, query).await
}
async fn list_subscriptions_inner(
State(state): State<AppState>,
Query(query): Query<ListSubscriptionsQuery>,
) -> Result<Json<SubscriptionPage>, HandlerError> {
let limit = query
.limit
.unwrap_or(state.config.database.subscriptions_list_limit)
.min(state.config.database.subscriptions_list_limit_cap);
let last_id = query.last_id.unwrap_or_default();
let subscriptions = state
.repository
.list_paginated_with_branches(last_id, limit as i64)
.await?;
let data: Vec<SubscriptionHal> = subscriptions.into_iter().map(map_to_hal).collect();
let next_id = data.last().map(|s| s.subscription.id).unwrap_or(last_id);
let remaining_count = state.repository.count_remaining(next_id).await?;
let next_link = data
.last()
.filter(|_| remaining_count > 0)
.map(|s| HalLink {
href: format!(
"/subscriptions?limit={}&last_id={}",
limit, s.subscription.id
),
});
Ok(Json(SubscriptionPage {
data,
remaining_count,
links: SubscriptionPageLinks { next: next_link },
}))
}
#[allow(rustdoc::invalid_html_tags)]
#[rovo]
pub async fn get_subscription(
state: State<AppState>,
Path(id): Path<i64>,
) -> Result<Json<SubscriptionHal>, HandlerError> {
get_subscription_inner(state, Path(id)).await
}
async fn get_subscription_inner(
State(state): State<AppState>,
Path(id): Path<i64>,
) -> Result<Json<SubscriptionHal>, HandlerError> {
let sub_with_branch = state
.repository
.get_by_id_with_branch(id)
.await?
.ok_or(HandlerError::NotFound)?;
Ok(Json(map_to_hal(sub_with_branch)))
}
#[allow(rustdoc::invalid_html_tags)]
#[rovo]
pub async fn update_subscription(
state: State<AppState>,
Path(id): Path<i64>,
payload: Json<UpdateSubscription>,
) -> Result<Json<SubscriptionHal>, HandlerError> {
update_subscription_inner(state, Path(id), payload).await
}
async fn update_subscription_inner(
State(state): State<AppState>,
Path(id): Path<i64>,
Json(payload): Json<UpdateSubscription>,
) -> Result<Json<SubscriptionHal>, HandlerError> {
state.repository.update(id, &payload).await?;
let sub_with_branch = state
.repository
.get_by_id_with_branch(id)
.await?
.ok_or(HandlerError::NotFound)?;
Ok(Json(map_to_hal(sub_with_branch)))
}
#[allow(rustdoc::invalid_html_tags)]
#[rovo]
pub async fn delete_subscription(
state: State<AppState>,
Path(id): Path<i64>,
) -> Result<(), HandlerError> {
delete_subscription_inner(state, Path(id)).await
}
async fn delete_subscription_inner(
State(state): State<AppState>,
Path(id): Path<i64>,
) -> Result<(), HandlerError> {
state.repository.delete_subscription_and_cascade(id).await?;
Ok(())
}
#[cfg(test)]
mod tests {
#![allow(
clippy::panic,
clippy::expect_used,
clippy::todo,
clippy::unimplemented,
clippy::indexing_slicing
)]
use super::*;
use crate::domain::{BranchName, EventType, RepoUrl, TargetRepo};
use crate::model::CreateSubscription;
use crate::state::AppState;
use crate::test_utils::create_test_db;
use axum::Json;
use axum::extract::State;
#[tokio::test]
async fn test_crud_subscription() {
let pool = create_test_db().await;
let config = crate::test_utils::create_test_config();
let state = AppState {
config: std::sync::Arc::new(config),
repository: std::sync::Arc::new(crate::repository::SqliteRepository::new(pool.clone())),
db_pool: pool.clone(),
};
let payload = CreateSubscription {
source_repo_url: RepoUrl::new("https://github.com/org/repo".to_string()).unwrap(),
source_branch_name: BranchName::new("main".to_string()).unwrap(),
target_repo: TargetRepo::new("org/target".to_string()).unwrap(),
event_type: EventType::new("dispatch".to_string()).unwrap(),
gh_app_installation_id: 1,
};
let res = create_subscription_inner(State(state.clone()), Json(payload))
.await
.unwrap();
let id = res.subscription.id;
assert_eq!(res.links.self_link.href, format!("/subscriptions/{}", id));
let list = list_subscriptions_inner(
State(state.clone()),
Query(ListSubscriptionsQuery {
limit: None,
last_id: None,
}),
)
.await
.unwrap();
assert_eq!(list.data.len(), 1);
assert_eq!(
list.data[0].links.self_link.href,
format!("/subscriptions/{}", id)
);
assert_eq!(list.remaining_count, 0);
let get = get_subscription_inner(State(state.clone()), Path(id))
.await
.unwrap();
assert_eq!(get.subscription.id, id);
assert_eq!(get.links.self_link.href, format!("/subscriptions/{}", id));
let update_payload = UpdateSubscription {
target_repo: Some(TargetRepo::new("org/new-target".to_string()).unwrap()),
event_type: None,
gh_app_installation_id: None,
};
let updated =
update_subscription_inner(State(state.clone()), Path(id), Json(update_payload))
.await
.unwrap();
assert_eq!(
updated.subscription.target_repo,
TargetRepo::new("org/new-target".to_string()).unwrap()
);
assert_eq!(
updated.links.self_link.href,
format!("/subscriptions/{}", id)
);
delete_subscription_inner(State(state.clone()), Path(id))
.await
.unwrap();
let get_after_delete = get_subscription_inner(State(state.clone()), Path(id)).await;
assert!(get_after_delete.is_err());
}
#[tokio::test]
async fn test_non_existent_subscription_returns_not_found() {
let pool = create_test_db().await;
let config = crate::test_utils::create_test_config();
let state = AppState {
config: std::sync::Arc::new(config),
repository: std::sync::Arc::new(crate::repository::SqliteRepository::new(pool.clone())),
db_pool: pool.clone(),
};
let get_res = get_subscription_inner(State(state.clone()), Path(999)).await;
assert!(matches!(get_res, Err(HandlerError::NotFound)));
let update_payload = UpdateSubscription {
target_repo: Some(TargetRepo::new("org/new-target".to_string()).unwrap()),
event_type: None,
gh_app_installation_id: None,
};
let update_res =
update_subscription_inner(State(state.clone()), Path(999), Json(update_payload)).await;
assert!(matches!(update_res, Err(HandlerError::NotFound)));
let delete_res = delete_subscription_inner(State(state.clone()), Path(999)).await;
assert!(matches!(delete_res, Err(HandlerError::NotFound)));
}
#[tokio::test]
async fn test_list_subscriptions_pagination() {
let pool = create_test_db().await;
let config = crate::test_utils::create_test_config();
let state = AppState {
config: std::sync::Arc::new(config),
repository: std::sync::Arc::new(crate::repository::SqliteRepository::new(pool.clone())),
db_pool: pool.clone(),
};
#[allow(clippy::excessive_nesting)]
for i in 0..3 {
let payload = CreateSubscription {
source_repo_url: RepoUrl::new(format!("https://github.com/org/repo{}", i)).unwrap(),
source_branch_name: BranchName::new("main".to_string()).unwrap(),
target_repo: TargetRepo::new("org/target".to_string()).unwrap(),
event_type: EventType::new("dispatch".to_string()).unwrap(),
gh_app_installation_id: 1,
};
let _ = create_subscription_inner(State(state.clone()), Json(payload))
.await
.unwrap();
}
let page1 = list_subscriptions_inner(
State(state.clone()),
Query(ListSubscriptionsQuery {
limit: Some(2),
last_id: None,
}),
)
.await
.unwrap();
assert_eq!(page1.data.len(), 2);
assert_eq!(page1.remaining_count, 1);
assert!(page1.links.next.is_some());
let last_id = page1.data.last().unwrap().subscription.id;
let page2 = list_subscriptions_inner(
State(state.clone()),
Query(ListSubscriptionsQuery {
limit: Some(2),
last_id: Some(last_id),
}),
)
.await
.unwrap();
assert_eq!(page2.data.len(), 1);
assert_eq!(page2.remaining_count, 0);
assert!(page2.links.next.is_none());
}
#[tokio::test]
async fn test_cascading_branch_cleanup() {
let pool = create_test_db().await;
let config = crate::test_utils::create_test_config();
let state = AppState {
config: std::sync::Arc::new(config),
repository: std::sync::Arc::new(crate::repository::SqliteRepository::new(pool.clone())),
db_pool: pool.clone(),
};
let payload = CreateSubscription {
source_repo_url: RepoUrl::new("https://github.com/org/repo".to_string()).unwrap(),
source_branch_name: BranchName::new("main".to_string()).unwrap(),
target_repo: TargetRepo::new("org/target".to_string()).unwrap(),
event_type: EventType::new("dispatch".to_string()).unwrap(),
gh_app_installation_id: 1,
};
let sub1 = create_subscription_inner(State(state.clone()), Json(payload.clone()))
.await
.unwrap();
let sub2 = create_subscription_inner(State(state.clone()), Json(payload))
.await
.unwrap();
let branch_id = sub1.subscription.branch_id;
let branch: Option<(i64,)> = sqlx::query_as("SELECT id FROM branches WHERE id = ?")
.bind(branch_id)
.fetch_optional(&pool)
.await
.unwrap();
assert!(branch.is_some());
delete_subscription_inner(State(state.clone()), Path(sub1.subscription.id))
.await
.unwrap();
let branch_still_exists: Option<(i64,)> =
sqlx::query_as("SELECT id FROM branches WHERE id = ?")
.bind(branch_id)
.fetch_optional(&pool)
.await
.unwrap();
assert!(branch_still_exists.is_some());
delete_subscription_inner(State(state.clone()), Path(sub2.subscription.id))
.await
.unwrap();
let branch_gone: Option<(i64,)> = sqlx::query_as("SELECT id FROM branches WHERE id = ?")
.bind(branch_id)
.fetch_optional(&pool)
.await
.unwrap();
assert!(branch_gone.is_none());
}
}