Compare commits

...

36 Commits

Author SHA1 Message Date
2698cb3ad4 cleanup: narrow SCTE-35 visibility to pub(crate), remove dead target_duration_secs 2026-07-12 15:31:59 +02:00
abd6bf4bc4 fix: add utoipa annotations to iCal export/import handlers 2026-07-12 15:30:45 +02:00
25ff9d2779 fix: rename recycle_policy column to rotation_policy + migration 2026-07-12 15:29:50 +02:00
2995aea606 fix: wire tags filter through LibrarySearchFilter to SQLite adapter 2026-07-12 15:29:06 +02:00
f52e024e98 refactor: deduplicate content_type/role string converters in MCP tools 2026-07-12 15:27:52 +02:00
5561b70e1b refactor: extract build_schedule helper, deduplicate generate/preview/preview_config 2026-07-12 15:26:30 +02:00
8de7d33007 fix: iCal duration parser tracks seconds, rounds up to minutes 2026-07-12 15:24:50 +02:00
8dabbdf280 fix: fill_weighted sorts by recency via playback history 2026-07-12 15:24:17 +02:00
9558f04f73 fix: add gap_filler to frontend ChannelResponse + UpdateChannelRequest 2026-07-12 15:21:20 +02:00
a823a79f6b fix: install curl in presentation image for healthcheck 2026-07-12 15:20:51 +02:00
5bc1e5e44b frontend: align types to backend, playout HLS, rm Jellyfin proxy (#12) 2026-07-12 15:11:50 +02:00
711f0e4411 docker: 3-binary deployment (presentation, worker, playout) 2026-07-12 14:56:38 +02:00
a7fa2ec4aa iCalendar import: parse_ical + API + MCP tool (#16) 2026-07-12 14:46:15 +02:00
6b0d060945 expand MCP tools: browse_library, stats, analyze, preview, suggest, config tools (#17) 2026-07-12 14:34:36 +02:00
6dbf7ab98c SCTE-35 markers + overlay metadata in playout (#10) 2026-07-12 14:33:22 +02:00
e27e2ab6d1 wire utoipa OpenAPI: #[utoipa::path] on all handlers, Scalar UI (#14) 2026-07-12 14:31:20 +02:00
b00272aceb iCalendar export: domain service + API + MCP tool (#15) 2026-07-12 14:27:20 +02:00
affd7f0223 merge feat/playout: HLS streaming, FFmpeg, multi-channel playout service (#9, #11) 2026-07-12 14:14:52 +02:00
8db70df0a0 merge feat/schedule-engine: interstitials, gap filler, mid-roll breaks (#3, #4, #8, #6) 2026-07-12 14:14:43 +02:00
c14e01e0f2 merge feat/library-sync: chapter extraction + MediaRole auto-detection (#5, #7) 2026-07-12 14:14:33 +02:00
874de68fb5 wire mid-roll breaks into schedule engine with chapter-aware splitting 2026-07-12 14:12:59 +02:00
79e975f057 wire gap_filler into schedule engine, API, MCP 2026-07-12 14:08:55 +02:00
8edd7a9c0b wire interstitial insertion into schedule engine resolve_block 2026-07-12 14:05:33 +02:00
8e4f724562 add integration tests for all 6 FillStrategy variants, role filter on LibrarySearchFilter 2026-07-12 14:02:45 +02:00
f21d70c559 MediaRole auto-detection + manual role API + sync wiring (#7)
role_detector: classify items as Interstitial by collection name/tag patterns.
Wire chapter extraction + role detection into sync adapter (worker + presentation).
SQLite: persist/read role column, migration.
PUT /library/items/{id}/role endpoint for manual override.
2026-07-12 14:00:28 +02:00
607d311375 chapter extraction via ffprobe during library sync (#5)
ffprobe module in adapter-common: parse JSON output into Vec<Chapter>,
should_probe_chapters gate (movie OR >45min), graceful failure.
Add set_chapters/set_role setters to MediaItem.
2026-07-12 14:00:17 +02:00
84dd05a8a6 multi-channel playout: concurrent tick, status endpoint, ffmpeg restart (#11) 2026-07-12 13:59:54 +02:00
1b3ecc10e1 add playout service: HLS streaming, FFmpeg, SegmentStore port (#9) 2026-07-12 13:58:09 +02:00
de7f3092d2 add parallel execution plan for roadmap issues 2026-07-12 07:54:01 +02:00
60524c56f7 add domain concepts: FillStrategy variants, Chapter, InterstitialRule, MidRollRule, gap_filler
- FillStrategy: +Alternating (round-robin series), +Weighted (fresh-first), +Marathon (always ep1, loops)
- Chapter value object on MediaItem (JSON column in library_items)
- InterstitialRule + MidRollRule on ProgrammingBlock (data model only)
- gap_filler: Option<MediaFilter> on Channel (persisted as JSON column)
- migration: 20260712000002_add_chapters_and_gap_filler.sql
2026-07-12 07:41:57 +02:00
39f5f99bfd refactor: replace get_stream_url with get_source_uri per ADR-0002
Providers now expose a source URI (for FFmpeg) instead of a
viewer-facing stream URL. Playout Service will own transcoding.

- Add SourceUri value object (NetworkUrl | FilePath)
- Remove StreamQuality, StreamingProtocol from domain
- Simplify ProviderCapabilities (drop streaming_protocol, transcode)
- Jellyfin: return static direct stream URL
- Local files: return absolute file path
- Rename use case get_stream_url -> get_source
- Stream endpoint returns JSON SourceUri instead of 307 redirect
2026-07-12 07:34:13 +02:00
abcf69ce7e extract background tasks into worker binary
Move auto_scheduler, library_sync, broadcast_poller, webhook_consumer
from presentation/background/ into crates/worker/src/jobs/.
Presentation is now a pure HTTP server.
2026-07-12 07:28:17 +02:00
826e824b58 replace tokio::broadcast event bus with SQLite-backed event queue (ADR-0003)
- DomainEvent: add Serialize/Deserialize
- EventConsumer port: poll_next/ack/nack replacing recv()
- EventEnvelope: wraps event with id/retry_count/created_at
- SqliteEventPublisher/SqliteEventConsumer: INSERT/poll/DLQ
- migration: event_queue + dead_letter_queue tables
- InMemoryEventBus: combined publisher+consumer test double
- NoopEventConsumer: test stub
- webhook_consumer: polls EventConsumer instead of broadcast::Receiver
- presentation+mcp wiring: create from SqlitePool
2026-07-12 07:23:21 +02:00
e2393be635 collapse 20 pass-through use cases; handlers call ports directly
delete get/list/list_by_owner channels, get_settings/activity_log admin,
get_item/get_sync_status/list_collections/list_seasons/list_shows/list_genres library,
get/list/delete providers, get/list/patch_label config_snapshots,
get_active/list_history/delete_after schedule — all single-delegation.

remove ChannelQueryDeps, LibraryQueryDeps, deleted query/command structs.
add direct port fields to AppState. update MCP crate accordingly.
2026-07-12 07:18:26 +02:00
a6558e15b2 kill LibraryItem, unify to MediaItem; decouple schedule engine from providers
ADR-0001: MediaItem absorbs LibraryItem fields (provider_id, external_id,
collection_name, collection_type, synced_at, role/MediaRole).
LibraryItem + LibraryItemRow deleted. All ports/adapters/tests updated.

ADR-0002: schedule engine takes LibraryQuery instead of IProviderRegistry.
Algorithmic blocks query library via search(), manual blocks via get_by_id().
get_stream_url removed from engine; provider_registry moved to ScheduleDeps
for playback-time stream URL resolution in application layer.

BlockContent provider_id field removed (meaningless when querying library).
2026-07-12 07:02:08 +02:00
773e228e21 rename RecyclePolicy→RotationPolicy, collapse AccessMode, clean OIDC refs 2026-07-12 06:50:25 +02:00
187 changed files with 7918 additions and 2860 deletions

71
Cargo.lock generated
View File

@@ -25,6 +25,7 @@ dependencies = [
"serde",
"serde_json",
"sqlx",
"tokio",
"tracing",
"uuid",
]
@@ -34,7 +35,11 @@ name = "adapter-event-publisher"
version = "0.1.0"
dependencies = [
"async-trait",
"chrono",
"domain",
"serde",
"serde_json",
"sqlx",
"tokio",
"tracing",
]
@@ -1762,6 +1767,29 @@ version = "0.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6"
[[package]]
name = "playout"
version = "0.1.0"
dependencies = [
"anyhow",
"async-trait",
"axum",
"bytes",
"chrono",
"domain",
"dotenvy",
"futures",
"serde",
"serde_json",
"thiserror",
"tokio",
"tower",
"tower-http",
"tracing",
"tracing-subscriber",
"uuid",
]
[[package]]
name = "potential_utf"
version = "0.1.5"
@@ -1791,6 +1819,7 @@ name = "presentation"
version = "0.1.0"
dependencies = [
"adapter-auth",
"adapter-common",
"adapter-event-publisher",
"adapter-jellyfin",
"adapter-local-files",
@@ -1804,9 +1833,7 @@ dependencies = [
"chrono",
"domain",
"dotenvy",
"handlebars",
"infra-wiring",
"reqwest",
"serde",
"serde_json",
"thiserror",
@@ -1816,6 +1843,8 @@ dependencies = [
"tower-http",
"tracing",
"tracing-subscriber",
"utoipa",
"utoipa-scalar",
"uuid",
]
@@ -3002,6 +3031,18 @@ dependencies = [
"uuid",
]
[[package]]
name = "utoipa-scalar"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "59559e1509172f6b26c1cdbc7247c4ddd1ac6560fe94b584f81ee489b141f719"
dependencies = [
"axum",
"serde",
"serde_json",
"utoipa",
]
[[package]]
name = "uuid"
version = "1.23.4"
@@ -3365,6 +3406,32 @@ version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
[[package]]
name = "worker"
version = "0.1.0"
dependencies = [
"adapter-auth",
"adapter-common",
"adapter-event-publisher",
"adapter-jellyfin",
"adapter-local-files",
"adapter-sqlite",
"anyhow",
"application",
"async-trait",
"chrono",
"domain",
"dotenvy",
"handlebars",
"infra-wiring",
"reqwest",
"serde_json",
"tokio",
"tracing",
"tracing-subscriber",
"uuid",
]
[[package]]
name = "writeable"
version = "0.6.3"

View File

@@ -1,5 +1,5 @@
[workspace]
members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files", "crates/adapters/event-publisher", "crates/presentation", "crates/mcp"]
members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files", "crates/adapters/event-publisher", "crates/presentation", "crates/worker", "crates/mcp", "crates/playout"]
exclude = ["k-tv-backend", "k-tv-frontend"]
resolver = "2"
@@ -24,6 +24,7 @@ tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
reqwest = { version = "0.12", features = ["json"] }
utoipa = { version = "5", features = ["chrono", "uuid"] }
utoipa-scalar = { version = "0.3", features = ["axum"] }
jsonwebtoken = "9"
# Internal crates

View File

@@ -11,24 +11,38 @@
# TRAEFIK_CERT_RESOLVER cert resolver name for TLS (default: letsencrypt)
# FRONTEND_HOST public hostname for the frontend e.g. tv.example.com
# BACKEND_HOST public hostname for the backend API e.g. tv-api.example.com
# PLAYOUT_HOST public hostname for playout streams e.g. tv-playout.example.com
#
# Remember: NEXT_PUBLIC_API_URL in .env must be the *public* backend URL,
# e.g. https://tv-api.example.com/api/v1, and you must rebuild after changing it.
services:
backend:
ports: [] # Traefik handles ingress; no direct port exposure needed
presentation:
ports: []
networks:
- default
- traefik
labels:
- "traefik.enable=true"
- "traefik.docker.network=${TRAEFIK_NETWORK:-traefik_proxy}"
- "traefik.http.routers.ktv-backend.rule=Host(`${BACKEND_HOST}`)"
- "traefik.http.routers.ktv-backend.entrypoints=${TRAEFIK_ENTRYPOINT:-websecure}"
- "traefik.http.routers.ktv-backend.tls.certresolver=${TRAEFIK_CERT_RESOLVER:-letsencrypt}"
- "traefik.http.services.ktv-backend.loadbalancer.server.port=3000"
- "traefik.http.routers.ktv-presentation.rule=Host(`${BACKEND_HOST}`)"
- "traefik.http.routers.ktv-presentation.entrypoints=${TRAEFIK_ENTRYPOINT:-websecure}"
- "traefik.http.routers.ktv-presentation.tls.certresolver=${TRAEFIK_CERT_RESOLVER:-letsencrypt}"
- "traefik.http.services.ktv-presentation.loadbalancer.server.port=3000"
playout:
ports: []
networks:
- default
- traefik
labels:
- "traefik.enable=true"
- "traefik.docker.network=${TRAEFIK_NETWORK:-traefik_proxy}"
- "traefik.http.routers.ktv-playout.rule=Host(`${PLAYOUT_HOST}`)"
- "traefik.http.routers.ktv-playout.entrypoints=${TRAEFIK_ENTRYPOINT:-websecure}"
- "traefik.http.routers.ktv-playout.tls.certresolver=${TRAEFIK_CERT_RESOLVER:-letsencrypt}"
- "traefik.http.services.ktv-playout.loadbalancer.server.port=9090"
frontend:
ports: []

View File

@@ -1,19 +1,19 @@
services:
# ── Backend (Rust / Axum) ──────────────────────────────────────────────────
backend:
build: ./k-tv-backend
# ── Presentation (Rust / Axum — HTTP API) ────────────────────────────────
presentation:
build:
context: ./k-tv-backend
target: presentation
image: registry.gabrielkaszewski.dev/k-tv-presentation:latest
ports:
- "${BACKEND_PORT:-3000}:3000"
environment:
- HOST=0.0.0.0
- PORT=3000
- DATABASE_URL=sqlite:///app/data/k-tv.db?mode=rwc
# Allow requests from the browser (the user-facing frontend URL)
- CORS_ALLOWED_ORIGINS=${CORS_ALLOWED_ORIGINS}
# Auth — generate with: openssl rand -hex 32
- JWT_SECRET=${JWT_SECRET}
# Cookie secret — generate with: openssl rand -base64 64
- COOKIE_SECRET=${COOKIE_SECRET}
- JWT_EXPIRY_HOURS=${JWT_EXPIRY_HOURS:-24}
- SECURE_COOKIE=${SECURE_COOKIE:-false}
@@ -21,7 +21,6 @@ services:
- ALLOW_REGISTRATION=${ALLOW_REGISTRATION:-true}
- DB_MAX_CONNECTIONS=${DB_MAX_CONNECTIONS:-5}
- DB_MIN_CONNECTIONS=${DB_MIN_CONNECTIONS:-1}
# Jellyfin — all three required for schedule generation
- JELLYFIN_BASE_URL=${JELLYFIN_BASE_URL}
- JELLYFIN_API_KEY=${JELLYFIN_API_KEY}
- JELLYFIN_USER_ID=${JELLYFIN_USER_ID}
@@ -34,40 +33,60 @@ services:
timeout: 5s
retries: 3
# ── Worker (background jobs) ─────────────────────────────────────────────
worker:
build:
context: ./k-tv-backend
target: worker
image: registry.gabrielkaszewski.dev/k-tv-worker:latest
environment:
- DATABASE_URL=sqlite:///app/data/k-tv.db?mode=rwc
- JELLYFIN_BASE_URL=${JELLYFIN_BASE_URL}
- JELLYFIN_API_KEY=${JELLYFIN_API_KEY}
- JELLYFIN_USER_ID=${JELLYFIN_USER_ID}
volumes:
- backend_data:/app/data
depends_on:
presentation:
condition: service_healthy
restart: unless-stopped
# ── Playout (HLS streaming) ──────────────────────────────────────────────
playout:
build:
context: ./k-tv-backend
target: playout
image: registry.gabrielkaszewski.dev/k-tv-playout:latest
ports:
- "${PLAYOUT_PORT:-9090}:9090"
environment:
- DATABASE_URL=sqlite:///app/data/k-tv.db?mode=rwc
- PLAYOUT_LISTEN_ADDR=0.0.0.0:9090
- PLAYOUT_STORAGE_PATH=/tmp/k-tv-playout
- PLAYOUT_SEGMENT_DURATION=${PLAYOUT_SEGMENT_DURATION:-6}
volumes:
- backend_data:/app/data
depends_on:
presentation:
condition: service_healthy
restart: unless-stopped
# ── Frontend (Next.js) ────────────────────────────────────────────────────
frontend:
build:
context: ./k-tv-frontend
args:
# Browser-visible backend URL — baked into the client bundle at build time.
# Rebuild the image after changing this.
NEXT_PUBLIC_API_URL: ${NEXT_PUBLIC_API_URL:-http://localhost:4000/api/v1}
NEXT_PUBLIC_PLAYOUT_URL: ${NEXT_PUBLIC_PLAYOUT_URL:-http://localhost:9090}
image: registry.gabrielkaszewski.dev/k-tv-frontend:latest
ports:
- "${FRONTEND_PORT:-3001}:3001"
environment:
# Server-side API URL — uses Docker's internal network, never exposed.
# Next.js API routes (e.g. /api/stream/[channelId]) use this.
API_URL: http://backend:3000/api/v1
API_URL: http://presentation:3000/api/v1
depends_on:
backend:
presentation:
condition: service_healthy
restart: unless-stopped
volumes:
backend_data:
# ── Optional: PostgreSQL ───────────────────────────────────────────────────
# Uncomment the db service and set DATABASE_URL in backend's environment:
# DATABASE_URL: postgres://ktv:${POSTGRES_PASSWORD}@db:5432/ktv
#
# db:
# image: postgres:16-alpine
# environment:
# POSTGRES_USER: ktv
# POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
# POSTGRES_DB: ktv
# volumes:
# - db_data:/var/lib/postgresql/data
# restart: unless-stopped
#
# db_data:

View File

@@ -11,3 +11,4 @@ uuid = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
tracing = { workspace = true }
tokio = { workspace = true }

View File

@@ -0,0 +1,180 @@
use domain::{Chapter, ContentType, SourceUri};
use serde::Deserialize;
const CHAPTER_PROBE_MIN_DURATION_SECS: u32 = 2700;
#[derive(Deserialize)]
struct FfprobeOutput {
#[serde(default)]
chapters: Vec<FfprobeChapter>,
}
#[derive(Deserialize)]
struct FfprobeChapter {
#[serde(default)]
start_time: String,
#[serde(default)]
end_time: String,
#[serde(default)]
tags: Option<FfprobeChapterTags>,
}
#[derive(Deserialize)]
struct FfprobeChapterTags {
title: Option<String>,
}
pub fn parse_chapters_json(json: &str) -> Vec<Chapter> {
let output: FfprobeOutput = match serde_json::from_str(json) {
Ok(o) => o,
Err(_) => return Vec::new(),
};
output
.chapters
.into_iter()
.map(|c| {
let title = c.tags.and_then(|t| t.title);
let start_secs = c.start_time.parse::<f64>().unwrap_or(0.0);
let end_secs = c.end_time.parse::<f64>().unwrap_or(0.0);
Chapter::new(title, start_secs, end_secs)
})
.collect()
}
pub fn should_probe_chapters(content_type: &ContentType, duration_secs: u32) -> bool {
matches!(content_type, ContentType::Movie) || duration_secs > CHAPTER_PROBE_MIN_DURATION_SECS
}
pub async fn extract_chapters(source_uri: &SourceUri) -> Vec<Chapter> {
let path = match source_uri {
SourceUri::FilePath { path } => path.clone(),
SourceUri::NetworkUrl { url } => url.clone(),
};
let result = tokio::process::Command::new("ffprobe")
.args([
"-v",
"quiet",
"-print_format",
"json",
"-show_chapters",
&path,
])
.output()
.await;
match result {
Ok(output) if output.status.success() => {
let json = String::from_utf8_lossy(&output.stdout);
parse_chapters_json(&json)
}
Ok(output) => {
tracing::warn!(
path = %path,
stderr = %String::from_utf8_lossy(&output.stderr),
"ffprobe exited with non-zero status"
);
Vec::new()
}
Err(e) => {
tracing::warn!(error = %e, "ffprobe not available or failed to execute");
Vec::new()
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parse_ffprobe_json_with_chapters() {
let json = r#"{
"chapters": [
{
"id": 0,
"time_base": "1/1000",
"start": 0,
"start_time": "0.000000",
"end": 300000,
"end_time": "300.000000",
"tags": { "title": "Opening" }
},
{
"id": 1,
"time_base": "1/1000",
"start": 300000,
"start_time": "300.000000",
"end": 1800000,
"end_time": "1800.000000",
"tags": { "title": "Main Feature" }
},
{
"id": 2,
"time_base": "1/1000",
"start": 1800000,
"start_time": "1800.000000",
"end": 2100000,
"end_time": "2100.000000"
}
]
}"#;
let chapters = parse_chapters_json(json);
assert_eq!(chapters.len(), 3);
assert_eq!(chapters[0].title(), Some("Opening"));
assert!((chapters[0].start_secs() - 0.0).abs() < f64::EPSILON);
assert!((chapters[0].end_secs() - 300.0).abs() < f64::EPSILON);
assert_eq!(chapters[1].title(), Some("Main Feature"));
assert!((chapters[1].start_secs() - 300.0).abs() < f64::EPSILON);
assert!((chapters[1].end_secs() - 1800.0).abs() < f64::EPSILON);
assert_eq!(chapters[2].title(), None);
assert!((chapters[2].start_secs() - 1800.0).abs() < f64::EPSILON);
assert!((chapters[2].end_secs() - 2100.0).abs() < f64::EPSILON);
}
#[test]
fn parse_ffprobe_json_no_chapters() {
let json = r#"{ "chapters": [] }"#;
let chapters = parse_chapters_json(json);
assert!(chapters.is_empty());
}
#[test]
fn parse_ffprobe_json_missing_chapters_key() {
let json = r#"{}"#;
let chapters = parse_chapters_json(json);
assert!(chapters.is_empty());
}
#[test]
fn parse_ffprobe_json_invalid() {
let chapters = parse_chapters_json("not json");
assert!(chapters.is_empty());
}
#[test]
fn short_items_skip_probe() {
assert!(!should_probe_chapters(&ContentType::Episode, 1800));
assert!(!should_probe_chapters(&ContentType::Short, 300));
}
#[test]
fn movie_always_probed() {
assert!(should_probe_chapters(&ContentType::Movie, 600));
assert!(should_probe_chapters(&ContentType::Movie, 7200));
}
#[test]
fn long_episode_probed() {
assert!(should_probe_chapters(&ContentType::Episode, 3600));
}
#[test]
fn episode_under_threshold_skipped() {
assert!(!should_probe_chapters(&ContentType::Episode, 2700));
}
}

View File

@@ -1,5 +1,8 @@
pub mod ffprobe;
pub mod role_detector;
use chrono::{DateTime, Utc};
use domain::{DomainError, RecyclePolicy, ScheduleConfig, ScheduleConfigCompat};
use domain::{DomainError, RotationPolicy, ScheduleConfig, ScheduleConfigCompat};
use serde::de::DeserializeOwned;
use uuid::Uuid;
@@ -32,8 +35,8 @@ pub fn parse_schedule_config(json: &str) -> Result<ScheduleConfig, DomainError>
Ok(ScheduleConfig::from(compat))
}
pub fn parse_recycle_policy(json: &str) -> Result<RecyclePolicy, DomainError> {
parse_json(json, "recycle_policy")
pub fn parse_rotation_policy(json: &str) -> Result<RotationPolicy, DomainError> {
parse_json(json, "rotation_policy")
}
pub fn parse_enum_or_default<T: DeserializeOwned + Default>(value: String) -> T {
@@ -151,9 +154,9 @@ mod tests {
}
#[test]
fn parse_recycle_policy_valid() {
fn parse_rotation_policy_valid() {
let json = r#"{"cooldown_days":7,"cooldown_generations":3,"min_available_ratio":0.3}"#;
let policy = parse_recycle_policy(json).unwrap();
let policy = parse_rotation_policy(json).unwrap();
assert_eq!(policy.cooldown_days, Some(7));
}

View File

@@ -0,0 +1,158 @@
use domain::{MediaItem, MediaRole};
#[derive(Debug, Clone)]
pub struct RoleDetectionConfig {
pub interstitial_collection_patterns: Vec<String>,
pub interstitial_tag_patterns: Vec<String>,
}
impl Default for RoleDetectionConfig {
fn default() -> Self {
Self {
interstitial_collection_patterns: vec![
"bumper".into(),
"bumpers".into(),
"ad".into(),
"ads".into(),
"interstitial".into(),
"interstitials".into(),
"promo".into(),
"promos".into(),
"ident".into(),
"idents".into(),
],
interstitial_tag_patterns: vec![
"bumper".into(),
"interstitial".into(),
"ad".into(),
"promo".into(),
"ident".into(),
],
}
}
}
pub fn detect_role(item: &MediaItem, config: &RoleDetectionConfig) -> MediaRole {
if matches_collection_pattern(item, &config.interstitial_collection_patterns) {
return MediaRole::Interstitial;
}
if matches_tag_pattern(item, &config.interstitial_tag_patterns) {
return MediaRole::Interstitial;
}
MediaRole::Program
}
fn matches_collection_pattern(item: &MediaItem, patterns: &[String]) -> bool {
let collection_name = match item.collection_name() {
Some(name) => name.to_lowercase(),
None => return false,
};
patterns
.iter()
.any(|pattern| collection_name == pattern.to_lowercase())
}
fn matches_tag_pattern(item: &MediaItem, patterns: &[String]) -> bool {
item.tags().iter().any(|tag| {
let lower_tag = tag.to_lowercase();
patterns
.iter()
.any(|pattern| lower_tag == pattern.to_lowercase())
})
}
#[cfg(test)]
mod tests {
use super::*;
use domain::{ContentType, MediaItemId, MediaItemRow};
fn make_item(
collection_name: Option<&str>,
tags: Vec<&str>,
) -> MediaItem {
MediaItem::from_persistence(MediaItemRow {
id: MediaItemId::new("test::1"),
provider_id: "test".into(),
external_id: "1".into(),
title: "Test Item".into(),
content_type: ContentType::Movie,
duration_secs: 3600,
description: None,
series_name: None,
season_number: None,
episode_number: None,
year: None,
genres: vec![],
tags: tags.into_iter().map(String::from).collect(),
collection_id: None,
collection_name: collection_name.map(String::from),
collection_type: None,
thumbnail_url: None,
synced_at: None,
role: MediaRole::default(),
chapters: vec![],
})
}
#[test]
fn item_from_bumpers_collection_gets_interstitial() {
let item = make_item(Some("Bumpers"), vec![]);
let config = RoleDetectionConfig::default();
assert_eq!(detect_role(&item, &config), MediaRole::Interstitial);
}
#[test]
fn item_with_bumper_tag_gets_interstitial() {
let item = make_item(None, vec!["bumper"]);
let config = RoleDetectionConfig::default();
assert_eq!(detect_role(&item, &config), MediaRole::Interstitial);
}
#[test]
fn item_from_ads_collection_gets_interstitial() {
let item = make_item(Some("Ads"), vec![]);
let config = RoleDetectionConfig::default();
assert_eq!(detect_role(&item, &config), MediaRole::Interstitial);
}
#[test]
fn normal_item_from_movies_gets_program() {
let item = make_item(Some("Movies"), vec![]);
let config = RoleDetectionConfig::default();
assert_eq!(detect_role(&item, &config), MediaRole::Program);
}
#[test]
fn item_with_no_collection_or_tags_gets_program() {
let item = make_item(None, vec![]);
let config = RoleDetectionConfig::default();
assert_eq!(detect_role(&item, &config), MediaRole::Program);
}
#[test]
fn case_insensitive_collection_match() {
let item = make_item(Some("INTERSTITIALS"), vec![]);
let config = RoleDetectionConfig::default();
assert_eq!(detect_role(&item, &config), MediaRole::Interstitial);
}
#[test]
fn case_insensitive_tag_match() {
let item = make_item(None, vec!["PROMO"]);
let config = RoleDetectionConfig::default();
assert_eq!(detect_role(&item, &config), MediaRole::Interstitial);
}
#[test]
fn custom_config_patterns() {
let item = make_item(Some("Station IDs"), vec![]);
let config = RoleDetectionConfig {
interstitial_collection_patterns: vec!["station ids".into()],
interstitial_tag_patterns: vec![],
};
assert_eq!(detect_role(&item, &config), MediaRole::Interstitial);
}
}

View File

@@ -8,3 +8,7 @@ domain = { workspace = true }
async-trait = { workspace = true }
tokio = { workspace = true }
tracing = { workspace = true }
sqlx = { workspace = true, features = ["sqlite"] }
serde = { workspace = true }
serde_json = { workspace = true }
chrono = { workspace = true }

View File

@@ -1,42 +1,170 @@
use async_trait::async_trait;
use domain::errors::{DomainError, DomainResult};
use domain::events::DomainEvent;
use domain::events::{DomainEvent, EventEnvelope};
use domain::ports::events::{EventConsumer, EventPublisher};
use tokio::sync::broadcast;
use sqlx::SqlitePool;
pub struct ChannelEventBus {
tx: broadcast::Sender<DomainEvent>,
fn event_type_label(event: &DomainEvent) -> &'static str {
match event {
DomainEvent::BroadcastTransition { .. } => "broadcast_transition",
DomainEvent::NoSignal { .. } => "no_signal",
DomainEvent::ScheduleGenerated { .. } => "schedule_generated",
DomainEvent::ChannelCreated { .. } => "channel_created",
DomainEvent::ChannelUpdated { .. } => "channel_updated",
DomainEvent::ChannelDeleted { .. } => "channel_deleted",
DomainEvent::UserRegistered { .. } => "user_registered",
_ => "unknown",
}
}
impl ChannelEventBus {
pub fn new(capacity: usize) -> Self {
let (tx, _) = broadcast::channel(capacity);
Self { tx }
pub struct SqliteEventPublisher {
pool: SqlitePool,
}
pub fn subscriber(&self) -> broadcast::Receiver<DomainEvent> {
self.tx.subscribe()
}
pub fn sender(&self) -> broadcast::Sender<DomainEvent> {
self.tx.clone()
impl SqliteEventPublisher {
pub fn new(pool: SqlitePool) -> Self {
Self { pool }
}
}
#[async_trait]
impl EventPublisher for ChannelEventBus {
impl EventPublisher for SqliteEventPublisher {
async fn publish(&self, event: DomainEvent) -> DomainResult<()> {
let _ = self.tx.send(event);
let event_type = event_type_label(&event);
let payload = serde_json::to_string(&event)
.map_err(|e| DomainError::InfrastructureError(format!("event serialize: {e}")))?;
sqlx::query(
"INSERT INTO event_queue (event_type, payload, status) VALUES (?, ?, 'pending')",
)
.bind(event_type)
.bind(&payload)
.execute(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(())
}
}
pub struct SqliteEventConsumer {
pool: SqlitePool,
}
impl SqliteEventConsumer {
pub fn new(pool: SqlitePool) -> Self {
Self { pool }
}
}
#[derive(sqlx::FromRow)]
struct EventRow {
id: i64,
payload: String,
retry_count: i32,
created_at: String,
max_retries: i32,
event_type: String,
}
#[async_trait]
impl EventConsumer for ChannelEventBus {
async fn recv(&self) -> DomainResult<DomainEvent> {
let mut rx = self.tx.subscribe();
rx.recv()
impl EventConsumer for SqliteEventConsumer {
async fn poll_next(&self) -> DomainResult<Option<EventEnvelope>> {
let row: Option<EventRow> = sqlx::query_as(
"SELECT id, event_type, payload, retry_count, created_at, max_retries \
FROM event_queue WHERE status = 'pending' ORDER BY id ASC LIMIT 1",
)
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
let row = match row {
Some(r) => r,
None => return Ok(None),
};
sqlx::query("UPDATE event_queue SET status = 'processing', updated_at = datetime('now') WHERE id = ?")
.bind(row.id)
.execute(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
match serde_json::from_str::<DomainEvent>(&row.payload) {
Ok(event) => Ok(Some(EventEnvelope::from_persistence(
row.id,
event,
row.retry_count as u32,
row.created_at,
))),
Err(e) => {
move_to_dlq(&self.pool, &row, &e.to_string()).await?;
Ok(None)
}
}
}
async fn ack(&self, event_id: i64) -> DomainResult<()> {
sqlx::query("DELETE FROM event_queue WHERE id = ?")
.bind(event_id)
.execute(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(())
}
async fn nack(&self, event_id: i64, error: &str) -> DomainResult<()> {
let row: Option<EventRow> = sqlx::query_as(
"SELECT id, event_type, payload, retry_count, created_at, max_retries FROM event_queue WHERE id = ?",
)
.bind(event_id)
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
let row = match row {
Some(r) => r,
None => return Ok(()),
};
let new_retry = row.retry_count + 1;
if new_retry >= row.max_retries {
move_to_dlq(&self.pool, &row, error).await?;
} else {
sqlx::query(
"UPDATE event_queue SET status = 'pending', retry_count = ?, error_message = ?, updated_at = datetime('now') WHERE id = ?",
)
.bind(new_retry)
.bind(error)
.bind(event_id)
.execute(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
}
Ok(())
}
}
async fn move_to_dlq(pool: &SqlitePool, row: &EventRow, error: &str) -> DomainResult<()> {
sqlx::query(
"INSERT INTO dead_letter_queue (original_event_id, event_type, payload, error_message, retry_count, original_created_at) \
VALUES (?, ?, ?, ?, ?, ?)",
)
.bind(row.id)
.bind(&row.event_type)
.bind(&row.payload)
.bind(error)
.bind(row.retry_count)
.bind(&row.created_at)
.execute(pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
sqlx::query("DELETE FROM event_queue WHERE id = ?")
.bind(row.id)
.execute(pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(())
}

View File

@@ -1,4 +1,4 @@
use domain::{ContentType, MediaItem, MediaItemId, MediaItemRow};
use domain::{ContentType, MediaItem, MediaItemId, MediaItemRow, MediaRole};
use crate::models::JellyfinItem;
@@ -17,7 +17,7 @@ pub(crate) fn map_jellyfin_item(item: JellyfinItem) -> Option<MediaItem> {
.unwrap_or(0);
Some(MediaItem::from_persistence(MediaItemRow {
id: MediaItemId::new(item.id),
id: MediaItemId::new(&item.id),
title: item.name,
content_type,
duration_secs,
@@ -30,5 +30,12 @@ pub(crate) fn map_jellyfin_item(item: JellyfinItem) -> Option<MediaItem> {
episode_number: item.index_number,
thumbnail_url: None,
collection_id: None,
provider_id: String::new(),
external_id: item.id,
collection_name: None,
collection_type: None,
synced_at: None,
role: MediaRole::default(),
chapters: Vec::new(),
}))
}

View File

@@ -37,20 +37,6 @@ pub(crate) struct JellyfinItem {
pub recursive_item_count: Option<u32>,
}
#[derive(Debug, Deserialize)]
pub(crate) struct JellyfinPlaybackInfoResponse {
#[serde(rename = "MediaSources")]
pub media_sources: Vec<JellyfinMediaSource>,
}
#[derive(Debug, Deserialize)]
pub(crate) struct JellyfinMediaSource {
#[serde(rename = "SupportsDirectStream")]
pub supports_direct_stream: bool,
#[serde(rename = "DirectStreamUrl")]
pub direct_stream_url: Option<String>,
}
pub(crate) fn jellyfin_item_type(ct: &ContentType) -> &'static str {
match ct {
ContentType::Movie => "Movie",

View File

@@ -1,18 +1,11 @@
use async_trait::async_trait;
use domain::ports::{
Collection, IMediaProvider, ProviderCapabilities, SeriesSummary, StreamQuality,
StreamingProtocol,
};
use domain::{ContentType, DomainError, DomainResult, MediaFilter, MediaItem, MediaItemId};
use domain::ports::{Collection, IMediaProvider, ProviderCapabilities, SeriesSummary};
use domain::{ContentType, DomainError, DomainResult, MediaFilter, MediaItem, MediaItemId, SourceUri};
use crate::config::JellyfinConfig;
use crate::mapping::{map_jellyfin_item, TICKS_PER_SEC};
use crate::models::{
jellyfin_item_type, JellyfinItemsResponse, JellyfinPlaybackInfoResponse,
};
const FALLBACK_HLS_BITRATE: u32 = 8_000_000;
use crate::models::{jellyfin_item_type, JellyfinItemsResponse};
pub struct JellyfinMediaProvider {
client: reqwest::Client,
@@ -129,16 +122,6 @@ impl JellyfinMediaProvider {
Ok(items)
}
fn hls_url(&self, item_id: &MediaItemId, bitrate: u32) -> String {
format!(
"{}/Videos/{}/master.m3u8?videoCodec=h264&audioCodec=aac&VideoBitRate={}&mediaSourceId={}&SubtitleMethod=Hls&subtitleCodec=vtt&api_key={}",
self.config.base_url,
item_id.as_ref(),
bitrate,
item_id.as_ref(),
self.config.api_key,
)
}
}
#[async_trait]
@@ -151,9 +134,7 @@ impl IMediaProvider for JellyfinMediaProvider {
tags: true,
decade: true,
search: true,
streaming_protocol: StreamingProtocol::Hls,
rescan: false,
transcode: false,
}
}
@@ -350,52 +331,14 @@ impl IMediaProvider for JellyfinMediaProvider {
Ok(body.items.into_iter().map(|item| item.name).collect())
}
async fn get_stream_url(
&self,
item_id: &MediaItemId,
quality: &StreamQuality,
) -> DomainResult<String> {
match quality {
StreamQuality::Direct => {
let url = format!(
"{}/Items/{}/PlaybackInfo",
async fn get_source_uri(&self, item_id: &MediaItemId) -> DomainResult<SourceUri> {
Ok(SourceUri::NetworkUrl {
url: format!(
"{}/Videos/{}/stream?static=true&api_key={}",
self.config.base_url,
item_id.as_ref()
);
let resp = self
.client
.post(&url)
.header("X-Emby-Token", &self.config.api_key)
.query(&[
("userId", &self.config.user_id),
("mediaSourceId", &item_id.as_ref().to_string()),
])
.json(&serde_json::json!({}))
.send()
.await
.map_err(|e| {
DomainError::InfrastructureError(format!("PlaybackInfo failed: {e}"))
})?;
if resp.status().is_success() {
let info: JellyfinPlaybackInfoResponse = resp.json().await.map_err(|e| {
DomainError::InfrastructureError(format!(
"PlaybackInfo parse failed: {e}"
))
})?;
if let Some(src) = info.media_sources.first()
&& src.supports_direct_stream
&& let Some(rel_url) = &src.direct_stream_url
{
return Ok(format!(
"{}{}&api_key={}",
self.config.base_url, rel_url, self.config.api_key
));
}
}
Ok(self.hls_url(item_id, FALLBACK_HLS_BITRATE))
}
StreamQuality::Transcode(bps) => Ok(self.hls_url(item_id, *bps)),
}
item_id.as_ref(),
self.config.api_key,
),
})
}
}

View File

@@ -30,7 +30,7 @@ impl LocalFilesBundle {
});
let provider =
LocalFilesProvider::new(Arc::clone(&local_index), &config, transcode_manager.clone());
LocalFilesProvider::new(Arc::clone(&local_index), &config);
Self {
provider,

View File

@@ -1,36 +1,23 @@
use std::sync::Arc;
use async_trait::async_trait;
use domain::ports::{
Collection, IMediaProvider, ProviderCapabilities, StreamQuality, StreamingProtocol,
};
use domain::{ContentType, DomainError, DomainResult, MediaFilter, MediaItem, MediaItemId, MediaItemRow};
use domain::ports::{Collection, IMediaProvider, ProviderCapabilities};
use domain::{ContentType, DomainError, DomainResult, MediaFilter, MediaItem, MediaItemId, MediaItemRow, MediaRole, SourceUri};
use crate::config::LocalFilesConfig;
use crate::index::{decode_id, LocalIndex};
use crate::scanner::LocalFileItem;
use crate::transcoder::TranscodeManager;
pub struct LocalFilesProvider {
pub index: Arc<LocalIndex>,
base_url: String,
transcode_manager: Option<Arc<TranscodeManager>>,
}
const SHORT_DURATION_SECS: u32 = 1200;
const DECADE_SPAN: u16 = 9;
impl LocalFilesProvider {
pub fn new(
index: Arc<LocalIndex>,
config: &LocalFilesConfig,
transcode_manager: Option<Arc<TranscodeManager>>,
) -> Self {
Self {
index,
base_url: config.base_url.trim_end_matches('/').to_string(),
transcode_manager,
}
pub fn new(index: Arc<LocalIndex>, _config: &LocalFilesConfig) -> Self {
Self { index }
}
}
@@ -54,6 +41,13 @@ fn to_media_item(id: MediaItemId, item: &LocalFileItem) -> MediaItem {
episode_number: None,
thumbnail_url: None,
collection_id: None,
provider_id: String::new(),
external_id: String::new(),
collection_name: None,
collection_type: None,
synced_at: None,
role: MediaRole::default(),
chapters: Vec::new(),
})
}
@@ -67,13 +61,7 @@ impl IMediaProvider for LocalFilesProvider {
tags: true,
decade: true,
search: true,
streaming_protocol: if self.transcode_manager.is_some() {
StreamingProtocol::Hls
} else {
StreamingProtocol::DirectFile
},
rescan: true,
transcode: self.transcode_manager.is_some(),
}
}
@@ -149,31 +137,14 @@ impl IMediaProvider for LocalFilesProvider {
.map(|item| to_media_item(item_id.clone(), &item)))
}
async fn get_stream_url(
&self,
item_id: &MediaItemId,
quality: &StreamQuality,
) -> DomainResult<String> {
match quality {
StreamQuality::Transcode(_) if self.transcode_manager.is_some() => {
let tm = self.transcode_manager.as_ref().unwrap();
async fn get_source_uri(&self, item_id: &MediaItemId) -> DomainResult<SourceUri> {
let rel = decode_id(item_id).ok_or_else(|| {
DomainError::InfrastructureError("invalid item id encoding".into())
})?;
let src = self.index.root_dir.join(&rel);
tm.ensure_transcoded(item_id.as_ref(), &src).await?;
Ok(format!(
"{}/api/v1/files/transcode/{}/playlist.m3u8",
self.base_url,
item_id.as_ref()
))
}
_ => Ok(format!(
"{}/api/v1/files/stream/{}",
self.base_url,
item_id.as_ref()
)),
}
let abs_path = self.index.root_dir.join(&rel);
Ok(SourceUri::FilePath {
path: abs_path.to_string_lossy().into_owned(),
})
}
async fn list_collections(&self) -> DomainResult<Vec<Collection>> {

View File

@@ -4,13 +4,13 @@ use sqlx::{Row, SqlitePool};
use uuid::Uuid;
use adapter_common::{
map_sqlx_error, parse_dt, parse_enum_or_default, parse_recycle_policy, parse_schedule_config,
map_sqlx_error, parse_dt, parse_enum_or_default, parse_rotation_policy, parse_schedule_config,
parse_uuid, serialize_enum_as_string,
};
use domain::{
ports::channel::{ChannelCommand, ChannelQuery},
AccessMode, Channel, ChannelConfigSnapshot, ChannelId, ChannelRow as DomainChannelRow,
DomainError, DomainResult, LogoPosition, ScheduleConfig, SnapshotId, UserId,
DomainError, DomainResult, LogoPosition, MediaFilter, ScheduleConfig, SnapshotId, UserId,
};
pub struct SqliteChannelRepository {
@@ -23,7 +23,7 @@ impl SqliteChannelRepository {
}
}
const SELECT_COLS: &str = "id, owner_id, name, description, timezone, schedule_config, recycle_policy, auto_schedule, access_mode, access_password_hash, logo, logo_position, logo_opacity, webhook_url, webhook_poll_interval_secs, webhook_body_template, webhook_headers, created_at, updated_at";
const SELECT_COLS: &str = "id, owner_id, name, description, timezone, schedule_config, rotation_policy, auto_schedule, access_mode, logo, logo_position, logo_opacity, webhook_url, webhook_poll_interval_secs, webhook_body_template, webhook_headers, gap_filler, created_at, updated_at";
#[derive(Debug, sqlx::FromRow)]
struct ChannelRow {
@@ -33,10 +33,9 @@ struct ChannelRow {
description: Option<String>,
timezone: String,
schedule_config: String,
recycle_policy: String,
rotation_policy: String,
auto_schedule: i64,
access_mode: String,
access_password_hash: Option<String>,
logo: Option<String>,
logo_position: String,
logo_opacity: f32,
@@ -44,6 +43,7 @@ struct ChannelRow {
webhook_poll_interval_secs: i64,
webhook_body_template: Option<String>,
webhook_headers: Option<String>,
gap_filler: Option<String>,
created_at: String,
updated_at: String,
}
@@ -53,10 +53,15 @@ impl ChannelRow {
let id = ChannelId::from_uuid(parse_uuid(&self.id, "channel id")?);
let owner_id = UserId::from_uuid(parse_uuid(&self.owner_id, "owner id")?);
let schedule_config = parse_schedule_config(&self.schedule_config)?;
let recycle_policy = parse_recycle_policy(&self.recycle_policy)?;
let rotation_policy = parse_rotation_policy(&self.rotation_policy)?;
let access_mode: AccessMode = parse_enum_or_default(self.access_mode);
let logo_position: LogoPosition = parse_enum_or_default(self.logo_position);
let gap_filler: Option<MediaFilter> = self
.gap_filler
.as_deref()
.and_then(|s| serde_json::from_str(s).ok());
Ok(Channel::from_persistence(DomainChannelRow {
id,
owner_id,
@@ -64,10 +69,9 @@ impl ChannelRow {
description: self.description,
timezone: self.timezone,
schedule_config,
recycle_policy,
rotation_policy,
auto_schedule: self.auto_schedule != 0,
access_mode,
access_password_hash: self.access_password_hash,
logo: self.logo,
logo_position,
logo_opacity: self.logo_opacity,
@@ -75,6 +79,7 @@ impl ChannelRow {
webhook_poll_interval_secs: self.webhook_poll_interval_secs as u32,
webhook_body_template: self.webhook_body_template,
webhook_headers: self.webhook_headers,
gap_filler,
created_at: parse_dt(&self.created_at)?,
updated_at: parse_dt(&self.updated_at)?,
}))
@@ -109,28 +114,31 @@ impl ChannelCommand for SqliteChannelRepository {
async fn save(&self, channel: &Channel) -> DomainResult<()> {
let schedule_config = serde_json::to_string(channel.schedule_config())
.map_err(|e| DomainError::RepositoryError(format!("serialize schedule_config: {e}")))?;
let recycle_policy = serde_json::to_string(channel.recycle_policy())
.map_err(|e| DomainError::RepositoryError(format!("serialize recycle_policy: {e}")))?;
let rotation_policy = serde_json::to_string(channel.rotation_policy())
.map_err(|e| DomainError::RepositoryError(format!("serialize rotation_policy: {e}")))?;
let access_mode = serialize_enum_as_string(channel.access_mode(), "public");
let logo_position = serialize_enum_as_string(channel.logo_position(), "top_right");
let gap_filler_json = channel
.gap_filler()
.map(|f| serde_json::to_string(f).unwrap_or_default());
sqlx::query(
r#"
INSERT INTO channels
(id, owner_id, name, description, timezone, schedule_config, recycle_policy,
auto_schedule, access_mode, access_password_hash, logo, logo_position,
(id, owner_id, name, description, timezone, schedule_config, rotation_policy,
auto_schedule, access_mode, logo, logo_position,
logo_opacity, webhook_url, webhook_poll_interval_secs, webhook_body_template,
webhook_headers, created_at, updated_at)
webhook_headers, gap_filler, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
name = excluded.name,
description = excluded.description,
timezone = excluded.timezone,
schedule_config = excluded.schedule_config,
recycle_policy = excluded.recycle_policy,
rotation_policy = excluded.rotation_policy,
auto_schedule = excluded.auto_schedule,
access_mode = excluded.access_mode,
access_password_hash = excluded.access_password_hash,
logo = excluded.logo,
logo_position = excluded.logo_position,
logo_opacity = excluded.logo_opacity,
@@ -138,6 +146,7 @@ impl ChannelCommand for SqliteChannelRepository {
webhook_poll_interval_secs = excluded.webhook_poll_interval_secs,
webhook_body_template = excluded.webhook_body_template,
webhook_headers = excluded.webhook_headers,
gap_filler = excluded.gap_filler,
updated_at = excluded.updated_at
"#,
)
@@ -147,10 +156,9 @@ impl ChannelCommand for SqliteChannelRepository {
.bind(channel.description())
.bind(channel.timezone())
.bind(&schedule_config)
.bind(&recycle_policy)
.bind(&rotation_policy)
.bind(channel.auto_schedule() as i64)
.bind(&access_mode)
.bind(channel.access_password_hash())
.bind(channel.logo())
.bind(&logo_position)
.bind(channel.logo_opacity())
@@ -158,6 +166,7 @@ impl ChannelCommand for SqliteChannelRepository {
.bind(channel.webhook_poll_interval_secs() as i64)
.bind(channel.webhook_body_template())
.bind(channel.webhook_headers())
.bind(&gap_filler_json)
.bind(channel.created_at().to_rfc3339())
.bind(channel.updated_at().to_rfc3339())
.execute(&self.pool)

View File

@@ -1,12 +1,13 @@
use async_trait::async_trait;
use sqlx::SqlitePool;
use adapter_common::{content_type_str, parse_content_type, parse_genres_blob};
use adapter_common::{content_type_str, parse_content_type, parse_enum_or_default, parse_genres_blob, serialize_enum_as_string};
use domain::{
ports::library::{LibraryCommand, LibraryQuery},
ContentType, DomainError, DomainResult, LibraryCollection, LibraryItem,
LibraryItemRow as DomainLibraryItemRow, LibrarySearchFilter, LibrarySyncLogEntry,
LibrarySyncResult, SeasonSummary, ShowSummary,
ContentType, DomainError, DomainResult, LibraryCollection,
LibrarySearchFilter, LibrarySyncLogEntry,
LibrarySyncResult, MediaItem, MediaItemRow as DomainMediaItemRow,
MediaRole, SeasonSummary, ShowSummary,
};
pub struct SqliteLibraryRepository {
@@ -38,17 +39,24 @@ struct LibraryItemRow {
collection_type: Option<String>,
thumbnail_url: Option<String>,
synced_at: String,
chapters: Option<String>,
role: Option<String>,
}
impl LibraryItemRow {
fn into_library_item(self) -> LibraryItem {
LibraryItem::from_persistence(DomainLibraryItemRow {
id: self.id,
fn into_media_item(self) -> MediaItem {
let role: MediaRole = self
.role
.map(parse_enum_or_default)
.unwrap_or_default();
MediaItem::from_persistence(DomainMediaItemRow {
id: domain::MediaItemId::new(&self.id),
provider_id: self.provider_id,
external_id: self.external_id,
title: self.title,
content_type: parse_content_type(&self.content_type),
duration_secs: self.duration_secs as u32,
description: None,
series_name: self.series_name,
season_number: self.season_number.map(|n| n as u32),
episode_number: self.episode_number.map(|n| n as u32),
@@ -59,7 +67,13 @@ impl LibraryItemRow {
collection_name: self.collection_name,
collection_type: self.collection_type,
thumbnail_url: self.thumbnail_url,
synced_at: self.synced_at,
synced_at: Some(self.synced_at),
role,
chapters: self
.chapters
.as_deref()
.and_then(|s| serde_json::from_str(s).ok())
.unwrap_or_default(),
})
}
}
@@ -93,7 +107,7 @@ struct SeasonSummaryRow {
#[async_trait]
impl LibraryCommand for SqliteLibraryRepository {
async fn upsert_items(&self, _provider_id: &str, items: Vec<LibraryItem>) -> DomainResult<()> {
async fn upsert_items(&self, _provider_id: &str, items: Vec<MediaItem>) -> DomainResult<()> {
let mut tx = self
.pool
.begin()
@@ -101,14 +115,22 @@ impl LibraryCommand for SqliteLibraryRepository {
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
for item in items {
let chapters_json = if item.chapters().is_empty() {
None
} else {
Some(serde_json::to_string(item.chapters()).unwrap_or_default())
};
let role_str = serialize_enum_as_string(item.role(), "program");
sqlx::query(
"INSERT OR REPLACE INTO library_items
(id, provider_id, external_id, title, content_type, duration_secs,
series_name, season_number, episode_number, year, genres, tags,
collection_id, collection_name, collection_type, thumbnail_url, synced_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
collection_id, collection_name, collection_type, thumbnail_url, synced_at, chapters, role)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
)
.bind(item.id())
.bind(item.id().value())
.bind(item.provider_id())
.bind(item.external_id())
.bind(item.title())
@@ -124,7 +146,9 @@ impl LibraryCommand for SqliteLibraryRepository {
.bind(item.collection_name())
.bind(item.collection_type())
.bind(item.thumbnail_url())
.bind(item.synced_at())
.bind(item.synced_at().unwrap_or(""))
.bind(&chapters_json)
.bind(&role_str)
.execute(&mut *tx)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
@@ -135,6 +159,22 @@ impl LibraryCommand for SqliteLibraryRepository {
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
async fn update_role(&self, item_id: &str, role: MediaRole) -> DomainResult<()> {
let role_str = serialize_enum_as_string(&role, "program");
let rows = sqlx::query("UPDATE library_items SET role = ? WHERE id = ?")
.bind(&role_str)
.bind(item_id)
.execute(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
if rows.rows_affected() == 0 {
return Err(DomainError::NotFound(format!(
"Library item {item_id} not found"
)));
}
Ok(())
}
async fn clear_provider(&self, provider_id: &str) -> DomainResult<()> {
sqlx::query("DELETE FROM library_items WHERE provider_id = ?")
.bind(provider_id)
@@ -187,7 +227,7 @@ impl LibraryQuery for SqliteLibraryRepository {
async fn search(
&self,
filter: &LibrarySearchFilter,
) -> DomainResult<(Vec<LibraryItem>, u32)> {
) -> DomainResult<(Vec<MediaItem>, u32)> {
let mut conditions: Vec<String> = vec![];
if let Some(p) = filter.provider_id() {
@@ -233,6 +273,19 @@ impl LibraryQuery for SqliteLibraryRepository {
.collect();
conditions.push(format!("({})", genre_conditions.join(" OR ")));
}
if !filter.tags().is_empty() {
let tag_conditions: Vec<String> = filter
.tags()
.iter()
.map(|t| {
format!(
"EXISTS (SELECT 1 FROM json_each(library_items.tags) WHERE LOWER(value) = LOWER('{}'))",
t.replace('\'', "''")
)
})
.collect();
conditions.push(format!("({})", tag_conditions.join(" OR ")));
}
if let Some(sn) = filter.season_number() {
conditions.push(format!("season_number = {}", sn));
}
@@ -263,19 +316,19 @@ impl LibraryQuery for SqliteLibraryRepository {
Ok((
rows.into_iter()
.map(LibraryItemRow::into_library_item)
.map(LibraryItemRow::into_media_item)
.collect(),
total as u32,
))
}
async fn get_by_id(&self, id: &str) -> DomainResult<Option<LibraryItem>> {
async fn get_by_id(&self, id: &str) -> DomainResult<Option<MediaItem>> {
let row = sqlx::query_as::<_, LibraryItemRow>("SELECT * FROM library_items WHERE id = ?")
.bind(id)
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(row.map(LibraryItemRow::into_library_item))
Ok(row.map(LibraryItemRow::into_media_item))
}
async fn list_collections(

View File

@@ -1,6 +1,6 @@
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
@@ -17,7 +17,7 @@ pub struct ActivityEventResponse {
pub channel_id: Option<Uuid>,
}
#[derive(Debug, Deserialize, ToSchema)]
#[derive(Debug, Deserialize, ToSchema, IntoParams)]
pub struct ActivityLogParams {
pub limit: Option<u32>,
}

View File

@@ -11,7 +11,6 @@ pub struct CreateChannelRequest {
pub description: Option<String>,
pub timezone: String,
pub access_mode: Option<String>,
pub access_password: Option<String>,
pub webhook_url: Option<String>,
pub webhook_poll_interval_secs: Option<u32>,
pub webhook_body_template: Option<String>,
@@ -26,10 +25,9 @@ pub struct UpdateChannelRequest {
#[schema(value_type = Option<Object>)]
pub schedule_config: Option<domain::models::ScheduleConfigCompat>,
#[schema(value_type = Option<Object>)]
pub recycle_policy: Option<domain::RecyclePolicy>,
pub rotation_policy: Option<domain::RotationPolicy>,
pub auto_schedule: Option<bool>,
pub access_mode: Option<String>,
pub access_password: Option<String>,
pub logo: Option<Option<String>>,
pub logo_position: Option<String>,
pub logo_opacity: Option<f32>,
@@ -37,6 +35,8 @@ pub struct UpdateChannelRequest {
pub webhook_poll_interval_secs: Option<u32>,
pub webhook_body_template: Option<Option<String>>,
pub webhook_headers: Option<Option<String>>,
#[schema(value_type = Option<Option<Object>>)]
pub gap_filler: Option<Option<domain::MediaFilter>>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
@@ -47,7 +47,7 @@ pub struct ChannelResponse {
pub description: Option<String>,
pub timezone: String,
pub schedule_config: serde_json::Value,
pub recycle_policy: serde_json::Value,
pub rotation_policy: serde_json::Value,
pub auto_schedule: bool,
pub access_mode: String,
pub logo: Option<String>,
@@ -57,6 +57,8 @@ pub struct ChannelResponse {
pub webhook_poll_interval_secs: u32,
pub webhook_body_template: Option<String>,
pub webhook_headers: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub gap_filler: Option<serde_json::Value>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
@@ -70,7 +72,7 @@ impl From<domain::Channel> for ChannelResponse {
description: c.description().map(|s| s.to_string()),
timezone: c.timezone().to_string(),
schedule_config: serde_json::to_value(c.schedule_config()).unwrap_or_default(),
recycle_policy: serde_json::to_value(c.recycle_policy()).unwrap_or_default(),
rotation_policy: serde_json::to_value(c.rotation_policy()).unwrap_or_default(),
auto_schedule: c.auto_schedule(),
access_mode: enum_to_string(c.access_mode()),
logo: c.logo().map(|s| s.to_string()),
@@ -80,6 +82,7 @@ impl From<domain::Channel> for ChannelResponse {
webhook_poll_interval_secs: c.webhook_poll_interval_secs(),
webhook_body_template: c.webhook_body_template().map(|s| s.to_string()),
webhook_headers: c.webhook_headers().map(|s| s.to_string()),
gap_filler: c.gap_filler().map(|f| serde_json::to_value(f).unwrap_or_default()),
created_at: c.created_at(),
updated_at: c.updated_at(),
}

View File

@@ -1,8 +1,6 @@
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
use crate::common::enum_to_string;
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct ProviderCapabilitiesResponse {
pub collections: bool,
@@ -11,9 +9,7 @@ pub struct ProviderCapabilitiesResponse {
pub tags: bool,
pub decade: bool,
pub search: bool,
pub streaming_protocol: String,
pub rescan: bool,
pub transcode: bool,
}
impl From<domain::ports::ProviderCapabilities> for ProviderCapabilitiesResponse {
@@ -25,9 +21,7 @@ impl From<domain::ports::ProviderCapabilities> for ProviderCapabilitiesResponse
tags: c.tags,
decade: c.decade,
search: c.search,
streaming_protocol: enum_to_string(&c.streaming_protocol),
rescan: c.rescan,
transcode: c.transcode,
}
}
}

View File

@@ -1,7 +1,7 @@
use serde::Deserialize;
use utoipa::ToSchema;
use utoipa::{IntoParams, ToSchema};
#[derive(Debug, Deserialize, ToSchema)]
#[derive(Debug, Deserialize, ToSchema, IntoParams)]
pub struct IptvParams {
pub token: Option<String>,
}

View File

@@ -20,7 +20,7 @@ pub use config::{ConfigResponse, ProviderCapabilitiesResponse, ProviderInfo};
pub use iptv::IptvParams;
pub use library::{
CollectionResponse, GenresParams, LibraryItemResponse, LibrarySearchParams, ProviderParam,
SeasonResponse, SeasonsParams, ShowResponse, ShowsParams, SyncStatusEntry,
SeasonResponse, SeasonsParams, ShowResponse, ShowsParams, SyncStatusEntry, UpdateRoleRequest,
};
pub use providers::{ProviderConfigRequest, ProviderConfigResponse};
pub use schedule::{

View File

@@ -1,5 +1,5 @@
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
use utoipa::{IntoParams, ToSchema};
use crate::common::enum_to_string;
@@ -21,13 +21,14 @@ pub struct LibraryItemResponse {
pub collection_name: Option<String>,
pub collection_type: Option<String>,
pub thumbnail_url: Option<String>,
pub synced_at: String,
pub synced_at: Option<String>,
pub role: String,
}
impl From<domain::LibraryItem> for LibraryItemResponse {
fn from(i: domain::LibraryItem) -> Self {
impl From<domain::MediaItem> for LibraryItemResponse {
fn from(i: domain::MediaItem) -> Self {
Self {
id: i.id().to_string(),
id: i.id().value().to_string(),
provider_id: i.provider_id().to_string(),
external_id: i.external_id().to_string(),
title: i.title().to_string(),
@@ -43,7 +44,8 @@ impl From<domain::LibraryItem> for LibraryItemResponse {
collection_name: i.collection_name().map(|s| s.to_string()),
collection_type: i.collection_type().map(|s| s.to_string()),
thumbnail_url: i.thumbnail_url().map(|s| s.to_string()),
synced_at: i.synced_at().to_string(),
synced_at: i.synced_at().map(|s| s.to_string()),
role: enum_to_string(i.role()),
}
}
}
@@ -126,7 +128,7 @@ impl From<domain::LibrarySyncLogEntry> for SyncStatusEntry {
}
}
#[derive(Debug, Deserialize, ToSchema)]
#[derive(Debug, Deserialize, ToSchema, IntoParams)]
pub struct LibrarySearchParams {
pub provider: Option<String>,
pub content_type: Option<String>,
@@ -142,12 +144,12 @@ pub struct LibrarySearchParams {
pub limit: Option<u32>,
}
#[derive(Debug, Deserialize, ToSchema)]
#[derive(Debug, Deserialize, ToSchema, IntoParams)]
pub struct ProviderParam {
pub provider: Option<String>,
}
#[derive(Debug, Deserialize, ToSchema)]
#[derive(Debug, Deserialize, ToSchema, IntoParams)]
pub struct ShowsParams {
pub provider: Option<String>,
pub search_term: Option<String>,
@@ -155,14 +157,19 @@ pub struct ShowsParams {
pub genres: Vec<String>,
}
#[derive(Debug, Deserialize, ToSchema)]
#[derive(Debug, Deserialize, ToSchema, IntoParams)]
pub struct SeasonsParams {
pub series_name: String,
pub provider: Option<String>,
}
#[derive(Debug, Deserialize, ToSchema)]
#[derive(Debug, Deserialize, ToSchema, IntoParams)]
pub struct GenresParams {
pub content_type: Option<String>,
pub provider: Option<String>,
}
#[derive(Debug, Deserialize, ToSchema)]
pub struct UpdateRoleRequest {
pub role: String,
}

View File

@@ -5,8 +5,6 @@ use uuid::Uuid;
use crate::common::enum_to_string;
const DEFAULT_ACCESS_MODE: &str = "public";
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct MediaItemResponse {
pub id: String,
@@ -47,8 +45,6 @@ pub struct SlotResponse {
pub end_at: DateTime<Utc>,
pub item: MediaItemResponse,
pub source_block_id: Uuid,
#[serde(default)]
pub block_access_mode: String,
}
impl From<domain::ScheduledSlot> for SlotResponse {
@@ -59,26 +55,6 @@ impl From<domain::ScheduledSlot> for SlotResponse {
end_at: s.end_at(),
item: s.item().clone().into(),
source_block_id: s.source_block_id().value(),
block_access_mode: String::from(DEFAULT_ACCESS_MODE),
}
}
}
impl SlotResponse {
pub fn with_block_access(slot: domain::ScheduledSlot, channel: &domain::Channel) -> Self {
let block_access_mode = channel
.schedule_config()
.all_blocks()
.find(|b| b.id() == slot.source_block_id())
.map(|b| enum_to_string(b.access_mode()))
.unwrap_or_else(|| String::from(DEFAULT_ACCESS_MODE));
Self {
id: slot.id().value(),
start_at: slot.start_at(),
end_at: slot.end_at(),
item: slot.item().clone().into(),
source_block_id: slot.source_block_id().value(),
block_access_mode,
}
}
}
@@ -87,7 +63,6 @@ impl SlotResponse {
pub struct CurrentBroadcastResponse {
pub slot: SlotResponse,
pub offset_secs: u32,
pub block_access_mode: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

View File

@@ -1,12 +0,0 @@
use domain::models::ActivityEvent;
use domain::DomainResult;
use super::deps::AdminDeps;
use super::queries::GetActivityLogQuery;
pub async fn execute(
deps: &AdminDeps,
query: GetActivityLogQuery,
) -> DomainResult<Vec<ActivityEvent>> {
deps.activity_query.recent(query.limit).await
}

View File

@@ -1,15 +0,0 @@
use domain::DomainResult;
use super::deps::AdminDeps;
use super::queries::GetSettingsQuery;
pub async fn execute(
deps: &AdminDeps,
_query: GetSettingsQuery,
) -> DomainResult<Vec<(String, String)>> {
deps.settings_repo.get_all().await
}
#[cfg(test)]
#[path = "tests/get_settings.rs"]
mod tests;

View File

@@ -1,10 +1,6 @@
pub mod activity_log;
pub mod commands;
pub mod deps;
pub mod get_settings;
pub mod queries;
pub mod update_settings;
pub use commands::UpdateSettingsCommand;
pub use deps::AdminDeps;
pub use queries::{GetActivityLogQuery, GetSettingsQuery};

View File

@@ -1,5 +0,0 @@
pub struct GetSettingsQuery;
pub struct GetActivityLogQuery {
pub limit: u32,
}

View File

@@ -1,46 +0,0 @@
use std::sync::Arc;
use domain::testing::{InMemoryActivityLog, InMemoryAppSettings};
use crate::admin::commands::UpdateSettingsCommand;
use crate::admin::deps::AdminDeps;
use crate::admin::queries::GetSettingsQuery;
use crate::admin::{get_settings, update_settings};
fn make_deps() -> AdminDeps {
AdminDeps {
settings_repo: Arc::new(InMemoryAppSettings::new()),
activity_query: Arc::new(InMemoryActivityLog::new()),
}
}
#[tokio::test]
async fn get_empty_settings() {
let deps = make_deps();
let settings = get_settings::execute(&deps, GetSettingsQuery).await.unwrap();
assert!(settings.is_empty());
}
#[tokio::test]
async fn get_returns_stored_settings() {
let deps = make_deps();
update_settings::execute(
&deps,
UpdateSettingsCommand {
settings: vec![
("a".into(), "1".into()),
("b".into(), "2".into()),
],
},
)
.await
.unwrap();
let settings = get_settings::execute(&deps, GetSettingsQuery).await.unwrap();
assert_eq!(settings.len(), 2);
let keys: Vec<&str> = settings.iter().map(|(k, _)| k.as_str()).collect();
assert!(keys.contains(&"a"));
assert!(keys.contains(&"b"));
}

View File

@@ -121,10 +121,10 @@ async fn login_fails_for_unknown_email() {
}
#[tokio::test]
async fn login_fails_for_oidc_only_user() {
async fn login_fails_for_user_without_password() {
let repo = Arc::new(InMemoryUserRepository::new());
let email = Email::new("oidc@example.com").unwrap();
let user = domain::models::User::new("oidc|subject", email);
let email = Email::new("external@example.com").unwrap();
let user = domain::models::User::new("external|subject", email);
repo.store.lock().unwrap().insert(user.id(), user);
let deps = AuthDeps {
@@ -138,7 +138,7 @@ async fn login_fails_for_oidc_only_user() {
let result = login::execute(
&deps,
LoginCommand {
email: "oidc@example.com".into(),
email: "external@example.com".into(),
password: "password123".into(),
remember_me: false,
},

View File

@@ -1,5 +1,5 @@
use domain::models::ScheduleConfig;
use domain::value_objects::{ChannelId, RecyclePolicy, UserId};
use domain::value_objects::{ChannelId, MediaFilter, RotationPolicy, UserId};
pub struct CreateChannelCommand {
pub owner_id: UserId,
@@ -14,8 +14,9 @@ pub struct UpdateChannelCommand {
pub description: Option<Option<String>>,
pub timezone: Option<String>,
pub schedule_config: Option<ScheduleConfig>,
pub recycle_policy: Option<RecyclePolicy>,
pub rotation_policy: Option<RotationPolicy>,
pub auto_schedule: Option<bool>,
pub gap_filler: Option<Option<MediaFilter>>,
}
pub struct DeleteChannelCommand {

View File

@@ -7,7 +7,3 @@ pub struct ChannelCommandDeps {
pub channel_query: Arc<dyn ChannelQuery>,
pub event_publisher: Arc<dyn EventPublisher>,
}
pub struct ChannelQueryDeps {
pub channel_query: Arc<dyn ChannelQuery>,
}

View File

@@ -1,13 +0,0 @@
use domain::models::Channel;
use domain::DomainResult;
use super::deps::ChannelQueryDeps;
use super::queries::GetChannelQuery;
pub async fn execute(deps: &ChannelQueryDeps, query: GetChannelQuery) -> DomainResult<Option<Channel>> {
deps.channel_query.find_by_id(query.channel_id).await
}
#[cfg(test)]
#[path = "tests/get.rs"]
mod tests;

View File

@@ -1,13 +0,0 @@
use domain::models::Channel;
use domain::DomainResult;
use super::deps::ChannelQueryDeps;
use super::queries::ListChannelsQuery;
pub async fn execute(deps: &ChannelQueryDeps, _query: ListChannelsQuery) -> DomainResult<Vec<Channel>> {
deps.channel_query.find_all().await
}
#[cfg(test)]
#[path = "tests/list.rs"]
mod tests;

View File

@@ -1,13 +0,0 @@
use domain::models::Channel;
use domain::DomainResult;
use super::deps::ChannelQueryDeps;
use super::queries::ListByOwnerQuery;
pub async fn execute(deps: &ChannelQueryDeps, query: ListByOwnerQuery) -> DomainResult<Vec<Channel>> {
deps.channel_query.find_by_owner(query.owner_id).await
}
#[cfg(test)]
#[path = "tests/list_by_owner.rs"]
mod tests;

View File

@@ -2,15 +2,10 @@ pub mod commands;
pub mod create;
pub mod delete;
pub mod deps;
pub mod get;
pub mod list;
pub mod list_by_owner;
pub mod queries;
pub mod update;
pub use commands::{CreateChannelCommand, DeleteChannelCommand, UpdateChannelCommand};
pub use deps::{ChannelCommandDeps, ChannelQueryDeps};
pub use queries::{GetChannelQuery, ListByOwnerQuery, ListChannelsQuery};
pub use deps::ChannelCommandDeps;
use domain::models::Channel;
use domain::value_objects::{ChannelId, UserId};

View File

@@ -1,11 +0,0 @@
use domain::value_objects::{ChannelId, UserId};
pub struct GetChannelQuery {
pub channel_id: ChannelId,
}
pub struct ListChannelsQuery;
pub struct ListByOwnerQuery {
pub owner_id: UserId,
}

View File

@@ -1,66 +0,0 @@
use std::sync::Arc;
use domain::testing::{InMemoryChannelRepository, NoopEventPublisher};
use domain::value_objects::{ChannelId, UserId};
use crate::channels::commands::CreateChannelCommand;
use crate::channels::deps::{ChannelCommandDeps, ChannelQueryDeps};
use crate::channels::queries::GetChannelQuery;
use crate::channels::{create, get};
fn make_deps() -> (ChannelCommandDeps, ChannelQueryDeps) {
let repo = Arc::new(InMemoryChannelRepository::new());
let cmd_deps = ChannelCommandDeps {
channel_command: repo.clone(),
channel_query: repo.clone(),
event_publisher: Arc::new(NoopEventPublisher::new()),
};
let query_deps = ChannelQueryDeps {
channel_query: repo,
};
(cmd_deps, query_deps)
}
#[tokio::test]
async fn get_existing_channel() {
let (cmd_deps, query_deps) = make_deps();
let channel = create::execute(
&cmd_deps,
CreateChannelCommand {
owner_id: UserId::generate(),
name: "Findable".into(),
timezone: "UTC".into(),
},
)
.await
.unwrap();
let found = get::execute(
&query_deps,
GetChannelQuery {
channel_id: channel.id(),
},
)
.await
.unwrap();
assert!(found.is_some());
assert_eq!(found.unwrap().name(), "Findable");
}
#[tokio::test]
async fn get_nonexistent_returns_none() {
let (_, query_deps) = make_deps();
let found = get::execute(
&query_deps,
GetChannelQuery {
channel_id: ChannelId::generate(),
},
)
.await
.unwrap();
assert!(found.is_none());
}

View File

@@ -1,51 +0,0 @@
use std::sync::Arc;
use domain::testing::{InMemoryChannelRepository, NoopEventPublisher};
use domain::value_objects::UserId;
use crate::channels::commands::CreateChannelCommand;
use crate::channels::deps::{ChannelCommandDeps, ChannelQueryDeps};
use crate::channels::queries::ListChannelsQuery;
use crate::channels::{create, list};
fn make_deps() -> (ChannelCommandDeps, ChannelQueryDeps) {
let repo = Arc::new(InMemoryChannelRepository::new());
let cmd_deps = ChannelCommandDeps {
channel_command: repo.clone(),
channel_query: repo.clone(),
event_publisher: Arc::new(NoopEventPublisher::new()),
};
let query_deps = ChannelQueryDeps {
channel_query: repo,
};
(cmd_deps, query_deps)
}
#[tokio::test]
async fn list_empty_returns_empty() {
let (_, query_deps) = make_deps();
let channels = list::execute(&query_deps, ListChannelsQuery).await.unwrap();
assert!(channels.is_empty());
}
#[tokio::test]
async fn list_returns_all_channels() {
let (cmd_deps, query_deps) = make_deps();
for name in ["A", "B", "C"] {
create::execute(
&cmd_deps,
CreateChannelCommand {
owner_id: UserId::generate(),
name: name.into(),
timezone: "UTC".into(),
},
)
.await
.unwrap();
}
let channels = list::execute(&query_deps, ListChannelsQuery).await.unwrap();
assert_eq!(channels.len(), 3);
}

View File

@@ -1,83 +0,0 @@
use std::sync::Arc;
use domain::testing::{InMemoryChannelRepository, NoopEventPublisher};
use domain::value_objects::UserId;
use crate::channels::commands::CreateChannelCommand;
use crate::channels::deps::{ChannelCommandDeps, ChannelQueryDeps};
use crate::channels::queries::ListByOwnerQuery;
use crate::channels::{create, list_by_owner};
fn make_deps() -> (ChannelCommandDeps, ChannelQueryDeps) {
let repo = Arc::new(InMemoryChannelRepository::new());
let cmd_deps = ChannelCommandDeps {
channel_command: repo.clone(),
channel_query: repo.clone(),
event_publisher: Arc::new(NoopEventPublisher::new()),
};
let query_deps = ChannelQueryDeps {
channel_query: repo,
};
(cmd_deps, query_deps)
}
#[tokio::test]
async fn filters_by_owner() {
let (cmd_deps, query_deps) = make_deps();
let alice = UserId::generate();
let bob = UserId::generate();
// Alice: 2 channels
for name in ["Alice-1", "Alice-2"] {
create::execute(
&cmd_deps,
CreateChannelCommand {
owner_id: alice,
name: name.into(),
timezone: "UTC".into(),
},
)
.await
.unwrap();
}
// Bob: 1 channel
create::execute(
&cmd_deps,
CreateChannelCommand {
owner_id: bob,
name: "Bob-1".into(),
timezone: "UTC".into(),
},
)
.await
.unwrap();
let alice_channels = list_by_owner::execute(
&query_deps,
ListByOwnerQuery {
owner_id: alice,
},
)
.await
.unwrap();
assert_eq!(alice_channels.len(), 2);
assert!(alice_channels.iter().all(|c| c.owner_id() == alice));
}
#[tokio::test]
async fn no_channels_returns_empty() {
let (_, query_deps) = make_deps();
let channels = list_by_owner::execute(
&query_deps,
ListByOwnerQuery {
owner_id: UserId::generate(),
},
)
.await
.unwrap();
assert!(channels.is_empty());
}

View File

@@ -43,8 +43,9 @@ async fn updates_channel_name() {
description: None,
timezone: None,
schedule_config: None,
recycle_policy: None,
rotation_policy: None,
auto_schedule: None,
gap_filler: None,
},
)
.await
@@ -80,8 +81,9 @@ async fn update_fails_if_not_owner() {
description: None,
timezone: None,
schedule_config: None,
recycle_policy: None,
rotation_policy: None,
auto_schedule: None,
gap_filler: None,
},
)
.await;
@@ -106,8 +108,9 @@ async fn update_nonexistent_channel_returns_not_found() {
description: None,
timezone: None,
schedule_config: None,
recycle_policy: None,
rotation_policy: None,
auto_schedule: None,
gap_filler: None,
},
)
.await;
@@ -146,8 +149,9 @@ async fn update_config_creates_snapshot() {
description: None,
timezone: None,
schedule_config: Some(new_config),
recycle_policy: None,
rotation_policy: None,
auto_schedule: None,
gap_filler: None,
},
)
.await
@@ -185,8 +189,9 @@ async fn update_without_config_skips_snapshot() {
description: None,
timezone: None,
schedule_config: None,
recycle_policy: None,
rotation_policy: None,
auto_schedule: None,
gap_filler: None,
},
)
.await
@@ -223,8 +228,9 @@ async fn update_description_clear() {
description: Some(Some("A description".into())),
timezone: None,
schedule_config: None,
recycle_policy: None,
rotation_policy: None,
auto_schedule: None,
gap_filler: None,
},
)
.await
@@ -241,8 +247,9 @@ async fn update_description_clear() {
description: Some(None),
timezone: None,
schedule_config: None,
recycle_policy: None,
rotation_policy: None,
auto_schedule: None,
gap_filler: None,
},
)
.await

View File

@@ -29,12 +29,15 @@ pub async fn execute(deps: &ChannelCommandDeps, cmd: UpdateChannelCommand) -> Do
if let Some(config) = cmd.schedule_config {
channel.set_schedule_config(config);
}
if let Some(policy) = cmd.recycle_policy {
channel.set_recycle_policy(policy);
if let Some(policy) = cmd.rotation_policy {
channel.set_rotation_policy(policy);
}
if let Some(auto) = cmd.auto_schedule {
channel.set_auto_schedule(auto);
}
if let Some(gap_filler) = cmd.gap_filler {
channel.set_gap_filler(gap_filler);
}
deps.channel_command.save(&channel).await?;

View File

@@ -41,9 +41,7 @@ pub fn execute(deps: &ConfigDeps, _query: GetConfigQuery) -> SystemConfig {
tags: false,
decade: false,
search: false,
streaming_protocol: domain::ports::StreamingProtocol::DirectFile,
rescan: false,
transcode: false,
});
SystemConfig {

View File

@@ -5,12 +5,6 @@ pub struct SaveSnapshotCommand {
pub label: Option<String>,
}
pub struct PatchLabelCommand {
pub channel_id: ChannelId,
pub snapshot_id: SnapshotId,
pub label: Option<String>,
}
pub struct RestoreSnapshotCommand {
pub channel_id: ChannelId,
pub snapshot_id: SnapshotId,

View File

@@ -1,14 +0,0 @@
use domain::models::ChannelConfigSnapshot;
use domain::DomainResult;
use super::deps::ConfigSnapshotDeps;
use super::queries::GetSnapshotQuery;
pub async fn execute(
deps: &ConfigSnapshotDeps,
query: GetSnapshotQuery,
) -> DomainResult<Option<ChannelConfigSnapshot>> {
deps.channel_query
.get_config_snapshot(query.channel_id, query.snapshot_id)
.await
}

View File

@@ -1,16 +0,0 @@
use domain::models::ChannelConfigSnapshot;
use domain::DomainResult;
use super::deps::ConfigSnapshotDeps;
use super::queries::ListSnapshotsQuery;
pub async fn execute(
deps: &ConfigSnapshotDeps,
query: ListSnapshotsQuery,
) -> DomainResult<Vec<ChannelConfigSnapshot>> {
deps.channel_query.list_config_snapshots(query.channel_id).await
}
#[cfg(test)]
#[path = "tests/list.rs"]
mod tests;

View File

@@ -1,12 +1,7 @@
pub mod commands;
pub mod deps;
pub mod get;
pub mod list;
pub mod patch_label;
pub mod queries;
pub mod restore;
pub mod save;
pub use commands::{PatchLabelCommand, RestoreSnapshotCommand, SaveSnapshotCommand};
pub use commands::{RestoreSnapshotCommand, SaveSnapshotCommand};
pub use deps::ConfigSnapshotDeps;
pub use queries::{GetSnapshotQuery, ListSnapshotsQuery};

View File

@@ -1,14 +0,0 @@
use domain::models::ChannelConfigSnapshot;
use domain::DomainResult;
use super::commands::PatchLabelCommand;
use super::deps::ConfigSnapshotDeps;
pub async fn execute(
deps: &ConfigSnapshotDeps,
cmd: PatchLabelCommand,
) -> DomainResult<Option<ChannelConfigSnapshot>> {
deps.channel_command
.patch_config_snapshot_label(cmd.channel_id, cmd.snapshot_id, cmd.label)
.await
}

View File

@@ -1,10 +0,0 @@
use domain::value_objects::{ChannelId, SnapshotId};
pub struct ListSnapshotsQuery {
pub channel_id: ChannelId,
}
pub struct GetSnapshotQuery {
pub channel_id: ChannelId,
pub snapshot_id: SnapshotId,
}

View File

@@ -1,77 +0,0 @@
use std::sync::Arc;
use domain::models::Channel;
use domain::testing::InMemoryChannelRepository;
use domain::value_objects::UserId;
use crate::config_snapshots::commands::SaveSnapshotCommand;
use crate::config_snapshots::deps::ConfigSnapshotDeps;
use crate::config_snapshots::queries::ListSnapshotsQuery;
use crate::config_snapshots::{list, save};
fn make_deps() -> (ConfigSnapshotDeps, Arc<InMemoryChannelRepository>) {
let repo = Arc::new(InMemoryChannelRepository::new());
let deps = ConfigSnapshotDeps {
channel_command: repo.clone(),
channel_query: repo.clone(),
};
(deps, repo)
}
async fn seed_channel(repo: &InMemoryChannelRepository) -> Channel {
let channel = Channel::new(UserId::generate(), "Test Channel", "UTC");
repo.channels
.lock()
.unwrap()
.insert(channel.id(), channel.clone());
channel
}
#[tokio::test]
async fn list_empty() {
let (deps, repo) = make_deps();
let channel = seed_channel(&repo).await;
let snaps = list::execute(
&deps,
ListSnapshotsQuery {
channel_id: channel.id(),
},
)
.await
.unwrap();
assert!(snaps.is_empty());
}
#[tokio::test]
async fn list_returns_saved_snapshots() {
let (deps, repo) = make_deps();
let channel = seed_channel(&repo).await;
for label in ["first", "second"] {
save::execute(
&deps,
SaveSnapshotCommand {
channel_id: channel.id(),
label: Some(label.into()),
},
)
.await
.unwrap();
}
let snaps = list::execute(
&deps,
ListSnapshotsQuery {
channel_id: channel.id(),
},
)
.await
.unwrap();
assert_eq!(snaps.len(), 2);
// Newest first
assert_eq!(snaps[0].version_num(), 2);
assert_eq!(snaps[1].version_num(), 1);
}

View File

@@ -9,7 +9,3 @@ pub struct LibraryCommandDeps {
pub provider_registry: Arc<dyn IProviderRegistry>,
pub event_publisher: Arc<dyn EventPublisher>,
}
pub struct LibraryQueryDeps {
pub library_query: Arc<dyn LibraryQuery>,
}

View File

@@ -1,16 +0,0 @@
use domain::models::LibraryItem;
use domain::DomainResult;
use super::deps::LibraryQueryDeps;
use super::queries::GetItemQuery;
pub async fn execute(
deps: &LibraryQueryDeps,
query: GetItemQuery,
) -> DomainResult<Option<LibraryItem>> {
deps.library_query.get_by_id(&query.item_id).await
}
#[cfg(test)]
#[path = "tests/get_item.rs"]
mod tests;

View File

@@ -1,16 +0,0 @@
use domain::models::LibrarySyncLogEntry;
use domain::DomainResult;
use super::deps::LibraryQueryDeps;
use super::queries::GetSyncStatusQuery;
pub async fn execute(
deps: &LibraryQueryDeps,
_query: GetSyncStatusQuery,
) -> DomainResult<Vec<LibrarySyncLogEntry>> {
deps.library_query.latest_sync_status().await
}
#[cfg(test)]
#[path = "tests/get_sync_status.rs"]
mod tests;

View File

@@ -1,18 +0,0 @@
use domain::models::LibraryCollection;
use domain::DomainResult;
use super::deps::LibraryQueryDeps;
use super::queries::ListCollectionsQuery;
pub async fn execute(
deps: &LibraryQueryDeps,
query: ListCollectionsQuery,
) -> DomainResult<Vec<LibraryCollection>> {
deps.library_query
.list_collections(query.provider_id.as_deref())
.await
}
#[cfg(test)]
#[path = "tests/list_collections.rs"]
mod tests;

View File

@@ -1,21 +0,0 @@
use domain::DomainResult;
use super::deps::LibraryQueryDeps;
use super::parse_content_type;
use super::queries::ListGenresQuery;
pub async fn execute(deps: &LibraryQueryDeps, query: ListGenresQuery) -> DomainResult<Vec<String>> {
let content_type = query
.content_type
.as_deref()
.map(parse_content_type)
.transpose()?;
deps.library_query
.list_genres(content_type.as_ref(), query.provider_id.as_deref())
.await
}
#[cfg(test)]
#[path = "tests/list_genres.rs"]
mod tests;

View File

@@ -1,18 +0,0 @@
use domain::models::SeasonSummary;
use domain::DomainResult;
use super::deps::LibraryQueryDeps;
use super::queries::ListSeasonsQuery;
pub async fn execute(
deps: &LibraryQueryDeps,
query: ListSeasonsQuery,
) -> DomainResult<Vec<SeasonSummary>> {
deps.library_query
.list_seasons(&query.series_name, query.provider_id.as_deref())
.await
}
#[cfg(test)]
#[path = "tests/list_seasons.rs"]
mod tests;

View File

@@ -1,22 +0,0 @@
use domain::models::ShowSummary;
use domain::DomainResult;
use super::deps::LibraryQueryDeps;
use super::queries::ListShowsQuery;
pub async fn execute(
deps: &LibraryQueryDeps,
query: ListShowsQuery,
) -> DomainResult<Vec<ShowSummary>> {
deps.library_query
.list_shows(
query.provider_id.as_deref(),
query.search_term.as_deref(),
&query.genres,
)
.await
}
#[cfg(test)]
#[path = "tests/list_shows.rs"]
mod tests;

View File

@@ -1,26 +1,17 @@
pub mod commands;
pub mod deps;
pub mod get_item;
pub mod get_sync_status;
pub mod list_collections;
pub mod list_genres;
pub mod list_seasons;
pub mod list_shows;
pub mod queries;
pub mod search;
pub mod sync;
pub use commands::TriggerSyncCommand;
pub use deps::{LibraryCommandDeps, LibraryQueryDeps};
pub use queries::{
GetItemQuery, GetSyncStatusQuery, ListCollectionsQuery, ListGenresQuery, ListSeasonsQuery,
ListShowsQuery, SearchItemsQuery,
};
pub use deps::LibraryCommandDeps;
pub use queries::SearchItemsQuery;
use domain::errors::{DomainError, DomainResult};
use domain::value_objects::ContentType;
pub(crate) fn parse_content_type(s: &str) -> DomainResult<ContentType> {
pub fn parse_content_type(s: &str) -> DomainResult<ContentType> {
match s {
"movie" => Ok(ContentType::Movie),
"episode" => Ok(ContentType::Episode),

View File

@@ -10,29 +10,3 @@ pub struct SearchItemsQuery {
pub offset: u32,
pub limit: u32,
}
pub struct ListCollectionsQuery {
pub provider_id: Option<String>,
}
pub struct ListShowsQuery {
pub provider_id: Option<String>,
pub search_term: Option<String>,
pub genres: Vec<String>,
}
pub struct ListSeasonsQuery {
pub series_name: String,
pub provider_id: Option<String>,
}
pub struct ListGenresQuery {
pub content_type: Option<String>,
pub provider_id: Option<String>,
}
pub struct GetItemQuery {
pub item_id: String,
}
pub struct GetSyncStatusQuery;

View File

@@ -1,15 +1,15 @@
use domain::DomainResult;
use domain::models::LibraryItem;
use domain::models::MediaItem;
use domain::value_objects::LibrarySearchFilter;
use super::deps::LibraryQueryDeps;
use super::deps::LibraryCommandDeps;
use super::parse_content_type;
use super::queries::SearchItemsQuery;
pub async fn execute(
deps: &LibraryQueryDeps,
deps: &LibraryCommandDeps,
query: SearchItemsQuery,
) -> DomainResult<(Vec<LibraryItem>, u32)> {
) -> DomainResult<(Vec<MediaItem>, u32)> {
let content_type = query
.content_type
.as_deref()

View File

@@ -1,50 +0,0 @@
use domain::models::LibraryItem;
use domain::value_objects::ContentType;
use crate::library::get_item;
use crate::library::queries::GetItemQuery;
#[path = "helpers.rs"]
mod helpers;
fn seed_item(repo: &std::sync::Arc<domain::testing::InMemoryLibraryRepository>) {
let item = LibraryItem::new("test", "m1", "Die Hard", ContentType::Movie, 7800, "2026-01-01");
repo.items
.lock()
.unwrap()
.insert(item.id().to_string(), item);
}
#[tokio::test]
async fn get_existing_item() {
let (_cmd, query, repo) = helpers::make_deps();
seed_item(&repo);
let item = get_item::execute(
&query,
GetItemQuery {
item_id: "test::m1".into(),
},
)
.await
.unwrap();
assert!(item.is_some());
assert_eq!(item.unwrap().title(), "Die Hard");
}
#[tokio::test]
async fn get_nonexistent_item_returns_none() {
let (_cmd, query, _repo) = helpers::make_deps();
let item = get_item::execute(
&query,
GetItemQuery {
item_id: "test::missing".into(),
},
)
.await
.unwrap();
assert!(item.is_none());
}

View File

@@ -1,41 +0,0 @@
use crate::library::get_sync_status;
use crate::library::queries::GetSyncStatusQuery;
#[path = "helpers.rs"]
mod helpers;
#[tokio::test]
async fn get_sync_status_empty() {
let (_cmd, query, _repo) = helpers::make_deps();
let entries = get_sync_status::execute(&query, GetSyncStatusQuery)
.await
.unwrap();
assert!(entries.is_empty());
}
#[tokio::test]
async fn get_sync_status_after_sync() {
let (cmd, query, _repo) = helpers::make_deps();
// Start a sync log entry
let log_id = domain::ports::LibraryCommand::log_sync_start(&*cmd.library_command, "test")
.await
.unwrap();
// Finish it
let result = domain::models::LibrarySyncResult::new("test", 42, 500);
domain::ports::LibraryCommand::log_sync_finish(&*cmd.library_command, log_id, &result)
.await
.unwrap();
let entries = get_sync_status::execute(&query, GetSyncStatusQuery)
.await
.unwrap();
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].provider_id(), "test");
assert_eq!(entries[0].items_found(), 42);
assert_eq!(entries[0].status(), "success");
}

View File

@@ -4,16 +4,13 @@ use async_trait::async_trait;
use domain::errors::DomainResult;
use domain::models::MediaItem;
use domain::ports::{
Collection, IProviderRegistry, ProviderCapabilities, SeriesSummary, StreamQuality,
StreamingProtocol,
};
use domain::ports::{Collection, IProviderRegistry, ProviderCapabilities, SeriesSummary};
use domain::SourceUri;
use domain::testing::{InMemoryLibraryRepository, NoopEventPublisher, NoopLibrarySync};
use domain::value_objects::{ContentType, MediaFilter, MediaItemId};
use crate::library::deps::{LibraryCommandDeps, LibraryQueryDeps};
use crate::library::deps::LibraryCommandDeps;
/// Minimal IProviderRegistry for library tests.
pub(crate) struct TestProviderRegistry;
#[async_trait]
@@ -30,11 +27,7 @@ impl IProviderRegistry for TestProviderRegistry {
Ok(None)
}
async fn get_stream_url(
&self,
_item_id: &MediaItemId,
_quality: &StreamQuality,
) -> DomainResult<String> {
async fn get_source_uri(&self, _item_id: &MediaItemId) -> DomainResult<SourceUri> {
Err(domain::DomainError::InfrastructureError(
"TestProviderRegistry does not support streaming".into(),
))
@@ -56,9 +49,7 @@ impl IProviderRegistry for TestProviderRegistry {
tags: false,
decade: false,
search: true,
streaming_protocol: StreamingProtocol::Hls,
rescan: false,
transcode: false,
})
}
@@ -83,24 +74,14 @@ impl IProviderRegistry for TestProviderRegistry {
}
}
/// Build LibraryCommandDeps and LibraryQueryDeps backed by InMemory repos.
///
/// Returns deps plus the underlying repo for seeding test data.
pub(crate) fn make_deps() -> (
LibraryCommandDeps,
LibraryQueryDeps,
Arc<InMemoryLibraryRepository>,
) {
pub(crate) fn make_deps() -> (LibraryCommandDeps, Arc<InMemoryLibraryRepository>) {
let repo = Arc::new(InMemoryLibraryRepository::new());
let cmd_deps = LibraryCommandDeps {
let deps = LibraryCommandDeps {
library_command: repo.clone(),
library_query: repo.clone(),
library_sync: Arc::new(NoopLibrarySync::new()),
provider_registry: Arc::new(TestProviderRegistry),
event_publisher: Arc::new(NoopEventPublisher::new()),
};
let query_deps = LibraryQueryDeps {
library_query: repo.clone(),
};
(cmd_deps, query_deps, repo)
(deps, repo)
}

View File

@@ -1,83 +0,0 @@
use domain::models::{LibraryItem, LibraryItemRow};
use domain::value_objects::ContentType;
use crate::library::list_collections;
use crate::library::queries::ListCollectionsQuery;
#[path = "helpers.rs"]
mod helpers;
fn seed_with_collections(repo: &std::sync::Arc<domain::testing::InMemoryLibraryRepository>) {
let mut store = repo.items.lock().unwrap();
let item = LibraryItem::from_persistence(LibraryItemRow {
id: "test::m1".into(),
provider_id: "test".into(),
external_id: "m1".into(),
title: "Die Hard".into(),
content_type: ContentType::Movie,
duration_secs: 7800,
series_name: None,
season_number: None,
episode_number: None,
year: None,
genres: vec![],
tags: vec![],
collection_id: Some("col-1".into()),
collection_name: Some("Movies".into()),
collection_type: Some("movies".into()),
thumbnail_url: None,
synced_at: "2026-01-01".into(),
});
store.insert(item.id().to_string(), item);
let item2 = LibraryItem::from_persistence(LibraryItemRow {
id: "test::e1".into(),
provider_id: "test".into(),
external_id: "e1".into(),
title: "BB S01E01".into(),
content_type: ContentType::Episode,
duration_secs: 2700,
series_name: Some("Breaking Bad".into()),
season_number: Some(1),
episode_number: Some(1),
year: None,
genres: vec![],
tags: vec![],
collection_id: Some("col-2".into()),
collection_name: Some("TV Shows".into()),
collection_type: Some("tvshows".into()),
thumbnail_url: None,
synced_at: "2026-01-01".into(),
});
store.insert(item2.id().to_string(), item2);
}
#[tokio::test]
async fn list_collections_returns_distinct() {
let (_cmd, query, repo) = helpers::make_deps();
seed_with_collections(&repo);
let cols = list_collections::execute(
&query,
ListCollectionsQuery { provider_id: None },
)
.await
.unwrap();
assert_eq!(cols.len(), 2);
}
#[tokio::test]
async fn list_collections_empty_library() {
let (_cmd, query, _repo) = helpers::make_deps();
let cols = list_collections::execute(
&query,
ListCollectionsQuery { provider_id: None },
)
.await
.unwrap();
assert!(cols.is_empty());
}

View File

@@ -1,108 +0,0 @@
use domain::models::{LibraryItem, LibraryItemRow};
use domain::value_objects::ContentType;
use crate::library::list_genres;
use crate::library::queries::ListGenresQuery;
#[path = "helpers.rs"]
mod helpers;
fn seed_with_genres(repo: &std::sync::Arc<domain::testing::InMemoryLibraryRepository>) {
let mut store = repo.items.lock().unwrap();
let item1 = LibraryItem::from_persistence(LibraryItemRow {
id: "test::m1".into(),
provider_id: "test".into(),
external_id: "m1".into(),
title: "Die Hard".into(),
content_type: ContentType::Movie,
duration_secs: 7800,
series_name: None,
season_number: None,
episode_number: None,
year: None,
genres: vec!["Action".into(), "Thriller".into()],
tags: vec![],
collection_id: None,
collection_name: None,
collection_type: None,
thumbnail_url: None,
synced_at: "2026-01-01".into(),
});
let item2 = LibraryItem::from_persistence(LibraryItemRow {
id: "test::m2".into(),
provider_id: "test".into(),
external_id: "m2".into(),
title: "Alien".into(),
content_type: ContentType::Movie,
duration_secs: 7020,
series_name: None,
season_number: None,
episode_number: None,
year: None,
genres: vec!["Sci-Fi".into(), "Action".into()],
tags: vec![],
collection_id: None,
collection_name: None,
collection_type: None,
thumbnail_url: None,
synced_at: "2026-01-01".into(),
});
store.insert(item1.id().to_string(), item1);
store.insert(item2.id().to_string(), item2);
}
#[tokio::test]
async fn list_genres_returns_unique() {
let (_cmd, query, repo) = helpers::make_deps();
seed_with_genres(&repo);
let genres = list_genres::execute(
&query,
ListGenresQuery {
content_type: None,
provider_id: None,
},
)
.await
.unwrap();
assert_eq!(genres.len(), 3); // Action, Sci-Fi, Thriller (deduped)
assert!(genres.contains(&"Action".to_string()));
assert!(genres.contains(&"Sci-Fi".to_string()));
assert!(genres.contains(&"Thriller".to_string()));
}
#[tokio::test]
async fn list_genres_empty_library() {
let (_cmd, query, _repo) = helpers::make_deps();
let genres = list_genres::execute(
&query,
ListGenresQuery {
content_type: None,
provider_id: None,
},
)
.await
.unwrap();
assert!(genres.is_empty());
}
#[tokio::test]
async fn list_genres_invalid_content_type_errors() {
let (_cmd, query, _repo) = helpers::make_deps();
let result = list_genres::execute(
&query,
ListGenresQuery {
content_type: Some("invalid".into()),
provider_id: None,
},
)
.await;
assert!(result.is_err());
}

View File

@@ -1,77 +0,0 @@
use domain::models::{LibraryItem, LibraryItemRow};
use domain::value_objects::ContentType;
use crate::library::list_seasons;
use crate::library::queries::ListSeasonsQuery;
#[path = "helpers.rs"]
mod helpers;
fn seed_episodes(repo: &std::sync::Arc<domain::testing::InMemoryLibraryRepository>) {
let mut store = repo.items.lock().unwrap();
for (i, season) in [1u32, 1, 2, 2, 2, 3].iter().enumerate() {
let item = LibraryItem::from_persistence(LibraryItemRow {
id: format!("test::e{i}"),
provider_id: "test".into(),
external_id: format!("e{i}"),
title: format!("BB S{season:02}E{:02}", i + 1),
content_type: ContentType::Episode,
duration_secs: 2700,
series_name: Some("Breaking Bad".into()),
season_number: Some(*season),
episode_number: Some(i as u32 + 1),
year: None,
genres: vec![],
tags: vec![],
collection_id: None,
collection_name: None,
collection_type: None,
thumbnail_url: None,
synced_at: "2026-01-01".into(),
});
store.insert(item.id().to_string(), item);
}
}
#[tokio::test]
async fn list_seasons_for_series() {
let (_cmd, query, repo) = helpers::make_deps();
seed_episodes(&repo);
let seasons = list_seasons::execute(
&query,
ListSeasonsQuery {
series_name: "Breaking Bad".into(),
provider_id: None,
},
)
.await
.unwrap();
assert_eq!(seasons.len(), 3);
// Sorted by season_number
assert_eq!(seasons[0].season_number(), 1);
assert_eq!(seasons[0].episode_count(), 2);
assert_eq!(seasons[1].season_number(), 2);
assert_eq!(seasons[1].episode_count(), 3);
assert_eq!(seasons[2].season_number(), 3);
assert_eq!(seasons[2].episode_count(), 1);
}
#[tokio::test]
async fn list_seasons_unknown_series() {
let (_cmd, query, _repo) = helpers::make_deps();
let seasons = list_seasons::execute(
&query,
ListSeasonsQuery {
series_name: "Nonexistent".into(),
provider_id: None,
},
)
.await
.unwrap();
assert!(seasons.is_empty());
}

View File

@@ -1,101 +0,0 @@
use domain::models::{LibraryItem, LibraryItemRow};
use domain::value_objects::ContentType;
use crate::library::list_shows;
use crate::library::queries::ListShowsQuery;
#[path = "helpers.rs"]
mod helpers;
fn seed_episodes(repo: &std::sync::Arc<domain::testing::InMemoryLibraryRepository>) {
let mut store = repo.items.lock().unwrap();
for (i, (series, season)) in [
("Breaking Bad", 1u32),
("Breaking Bad", 1),
("Breaking Bad", 2),
("The Wire", 1),
("The Wire", 1),
]
.iter()
.enumerate()
{
let item = LibraryItem::from_persistence(LibraryItemRow {
id: format!("test::e{i}"),
provider_id: "test".into(),
external_id: format!("e{i}"),
title: format!("{series} S{season:02}E{i:02}"),
content_type: ContentType::Episode,
duration_secs: 2700,
series_name: Some(series.to_string()),
season_number: Some(*season),
episode_number: Some(i as u32 + 1),
year: None,
genres: vec![],
tags: vec![],
collection_id: None,
collection_name: None,
collection_type: None,
thumbnail_url: None,
synced_at: "2026-01-01".into(),
});
store.insert(item.id().to_string(), item);
}
}
#[tokio::test]
async fn list_shows_returns_summaries() {
let (_cmd, query, repo) = helpers::make_deps();
seed_episodes(&repo);
let shows = list_shows::execute(
&query,
ListShowsQuery {
provider_id: None,
search_term: None,
genres: vec![],
},
)
.await
.unwrap();
assert_eq!(shows.len(), 2);
}
#[tokio::test]
async fn list_shows_with_search_term() {
let (_cmd, query, repo) = helpers::make_deps();
seed_episodes(&repo);
let shows = list_shows::execute(
&query,
ListShowsQuery {
provider_id: None,
search_term: Some("breaking".into()),
genres: vec![],
},
)
.await
.unwrap();
assert_eq!(shows.len(), 1);
assert_eq!(shows[0].series_name(), "Breaking Bad");
}
#[tokio::test]
async fn list_shows_empty() {
let (_cmd, query, _repo) = helpers::make_deps();
let shows = list_shows::execute(
&query,
ListShowsQuery {
provider_id: None,
search_term: None,
genres: vec![],
},
)
.await
.unwrap();
assert!(shows.is_empty());
}

View File

@@ -1,5 +1,5 @@
use domain::models::{LibraryItem, LibraryItemRow};
use domain::value_objects::ContentType;
use domain::models::{MediaItem, MediaItemRow};
use domain::value_objects::{ContentType, MediaItemId, MediaRole};
use crate::library::queries::SearchItemsQuery;
use crate::library::search;
@@ -10,25 +10,26 @@ mod helpers;
fn seed_items(repo: &std::sync::Arc<domain::testing::InMemoryLibraryRepository>) {
let mut store = repo.items.lock().unwrap();
let items = vec![
LibraryItem::new("test", "m1", "Die Hard", ContentType::Movie, 7800, "2026-01-01"),
LibraryItem::new("test", "m2", "Alien", ContentType::Movie, 7020, "2026-01-01"),
LibraryItem::new("test", "e1", "BB S01E01", ContentType::Episode, 2700, "2026-01-01"),
MediaItem::new_library("test", "m1", "Die Hard", ContentType::Movie, 7800, "2026-01-01"),
MediaItem::new_library("test", "m2", "Alien", ContentType::Movie, 7020, "2026-01-01"),
MediaItem::new_library("test", "e1", "BB S01E01", ContentType::Episode, 2700, "2026-01-01"),
];
for item in items {
store.insert(item.id().to_string(), item);
store.insert(item.id().value().to_string(), item);
}
}
fn seed_items_with_genres(repo: &std::sync::Arc<domain::testing::InMemoryLibraryRepository>) {
let mut store = repo.items.lock().unwrap();
let action = LibraryItem::from_persistence(LibraryItemRow {
id: "test::m1".into(),
let action = MediaItem::from_persistence(MediaItemRow {
id: MediaItemId::new("test::m1"),
provider_id: "test".into(),
external_id: "m1".into(),
title: "Die Hard".into(),
content_type: ContentType::Movie,
duration_secs: 7800,
description: None,
series_name: None,
season_number: None,
episode_number: None,
@@ -39,15 +40,18 @@ fn seed_items_with_genres(repo: &std::sync::Arc<domain::testing::InMemoryLibrary
collection_name: None,
collection_type: None,
thumbnail_url: None,
synced_at: "2026-01-01".into(),
synced_at: Some("2026-01-01".into()),
role: MediaRole::default(),
chapters: Vec::new(),
});
let scifi = LibraryItem::from_persistence(LibraryItemRow {
id: "test::m2".into(),
let scifi = MediaItem::from_persistence(MediaItemRow {
id: MediaItemId::new("test::m2"),
provider_id: "test".into(),
external_id: "m2".into(),
title: "Alien".into(),
content_type: ContentType::Movie,
duration_secs: 7020,
description: None,
series_name: None,
season_number: None,
episode_number: None,
@@ -58,15 +62,18 @@ fn seed_items_with_genres(repo: &std::sync::Arc<domain::testing::InMemoryLibrary
collection_name: None,
collection_type: None,
thumbnail_url: None,
synced_at: "2026-01-01".into(),
synced_at: Some("2026-01-01".into()),
role: MediaRole::default(),
chapters: Vec::new(),
});
let comedy = LibraryItem::from_persistence(LibraryItemRow {
id: "test::m3".into(),
let comedy = MediaItem::from_persistence(MediaItemRow {
id: MediaItemId::new("test::m3"),
provider_id: "test".into(),
external_id: "m3".into(),
title: "Airplane!".into(),
content_type: ContentType::Movie,
duration_secs: 5280,
description: None,
series_name: None,
season_number: None,
episode_number: None,
@@ -77,21 +84,23 @@ fn seed_items_with_genres(repo: &std::sync::Arc<domain::testing::InMemoryLibrary
collection_name: None,
collection_type: None,
thumbnail_url: None,
synced_at: "2026-01-01".into(),
synced_at: Some("2026-01-01".into()),
role: MediaRole::default(),
chapters: Vec::new(),
});
store.insert(action.id().to_string(), action);
store.insert(scifi.id().to_string(), scifi);
store.insert(comedy.id().to_string(), comedy);
store.insert(action.id().value().to_string(), action);
store.insert(scifi.id().value().to_string(), scifi);
store.insert(comedy.id().value().to_string(), comedy);
}
#[tokio::test]
async fn search_empty_filter_returns_all() {
let (_cmd, query, repo) = helpers::make_deps();
let (deps, repo) = helpers::make_deps();
seed_items(&repo);
let (items, total) = search::execute(
&query,
&deps,
SearchItemsQuery {
provider_id: None,
content_type: None,
@@ -114,11 +123,11 @@ async fn search_empty_filter_returns_all() {
#[tokio::test]
async fn search_by_content_type() {
let (_cmd, query, repo) = helpers::make_deps();
let (deps, repo) = helpers::make_deps();
seed_items(&repo);
let (items, total) = search::execute(
&query,
&deps,
SearchItemsQuery {
provider_id: None,
content_type: Some("movie".into()),
@@ -141,11 +150,11 @@ async fn search_by_content_type() {
#[tokio::test]
async fn search_by_genre() {
let (_cmd, query, repo) = helpers::make_deps();
let (deps, repo) = helpers::make_deps();
seed_items_with_genres(&repo);
let (items, total) = search::execute(
&query,
&deps,
SearchItemsQuery {
provider_id: None,
content_type: None,
@@ -169,11 +178,11 @@ async fn search_by_genre() {
#[tokio::test]
async fn search_by_search_term() {
let (_cmd, query, repo) = helpers::make_deps();
let (deps, repo) = helpers::make_deps();
seed_items(&repo);
let (items, total) = search::execute(
&query,
&deps,
SearchItemsQuery {
provider_id: None,
content_type: None,
@@ -197,10 +206,10 @@ async fn search_by_search_term() {
#[tokio::test]
async fn search_invalid_content_type_errors() {
let (_cmd, query, _repo) = helpers::make_deps();
let (deps, _repo) = helpers::make_deps();
let result = search::execute(
&query,
&deps,
SearchItemsQuery {
provider_id: None,
content_type: Some("bogus".into()),
@@ -221,11 +230,11 @@ async fn search_invalid_content_type_errors() {
#[tokio::test]
async fn search_pagination() {
let (_cmd, query, repo) = helpers::make_deps();
let (deps, repo) = helpers::make_deps();
seed_items(&repo);
let (items, total) = search::execute(
&query,
&deps,
SearchItemsQuery {
provider_id: None,
content_type: None,

View File

@@ -6,10 +6,10 @@ mod helpers;
#[tokio::test]
async fn trigger_sync_returns_provider_ids() {
let (cmd, _query, _repo) = helpers::make_deps();
let (deps, _repo) = helpers::make_deps();
let ids = sync::execute(
&cmd,
&deps,
TriggerSyncCommand { provider_id: None },
)
.await
@@ -20,10 +20,10 @@ async fn trigger_sync_returns_provider_ids() {
#[tokio::test]
async fn trigger_sync_specific_provider() {
let (cmd, _query, _repo) = helpers::make_deps();
let (deps, _repo) = helpers::make_deps();
let ids = sync::execute(
&cmd,
&deps,
TriggerSyncCommand {
provider_id: Some("test".into()),
},
@@ -36,16 +36,15 @@ async fn trigger_sync_specific_provider() {
#[tokio::test]
async fn trigger_sync_while_running_errors() {
let (cmd, _query, repo) = helpers::make_deps();
let (deps, repo) = helpers::make_deps();
// Simulate a running sync by inserting a log entry with "running" status
repo.items.lock().unwrap(); // just verify repo is accessible
let _log_id = domain::ports::LibraryCommand::log_sync_start(&*cmd.library_command, "test")
repo.items.lock().unwrap();
let _log_id = domain::ports::LibraryCommand::log_sync_start(&*deps.library_command, "test")
.await
.unwrap();
let result = sync::execute(
&cmd,
&deps,
TriggerSyncCommand { provider_id: None },
)
.await;

View File

@@ -4,7 +4,3 @@ pub struct UpsertProviderCommand {
pub config: serde_json::Value,
pub enabled: bool,
}
pub struct DeleteProviderCommand {
pub id: String,
}

View File

@@ -1,8 +0,0 @@
use domain::DomainResult;
use super::commands::DeleteProviderCommand;
use super::deps::ProviderDeps;
pub async fn execute(deps: &ProviderDeps, cmd: DeleteProviderCommand) -> DomainResult<()> {
deps.provider_config_command.delete(&cmd.id).await
}

View File

@@ -1,12 +0,0 @@
use domain::models::ProviderConfigRow;
use domain::DomainResult;
use super::deps::ProviderDeps;
use super::queries::GetProviderQuery;
pub async fn execute(
deps: &ProviderDeps,
query: GetProviderQuery,
) -> DomainResult<Option<ProviderConfigRow>> {
deps.provider_config_query.get_by_id(&query.id).await
}

View File

@@ -1,16 +0,0 @@
use domain::models::ProviderConfigRow;
use domain::DomainResult;
use super::deps::ProviderDeps;
use super::queries::ListProvidersQuery;
pub async fn execute(
deps: &ProviderDeps,
_query: ListProvidersQuery,
) -> DomainResult<Vec<ProviderConfigRow>> {
deps.provider_config_query.get_all().await
}
#[cfg(test)]
#[path = "tests/list.rs"]
mod tests;

View File

@@ -1,11 +1,6 @@
pub mod commands;
pub mod delete;
pub mod deps;
pub mod get;
pub mod list;
pub mod queries;
pub mod upsert;
pub use commands::{DeleteProviderCommand, UpsertProviderCommand};
pub use commands::UpsertProviderCommand;
pub use deps::ProviderDeps;
pub use queries::{GetProviderQuery, ListProvidersQuery};

View File

@@ -1,5 +0,0 @@
pub struct ListProvidersQuery;
pub struct GetProviderQuery {
pub id: String,
}

View File

@@ -1,45 +0,0 @@
use std::sync::Arc;
use domain::testing::InMemoryProviderConfig;
use crate::providers::commands::UpsertProviderCommand;
use crate::providers::deps::ProviderDeps;
use crate::providers::queries::ListProvidersQuery;
use crate::providers::{list, upsert};
fn make_deps() -> ProviderDeps {
let repo = Arc::new(InMemoryProviderConfig::new());
ProviderDeps {
provider_config_command: repo.clone(),
provider_config_query: repo,
}
}
#[tokio::test]
async fn list_empty() {
let deps = make_deps();
let providers = list::execute(&deps, ListProvidersQuery).await.unwrap();
assert!(providers.is_empty());
}
#[tokio::test]
async fn list_returns_all_providers() {
let deps = make_deps();
for id in ["jf-1", "local-1"] {
upsert::execute(
&deps,
UpsertProviderCommand {
id: id.into(),
provider_type: "jellyfin".into(),
config: serde_json::json!({}),
enabled: true,
},
)
.await
.unwrap();
}
let providers = list::execute(&deps, ListProvidersQuery).await.unwrap();
assert_eq!(providers.len(), 2);
}

View File

@@ -3,8 +3,3 @@ use uuid::Uuid;
pub struct GenerateScheduleCommand {
pub channel_id: Uuid,
}
pub struct DeleteSchedulesAfterCommand {
pub channel_id: Uuid,
pub target_generation: u32,
}

View File

@@ -1,19 +0,0 @@
use domain::value_objects::ChannelId;
use domain::DomainResult;
use super::commands::DeleteSchedulesAfterCommand;
use super::deps::ScheduleDeps;
pub async fn execute(
deps: &ScheduleDeps,
cmd: DeleteSchedulesAfterCommand,
) -> DomainResult<()> {
let channel_id = ChannelId::from(cmd.channel_id);
deps.schedule_command
.delete_schedules_after(channel_id, cmd.target_generation)
.await
}
#[cfg(test)]
#[path = "tests/delete_after.rs"]
mod tests;

View File

@@ -1,6 +1,6 @@
use std::sync::Arc;
use domain::ports::{ChannelQuery, EventPublisher, ScheduleCommand, ScheduleQuery};
use domain::ports::{ChannelQuery, EventPublisher, IProviderRegistry, ScheduleCommand, ScheduleQuery};
use domain::ScheduleEngineService;
pub struct ScheduleDeps {
@@ -9,4 +9,5 @@ pub struct ScheduleDeps {
pub schedule_query: Arc<dyn ScheduleQuery>,
pub schedule_command: Arc<dyn ScheduleCommand>,
pub event_publisher: Arc<dyn EventPublisher>,
pub provider_registry: Arc<dyn IProviderRegistry>,
}

View File

@@ -1,20 +0,0 @@
use chrono::Utc;
use domain::models::GeneratedSchedule;
use domain::value_objects::ChannelId;
use domain::DomainResult;
use super::deps::ScheduleDeps;
use super::queries::GetActiveScheduleQuery;
pub async fn execute(
deps: &ScheduleDeps,
query: GetActiveScheduleQuery,
) -> DomainResult<Option<GeneratedSchedule>> {
let channel_id = ChannelId::from(query.channel_id);
deps.schedule_query.find_active(channel_id, Utc::now()).await
}
#[cfg(test)]
#[path = "tests/get_active.rs"]
mod tests;

View File

@@ -1,13 +1,12 @@
use chrono::Utc;
use domain::ports::StreamQuality;
use domain::value_objects::ChannelId;
use domain::value_objects::{ChannelId, SourceUri};
use domain::{DomainResult, ScheduleEngineService};
use super::deps::ScheduleDeps;
use super::queries::GetStreamUrlQuery;
use super::queries::GetSourceQuery;
pub async fn execute(deps: &ScheduleDeps, query: GetStreamUrlQuery) -> DomainResult<Option<String>> {
pub async fn execute(deps: &ScheduleDeps, query: GetSourceQuery) -> DomainResult<Option<SourceUri>> {
let channel_id = ChannelId::from(query.channel_id);
let now = Utc::now();
@@ -22,9 +21,9 @@ pub async fn execute(deps: &ScheduleDeps, query: GetStreamUrlQuery) -> DomainRes
};
let item_id = broadcast.slot().item().id().clone();
let url = deps
.schedule_engine
.get_stream_url(&item_id, &StreamQuality::Direct)
let uri = deps
.provider_registry
.get_source_uri(&item_id)
.await?;
Ok(Some(url))
Ok(Some(uri))
}

View File

@@ -1,20 +0,0 @@
use domain::models::GeneratedSchedule;
use domain::value_objects::ChannelId;
use domain::DomainResult;
use super::deps::ScheduleDeps;
use super::queries::ListHistoryQuery;
pub async fn execute(
deps: &ScheduleDeps,
query: ListHistoryQuery,
) -> DomainResult<Vec<GeneratedSchedule>> {
let channel_id = ChannelId::from(query.channel_id);
deps.schedule_query
.list_schedule_history(channel_id)
.await
}
#[cfg(test)]
#[path = "tests/list_history.rs"]
mod tests;

View File

@@ -1,17 +1,11 @@
pub mod commands;
pub mod delete_after;
pub mod deps;
pub mod generate;
pub mod get_active;
pub mod get_current_broadcast;
pub mod get_epg;
pub mod get_stream_url;
pub mod list_history;
pub mod get_source;
pub mod queries;
pub use commands::{DeleteSchedulesAfterCommand, GenerateScheduleCommand};
pub use commands::GenerateScheduleCommand;
pub use deps::ScheduleDeps;
pub use queries::{
GetActiveScheduleQuery, GetCurrentBroadcastQuery, GetEpgQuery, GetStreamUrlQuery,
ListHistoryQuery,
};
pub use queries::{GetCurrentBroadcastQuery, GetEpgQuery, GetSourceQuery};

View File

@@ -1,9 +1,5 @@
use uuid::Uuid;
pub struct GetActiveScheduleQuery {
pub channel_id: Uuid,
}
pub struct GetCurrentBroadcastQuery {
pub channel_id: Uuid,
}
@@ -12,10 +8,6 @@ pub struct GetEpgQuery {
pub channel_id: Uuid,
}
pub struct GetStreamUrlQuery {
pub channel_id: Uuid,
}
pub struct ListHistoryQuery {
pub struct GetSourceQuery {
pub channel_id: Uuid,
}

View File

@@ -1,64 +0,0 @@
use domain::models::{Channel, GeneratedSchedule};
use domain::value_objects::UserId;
use crate::schedule::commands::DeleteSchedulesAfterCommand;
use crate::schedule::delete_after;
use crate::schedule::queries::ListHistoryQuery;
use crate::schedule::list_history;
#[path = "helpers.rs"]
mod helpers;
use helpers::make_schedule_deps;
#[tokio::test]
async fn delete_after_removes_later_generations() {
let (deps, channel_repo, schedule_repo) = make_schedule_deps();
let channel = Channel::new(UserId::generate(), "Cleanup", "UTC");
let channel_id = channel.id();
channel_repo
.channels
.lock()
.unwrap()
.insert(channel_id, channel);
// Manually insert schedules with different generations.
let now = chrono::Utc::now();
for generation in 1..=3 {
let sched = GeneratedSchedule::new(
channel_id,
now,
now + chrono::Duration::hours(24),
generation,
vec![],
);
schedule_repo
.schedules
.lock()
.unwrap()
.insert(sched.id(), sched);
}
// Delete generations > 1.
delete_after::execute(
&deps,
DeleteSchedulesAfterCommand {
channel_id: channel_id.value(),
target_generation: 1,
},
)
.await
.unwrap();
let remaining = list_history::execute(
&deps,
ListHistoryQuery {
channel_id: channel_id.value(),
},
)
.await
.unwrap();
assert_eq!(remaining.len(), 1);
assert_eq!(remaining[0].generation(), 1);
}

View File

@@ -1,22 +0,0 @@
use crate::schedule::get_active;
use crate::schedule::queries::GetActiveScheduleQuery;
#[path = "helpers.rs"]
mod helpers;
use helpers::make_schedule_deps;
#[tokio::test]
async fn returns_none_when_no_schedule_exists() {
let (deps, _, _) = make_schedule_deps();
let result = get_active::execute(
&deps,
GetActiveScheduleQuery {
channel_id: uuid::Uuid::new_v4(),
},
)
.await
.unwrap();
assert!(result.is_none());
}

View File

@@ -4,17 +4,14 @@ use async_trait::async_trait;
use domain::errors::DomainResult;
use domain::models::MediaItem;
use domain::ports::{
Collection, IProviderRegistry, ProviderCapabilities,
SeriesSummary, StreamQuality, StreamingProtocol,
};
use domain::testing::{InMemoryChannelRepository, InMemoryScheduleRepository, NoopEventPublisher};
use domain::ports::{Collection, IProviderRegistry, ProviderCapabilities, SeriesSummary};
use domain::SourceUri;
use domain::testing::{InMemoryChannelRepository, InMemoryLibraryRepository, InMemoryScheduleRepository, NoopEventPublisher};
use domain::value_objects::{ContentType, MediaFilter, MediaItemId};
use domain::ScheduleEngineService;
use crate::schedule::deps::ScheduleDeps;
/// Minimal IProviderRegistry backed by a NoopMediaProvider.
pub(crate) struct TestProviderRegistry;
#[async_trait]
@@ -31,11 +28,7 @@ impl IProviderRegistry for TestProviderRegistry {
Ok(None)
}
async fn get_stream_url(
&self,
_item_id: &MediaItemId,
_quality: &StreamQuality,
) -> DomainResult<String> {
async fn get_source_uri(&self, _item_id: &MediaItemId) -> DomainResult<SourceUri> {
Err(domain::DomainError::InfrastructureError(
"TestProviderRegistry does not support streaming".into(),
))
@@ -57,9 +50,7 @@ impl IProviderRegistry for TestProviderRegistry {
tags: false,
decade: false,
search: false,
streaming_protocol: StreamingProtocol::Hls,
rescan: false,
transcode: false,
})
}
@@ -84,9 +75,6 @@ impl IProviderRegistry for TestProviderRegistry {
}
}
/// Build ScheduleDeps backed by InMemory repos and a test provider registry.
///
/// Returns the deps plus the underlying repos for test assertions.
pub(crate) fn make_schedule_deps() -> (
ScheduleDeps,
Arc<InMemoryChannelRepository>,
@@ -94,10 +82,11 @@ pub(crate) fn make_schedule_deps() -> (
) {
let channel_repo = Arc::new(InMemoryChannelRepository::new());
let schedule_repo = Arc::new(InMemoryScheduleRepository::new());
let library_repo = Arc::new(InMemoryLibraryRepository::new());
let provider_registry = Arc::new(TestProviderRegistry);
let engine = Arc::new(ScheduleEngineService::new(
provider_registry,
library_repo,
channel_repo.clone(),
schedule_repo.clone(),
schedule_repo.clone(),
@@ -109,6 +98,7 @@ pub(crate) fn make_schedule_deps() -> (
schedule_query: schedule_repo.clone(),
schedule_command: schedule_repo.clone(),
event_publisher: Arc::new(NoopEventPublisher::new()),
provider_registry,
};
(deps, channel_repo, schedule_repo)

View File

@@ -1,22 +0,0 @@
use crate::schedule::list_history;
use crate::schedule::queries::ListHistoryQuery;
#[path = "helpers.rs"]
mod helpers;
use helpers::make_schedule_deps;
#[tokio::test]
async fn returns_empty_for_new_channel() {
let (deps, _, _) = make_schedule_deps();
let result = list_history::execute(
&deps,
ListHistoryQuery {
channel_id: uuid::Uuid::new_v4(),
},
)
.await
.unwrap();
assert!(result.is_empty());
}

View File

@@ -1,6 +1,8 @@
use serde::{Deserialize, Serialize};
use crate::value_objects::{ChannelId, ScheduleId, SlotId};
#[derive(Clone, Debug)]
#[derive(Clone, Debug, Serialize, Deserialize)]
#[non_exhaustive]
pub enum DomainEvent {
BroadcastTransition {
@@ -18,6 +20,58 @@ pub enum DomainEvent {
UserRegistered { user_id: crate::value_objects::UserId },
}
pub struct EventEnvelope {
id: i64,
event: DomainEvent,
retry_count: u32,
created_at: String,
}
impl EventEnvelope {
pub fn new(id: i64, event: DomainEvent) -> Self {
Self {
id,
event,
retry_count: 0,
created_at: String::new(),
}
}
pub fn from_persistence(
id: i64,
event: DomainEvent,
retry_count: u32,
created_at: String,
) -> Self {
Self {
id,
event,
retry_count,
created_at,
}
}
pub fn id(&self) -> i64 {
self.id
}
pub fn event(&self) -> &DomainEvent {
&self.event
}
pub fn into_event(self) -> DomainEvent {
self.event
}
pub fn retry_count(&self) -> u32 {
self.retry_count
}
pub fn created_at(&self) -> &str {
&self.created_at
}
}
#[cfg(test)]
#[path = "tests/mod.rs"]
mod tests;

View File

@@ -10,5 +10,5 @@ pub mod value_objects;
pub use errors::{DomainError, DomainResult};
pub use events::DomainEvent;
pub use models::*;
pub use services::{generate_m3u, generate_xmltv, ScheduleEngineService};
pub use services::{generate_ical, generate_m3u, generate_xmltv, parse_ical, ScheduleEngineService};
pub use value_objects::*;

View File

@@ -3,8 +3,8 @@ use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use crate::value_objects::{
AccessMode, BlockId, ChannelId, FillStrategy, LogoPosition, MediaFilter, MediaItemId,
RecyclePolicy, UserId, Weekday,
AccessMode, BlockId, ChannelId, FillStrategy, InterstitialRule, LogoPosition, MediaFilter,
MediaItemId, MidRollRule, RotationPolicy, UserId, Weekday,
};
const SECONDS_IN_DAY: u32 = 86_400;
@@ -19,10 +19,9 @@ pub struct Channel {
description: Option<String>,
timezone: String,
schedule_config: ScheduleConfig,
recycle_policy: RecyclePolicy,
rotation_policy: RotationPolicy,
auto_schedule: bool,
access_mode: AccessMode,
access_password_hash: Option<String>,
logo: Option<String>,
logo_position: LogoPosition,
logo_opacity: f32,
@@ -30,6 +29,8 @@ pub struct Channel {
webhook_poll_interval_secs: u32,
webhook_body_template: Option<String>,
webhook_headers: Option<String>,
#[serde(default)]
gap_filler: Option<MediaFilter>,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
}
@@ -41,10 +42,9 @@ pub struct ChannelRow {
pub description: Option<String>,
pub timezone: String,
pub schedule_config: ScheduleConfig,
pub recycle_policy: RecyclePolicy,
pub rotation_policy: RotationPolicy,
pub auto_schedule: bool,
pub access_mode: AccessMode,
pub access_password_hash: Option<String>,
pub logo: Option<String>,
pub logo_position: LogoPosition,
pub logo_opacity: f32,
@@ -52,6 +52,7 @@ pub struct ChannelRow {
pub webhook_poll_interval_secs: u32,
pub webhook_body_template: Option<String>,
pub webhook_headers: Option<String>,
pub gap_filler: Option<MediaFilter>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
@@ -70,10 +71,9 @@ impl Channel {
description: None,
timezone: timezone.into(),
schedule_config: ScheduleConfig::default(),
recycle_policy: RecyclePolicy::default(),
rotation_policy: RotationPolicy::default(),
auto_schedule: false,
access_mode: AccessMode::default(),
access_password_hash: None,
logo: None,
logo_position: LogoPosition::default(),
logo_opacity: DEFAULT_LOGO_OPACITY,
@@ -81,6 +81,7 @@ impl Channel {
webhook_poll_interval_secs: DEFAULT_WEBHOOK_POLL_INTERVAL_SECS,
webhook_body_template: None,
webhook_headers: None,
gap_filler: None,
created_at: now,
updated_at: now,
}
@@ -94,10 +95,9 @@ impl Channel {
description: row.description,
timezone: row.timezone,
schedule_config: row.schedule_config,
recycle_policy: row.recycle_policy,
rotation_policy: row.rotation_policy,
auto_schedule: row.auto_schedule,
access_mode: row.access_mode,
access_password_hash: row.access_password_hash,
logo: row.logo,
logo_position: row.logo_position,
logo_opacity: row.logo_opacity,
@@ -105,6 +105,7 @@ impl Channel {
webhook_poll_interval_secs: row.webhook_poll_interval_secs,
webhook_body_template: row.webhook_body_template,
webhook_headers: row.webhook_headers,
gap_filler: row.gap_filler,
created_at: row.created_at,
updated_at: row.updated_at,
}
@@ -134,8 +135,8 @@ impl Channel {
&self.schedule_config
}
pub fn recycle_policy(&self) -> &RecyclePolicy {
&self.recycle_policy
pub fn rotation_policy(&self) -> &RotationPolicy {
&self.rotation_policy
}
pub fn auto_schedule(&self) -> bool {
@@ -146,10 +147,6 @@ impl Channel {
&self.access_mode
}
pub fn access_password_hash(&self) -> Option<&str> {
self.access_password_hash.as_deref()
}
pub fn logo(&self) -> Option<&str> {
self.logo.as_deref()
}
@@ -206,8 +203,8 @@ impl Channel {
self.updated_at = Utc::now();
}
pub fn set_recycle_policy(&mut self, policy: RecyclePolicy) {
self.recycle_policy = policy;
pub fn set_rotation_policy(&mut self, policy: RotationPolicy) {
self.rotation_policy = policy;
self.updated_at = Utc::now();
}
@@ -215,6 +212,15 @@ impl Channel {
self.auto_schedule = enabled;
self.updated_at = Utc::now();
}
pub fn gap_filler(&self) -> Option<&MediaFilter> {
self.gap_filler.as_ref()
}
pub fn set_gap_filler(&mut self, gap_filler: Option<MediaFilter>) {
self.gap_filler = gap_filler;
self.updated_at = Utc::now();
}
}
// deny_unknown_fields required so #[serde(untagged)] compat enum correctly rejects V1 payloads
@@ -276,6 +282,13 @@ impl ScheduleConfig {
&self.day_blocks
}
pub fn find_block_mut(&mut self, block_id: BlockId) -> Option<&mut ProgrammingBlock> {
self.day_blocks
.values_mut()
.flatten()
.find(|b| b.id() == block_id)
}
pub fn insert_day(&mut self, day: Weekday, blocks: Vec<ProgrammingBlock>) {
self.day_blocks.insert(day, blocks);
}
@@ -331,13 +344,13 @@ pub struct ProgrammingBlock {
loop_on_finish: bool,
#[serde(default)]
ignore_recycle_policy: bool,
ignore_rotation_policy: bool,
#[serde(default)]
access_mode: AccessMode,
interstitial_rule: Option<InterstitialRule>,
#[serde(default, skip_serializing_if = "Option::is_none")]
access_password_hash: Option<String>,
#[serde(default)]
mid_roll_rule: Option<MidRollRule>,
}
impl ProgrammingBlock {
@@ -356,12 +369,33 @@ impl ProgrammingBlock {
content: BlockContent::Algorithmic {
filter,
strategy,
provider_id: String::new(),
},
loop_on_finish: true,
ignore_recycle_policy: false,
access_mode: AccessMode::default(),
access_password_hash: None,
ignore_rotation_policy: false,
interstitial_rule: None,
mid_roll_rule: None,
}
}
pub fn from_parts(
id: BlockId,
name: impl Into<String>,
start_time: NaiveTime,
duration_mins: u32,
content: BlockContent,
interstitial_rule: Option<InterstitialRule>,
mid_roll_rule: Option<MidRollRule>,
) -> Self {
Self {
id,
name: name.into(),
start_time,
duration_mins,
content,
loop_on_finish: true,
ignore_rotation_policy: false,
interstitial_rule,
mid_roll_rule,
}
}
@@ -378,12 +412,11 @@ impl ProgrammingBlock {
duration_mins,
content: BlockContent::Manual {
items,
provider_id: String::new(),
},
loop_on_finish: true,
ignore_recycle_policy: false,
access_mode: AccessMode::default(),
access_password_hash: None,
ignore_rotation_policy: false,
interstitial_rule: None,
mid_roll_rule: None,
}
}
@@ -411,16 +444,34 @@ impl ProgrammingBlock {
self.loop_on_finish
}
pub fn ignore_recycle_policy(&self) -> bool {
self.ignore_recycle_policy
pub fn ignore_rotation_policy(&self) -> bool {
self.ignore_rotation_policy
}
pub fn access_mode(&self) -> &AccessMode {
&self.access_mode
pub fn interstitial_rule(&self) -> Option<&InterstitialRule> {
self.interstitial_rule.as_ref()
}
pub fn access_password_hash(&self) -> Option<&str> {
self.access_password_hash.as_deref()
pub fn mid_roll_rule(&self) -> Option<&MidRollRule> {
self.mid_roll_rule.as_ref()
}
pub fn with_interstitial_rule(mut self, rule: InterstitialRule) -> Self {
self.interstitial_rule = Some(rule);
self
}
pub fn with_mid_roll_rule(mut self, rule: MidRollRule) -> Self {
self.mid_roll_rule = Some(rule);
self
}
pub fn set_interstitial_rule(&mut self, rule: Option<InterstitialRule>) {
self.interstitial_rule = rule;
}
pub fn set_mid_roll_rule(&mut self, rule: Option<MidRollRule>) {
self.mid_roll_rule = rule;
}
}
@@ -429,14 +480,10 @@ impl ProgrammingBlock {
pub enum BlockContent {
Manual {
items: Vec<MediaItemId>,
#[serde(default)]
provider_id: String,
},
Algorithmic {
filter: MediaFilter,
strategy: FillStrategy,
#[serde(default)]
provider_id: String,
},
}

View File

@@ -1,172 +1,5 @@
use crate::value_objects::ContentType;
const SYNC_STATUS_RUNNING: &str = "running";
#[derive(Debug, Clone)]
pub struct LibraryItem {
id: String,
provider_id: String,
external_id: String,
title: String,
content_type: ContentType,
duration_secs: u32,
series_name: Option<String>,
season_number: Option<u32>,
episode_number: Option<u32>,
year: Option<u16>,
genres: Vec<String>,
tags: Vec<String>,
collection_id: Option<String>,
collection_name: Option<String>,
collection_type: Option<String>,
thumbnail_url: Option<String>,
synced_at: String,
}
pub struct LibraryItemRow {
pub id: String,
pub provider_id: String,
pub external_id: String,
pub title: String,
pub content_type: ContentType,
pub duration_secs: u32,
pub series_name: Option<String>,
pub season_number: Option<u32>,
pub episode_number: Option<u32>,
pub year: Option<u16>,
pub genres: Vec<String>,
pub tags: Vec<String>,
pub collection_id: Option<String>,
pub collection_name: Option<String>,
pub collection_type: Option<String>,
pub thumbnail_url: Option<String>,
pub synced_at: String,
}
impl LibraryItem {
pub fn new(
provider_id: impl Into<String>,
external_id: impl Into<String>,
title: impl Into<String>,
content_type: ContentType,
duration_secs: u32,
synced_at: impl Into<String>,
) -> Self {
let provider_id = provider_id.into();
let external_id = external_id.into();
let id = format!("{}::{}", provider_id, external_id);
Self {
id,
provider_id,
external_id,
title: title.into(),
content_type,
duration_secs,
series_name: None,
season_number: None,
episode_number: None,
year: None,
genres: Vec::new(),
tags: Vec::new(),
collection_id: None,
collection_name: None,
collection_type: None,
thumbnail_url: None,
synced_at: synced_at.into(),
}
}
pub fn from_persistence(row: LibraryItemRow) -> Self {
Self {
id: row.id,
provider_id: row.provider_id,
external_id: row.external_id,
title: row.title,
content_type: row.content_type,
duration_secs: row.duration_secs,
series_name: row.series_name,
season_number: row.season_number,
episode_number: row.episode_number,
year: row.year,
genres: row.genres,
tags: row.tags,
collection_id: row.collection_id,
collection_name: row.collection_name,
collection_type: row.collection_type,
thumbnail_url: row.thumbnail_url,
synced_at: row.synced_at,
}
}
pub fn id(&self) -> &str {
&self.id
}
pub fn provider_id(&self) -> &str {
&self.provider_id
}
pub fn external_id(&self) -> &str {
&self.external_id
}
pub fn title(&self) -> &str {
&self.title
}
pub fn content_type(&self) -> &ContentType {
&self.content_type
}
pub fn duration_secs(&self) -> u32 {
self.duration_secs
}
pub fn series_name(&self) -> Option<&str> {
self.series_name.as_deref()
}
pub fn season_number(&self) -> Option<u32> {
self.season_number
}
pub fn episode_number(&self) -> Option<u32> {
self.episode_number
}
pub fn year(&self) -> Option<u16> {
self.year
}
pub fn genres(&self) -> &[String] {
&self.genres
}
pub fn tags(&self) -> &[String] {
&self.tags
}
pub fn collection_id(&self) -> Option<&str> {
self.collection_id.as_deref()
}
pub fn collection_name(&self) -> Option<&str> {
self.collection_name.as_deref()
}
pub fn collection_type(&self) -> Option<&str> {
self.collection_type.as_deref()
}
pub fn thumbnail_url(&self) -> Option<&str> {
self.thumbnail_url.as_deref()
}
pub fn synced_at(&self) -> &str {
&self.synced_at
}
}
#[derive(Debug, Clone)]
pub struct LibraryCollection {
id: String,

View File

@@ -1,7 +1,7 @@
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use crate::value_objects::{ChannelId, ContentType, MediaItemId, PlaybackRecordId};
use crate::value_objects::{Chapter, ChannelId, ContentType, MediaItemId, MediaRole, PlaybackRecordId};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MediaItem {
@@ -10,14 +10,27 @@ pub struct MediaItem {
content_type: ContentType,
duration_secs: u32,
description: Option<String>,
#[serde(default)]
genres: Vec<String>,
year: Option<u16>,
#[serde(default)]
tags: Vec<String>,
series_name: Option<String>,
season_number: Option<u32>,
episode_number: Option<u32>,
thumbnail_url: Option<String>,
collection_id: Option<String>,
#[serde(default)]
provider_id: String,
#[serde(default)]
external_id: String,
collection_name: Option<String>,
collection_type: Option<String>,
synced_at: Option<String>,
#[serde(default)]
role: MediaRole,
#[serde(default)]
chapters: Vec<Chapter>,
}
pub struct MediaItemRow {
@@ -34,6 +47,13 @@ pub struct MediaItemRow {
pub episode_number: Option<u32>,
pub thumbnail_url: Option<String>,
pub collection_id: Option<String>,
pub provider_id: String,
pub external_id: String,
pub collection_name: Option<String>,
pub collection_type: Option<String>,
pub synced_at: Option<String>,
pub role: MediaRole,
pub chapters: Vec<Chapter>,
}
impl MediaItem {
@@ -57,6 +77,48 @@ impl MediaItem {
episode_number: None,
thumbnail_url: None,
collection_id: None,
provider_id: String::new(),
external_id: String::new(),
collection_name: None,
collection_type: None,
synced_at: None,
role: MediaRole::default(),
chapters: Vec::new(),
}
}
pub fn new_library(
provider_id: impl Into<String>,
external_id: impl Into<String>,
title: impl Into<String>,
content_type: ContentType,
duration_secs: u32,
synced_at: impl Into<String>,
) -> Self {
let provider_id = provider_id.into();
let external_id = external_id.into();
let id = MediaItemId::new(format!("{}::{}", provider_id, external_id));
Self {
id,
title: title.into(),
content_type,
duration_secs,
description: None,
genres: Vec::new(),
year: None,
tags: Vec::new(),
series_name: None,
season_number: None,
episode_number: None,
thumbnail_url: None,
collection_id: None,
provider_id,
external_id,
collection_name: None,
collection_type: None,
synced_at: Some(synced_at.into()),
role: MediaRole::default(),
chapters: Vec::new(),
}
}
@@ -75,6 +137,13 @@ impl MediaItem {
episode_number: row.episode_number,
thumbnail_url: row.thumbnail_url,
collection_id: row.collection_id,
provider_id: row.provider_id,
external_id: row.external_id,
collection_name: row.collection_name,
collection_type: row.collection_type,
synced_at: row.synced_at,
role: row.role,
chapters: row.chapters,
}
}
@@ -129,6 +198,42 @@ impl MediaItem {
pub fn collection_id(&self) -> Option<&str> {
self.collection_id.as_deref()
}
pub fn provider_id(&self) -> &str {
&self.provider_id
}
pub fn external_id(&self) -> &str {
&self.external_id
}
pub fn collection_name(&self) -> Option<&str> {
self.collection_name.as_deref()
}
pub fn collection_type(&self) -> Option<&str> {
self.collection_type.as_deref()
}
pub fn synced_at(&self) -> Option<&str> {
self.synced_at.as_deref()
}
pub fn role(&self) -> &MediaRole {
&self.role
}
pub fn chapters(&self) -> &[Chapter] {
&self.chapters
}
pub fn set_chapters(&mut self, chapters: Vec<Chapter>) {
self.chapters = chapters;
}
pub fn set_role(&mut self, role: MediaRole) {
self.role = role;
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]

View File

@@ -16,7 +16,7 @@ pub use channel::{
pub use collections::{PageParams, Paginated};
pub use config_snapshot::ChannelConfigSnapshot;
pub use library::{
LibraryCollection, LibraryItem, LibraryItemRow, LibrarySyncLogEntry, LibrarySyncResult,
LibraryCollection, LibrarySyncLogEntry, LibrarySyncResult,
SeasonSummary, ShowSummary,
};
pub use media::{MediaItem, MediaItemRow, PlaybackRecord};

View File

@@ -94,7 +94,7 @@ fn programming_block_getters() {
assert_eq!(block.start_time(), t(8, 0));
assert_eq!(block.duration_mins(), 120);
assert!(block.loop_on_finish());
assert!(!block.ignore_recycle_policy());
assert!(!block.ignore_rotation_policy());
}
#[test]
@@ -102,9 +102,8 @@ fn manual_block_creation() {
let items = vec![MediaItemId::new("item1"), MediaItemId::new("item2")];
let block = ProgrammingBlock::new_manual("Manual Block", t(20, 0), 60, items);
match block.content() {
BlockContent::Manual { items, provider_id } => {
BlockContent::Manual { items } => {
assert_eq!(items.len(), 2);
assert!(provider_id.is_empty());
}
_ => panic!("Expected Manual content"),
}

View File

@@ -1,51 +1,5 @@
use super::*;
#[test]
fn library_item_new_generates_composite_id() {
let item = LibraryItem::new("jellyfin", "abc123", "Test Movie", ContentType::Movie, 7200, "2026-03-19T00:00:00Z");
assert_eq!(item.id(), "jellyfin::abc123");
assert_eq!(item.provider_id(), "jellyfin");
assert_eq!(item.external_id(), "abc123");
}
#[test]
fn library_item_new_defaults_optional_fields() {
let item = LibraryItem::new("jf", "1", "Movie", ContentType::Movie, 3600, "2026-01-01");
assert!(item.series_name().is_none());
assert!(item.season_number().is_none());
assert!(item.genres().is_empty());
assert!(item.tags().is_empty());
assert!(item.collection_id().is_none());
assert!(item.thumbnail_url().is_none());
}
#[test]
fn library_item_from_persistence_all_fields() {
let item = LibraryItem::from_persistence(LibraryItemRow {
id: "jf::abc".into(),
provider_id: "jf".into(),
external_id: "abc".into(),
title: "Breaking Bad S01E01".into(),
content_type: ContentType::Episode,
duration_secs: 2700,
series_name: Some("Breaking Bad".into()),
season_number: Some(1),
episode_number: Some(1),
year: Some(2008),
genres: vec!["Drama".into()],
tags: vec!["tv".into()],
collection_id: Some("col-1".into()),
collection_name: Some("TV Shows".into()),
collection_type: Some("tvshows".into()),
thumbnail_url: Some("http://thumb.jpg".into()),
synced_at: "2026-03-19T00:00:00Z".into(),
});
assert_eq!(item.series_name(), Some("Breaking Bad"));
assert_eq!(item.season_number(), Some(1));
assert_eq!(item.year(), Some(2008));
assert_eq!(item.collection_name(), Some("TV Shows"));
}
#[test]
fn library_collection_new_and_getters() {
let col = LibraryCollection::new("col-1", "Movies");

View File

@@ -1,5 +1,5 @@
use super::*;
use crate::value_objects::PlaybackRecordId;
use crate::value_objects::{Chapter, PlaybackRecordId};
#[test]
fn media_item_new_defaults() {
@@ -15,6 +15,33 @@ fn media_item_new_defaults() {
assert!(item.genres().is_empty());
assert!(item.year().is_none());
assert!(item.series_name().is_none());
assert_eq!(item.provider_id(), "");
assert_eq!(item.external_id(), "");
assert!(item.synced_at().is_none());
assert_eq!(item.role(), &MediaRole::Program);
assert!(item.chapters().is_empty());
}
#[test]
fn media_item_new_library_generates_composite_id() {
let item = MediaItem::new_library("jellyfin", "abc123", "Test Movie", ContentType::Movie, 7200, "2026-03-19T00:00:00Z");
assert_eq!(item.id().value(), "jellyfin::abc123");
assert_eq!(item.provider_id(), "jellyfin");
assert_eq!(item.external_id(), "abc123");
assert_eq!(item.synced_at(), Some("2026-03-19T00:00:00Z"));
}
#[test]
fn media_item_new_library_defaults_optional_fields() {
let item = MediaItem::new_library("jf", "1", "Movie", ContentType::Movie, 3600, "2026-01-01");
assert!(item.series_name().is_none());
assert!(item.season_number().is_none());
assert!(item.genres().is_empty());
assert!(item.tags().is_empty());
assert!(item.collection_id().is_none());
assert!(item.thumbnail_url().is_none());
assert!(item.collection_name().is_none());
assert!(item.collection_type().is_none());
}
#[test]
@@ -33,6 +60,13 @@ fn media_item_from_persistence_round_trip() {
episode_number: Some(1),
thumbnail_url: Some("http://thumb.jpg".into()),
collection_id: Some("col-1".into()),
provider_id: "jf".into(),
external_id: "abc".into(),
collection_name: Some("TV Shows".into()),
collection_type: Some("tvshows".into()),
synced_at: Some("2026-03-19T00:00:00Z".into()),
role: MediaRole::Program,
chapters: vec![Chapter::new(Some("Intro".into()), 0.0, 30.0)],
});
assert_eq!(item.title(), "Breaking Bad S01E01");
assert_eq!(item.series_name(), Some("Breaking Bad"));
@@ -40,6 +74,11 @@ fn media_item_from_persistence_round_trip() {
assert_eq!(item.episode_number(), Some(1));
assert_eq!(item.year(), Some(2008));
assert_eq!(item.collection_id(), Some("col-1"));
assert_eq!(item.provider_id(), "jf");
assert_eq!(item.external_id(), "abc");
assert_eq!(item.collection_name(), Some("TV Shows"));
assert_eq!(item.collection_type(), Some("tvshows"));
assert_eq!(item.synced_at(), Some("2026-03-19T00:00:00Z"));
}
#[test]

View File

@@ -3,10 +3,10 @@ use super::*;
#[test]
fn new_generates_id_and_timestamp() {
let email = Email::new("test@example.com").unwrap();
let user = User::new("oidc|123", email);
let user = User::new("external|123", email);
assert!(!user.is_admin());
assert!(user.password_hash().is_none());
assert_eq!(user.subject(), "oidc|123");
assert_eq!(user.subject(), "external|123");
}
#[test]

View File

@@ -1,7 +1,7 @@
use async_trait::async_trait;
use crate::errors::DomainResult;
pub use crate::events::DomainEvent;
pub use crate::events::{DomainEvent, EventEnvelope};
#[async_trait]
pub trait EventPublisher: Send + Sync {
@@ -10,7 +10,9 @@ pub trait EventPublisher: Send + Sync {
#[async_trait]
pub trait EventConsumer: Send + Sync {
async fn recv(&self) -> DomainResult<DomainEvent>;
async fn poll_next(&self) -> DomainResult<Option<EventEnvelope>>;
async fn ack(&self, event_id: i64) -> DomainResult<()>;
async fn nack(&self, event_id: i64, error: &str) -> DomainResult<()>;
}
#[async_trait]

View File

@@ -2,22 +2,24 @@ use async_trait::async_trait;
use crate::errors::DomainResult;
use crate::models::{
LibraryCollection, LibraryItem, LibrarySyncLogEntry, LibrarySyncResult,
LibraryCollection, LibrarySyncLogEntry, LibrarySyncResult, MediaItem,
SeasonSummary, ShowSummary,
};
use crate::value_objects::{ContentType, LibrarySearchFilter};
use crate::value_objects::{ContentType, LibrarySearchFilter, MediaRole};
use super::media::IMediaProvider;
#[async_trait]
pub trait LibraryCommand: Send + Sync {
async fn upsert_items(&self, provider_id: &str, items: Vec<LibraryItem>) -> DomainResult<()>;
async fn upsert_items(&self, provider_id: &str, items: Vec<MediaItem>) -> DomainResult<()>;
async fn clear_provider(&self, provider_id: &str) -> DomainResult<()>;
async fn log_sync_start(&self, provider_id: &str) -> DomainResult<i64>;
async fn log_sync_finish(&self, log_id: i64, result: &LibrarySyncResult) -> DomainResult<()>;
async fn update_role(&self, item_id: &str, role: MediaRole) -> DomainResult<()>;
}
#[async_trait]
@@ -25,9 +27,9 @@ pub trait LibraryQuery: Send + Sync {
async fn search(
&self,
filter: &LibrarySearchFilter,
) -> DomainResult<(Vec<LibraryItem>, u32)>;
) -> DomainResult<(Vec<MediaItem>, u32)>;
async fn get_by_id(&self, id: &str) -> DomainResult<Option<LibraryItem>>;
async fn get_by_id(&self, id: &str) -> DomainResult<Option<MediaItem>>;
async fn list_collections(
&self,

Some files were not shown because too many files have changed in this diff Show More