GKaszewski/k-tv
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases Wiki Activity

clean(presentation): strip comments, kill dead code, extract constants

Browse Source 
- remove all doc/inline comments
- delete unused ApiError variants (AuthRequired, NotImplemented) and helpers (internal, not_implemented)
- prefix lifetime-only AppState fields with _ to suppress dead_code warning
- extract magic numbers: TICK_INTERVAL_SECS, EXPIRY_THRESHOLD_HOURS, POLL_INTERVAL_SECS, EVENT_BUS_CAPACITY, DEFAULT_ACTIVITY_LIMIT, DEFAULT_SEARCH_LIMIT, etc
- replace inline serde_json::json! in sync_status with typed SyncStatusEntry
- fix mid-file import in schedule handler
- fix extractors: strip_prefix instead of magic offset 6
- fix unreachable pattern in wire_repositories with cfg guard
- use eq_ignore_ascii_case for content-type header check
This commit is contained in:
Gabriel Kaszewski
2026-07-12 04:35:54 +02:00
parent 25b33b6a0e
commit ff5f299a84
21 changed files with 91 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
crates/presentation/src/background/auto_scheduler.rs
View File

@@ -1,26 +1,21 @@
//! Background auto-scheduler task.
//!
//! Runs every hour, finds channels with `auto_schedule = true`, and regenerates
//! their schedule if it is within 24 hours of expiry.
use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use application::schedule::ScheduleDeps;
use application::schedule::GenerateScheduleCommand;
use application::schedule::{GenerateScheduleCommand, ScheduleDeps};
const TICK_INTERVAL_SECS: u64 = 3600;
const EXPIRY_THRESHOLD_HOURS: i64 = 24;
/// Run the auto-scheduler loop.
pub async fn run(deps: Arc<ScheduleDeps>) {
loop {
tokio::time::sleep(Duration::from_secs(3600)).await;
tokio::time::sleep(Duration::from_secs(TICK_INTERVAL_SECS)).await;
tick(&deps).await;
}
}
async fn tick(deps: &ScheduleDeps) {
// List all channels, find those with auto_schedule
let channels = match deps.channel_query.find_all().await {
Ok(c) => c,
Err(e) => {
@@ -36,7 +31,6 @@ async fn tick(deps: &ScheduleDeps) {
continue;
}
// Check latest schedule
let latest = match deps.schedule_query.find_latest(channel.id()).await {
Ok(s) => s,
Err(e) => {
@@ -52,7 +46,7 @@ async fn tick(deps: &ScheduleDeps) {
let should_generate = match &latest {
Some(s) => {
let remaining = s.valid_until() - now;
remaining < chrono::Duration::hours(24)
remaining < chrono::Duration::hours(EXPIRY_THRESHOLD_HOURS)
}
None => true,
};

11
crates/presentation/src/background/broadcast_poller.rs
View File

@@ -1,8 +1,3 @@
//! BroadcastPoller background task.
//!
//! Polls channels with webhook_url configured and emits domain events
//! when the current slot changes.
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -16,18 +11,18 @@ use domain::value_objects::{ChannelId, SlotId};
use application::schedule::ScheduleDeps;
/// Per-channel poll state.
const POLL_INTERVAL_SECS: u64 = 1;
struct ChannelPollState {
last_slot_id: Option<Uuid>,
last_checked: Instant,
}
/// Polls channels and emits broadcast transition events.
pub async fn run(deps: Arc<ScheduleDeps>, event_publisher: Arc<dyn EventPublisher>) {
let mut state: HashMap<Uuid, ChannelPollState> = HashMap::new();
loop {
tokio::time::sleep(Duration::from_secs(1)).await;
tokio::time::sleep(Duration::from_secs(POLL_INTERVAL_SECS)).await;
tick(&deps, &event_publisher, &mut state).await;
}
}

13
crates/presentation/src/background/library_sync.rs
View File

@@ -1,8 +1,3 @@
//! Background library sync task.
//!
//! Fires 10 seconds after startup, then every N hours (read from app_settings).
//! Can be triggered on-demand via the sync_trigger watch channel.
use std::sync::Arc;
use std::time::Duration;
@@ -11,6 +6,7 @@ use tokio::sync::watch;
const STARTUP_DELAY_SECS: u64 = 10;
const DEFAULT_INTERVAL_HOURS: u64 = 6;
const SYNC_INTERVAL_SETTING_KEY: &str = "library_sync_interval_hours";
pub async fn run(
sync_adapter: Arc<dyn LibrarySyncAdapter>,
@@ -36,7 +32,7 @@ pub async fn run(
}
async fn load_interval_hours(repo: &Arc<dyn AppSettingsRepository>) -> u64 {
repo.get("library_sync_interval_hours")
repo.get(SYNC_INTERVAL_SETTING_KEY)
.await
.ok()
.flatten()
@@ -51,9 +47,6 @@ async fn do_sync(
let provider_ids = registry.provider_ids();
for provider_id in provider_ids {
// We need a &dyn IMediaProvider, but IProviderRegistry doesn't expose one.
// The sync adapter will use the registry's fetch_items internally via its
// own stored reference to the provider. For now, we create a thin adapter.
tracing::info!("library-sync: syncing provider '{}'", provider_id);
let wrapper = RegistryProviderAdapter {
@@ -76,8 +69,6 @@ async fn do_sync(
}
}
/// Adapter that wraps IProviderRegistry calls for a specific provider_id,
/// implementing IMediaProvider so it can be passed to LibrarySyncAdapter.
struct RegistryProviderAdapter {
registry: Arc<dyn IProviderRegistry>,
provider_id: String,

2
crates/presentation/src/background/mod.rs
View File

@@ -1,5 +1,3 @@
//! Background tasks spawned at server startup.
pub mod auto_scheduler;
pub mod broadcast_poller;
pub mod library_sync;

11
crates/presentation/src/background/webhook_consumer.rs
View File

@@ -1,7 +1,3 @@
//! WebhookConsumer background task.
//!
//! Subscribes to domain events and delivers them to per-channel webhook URLs.
use std::sync::Arc;
use chrono::Utc;
@@ -13,7 +9,8 @@ use uuid::Uuid;
use domain::events::DomainEvent;
use domain::ports::ChannelQuery;
/// Consumes domain events and delivers them to per-channel webhook URLs.
const DEFAULT_CONTENT_TYPE: &str = "application/json";
pub async fn run(
mut rx: broadcast::Receiver<DomainEvent>,
channel_query: Arc<dyn ChannelQuery>,
@@ -177,7 +174,7 @@ async fn post_webhook(
if let Some(h) = headers_json {
if let Ok(map) = serde_json::from_str::<serde_json::Map<String, Value>>(h) {
for (k, v) in &map {
if k.to_lowercase() == "content-type" {
if k.eq_ignore_ascii_case("content-type") {
has_content_type = true;
}
if let Some(v_str) = v.as_str() {
@@ -188,7 +185,7 @@ async fn post_webhook(
}
if !has_content_type {
req = req.header("Content-Type", "application/json");
req = req.header("Content-Type", DEFAULT_CONTENT_TYPE);
}
match req.send().await {

34
crates/presentation/src/errors.rs
View File

@@ -1,5 +1,3 @@
//! API error handling — maps domain errors to HTTP responses.
use axum::{
Json,
http::StatusCode,
@@ -10,7 +8,6 @@ use thiserror::Error;
use domain::DomainError;
/// API-level errors.
#[derive(Debug, Error)]
pub enum ApiError {
#[error("{0}")]
@@ -28,20 +25,13 @@ pub enum ApiError {
#[error("Unauthorized: {0}")]
Unauthorized(String),
#[error("auth_required")]
AuthRequired,
#[error("Not found: {0}")]
NotFound(String),
#[error("Not implemented: {0}")]
NotImplemented(String),
#[error("Conflict: {0}")]
Conflict(String),
}
/// Error response body.
#[derive(Debug, Serialize)]
pub struct ErrorResponse {
pub error: String,
@@ -118,14 +108,6 @@ impl IntoResponse for ApiError {
},
),
ApiError::AuthRequired => (
StatusCode::UNAUTHORIZED,
ErrorResponse {
error: "auth_required".to_string(),
details: None,
},
),
ApiError::NotFound(msg) => (
StatusCode::NOT_FOUND,
ErrorResponse {
@@ -134,14 +116,6 @@ impl IntoResponse for ApiError {
},
),
ApiError::NotImplemented(msg) => (
StatusCode::NOT_IMPLEMENTED,
ErrorResponse {
error: "Not implemented".to_string(),
details: Some(msg.clone()),
},
),
ApiError::Conflict(msg) => (
StatusCode::CONFLICT,
ErrorResponse {
@@ -160,10 +134,6 @@ impl ApiError {
Self::Validation(msg.into())
}
pub fn internal(msg: impl Into<String>) -> Self {
Self::Internal(msg.into())
}
pub fn not_found(msg: impl Into<String>) -> Self {
Self::NotFound(msg.into())
}
@@ -171,8 +141,4 @@ impl ApiError {
pub fn conflict(msg: impl Into<String>) -> Self {
Self::Conflict(msg.into())
}
pub fn not_implemented(msg: impl Into<String>) -> Self {
Self::NotImplemented(msg.into())
}
}

15
crates/presentation/src/extractors.rs
View File

@@ -1,7 +1,3 @@
//! Auth extractors for API handlers.
//!
//! Provides `CurrentUser`, `OptionalCurrentUser`, and `AdminUser` extractors.
use axum::extract::FromRequestParts;
use axum::http::request::Parts;
use domain::User;
@@ -9,7 +5,6 @@ use domain::User;
use crate::errors::ApiError;
use crate::state::AppState;
/// Extracted current user from JWT Bearer token.
pub struct CurrentUser(pub User);
impl FromRequestParts<AppState> for CurrentUser {
@@ -37,9 +32,6 @@ impl FromRequestParts<AppState> for CurrentUser {
}
}
/// Optional current user — returns None instead of error when auth missing.
///
/// Checks `Authorization: Bearer <token>` first; falls back to `?token=<jwt>`.
pub struct OptionalCurrentUser(pub Option<User>);
impl FromRequestParts<AppState> for OptionalCurrentUser {
@@ -56,8 +48,8 @@ impl FromRequestParts<AppState> for OptionalCurrentUser {
}
let query_token = parts.uri.query().and_then(|q| {
q.split('&')
.find(|seg| seg.starts_with("token="))
.map(|seg| seg[6..].to_owned())
.find_map(|seg| seg.strip_prefix("token="))
.map(|v| v.to_owned())
});
if let Some(token) = query_token {
let user = validate_jwt_token(&token, state).await.ok();
@@ -74,7 +66,6 @@ impl FromRequestParts<AppState> for OptionalCurrentUser {
}
}
/// Extracted admin user — returns 403 if user is not an admin.
pub struct AdminUser(pub User);
impl FromRequestParts<AppState> for AdminUser {
@@ -92,7 +83,6 @@ impl FromRequestParts<AppState> for AdminUser {
}
}
/// Authenticate via JWT Bearer token from `Authorization` header.
#[cfg(feature = "auth-jwt")]
async fn try_jwt_auth(parts: &mut Parts, state: &AppState) -> Result<User, ApiError> {
use axum::http::header::AUTHORIZATION;
@@ -113,7 +103,6 @@ async fn try_jwt_auth(parts: &mut Parts, state: &AppState) -> Result<User, ApiEr
validate_jwt_token(token, state).await
}
/// Validate a raw JWT string and return the corresponding `User`.
#[cfg(feature = "auth-jwt")]
pub(crate) async fn validate_jwt_token(token: &str, state: &AppState) -> Result<User, ApiError> {
let validator = state

62
crates/presentation/src/factory.rs
View File

@@ -1,8 +1,3 @@
//! Factory — builds AppState from Config + DbPool.
//!
//! Connects to the database, runs migrations, creates all adapter instances,
//! constructs Deps structs, and returns a fully-wired AppState.
use std::sync::Arc;
use application::{
@@ -21,31 +16,26 @@ use infra_wiring::{Config, ConfigSource, DbPool};
use crate::state::AppState;
/// Build a fully-wired AppState ready for the HTTP server.
const EVENT_BUS_CAPACITY: usize = 64;
const DEV_JWT_SECRET: &str = "k-template-dev-secret-not-for-production-use-only";
pub async fn build_app_state(config: Config) -> anyhow::Result<AppState> {
// Connect to database
let pool = DbPool::connect(&config.database_url).await?;
pool.run_migrations().await?;
// Wire up all repositories from the database pool
let wire_output = wire_repositories(&pool)?;
// Auth service
let auth_service: Arc<dyn domain::ports::AuthService> =
Arc::new(adapter_auth::PasswordAuthService);
// Event bus
let event_bus = Arc::new(adapter_event_publisher::ChannelEventBus::new(64));
let event_bus = Arc::new(adapter_event_publisher::ChannelEventBus::new(EVENT_BUS_CAPACITY));
let event_publisher: Arc<dyn domain::ports::EventPublisher> = event_bus.clone();
// Provider registry
let provider_registry = build_provider_registry(&config).await;
// Library sync adapter — uses the LibraryCommand port internally
let library_sync: Arc<dyn domain::ports::LibrarySyncAdapter> =
build_library_sync(wire_output.library_command.clone());
// Schedule engine
let schedule_engine = Arc::new(ScheduleEngineService::new(
provider_registry.clone(),
wire_output.channel_query.clone(),
@@ -53,14 +43,11 @@ pub async fn build_app_state(config: Config) -> anyhow::Result<AppState> {
wire_output.schedule_command.clone(),
));
// JWT validator
#[cfg(feature = "auth-jwt")]
let jwt_validator = build_jwt_validator(&config)?;
// Sync trigger channel
let (sync_tx, sync_rx) = tokio::sync::watch::channel(());
// Build all deps structs
let auth_deps = Arc::new(AuthDeps {
user_command: wire_output.user_command.clone(),
user_query: wire_output.user_query.clone(),
@@ -120,7 +107,6 @@ pub async fn build_app_state(config: Config) -> anyhow::Result<AppState> {
let config_arc = Arc::new(config);
// Spawn background tasks
let bg_schedule_deps = schedule_deps.clone();
tokio::spawn(crate::background::auto_scheduler::run(bg_schedule_deps));
@@ -163,19 +149,14 @@ pub async fn build_app_state(config: Config) -> anyhow::Result<AppState> {
#[cfg(feature = "auth-jwt")]
jwt_validator,
provider_registry,
library_sync,
settings_repo: wire_output.settings,
event_bus,
_library_sync: library_sync,
_settings_repo: wire_output.settings,
_event_bus: event_bus,
config: config_arc,
sync_trigger: sync_tx,
})
}
// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------
/// Repository wiring output — trait objects ready for dependency injection.
struct WireOutput {
user_command: Arc<dyn domain::ports::UserCommand>,
user_query: Arc<dyn domain::ports::UserQuery>,
@@ -229,12 +210,12 @@ fn wire_repositories(pool: &DbPool) -> anyhow::Result<WireOutput> {
provider_config_query: w.provider_config_query,
})
}
#[cfg(not(any(feature = "sqlite", feature = "postgres")))]
_ => anyhow::bail!("database backend not compiled into this binary"),
}
}
async fn build_provider_registry(config: &Config) -> Arc<dyn IProviderRegistry> {
// Build a concrete registry that routes to configured providers.
let mut providers: Vec<(String, Arc<dyn IMediaProvider>)> = Vec::new();
match config.config_source {
@@ -259,8 +240,6 @@ async fn build_provider_registry(config: &Config) -> Arc<dyn IProviderRegistry>
}
}
ConfigSource::Db => {
// DB-based provider configs loaded elsewhere at runtime.
// For now, fall through to noop if nothing configured via env.
tracing::info!("CONFIG_SOURCE=db: provider configs loaded from database at runtime");
}
}
@@ -288,7 +267,7 @@ fn build_jwt_validator(config: &Config) -> anyhow::Result<Option<Arc<adapter_aut
anyhow::bail!("JWT_SECRET is required in production");
}
tracing::warn!("JWT_SECRET not set — using insecure development secret");
"k-template-dev-secret-not-for-production-use-only".to_string()
DEV_JWT_SECRET.to_string()
}
};
@@ -305,10 +284,6 @@ fn build_jwt_validator(config: &Config) -> anyhow::Result<Option<Arc<adapter_aut
Ok(Some(Arc::new(adapter_auth::JwtValidator::new(jwt_config))))
}
// ---------------------------------------------------------------------------
// NoopMediaProvider — fallback when nothing is configured
// ---------------------------------------------------------------------------
struct NoopMediaProvider;
#[async_trait::async_trait]
@@ -356,10 +331,6 @@ impl IMediaProvider for NoopMediaProvider {
}
}
// ---------------------------------------------------------------------------
// SimpleProviderRegistry — implements IProviderRegistry for N providers
// ---------------------------------------------------------------------------
struct SimpleProviderRegistry {
providers: Vec<(String, Arc<dyn IMediaProvider>)>,
}
@@ -377,7 +348,6 @@ impl SimpleProviderRegistry {
self.providers.first().map(|(_, v)| v)
}
/// Extract provider_id from a prefixed item ID (e.g. "jellyfin::abc123" → "jellyfin").
fn extract_provider_id(item_id: &str) -> Option<&str> {
item_id.find("::").map(|pos| &item_id[..pos])
}
@@ -411,7 +381,6 @@ impl IProviderRegistry for SimpleProviderRegistry {
return provider.fetch_by_id(item_id).await;
}
}
// Fall back to primary
if let Some(provider) = self.primary() {
provider.fetch_by_id(item_id).await
} else {
@@ -502,11 +471,6 @@ impl IProviderRegistry for SimpleProviderRegistry {
}
}
// ---------------------------------------------------------------------------
// SimpleSyncAdapter — wraps LibraryCommand for sync operations
// ---------------------------------------------------------------------------
/// Convert a MediaItem from a provider into a LibraryItem for persistence.
fn media_item_to_library_item(item: domain::MediaItem, provider_id: &str) -> domain::LibraryItem {
let external_id = item.id().value().to_string();
let id = format!("{}::{}", provider_id, external_id);
@@ -526,8 +490,8 @@ fn media_item_to_library_item(item: domain::MediaItem, provider_id: &str) -> dom
item.genres().to_vec(),
item.tags().to_vec(),
item.collection_id().map(|s| s.to_string()),
None, // collection_name not in MediaItem
None, // collection_type not in MediaItem
None,
None,
item.thumbnail_url().map(|s| s.to_string()),
now,
)
@@ -564,7 +528,6 @@ impl domain::ports::LibrarySyncAdapter for SimpleSyncAdapter {
}
};
// Fetch all items from provider
let filter = domain::MediaFilter::default();
let items = match provider.fetch_items(&filter).await {
Ok(items) => items,
@@ -581,8 +544,6 @@ impl domain::ports::LibrarySyncAdapter for SimpleSyncAdapter {
let items_found = items.len() as u32;
// Clear + insert (items are MediaItem; LibrarySyncAdapter implementations
// typically handle the conversion. Here we delegate to library_command directly.)
if let Err(e) = self.library_command.clear_provider(provider_id).await {
let result = domain::LibrarySyncResult::with_error(
provider_id,
@@ -593,7 +554,6 @@ impl domain::ports::LibrarySyncAdapter for SimpleSyncAdapter {
return result;
}
// Convert MediaItems to LibraryItems for storage
let library_items: Vec<domain::LibraryItem> = items
.into_iter()
.map(|item| media_item_to_library_item(item, provider_id))

10
crates/presentation/src/handlers/admin.rs
View File

@@ -1,5 +1,3 @@
//! Admin handlers.
use axum::Json;
use axum::extract::{Query, State};
use serde::Deserialize;
@@ -12,7 +10,8 @@ use crate::errors::ApiError;
use crate::extractors::AdminUser;
use crate::state::AppState;
/// GET /admin/settings
const DEFAULT_ACTIVITY_LIMIT: u32 = 50;
pub async fn get_settings(
State(state): State<AppState>,
AdminUser(_user): AdminUser,
@@ -23,7 +22,6 @@ pub async fn get_settings(
Ok(Json(SettingsResponse { settings }))
}
/// PUT /admin/settings
pub async fn update_settings(
State(state): State<AppState>,
AdminUser(_user): AdminUser,
@@ -35,7 +33,6 @@ pub async fn update_settings(
};
application::admin::update_settings::execute(&state.admin_deps, cmd).await?;
// Re-read after update
let pairs =
application::admin::get_settings::execute(&state.admin_deps, GetSettingsQuery).await?;
let settings: HashMap<String, String> = pairs.into_iter().collect();
@@ -47,14 +44,13 @@ pub struct ActivityLogParams {
pub limit: Option<u32>,
}
/// GET /admin/activity
pub async fn get_activity_log(
State(state): State<AppState>,
AdminUser(_user): AdminUser,
Query(params): Query<ActivityLogParams>,
) -> Result<Json<Vec<ActivityEventResponse>>, ApiError> {
let query = GetActivityLogQuery {
limit: params.limit.unwrap_or(50),
limit: params.limit.unwrap_or(DEFAULT_ACTIVITY_LIMIT),
};
let events = application::admin::activity_log::execute(&state.admin_deps, query).await?;
Ok(Json(

23
crates/presentation/src/handlers/auth.rs
View File

@@ -1,5 +1,3 @@
//! Authentication handlers.
use axum::Json;
use axum::extract::State;
@@ -10,7 +8,9 @@ use crate::errors::ApiError;
use crate::extractors::CurrentUser;
use crate::state::AppState;
/// POST /auth/register
const TOKEN_TYPE_BEARER: &str = "Bearer";
const SECS_PER_HOUR: u64 = 3600;
pub async fn register(
State(state): State<AppState>,
Json(req): Json<RegisterRequest>,
@@ -23,7 +23,6 @@ pub async fn register(
Ok(Json(UserResponse::from(user)))
}
/// POST /auth/login
pub async fn login(
State(state): State<AppState>,
Json(req): Json<LoginRequest>,
@@ -36,25 +35,22 @@ pub async fn login(
let (access_token, refresh_token) = create_tokens(&user, &state, req.remember_me)?;
Ok(Json(TokenResponse {
access_token,
token_type: "Bearer".to_string(),
expires_in: state.config.jwt_expiry_hours * 3600,
token_type: TOKEN_TYPE_BEARER.to_string(),
expires_in: state.config.jwt_expiry_hours * SECS_PER_HOUR,
refresh_token,
}))
}
/// POST /auth/logout — no-op for JWT (stateless)
pub async fn logout() -> Result<Json<serde_json::Value>, ApiError> {
Ok(Json(serde_json::json!({"message": "logged out"})))
}
/// GET /auth/me
pub async fn me(
CurrentUser(user): CurrentUser,
) -> Result<Json<UserResponse>, ApiError> {
Ok(Json(UserResponse::from(user)))
}
/// POST /auth/token — exchange credentials for tokens
#[cfg(feature = "auth-jwt")]
pub async fn get_token(
State(state): State<AppState>,
@@ -68,13 +64,12 @@ pub async fn get_token(
let (access_token, refresh_token) = create_tokens(&user, &state, req.remember_me)?;
Ok(Json(TokenResponse {
access_token,
token_type: "Bearer".to_string(),
expires_in: state.config.jwt_expiry_hours * 3600,
token_type: TOKEN_TYPE_BEARER.to_string(),
expires_in: state.config.jwt_expiry_hours * SECS_PER_HOUR,
refresh_token,
}))
}
/// POST /auth/refresh — refresh an access token
#[cfg(feature = "auth-jwt")]
pub async fn refresh_token(
State(state): State<AppState>,
@@ -106,8 +101,8 @@ pub async fn refresh_token(
let (access_token, refresh_token) = create_tokens(&user, &state, true)?;
Ok(Json(TokenResponse {
access_token,
token_type: "Bearer".to_string(),
expires_in: state.config.jwt_expiry_hours * 3600,
token_type: TOKEN_TYPE_BEARER.to_string(),
expires_in: state.config.jwt_expiry_hours * SECS_PER_HOUR,
refresh_token,
}))
}

15
crates/presentation/src/handlers/channels.rs
View File

@@ -1,5 +1,3 @@
//! Channel CRUD handlers.
use axum::Json;
use axum::extract::{Path, State};
@@ -20,7 +18,6 @@ use crate::errors::ApiError;
use crate::extractors::CurrentUser;
use crate::state::AppState;
/// GET /channels
pub async fn list_channels(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -30,7 +27,6 @@ pub async fn list_channels(
Ok(Json(channels.into_iter().map(ChannelResponse::from).collect()))
}
/// GET /channels/mine
pub async fn list_my_channels(
State(state): State<AppState>,
CurrentUser(user): CurrentUser,
@@ -43,7 +39,6 @@ pub async fn list_my_channels(
Ok(Json(channels.into_iter().map(ChannelResponse::from).collect()))
}
/// POST /channels
pub async fn create_channel(
State(state): State<AppState>,
CurrentUser(user): CurrentUser,
@@ -58,7 +53,6 @@ pub async fn create_channel(
Ok(Json(ChannelResponse::from(channel)))
}
/// GET /channels/:id
pub async fn get_channel(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -71,7 +65,6 @@ pub async fn get_channel(
Ok(Json(ChannelResponse::from(channel)))
}
/// PUT /channels/:id
pub async fn update_channel(
State(state): State<AppState>,
CurrentUser(user): CurrentUser,
@@ -108,7 +101,6 @@ pub async fn update_channel(
Ok(Json(ChannelResponse::from(channel)))
}
/// DELETE /channels/:id
pub async fn delete_channel(
State(state): State<AppState>,
CurrentUser(user): CurrentUser,
@@ -122,9 +114,6 @@ pub async fn delete_channel(
Ok(axum::http::StatusCode::NO_CONTENT)
}
// ── Config snapshots ─────────────────────────────────────────────────────
/// POST /channels/:id/snapshots
pub async fn save_snapshot(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -138,7 +127,6 @@ pub async fn save_snapshot(
Ok(Json(ConfigSnapshotResponse::from(snap)))
}
/// GET /channels/:id/snapshots
pub async fn list_snapshots(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -149,7 +137,6 @@ pub async fn list_snapshots(
Ok(Json(snaps.into_iter().map(ConfigSnapshotResponse::from).collect()))
}
/// GET /channels/:id/snapshots/:snapshot_id
pub async fn get_snapshot(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -165,7 +152,6 @@ pub async fn get_snapshot(
Ok(Json(ConfigSnapshotResponse::from(snap)))
}
/// PATCH /channels/:id/snapshots/:snapshot_id
pub async fn patch_snapshot(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -184,7 +170,6 @@ pub async fn patch_snapshot(
Ok(Json(ConfigSnapshotResponse::from(snap)))
}
/// POST /channels/:id/snapshots/:snapshot_id/restore
pub async fn restore_snapshot(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,

7
crates/presentation/src/handlers/config.rs
View File

@@ -1,5 +1,3 @@
//! System configuration handler.
use axum::Json;
use axum::extract::State;
@@ -8,7 +6,8 @@ use api_types::{ConfigResponse, ProviderCapabilitiesResponse, ProviderInfo};
use crate::errors::ApiError;
use crate::state::AppState;
/// GET /config — public system configuration
const FALLBACK_STREAMING_PROTOCOL: &str = "direct_file";
pub async fn get_config(
State(state): State<AppState>,
) -> Result<Json<ConfigResponse>, ApiError> {
@@ -36,7 +35,7 @@ pub async fn get_config(
tags: false,
decade: false,
search: false,
streaming_protocol: "direct_file".to_string(),
streaming_protocol: FALLBACK_STREAMING_PROTOCOL.to_string(),
rescan: false,
transcode: false,
});

17
crates/presentation/src/handlers/files.rs
View File

@@ -1,9 +1,3 @@
//! Local file streaming handlers (feature-gated).
//!
//! Placeholder — the actual streaming logic requires the local-files adapter
//! which provides file index and transcoding. This will be fleshed out once
//! the local-files adapter integration is complete.
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::IntoResponse;
@@ -11,22 +5,19 @@ use axum::response::IntoResponse;
use crate::errors::ApiError;
use crate::state::AppState;
/// GET /files/stream/:id — stream a local file
pub async fn stream_file(
State(_state): State<AppState>,
Path(_id): Path<String>,
) -> Result<impl IntoResponse, ApiError> {
// TODO: integrate with adapter-local-files for actual streaming
Err::<StatusCode, _>(ApiError::not_implemented(
"Local file streaming not yet wired in presentation crate",
Err::<StatusCode, _>(ApiError::NotFound(
"Local file streaming not yet wired in presentation crate".to_string(),
))
}
/// POST /files/rescan — rescan local files
pub async fn rescan(
State(_state): State<AppState>,
) -> Result<impl IntoResponse, ApiError> {
Err::<StatusCode, _>(ApiError::not_implemented(
"Local file rescan not yet wired in presentation crate",
Err::<StatusCode, _>(ApiError::NotFound(
"Local file rescan not yet wired in presentation crate".to_string(),
))
}

17
crates/presentation/src/handlers/iptv.rs
View File

@@ -1,5 +1,3 @@
//! IPTV export handlers (M3U, XMLTV).
use axum::extract::{Query, State};
use axum::http::header;
use axum::response::IntoResponse;
@@ -11,12 +9,14 @@ use crate::errors::ApiError;
use crate::extractors::OptionalCurrentUser;
use crate::state::AppState;
const M3U_CONTENT_TYPE: &str = "audio/x-mpegurl; charset=utf-8";
const XML_CONTENT_TYPE: &str = "application/xml; charset=utf-8";
#[derive(Debug, Deserialize)]
pub struct IptvParams {
pub token: Option<String>,
}
/// GET /iptv/playlist.m3u — M3U playlist
pub async fn m3u_playlist(
State(state): State<AppState>,
OptionalCurrentUser(_user): OptionalCurrentUser,
@@ -27,20 +27,13 @@ pub async fn m3u_playlist(
token: params.token,
};
let content = application::iptv::m3u::execute(&state.iptv_deps, query).await?;
Ok((
[(header::CONTENT_TYPE, "audio/x-mpegurl; charset=utf-8")],
content,
))
Ok(([(header::CONTENT_TYPE, M3U_CONTENT_TYPE)], content))
}
/// GET /iptv/epg.xml — XMLTV electronic program guide
pub async fn xmltv_epg(
State(state): State<AppState>,
OptionalCurrentUser(_user): OptionalCurrentUser,
) -> Result<impl IntoResponse, ApiError> {
let content = application::iptv::xmltv::execute(&state.iptv_deps, GetXmltvQuery).await?;
Ok((
[(header::CONTENT_TYPE, "application/xml; charset=utf-8")],
content,
))
Ok(([(header::CONTENT_TYPE, XML_CONTENT_TYPE)], content))
}

53
crates/presentation/src/handlers/library.rs
View File

@@ -1,8 +1,6 @@
//! Library browsing handlers.
use axum::Json;
use axum::extract::{Path, Query, State};
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use api_types::{CollectionResponse, LibraryItemResponse, PaginatedResponse, SeasonResponse, ShowResponse};
use application::library::{
@@ -14,6 +12,8 @@ use crate::errors::ApiError;
use crate::extractors::{AdminUser, CurrentUser};
use crate::state::AppState;
const DEFAULT_SEARCH_LIMIT: u32 = 50;
#[derive(Debug, Deserialize)]
pub struct LibrarySearchParams {
pub provider: Option<String>,
@@ -30,7 +30,6 @@ pub struct LibrarySearchParams {
pub limit: Option<u32>,
}
/// GET /library/items
pub async fn search_items(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -46,7 +45,7 @@ pub async fn search_items(
season_number: params.season_number,
decade: params.decade,
offset: params.offset.unwrap_or(0),
limit: params.limit.unwrap_or(50),
limit: params.limit.unwrap_or(DEFAULT_SEARCH_LIMIT),
};
let (items, total) = application::library::search::execute(&state.library_query_deps, query).await?;
Ok(Json(PaginatedResponse::new(
@@ -55,7 +54,6 @@ pub async fn search_items(
)))
}
/// GET /library/items/:id
pub async fn get_item(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -68,7 +66,6 @@ pub async fn get_item(
Ok(Json(LibraryItemResponse::from(item)))
}
/// GET /library/collections
pub async fn list_collections(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -100,7 +97,6 @@ pub struct ShowsParams {
pub genres: Vec<String>,
}
/// GET /library/shows
pub async fn list_shows(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -121,7 +117,6 @@ pub struct SeasonsParams {
pub provider: Option<String>,
}
/// GET /library/seasons
pub async fn list_seasons(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -144,7 +139,6 @@ pub struct GenresParams {
pub provider: Option<String>,
}
/// GET /library/genres
pub async fn list_genres(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -159,40 +153,42 @@ pub async fn list_genres(
Ok(Json(genres))
}
/// GET /library/sync/status
#[derive(Debug, Serialize)]
pub(crate) struct SyncStatusEntry {
provider_id: String,
started_at: String,
finished_at: String,
items_found: u32,
status: String,
error_msg: String,
}
pub async fn sync_status(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
) -> Result<Json<serde_json::Value>, ApiError> {
) -> Result<Json<Vec<SyncStatusEntry>>, ApiError> {
let entries =
application::library::get_sync_status::execute(&state.library_query_deps, GetSyncStatusQuery)
.await?;
let result: Vec<serde_json::Value> = entries
let result: Vec<SyncStatusEntry> = entries
.into_iter()
.map(|e| {
serde_json::json!({
"provider_id": e.provider_id(),
"started_at": e.started_at(),
"finished_at": e.finished_at().unwrap_or(""),
"items_found": e.items_found(),
"status": e.status(),
"error_msg": e.error_msg().unwrap_or(""),
})
.map(|e| SyncStatusEntry {
provider_id: e.provider_id().to_string(),
started_at: e.started_at().to_string(),
finished_at: e.finished_at().unwrap_or("").to_string(),
items_found: e.items_found(),
status: e.status().to_string(),
error_msg: e.error_msg().unwrap_or("").to_string(),
})
.collect();
Ok(Json(serde_json::Value::Array(result)))
Ok(Json(result))
}
/// POST /library/sync — trigger sync (admin only)
///
/// Validates that no sync is already running, then sends a signal to the
/// background sync task to start a sync cycle immediately.
pub async fn trigger_sync(
State(state): State<AppState>,
AdminUser(_user): AdminUser,
) -> Result<axum::http::StatusCode, ApiError> {
let cmd = TriggerSyncCommand { provider_id: None };
let _provider_ids =
application::library::sync::execute(&state.library_command_deps, cmd)
.await
.map_err(|e| {
@@ -203,7 +199,6 @@ pub async fn trigger_sync(
}
})?;
// Signal the background sync task to run immediately
let _ = state.sync_trigger.send(());
Ok(axum::http::StatusCode::ACCEPTED)

6
crates/presentation/src/handlers/providers.rs
View File

@@ -1,5 +1,3 @@
//! Provider configuration CRUD handlers.
use axum::Json;
use axum::extract::{Path, State};
@@ -12,7 +10,6 @@ use crate::errors::ApiError;
use crate::extractors::AdminUser;
use crate::state::AppState;
/// GET /admin/providers
pub async fn list_providers(
State(state): State<AppState>,
AdminUser(_user): AdminUser,
@@ -27,7 +24,6 @@ pub async fn list_providers(
))
}
/// GET /admin/providers/:id
pub async fn get_provider(
State(state): State<AppState>,
AdminUser(_user): AdminUser,
@@ -40,7 +36,6 @@ pub async fn get_provider(
Ok(Json(ProviderConfigResponse::from(provider)))
}
/// PUT /admin/providers/:id
pub async fn upsert_provider(
State(state): State<AppState>,
AdminUser(_user): AdminUser,
@@ -60,7 +55,6 @@ pub async fn upsert_provider(
Ok(Json(serde_json::json!({"status": "ok"})))
}
/// DELETE /admin/providers/:id
pub async fn delete_provider(
State(state): State<AppState>,
AdminUser(_user): AdminUser,

13
crates/presentation/src/handlers/schedule.rs
View File

@@ -1,8 +1,7 @@
//! Schedule, broadcast, and stream handlers.
use axum::Json;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::IntoResponse;
use api_types::{
CurrentBroadcastResponse, ScheduleHistoryEntry, ScheduleResponse, SlotResponse,
@@ -16,7 +15,6 @@ use crate::errors::ApiError;
use crate::extractors::CurrentUser;
use crate::state::AppState;
/// POST /channels/:id/schedule — generate a new schedule
pub async fn generate_schedule(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -27,7 +25,6 @@ pub async fn generate_schedule(
Ok(Json(ScheduleResponse::from(schedule)))
}
/// GET /channels/:id/schedule — get the active schedule
pub async fn get_active_schedule(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,
@@ -40,9 +37,6 @@ pub async fn get_active_schedule(
}
}
use axum::response::IntoResponse;
/// GET /channels/:id/now — what's currently playing
pub async fn get_current_broadcast(
State(state): State<AppState>,
Path(id): Path<uuid::Uuid>,
@@ -51,7 +45,6 @@ pub async fn get_current_broadcast(
match application::schedule::get_current_broadcast::execute(&state.schedule_deps, query).await?
{
Some(broadcast) => {
// Look up the channel to resolve block access mode
let channel_query = application::channels::GetChannelQuery { channel_id: id };
let channel = application::channels::get::execute(&state.channel_query_deps, channel_query).await?;
@@ -73,7 +66,6 @@ pub async fn get_current_broadcast(
}
}
/// GET /channels/:id/epg — electronic program guide
pub async fn get_epg(
State(state): State<AppState>,
Path(id): Path<uuid::Uuid>,
@@ -83,12 +75,10 @@ pub async fn get_epg(
Ok(Json(slots.into_iter().map(SlotResponse::from).collect()))
}
/// GET /channels/:id/stream — redirect to stream URL (307)
pub async fn get_stream(
State(state): State<AppState>,
Path(id): Path<uuid::Uuid>,
) -> Result<axum::response::Response, ApiError> {
// Find the current broadcast first to get the item ID
let broadcast_query = GetCurrentBroadcastQuery { channel_id: id };
let broadcast =
application::schedule::get_current_broadcast::execute(&state.schedule_deps, broadcast_query)
@@ -113,7 +103,6 @@ pub async fn get_stream(
}
}
/// GET /channels/:id/schedule/history — list schedule generations
pub async fn list_schedule_history(
State(state): State<AppState>,
CurrentUser(_user): CurrentUser,

9
crates/presentation/src/main.rs
View File

@@ -1,5 +1,3 @@
//! k-tv server entry point.
use std::net::SocketAddr;
use tower_http::cors::{Any, CorsLayer};
@@ -17,10 +15,8 @@ mod state;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
// Load .env file if present
let _ = dotenvy::dotenv();
// Initialize tracing
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
@@ -28,7 +24,6 @@ async fn main() -> anyhow::Result<()> {
)
.init();
// Load config
let config = infra_wiring::Config::from_env()
.map_err(|e| anyhow::anyhow!("Config error: {}", e))?;
@@ -38,10 +33,8 @@ async fn main() -> anyhow::Result<()> {
info!("Starting k-tv server on {}:{}", host, port);
// Build the application state (connects DB, creates adapters, spawns background tasks)
let app_state = factory::build_app_state(config).await?;
// Build CORS layer
let cors = if cors_origins.iter().any(|o| o == "*") {
CorsLayer::new()
.allow_origin(Any)
@@ -58,14 +51,12 @@ async fn main() -> anyhow::Result<()> {
.allow_headers(Any)
};
// Build the router
let app = axum::Router::new()
.nest("/api/v1", routes::api_v1_router())
.layer(cors)
.layer(TraceLayer::new_for_http())
.with_state(app_state);
// Start serving
let addr: SocketAddr = format!("{}:{}", host, port).parse()?;
let listener = tokio::net::TcpListener::bind(addr).await?;
info!("Listening on {}", addr);

7
crates/presentation/src/mappers/mod.rs
View File

@@ -1,6 +1 @@
//! Domain → DTO mappings.
//!
//! Most conversions are already handled by `From` impls in the `api-types` crate.
//! This module is reserved for any presentation-layer-specific mappings that
//! don't belong in `api-types` (e.g., combining multiple domain objects into a
//! single response).

6
crates/presentation/src/routes.rs
View File

@@ -1,11 +1,8 @@
//! Router construction.
use axum::{Router, routing::{delete, get, post, put}};
use crate::handlers;
use crate::state::AppState;
/// Construct the API v1 router.
pub fn api_v1_router() -> Router<AppState> {
Router::new()
.nest("/auth", auth_router())
@@ -41,15 +38,12 @@ fn channel_router() -> Router<AppState> {
.route("/{id}", get(handlers::channels::get_channel))
.route("/{id}", put(handlers::channels::update_channel))
.route("/{id}", delete(handlers::channels::delete_channel))
// Schedule
.route("/{id}/schedule", post(handlers::schedule::generate_schedule))
.route("/{id}/schedule", get(handlers::schedule::get_active_schedule))
.route("/{id}/schedule/history", get(handlers::schedule::list_schedule_history))
// Broadcast
.route("/{id}/now", get(handlers::schedule::get_current_broadcast))
.route("/{id}/epg", get(handlers::schedule::get_epg))
.route("/{id}/stream", get(handlers::schedule::get_stream))
// Config snapshots
.route("/{id}/snapshots", post(handlers::channels::save_snapshot))
.route("/{id}/snapshots", get(handlers::channels::list_snapshots))
.route("/{id}/snapshots/{snapshot_id}", get(handlers::channels::get_snapshot))

20
crates/presentation/src/state.rs
View File

@@ -1,5 +1,3 @@
//! Application state — holds pre-built Deps structs from the application layer.
use std::sync::Arc;
use application::{
@@ -13,7 +11,6 @@ use application::{
schedule::ScheduleDeps,
};
/// Shared application state, passed to all handlers via `State<AppState>`.
#[derive(Clone)]
pub struct AppState {
pub auth_deps: Arc<AuthDeps>,
@@ -27,25 +24,16 @@ pub struct AppState {
pub iptv_deps: Arc<IptvDeps>,
pub provider_deps: Arc<ProviderDeps>,
/// JWT validator for token creation/validation in auth handlers.
#[cfg(feature = "auth-jwt")]
pub jwt_validator: Option<Arc<adapter_auth::JwtValidator>>,
/// Provider registry for config/capabilities endpoints.
pub provider_registry: Arc<dyn domain::ports::IProviderRegistry>,
/// Library sync adapter — needed for spawning background sync tasks.
pub library_sync: Arc<dyn domain::ports::LibrarySyncAdapter>,
// kept alive for background tasks — not read from handlers
pub _library_sync: Arc<dyn domain::ports::LibrarySyncAdapter>,
pub _settings_repo: Arc<dyn domain::ports::AppSettingsRepository>,
pub _event_bus: Arc<adapter_event_publisher::ChannelEventBus>,
/// App settings — read by library sync background task.
pub settings_repo: Arc<dyn domain::ports::AppSettingsRepository>,
/// Event bus for domain events.
pub event_bus: Arc<adapter_event_publisher::ChannelEventBus>,
/// Application config.
pub config: Arc<infra_wiring::Config>,
/// Trigger for on-demand library sync (sends () to wake the background task).
pub sync_trigger: tokio::sync::watch::Sender<()>,
}