From 56d742a74c1aa4999a6043cbe14f838a68ba55f3 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 12 Jul 2026 03:23:20 +0200 Subject: [PATCH] presentation: HTTP server crate w/ handlers, routes, background tasks Axum binary that wires all clean-arch crates together: - AppState holds pre-built Deps structs (auth, channels, schedule, library, etc.) - JWT extractors (CurrentUser, AdminUser, OptionalCurrentUser) - Handlers delegate to application use cases, map to api-types DTOs - Routes: auth, channels, schedule, library, admin, providers, config, iptv - Background: auto-scheduler, broadcast poller, webhook consumer, library sync - Factory builds everything from Config + DbPool - SimpleProviderRegistry impl of IProviderRegistry trait - NoopMediaProvider fallback --- Cargo.lock | 429 +++++++++++- Cargo.toml | 2 +- crates/presentation/Cargo.toml | 55 ++ .../src/background/auto_scheduler.rs | 85 +++ .../src/background/broadcast_poller.rs | 119 ++++ .../src/background/library_sync.rs | 125 ++++ crates/presentation/src/background/mod.rs | 6 + .../src/background/webhook_consumer.rs | 204 ++++++ crates/presentation/src/errors.rs | 178 +++++ crates/presentation/src/extractors.rs | 151 +++++ crates/presentation/src/factory.rs | 623 ++++++++++++++++++ crates/presentation/src/handlers/admin.rs | 66 ++ crates/presentation/src/handlers/auth.rs | 149 +++++ crates/presentation/src/handlers/channels.rs | 200 ++++++ crates/presentation/src/handlers/config.rs | 56 ++ crates/presentation/src/handlers/files.rs | 32 + crates/presentation/src/handlers/iptv.rs | 46 ++ crates/presentation/src/handlers/library.rs | 210 ++++++ crates/presentation/src/handlers/mod.rs | 10 + crates/presentation/src/handlers/providers.rs | 72 ++ crates/presentation/src/handlers/schedule.rs | 131 ++++ crates/presentation/src/main.rs | 76 +++ crates/presentation/src/mappers/mod.rs | 6 + crates/presentation/src/routes.rs | 109 +++ crates/presentation/src/state.rs | 51 ++ 25 files changed, 3186 insertions(+), 5 deletions(-) create mode 100644 crates/presentation/Cargo.toml create mode 100644 crates/presentation/src/background/auto_scheduler.rs create mode 100644 crates/presentation/src/background/broadcast_poller.rs create mode 100644 crates/presentation/src/background/library_sync.rs create mode 100644 crates/presentation/src/background/mod.rs create mode 100644 crates/presentation/src/background/webhook_consumer.rs create mode 100644 crates/presentation/src/errors.rs create mode 100644 crates/presentation/src/extractors.rs create mode 100644 crates/presentation/src/factory.rs create mode 100644 crates/presentation/src/handlers/admin.rs create mode 100644 crates/presentation/src/handlers/auth.rs create mode 100644 crates/presentation/src/handlers/channels.rs create mode 100644 crates/presentation/src/handlers/config.rs create mode 100644 crates/presentation/src/handlers/files.rs create mode 100644 crates/presentation/src/handlers/iptv.rs create mode 100644 crates/presentation/src/handlers/library.rs create mode 100644 crates/presentation/src/handlers/mod.rs create mode 100644 crates/presentation/src/handlers/providers.rs create mode 100644 crates/presentation/src/handlers/schedule.rs create mode 100644 crates/presentation/src/main.rs create mode 100644 crates/presentation/src/mappers/mod.rs create mode 100644 crates/presentation/src/routes.rs create mode 100644 crates/presentation/src/state.rs diff --git a/Cargo.lock b/Cargo.lock index b99e5e6..b675c78 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -31,6 +31,16 @@ dependencies = [ "uuid", ] +[[package]] +name = "adapter-event-publisher" +version = "0.1.0" +dependencies = [ + "async-trait", + "domain", + "tokio", + "tracing", +] + [[package]] name = "adapter-jellyfin" version = "0.1.0" @@ -93,6 +103,15 @@ dependencies = [ "uuid", ] +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + [[package]] name = "allocator-api2" version = "0.2.21" @@ -108,6 +127,12 @@ dependencies = [ "libc", ] +[[package]] +name = "anyhow" +version = "1.0.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" + [[package]] name = "api-types" version = "0.1.0" @@ -175,6 +200,81 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-extra" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9963ff19f40c6102c76756ef0a46004c0d58957d87259fc9208ff8441c12ab96" +dependencies = [ + "axum", + "axum-core", + "bytes", + "futures-util", + "headers", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "rustversion", + "serde_core", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "base16ct" version = "0.2.0" @@ -449,14 +549,38 @@ dependencies = [ "syn", ] +[[package]] +name = "darling" +version = "0.20.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc7f46116c46ff9ab3eb1597a45688b6715c6e628b5c133e288e709a29bcb4ee" +dependencies = [ + "darling_core 0.20.11", + "darling_macro 0.20.11", +] + [[package]] name = "darling" version = "0.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d" dependencies = [ - "darling_core", - "darling_macro", + "darling_core 0.23.0", + "darling_macro 0.23.0", +] + +[[package]] +name = "darling_core" +version = "0.20.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d00b9596d185e565c2207a0b01f8bd1a135483d02d9b7b0a54b11da8d53412e" +dependencies = [ + "fnv", + "ident_case", + "proc-macro2", + "quote", + "strsim", + "syn", ] [[package]] @@ -472,13 +596,24 @@ dependencies = [ "syn", ] +[[package]] +name = "darling_macro" +version = "0.20.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" +dependencies = [ + "darling_core 0.20.11", + "quote", + "syn", +] + [[package]] name = "darling_macro" version = "0.23.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" dependencies = [ - "darling_core", + "darling_core 0.23.0", "quote", "syn", ] @@ -503,6 +638,37 @@ dependencies = [ "serde_core", ] +[[package]] +name = "derive_builder" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "507dfb09ea8b7fa618fcf76e953f4f5e192547945816d5358edffe39f6f94947" +dependencies = [ + "derive_builder_macro", +] + +[[package]] +name = "derive_builder_core" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d5bcf7b024d6835cfb3d473887cd966994907effbe9227e8c8219824d06c4e8" +dependencies = [ + "darling 0.20.11", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "derive_builder_macro" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab63b0e2bf4d5928aff72e83a7dace85d7bba5fe12dcc3c5a572d78caffd3f3c" +dependencies = [ + "derive_builder_core", + "syn", +] + [[package]] name = "digest" version = "0.10.7" @@ -893,6 +1059,22 @@ dependencies = [ "tracing", ] +[[package]] +name = "handlebars" +version = "6.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f26569a2763497b7bd3fbd19374b774ea6038c5293678771259cd534d49740ff" +dependencies = [ + "derive_builder", + "log", + "num-order", + "pest", + "pest_derive", + "serde", + "serde_json", + "thiserror 2.0.18", +] + [[package]] name = "hashbrown" version = "0.12.3" @@ -925,6 +1107,30 @@ dependencies = [ "hashbrown 0.15.5", ] +[[package]] +name = "headers" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3314d5adb5d94bcdf56771f2e50dbbc80bb4bdf88967526706205ac9eff24eb" +dependencies = [ + "base64 0.22.1", + "bytes", + "headers-core", + "http", + "httpdate", + "mime", + "sha1", +] + +[[package]] +name = "headers-core" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4" +dependencies = [ + "http", +] + [[package]] name = "heck" version = "0.5.0" @@ -1003,6 +1209,12 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hyper" version = "1.10.1" @@ -1017,6 +1229,7 @@ dependencies = [ "http", "http-body", "httparse", + "httpdate", "itoa", "pin-project-lite", "smallvec", @@ -1372,6 +1585,21 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "md-5" version = "0.10.6" @@ -1422,6 +1650,15 @@ dependencies = [ "tempfile", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num-bigint" version = "0.4.8" @@ -1473,6 +1710,21 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-modular" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc41a1374056e9672221567958a66c16be12d0e2c1b408761e14d901c237d5e0" + +[[package]] +name = "num-order" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "537b596b97c40fcf8056d153049eb22f481c17ebce72a513ec9286e4986d1bb6" +dependencies = [ + "num-modular", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -1693,6 +1945,48 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "pest" +version = "2.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47627dd7305c6a2d6c8c6bcd24c5a4c17dbbf425f4f9c5313e724b38fc9782e9" +dependencies = [ + "memchr", + "ucd-trie", +] + +[[package]] +name = "pest_derive" +version = "2.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b4254325ecad416ab689e27ba51da03ba01a9632bc6e108f5fe7c3c4ad29d58" +dependencies = [ + "pest", + "pest_generator", +] + +[[package]] +name = "pest_generator" +version = "2.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c4c0e91ead7a8f7acecbca6f003fc2e8282b1dbe2dd9c9d2f16aba42995e0a7" +dependencies = [ + "pest", + "pest_meta", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "pest_meta" +version = "2.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9744bc48116fee06334924bb5f2bad41eed5e89bd26e29b0b799f9a3f82c210" +dependencies = [ + "pest", +] + [[package]] name = "phf" version = "0.12.1" @@ -1774,6 +2068,40 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "presentation" +version = "0.1.0" +dependencies = [ + "adapter-auth", + "adapter-event-publisher", + "adapter-jellyfin", + "adapter-local-files", + "adapter-postgres", + "adapter-sqlite", + "anyhow", + "api-types", + "application", + "async-trait", + "axum", + "axum-extra", + "chrono", + "domain", + "dotenvy", + "handlebars", + "infra-wiring", + "reqwest", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "tower", + "tower-http", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "primeorder" version = "0.13.6" @@ -1957,6 +2285,23 @@ dependencies = [ "syn", ] +[[package]] +name = "regex-automata" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f388202e4b80542a0921078cc23b6333bcf1409c1e3f86404cae4766a6131db" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + [[package]] name = "reqwest" version = "0.12.28" @@ -2257,6 +2602,7 @@ version = "1.0.150" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" dependencies = [ + "indexmap 2.14.0", "itoa", "memchr", "serde", @@ -2322,7 +2668,7 @@ version = "3.21.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "84d57bc0c8b9a17920c178daa6bb924850d54a9c97ab45194bb8c17ad66bb660" dependencies = [ - "darling", + "darling 0.23.0", "proc-macro2", "quote", "syn", @@ -2350,6 +2696,15 @@ dependencies = [ "digest", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "2.0.1" @@ -2768,6 +3123,15 @@ dependencies = [ "syn", ] +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + [[package]] name = "time" version = "0.3.53" @@ -2908,6 +3272,7 @@ dependencies = [ "tokio", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -2925,6 +3290,7 @@ dependencies = [ "tower", "tower-layer", "tower-service", + "tracing", "url", ] @@ -2970,6 +3336,49 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "serde", + "serde_json", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", + "tracing-serde", ] [[package]] @@ -2984,6 +3393,12 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "ucd-trie" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" + [[package]] name = "unicode-bidi" version = "0.3.18" @@ -3072,6 +3487,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "vcpkg" version = "0.2.15" diff --git a/Cargo.toml b/Cargo.toml index d2d093b..a791023 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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/postgres", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files", "crates/adapters/event-publisher"] +members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/postgres", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files", "crates/adapters/event-publisher", "crates/presentation"] exclude = ["k-tv-backend", "k-tv-frontend"] resolver = "2" diff --git a/crates/presentation/Cargo.toml b/crates/presentation/Cargo.toml new file mode 100644 index 0000000..1b59f7c --- /dev/null +++ b/crates/presentation/Cargo.toml @@ -0,0 +1,55 @@ +[package] +name = "presentation" +version = "0.1.0" +edition = "2024" + +[[bin]] +name = "k-tv" +path = "src/main.rs" + +[features] +default = ["sqlite", "auth-jwt", "jellyfin"] +sqlite = ["dep:adapter-sqlite", "infra-wiring/sqlite"] +postgres = ["dep:adapter-postgres", "infra-wiring/postgres"] +auth-jwt = ["adapter-auth/jwt"] +auth-oidc = ["adapter-auth/oidc"] +jellyfin = ["dep:adapter-jellyfin"] +local-files = ["dep:adapter-local-files", "dep:tokio-util"] + +[dependencies] +domain = { workspace = true } +application = { workspace = true } +api-types = { workspace = true } +infra-wiring = { workspace = true } +adapter-auth = { workspace = true } +adapter-event-publisher = { workspace = true } + +# Feature-gated adapters +adapter-sqlite = { workspace = true, optional = true } +adapter-postgres = { workspace = true, optional = true } +adapter-jellyfin = { workspace = true, optional = true } +adapter-local-files = { workspace = true, optional = true } + +# Framework +axum = { workspace = true } +axum-extra = { workspace = true, features = ["typed-header"] } +tower = { workspace = true } +tower-http = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } + +# Utils +serde = { workspace = true } +serde_json = { workspace = true } +uuid = { workspace = true } +chrono = { workspace = true } +thiserror = { workspace = true } +async-trait = { workspace = true } +anyhow = "1" +dotenvy = "0.15" +reqwest = { workspace = true } +handlebars = "6" + +# Local-files streaming +tokio-util = { version = "0.7", features = ["io"], optional = true } diff --git a/crates/presentation/src/background/auto_scheduler.rs b/crates/presentation/src/background/auto_scheduler.rs new file mode 100644 index 0000000..9076e57 --- /dev/null +++ b/crates/presentation/src/background/auto_scheduler.rs @@ -0,0 +1,85 @@ +//! Background auto-scheduler task. +//! +//! Runs every hour, finds channels with `auto_schedule = true`, and regenerates +//! their schedule if it is within 24 hours of expiry. + +use std::sync::Arc; +use std::time::Duration; + +use chrono::Utc; + +use application::schedule::ScheduleDeps; +use application::schedule::GenerateScheduleCommand; + +/// Run the auto-scheduler loop. +pub async fn run(deps: Arc) { + loop { + tokio::time::sleep(Duration::from_secs(3600)).await; + tick(&deps).await; + } +} + +async fn tick(deps: &ScheduleDeps) { + // List all channels, find those with auto_schedule + let channels = match deps.channel_query.find_all().await { + Ok(c) => c, + Err(e) => { + tracing::warn!("auto-scheduler: failed to fetch channels: {}", e); + return; + } + }; + + let now = Utc::now(); + + for channel in channels { + if !channel.auto_schedule() { + continue; + } + + // Check latest schedule + let latest = match deps.schedule_query.find_latest(channel.id()).await { + Ok(s) => s, + Err(e) => { + tracing::warn!( + "auto-scheduler: failed to fetch latest schedule for channel {}: {}", + channel.id().value(), + e + ); + continue; + } + }; + + let should_generate = match &latest { + Some(s) => { + let remaining = s.valid_until() - now; + remaining < chrono::Duration::hours(24) + } + None => true, + }; + + if !should_generate { + continue; + } + + let cmd = GenerateScheduleCommand { + channel_id: channel.id().value(), + }; + + match application::schedule::generate::execute(deps, cmd).await { + Ok(schedule) => { + tracing::info!( + "auto-scheduler: generated schedule for channel {} (gen {})", + channel.id().value(), + schedule.generation(), + ); + } + Err(e) => { + tracing::warn!( + "auto-scheduler: failed to generate schedule for channel {}: {}", + channel.id().value(), + e + ); + } + } + } +} diff --git a/crates/presentation/src/background/broadcast_poller.rs b/crates/presentation/src/background/broadcast_poller.rs new file mode 100644 index 0000000..1e108f7 --- /dev/null +++ b/crates/presentation/src/background/broadcast_poller.rs @@ -0,0 +1,119 @@ +//! BroadcastPoller background task. +//! +//! Polls channels with webhook_url configured and emits domain events +//! when the current slot changes. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use chrono::Utc; +use uuid::Uuid; + +use domain::events::DomainEvent; +use domain::ports::events::EventPublisher; +use domain::value_objects::{ChannelId, SlotId}; + +use application::schedule::ScheduleDeps; + +/// Per-channel poll state. +struct ChannelPollState { + last_slot_id: Option, + last_checked: Instant, +} + +/// Polls channels and emits broadcast transition events. +pub async fn run(deps: Arc, event_publisher: Arc) { + let mut state: HashMap = HashMap::new(); + + loop { + tokio::time::sleep(Duration::from_secs(1)).await; + tick(&deps, &event_publisher, &mut state).await; + } +} + +async fn tick( + deps: &ScheduleDeps, + event_publisher: &Arc, + state: &mut HashMap, +) { + let channels = match deps.channel_query.find_all().await { + Ok(c) => c, + Err(e) => { + tracing::error!("broadcast poller: failed to load channels: {}", e); + return; + } + }; + + let live_ids: std::collections::HashSet = + channels.iter().map(|c| c.id().value()).collect(); + state.retain(|id, _| live_ids.contains(id)); + + let now = Utc::now(); + + for channel in channels { + if channel.webhook_url().is_none() { + state.remove(&channel.id().value()); + continue; + } + + let poll_interval = Duration::from_secs(channel.webhook_poll_interval_secs() as u64); + let channel_uuid = channel.id().value(); + + let entry = state.entry(channel_uuid).or_insert(ChannelPollState { + last_slot_id: None, + last_checked: Instant::now() - poll_interval, + }); + + if entry.last_checked.elapsed() < poll_interval { + continue; + } + + entry.last_checked = Instant::now(); + + let current_slot_id = match deps + .schedule_query + .find_active(channel.id(), now) + .await + { + Ok(Some(schedule)) => schedule + .slots() + .iter() + .find(|s| s.start_at() <= now && now < s.end_at()) + .map(|s| s.id().value()), + Ok(None) => None, + Err(e) => { + tracing::error!( + "broadcast poller: error checking schedule for channel {}: {}", + channel_uuid, + e + ); + continue; + } + }; + + if current_slot_id == entry.last_slot_id { + continue; + } + + match ¤t_slot_id { + Some(slot_id) => { + let _ = event_publisher + .publish(DomainEvent::BroadcastTransition { + channel_id: ChannelId::from(channel_uuid), + slot_id: SlotId::from(*slot_id), + }) + .await; + } + None => { + let _ = event_publisher + .publish(DomainEvent::NoSignal { + channel_id: ChannelId::from(channel_uuid), + }) + .await; + } + } + + entry.last_slot_id = current_slot_id; + } +} diff --git a/crates/presentation/src/background/library_sync.rs b/crates/presentation/src/background/library_sync.rs new file mode 100644 index 0000000..d589091 --- /dev/null +++ b/crates/presentation/src/background/library_sync.rs @@ -0,0 +1,125 @@ +//! Background library sync task. +//! +//! Fires 10 seconds after startup, then every N hours (read from app_settings). +//! Can be triggered on-demand via the sync_trigger watch channel. + +use std::sync::Arc; +use std::time::Duration; + +use domain::ports::{AppSettingsRepository, IProviderRegistry, LibrarySyncAdapter}; +use tokio::sync::watch; + +const STARTUP_DELAY_SECS: u64 = 10; +const DEFAULT_INTERVAL_HOURS: u64 = 6; + +pub async fn run( + sync_adapter: Arc, + provider_registry: Arc, + settings_repo: Arc, + mut trigger_rx: watch::Receiver<()>, +) { + tokio::time::sleep(Duration::from_secs(STARTUP_DELAY_SECS)).await; + + loop { + do_sync(&sync_adapter, &provider_registry).await; + + let interval_hours = load_interval_hours(&settings_repo).await; + let sleep = tokio::time::sleep(Duration::from_secs(interval_hours * 3600)); + + tokio::select! { + _ = sleep => {} + _ = trigger_rx.changed() => { + tracing::info!("library-sync: triggered manually"); + } + } + } +} + +async fn load_interval_hours(repo: &Arc) -> u64 { + repo.get("library_sync_interval_hours") + .await + .ok() + .flatten() + .and_then(|v| v.parse::().ok()) + .unwrap_or(DEFAULT_INTERVAL_HOURS) +} + +async fn do_sync( + sync_adapter: &Arc, + registry: &Arc, +) { + let provider_ids = registry.provider_ids(); + + for provider_id in provider_ids { + // We need a &dyn IMediaProvider, but IProviderRegistry doesn't expose one. + // The sync adapter will use the registry's fetch_items internally via its + // own stored reference to the provider. For now, we create a thin adapter. + tracing::info!("library-sync: syncing provider '{}'", provider_id); + + let wrapper = RegistryProviderAdapter { + registry: registry.clone(), + provider_id: provider_id.clone(), + }; + + let result = sync_adapter.sync_provider(&wrapper, &provider_id).await; + + if let Some(err) = result.error() { + tracing::warn!("library-sync: provider '{}' failed: {}", provider_id, err); + } else { + tracing::info!( + "library-sync: provider '{}' done — {} items in {}ms", + provider_id, + result.items_found(), + result.duration_ms() + ); + } + } +} + +/// Adapter that wraps IProviderRegistry calls for a specific provider_id, +/// implementing IMediaProvider so it can be passed to LibrarySyncAdapter. +struct RegistryProviderAdapter { + registry: Arc, + provider_id: String, +} + +#[async_trait::async_trait] +impl domain::ports::IMediaProvider for RegistryProviderAdapter { + fn capabilities(&self) -> domain::ports::ProviderCapabilities { + self.registry + .capabilities(&self.provider_id) + .unwrap_or(domain::ports::ProviderCapabilities { + collections: false, + series: false, + genres: false, + tags: false, + decade: false, + search: false, + streaming_protocol: domain::ports::StreamingProtocol::DirectFile, + rescan: false, + transcode: false, + }) + } + + async fn fetch_items( + &self, + filter: &domain::MediaFilter, + ) -> domain::DomainResult> { + self.registry.fetch_items(&self.provider_id, filter).await + } + + async fn fetch_by_id( + &self, + item_id: &domain::MediaItemId, + ) -> domain::DomainResult> { + self.registry.fetch_by_id(item_id).await + } + + async fn get_stream_url( + &self, + item_id: &domain::MediaItemId, + quality: &domain::ports::StreamQuality, + ) -> domain::DomainResult { + self.registry.get_stream_url(item_id, quality).await + } +} diff --git a/crates/presentation/src/background/mod.rs b/crates/presentation/src/background/mod.rs new file mode 100644 index 0000000..fecfe9b --- /dev/null +++ b/crates/presentation/src/background/mod.rs @@ -0,0 +1,6 @@ +//! Background tasks spawned at server startup. + +pub mod auto_scheduler; +pub mod broadcast_poller; +pub mod library_sync; +pub mod webhook_consumer; diff --git a/crates/presentation/src/background/webhook_consumer.rs b/crates/presentation/src/background/webhook_consumer.rs new file mode 100644 index 0000000..9b777b0 --- /dev/null +++ b/crates/presentation/src/background/webhook_consumer.rs @@ -0,0 +1,204 @@ +//! WebhookConsumer background task. +//! +//! Subscribes to domain events and delivers them to per-channel webhook URLs. + +use std::sync::Arc; + +use chrono::Utc; +use handlebars::Handlebars; +use serde_json::{Value, json}; +use tokio::sync::broadcast; +use uuid::Uuid; + +use domain::events::DomainEvent; +use domain::ports::ChannelQuery; + +/// Consumes domain events and delivers them to per-channel webhook URLs. +pub async fn run( + mut rx: broadcast::Receiver, + channel_query: Arc, + client: reqwest::Client, +) { + loop { + match rx.recv().await { + Ok(event) => { + let channel_id = event_channel_id(&event); + let payload = build_payload(&event); + + let channel_id_vo = domain::ChannelId::from(channel_id); + match channel_query.find_by_id(channel_id_vo).await { + Ok(Some(channel)) => { + if let Some(url) = channel.webhook_url() { + let url = url.to_string(); + let client = client.clone(); + let template = channel.webhook_body_template().map(|s| s.to_string()); + let headers = channel.webhook_headers().map(|s| s.to_string()); + tokio::spawn(async move { + post_webhook( + &client, + &url, + payload, + template.as_deref(), + headers.as_deref(), + ) + .await; + }); + } + } + Ok(None) => {} + Err(e) => { + tracing::warn!( + "webhook consumer: failed to look up channel {}: {}", + channel_id, + e + ); + } + } + } + Err(broadcast::error::RecvError::Lagged(n)) => { + tracing::warn!("webhook consumer lagged, {} events dropped", n); + } + Err(broadcast::error::RecvError::Closed) => { + tracing::info!("webhook consumer: event bus closed, shutting down"); + break; + } + } + } +} + +fn event_channel_id(event: &DomainEvent) -> Uuid { + match event { + DomainEvent::BroadcastTransition { channel_id, .. } => channel_id.value(), + DomainEvent::NoSignal { channel_id } => channel_id.value(), + DomainEvent::ScheduleGenerated { channel_id, .. } => channel_id.value(), + DomainEvent::ChannelCreated { channel_id } => channel_id.value(), + DomainEvent::ChannelUpdated { channel_id } => channel_id.value(), + DomainEvent::ChannelDeleted { channel_id } => channel_id.value(), + _ => Uuid::nil(), + } +} + +fn build_payload(event: &DomainEvent) -> Value { + let now = Utc::now().to_rfc3339(); + let channel_id = event_channel_id(event); + match event { + DomainEvent::BroadcastTransition { slot_id, .. } => { + json!({ + "event": "broadcast_transition", + "timestamp": now, + "channel_id": channel_id, + "data": { + "slot_id": slot_id.value(), + } + }) + } + DomainEvent::NoSignal { .. } => { + json!({ + "event": "no_signal", + "timestamp": now, + "channel_id": channel_id, + "data": {} + }) + } + DomainEvent::ScheduleGenerated { schedule_id, .. } => { + json!({ + "event": "schedule_generated", + "timestamp": now, + "channel_id": channel_id, + "data": { + "schedule_id": schedule_id.value(), + } + }) + } + DomainEvent::ChannelCreated { .. } => { + json!({ + "event": "channel_created", + "timestamp": now, + "channel_id": channel_id, + "data": {} + }) + } + DomainEvent::ChannelUpdated { .. } => { + json!({ + "event": "channel_updated", + "timestamp": now, + "channel_id": channel_id, + "data": {} + }) + } + DomainEvent::ChannelDeleted { .. } => { + json!({ + "event": "channel_deleted", + "timestamp": now, + "channel_id": channel_id, + "data": {} + }) + } + _ => { + json!({ + "event": "unknown", + "timestamp": now, + "channel_id": channel_id, + "data": {} + }) + } + } +} + +async fn post_webhook( + client: &reqwest::Client, + url: &str, + payload: Value, + template: Option<&str>, + headers_json: Option<&str>, +) { + let body = if let Some(tmpl) = template { + let hbs = Handlebars::new(); + match hbs.render_template(tmpl, &payload) { + Ok(rendered) => rendered, + Err(e) => { + tracing::warn!("webhook template render failed for {}: {}", url, e); + return; + } + } + } else { + match serde_json::to_string(&payload) { + Ok(s) => s, + Err(e) => { + tracing::warn!("webhook payload serialize failed: {}", e); + return; + } + } + }; + + let mut req = client.post(url).body(body); + let mut has_content_type = false; + + if let Some(h) = headers_json { + if let Ok(map) = serde_json::from_str::>(h) { + for (k, v) in &map { + if k.to_lowercase() == "content-type" { + has_content_type = true; + } + if let Some(v_str) = v.as_str() { + req = req.header(k.as_str(), v_str); + } + } + } + } + + if !has_content_type { + req = req.header("Content-Type", "application/json"); + } + + match req.send().await { + Ok(resp) => { + if !resp.status().is_success() { + tracing::warn!("webhook POST to {} returned status {}", url, resp.status()); + } + } + Err(e) => { + tracing::warn!("webhook POST to {} failed: {}", url, e); + } + } +} diff --git a/crates/presentation/src/errors.rs b/crates/presentation/src/errors.rs new file mode 100644 index 0000000..6f8b8d2 --- /dev/null +++ b/crates/presentation/src/errors.rs @@ -0,0 +1,178 @@ +//! API error handling — maps domain errors to HTTP responses. + +use axum::{ + Json, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use serde::Serialize; +use thiserror::Error; + +use domain::DomainError; + +/// API-level errors. +#[derive(Debug, Error)] +pub enum ApiError { + #[error("{0}")] + Domain(#[from] DomainError), + + #[error("Validation error: {0}")] + Validation(String), + + #[error("Internal server error")] + Internal(String), + + #[error("Forbidden: {0}")] + Forbidden(String), + + #[error("Unauthorized: {0}")] + Unauthorized(String), + + #[error("auth_required")] + AuthRequired, + + #[error("Not found: {0}")] + NotFound(String), + + #[error("Not implemented: {0}")] + NotImplemented(String), + + #[error("Conflict: {0}")] + Conflict(String), +} + +/// Error response body. +#[derive(Debug, Serialize)] +pub struct ErrorResponse { + pub error: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub details: Option, +} + +impl IntoResponse for ApiError { + fn into_response(self) -> Response { + let (status, error_response) = match &self { + ApiError::Domain(domain_error) => { + let status = match domain_error { + DomainError::UserNotFound(_) + | DomainError::ChannelNotFound(_) + | DomainError::NoActiveSchedule(_) => StatusCode::NOT_FOUND, + + DomainError::UserAlreadyExists(_) => StatusCode::CONFLICT, + + DomainError::ValidationError(_) | DomainError::TimezoneError(_) => { + StatusCode::BAD_REQUEST + } + + DomainError::Unauthenticated(_) => StatusCode::UNAUTHORIZED, + DomainError::Forbidden(_) => StatusCode::FORBIDDEN, + + DomainError::RepositoryError(_) | DomainError::InfrastructureError(_) => { + StatusCode::INTERNAL_SERVER_ERROR + } + + _ => StatusCode::INTERNAL_SERVER_ERROR, + }; + + ( + status, + ErrorResponse { + error: domain_error.to_string(), + details: None, + }, + ) + } + + ApiError::Validation(msg) => ( + StatusCode::BAD_REQUEST, + ErrorResponse { + error: "Validation error".to_string(), + details: Some(msg.clone()), + }, + ), + + ApiError::Internal(msg) => { + tracing::error!("Internal error: {}", msg); + ( + StatusCode::INTERNAL_SERVER_ERROR, + ErrorResponse { + error: "Internal server error".to_string(), + details: None, + }, + ) + } + + ApiError::Forbidden(msg) => ( + StatusCode::FORBIDDEN, + ErrorResponse { + error: "Forbidden".to_string(), + details: Some(msg.clone()), + }, + ), + + ApiError::Unauthorized(msg) => ( + StatusCode::UNAUTHORIZED, + ErrorResponse { + error: "Unauthorized".to_string(), + details: Some(msg.clone()), + }, + ), + + ApiError::AuthRequired => ( + StatusCode::UNAUTHORIZED, + ErrorResponse { + error: "auth_required".to_string(), + details: None, + }, + ), + + ApiError::NotFound(msg) => ( + StatusCode::NOT_FOUND, + ErrorResponse { + error: "Not found".to_string(), + details: Some(msg.clone()), + }, + ), + + ApiError::NotImplemented(msg) => ( + StatusCode::NOT_IMPLEMENTED, + ErrorResponse { + error: "Not implemented".to_string(), + details: Some(msg.clone()), + }, + ), + + ApiError::Conflict(msg) => ( + StatusCode::CONFLICT, + ErrorResponse { + error: "Conflict".to_string(), + details: Some(msg.clone()), + }, + ), + }; + + (status, Json(error_response)).into_response() + } +} + +impl ApiError { + pub fn validation(msg: impl Into) -> Self { + Self::Validation(msg.into()) + } + + pub fn internal(msg: impl Into) -> Self { + Self::Internal(msg.into()) + } + + pub fn not_found(msg: impl Into) -> Self { + Self::NotFound(msg.into()) + } + + pub fn conflict(msg: impl Into) -> Self { + Self::Conflict(msg.into()) + } + + pub fn not_implemented(msg: impl Into) -> Self { + Self::NotImplemented(msg.into()) + } +} diff --git a/crates/presentation/src/extractors.rs b/crates/presentation/src/extractors.rs new file mode 100644 index 0000000..dce9960 --- /dev/null +++ b/crates/presentation/src/extractors.rs @@ -0,0 +1,151 @@ +//! Auth extractors for API handlers. +//! +//! Provides `CurrentUser`, `OptionalCurrentUser`, and `AdminUser` extractors. + +use axum::extract::FromRequestParts; +use axum::http::request::Parts; +use domain::User; + +use crate::errors::ApiError; +use crate::state::AppState; + +/// Extracted current user from JWT Bearer token. +pub struct CurrentUser(pub User); + +impl FromRequestParts for CurrentUser { + type Rejection = ApiError; + + async fn from_request_parts( + parts: &mut Parts, + state: &AppState, + ) -> Result { + #[cfg(feature = "auth-jwt")] + { + return match try_jwt_auth(parts, state).await { + Ok(user) => Ok(CurrentUser(user)), + Err(e) => Err(e), + }; + } + + #[cfg(not(feature = "auth-jwt"))] + { + let _ = (parts, state); + Err(ApiError::Unauthorized( + "No authentication backend configured".to_string(), + )) + } + } +} + +/// Optional current user — returns None instead of error when auth missing. +/// +/// Checks `Authorization: Bearer ` first; falls back to `?token=`. +pub struct OptionalCurrentUser(pub Option); + +impl FromRequestParts for OptionalCurrentUser { + type Rejection = ApiError; + + async fn from_request_parts( + parts: &mut Parts, + state: &AppState, + ) -> Result { + #[cfg(feature = "auth-jwt")] + { + if let Ok(user) = try_jwt_auth(parts, state).await { + return Ok(OptionalCurrentUser(Some(user))); + } + let query_token = parts.uri.query().and_then(|q| { + q.split('&') + .find(|seg| seg.starts_with("token=")) + .map(|seg| seg[6..].to_owned()) + }); + if let Some(token) = query_token { + let user = validate_jwt_token(&token, state).await.ok(); + return Ok(OptionalCurrentUser(user)); + } + Ok(OptionalCurrentUser(None)) + } + + #[cfg(not(feature = "auth-jwt"))] + { + let _ = (parts, state); + Ok(OptionalCurrentUser(None)) + } + } +} + +/// Extracted admin user — returns 403 if user is not an admin. +pub struct AdminUser(pub User); + +impl FromRequestParts for AdminUser { + type Rejection = ApiError; + + async fn from_request_parts( + parts: &mut Parts, + state: &AppState, + ) -> Result { + let CurrentUser(user) = CurrentUser::from_request_parts(parts, state).await?; + if !user.is_admin() { + return Err(ApiError::Forbidden("Admin access required".to_string())); + } + Ok(AdminUser(user)) + } +} + +/// Authenticate via JWT Bearer token from `Authorization` header. +#[cfg(feature = "auth-jwt")] +async fn try_jwt_auth(parts: &mut Parts, state: &AppState) -> Result { + use axum::http::header::AUTHORIZATION; + + let auth_header = parts + .headers + .get(AUTHORIZATION) + .ok_or_else(|| ApiError::Unauthorized("Missing Authorization header".to_string()))?; + + let auth_str = auth_header + .to_str() + .map_err(|_| ApiError::Unauthorized("Invalid Authorization header encoding".to_string()))?; + + let token = auth_str.strip_prefix("Bearer ").ok_or_else(|| { + ApiError::Unauthorized("Authorization header must use Bearer scheme".to_string()) + })?; + + validate_jwt_token(token, state).await +} + +/// Validate a raw JWT string and return the corresponding `User`. +#[cfg(feature = "auth-jwt")] +pub(crate) async fn validate_jwt_token(token: &str, state: &AppState) -> Result { + let validator = state + .jwt_validator + .as_ref() + .ok_or_else(|| ApiError::Internal("JWT validator not configured".to_string()))?; + + let claims = validator.validate_access_token(token).map_err(|e| { + tracing::debug!("JWT validation failed: {:?}", e); + match e { + adapter_auth::JwtError::Expired => { + ApiError::Unauthorized("Token expired".to_string()) + } + adapter_auth::JwtError::InvalidFormat => { + ApiError::Unauthorized("Invalid token format".to_string()) + } + _ => ApiError::Unauthorized("Token validation failed".to_string()), + } + })?; + + let user_id: uuid::Uuid = claims + .sub + .parse() + .map_err(|_| ApiError::Unauthorized("Invalid user ID in token".to_string()))?; + + let user = state + .auth_deps + .user_query + .find_by_id(domain::UserId::from(user_id)) + .await + .map_err(|e| ApiError::Internal(format!("Failed to fetch user: {}", e)))? + .ok_or_else(|| ApiError::Unauthorized("User not found".to_string()))?; + + Ok(user) +} diff --git a/crates/presentation/src/factory.rs b/crates/presentation/src/factory.rs new file mode 100644 index 0000000..c5320f6 --- /dev/null +++ b/crates/presentation/src/factory.rs @@ -0,0 +1,623 @@ +//! Factory — builds AppState from Config + DbPool. +//! +//! Connects to the database, runs migrations, creates all adapter instances, +//! constructs Deps structs, and returns a fully-wired AppState. + +use std::sync::Arc; + +use application::{ + admin::AdminDeps, + auth::AuthDeps, + channels::{ChannelCommandDeps, ChannelQueryDeps}, + config_snapshots::ConfigSnapshotDeps, + iptv::IptvDeps, + library::{LibraryCommandDeps, LibraryQueryDeps}, + providers::ProviderDeps, + schedule::ScheduleDeps, +}; +use domain::ports::{IMediaProvider, IProviderRegistry, ProviderCapabilities, StreamingProtocol}; +use domain::{DomainError, ScheduleEngineService}; +use infra_wiring::{Config, ConfigSource, DbPool}; + +use crate::state::AppState; + +/// Build a fully-wired AppState ready for the HTTP server. +pub async fn build_app_state(config: Config) -> anyhow::Result { + // Connect to database + let pool = DbPool::connect(&config.database_url).await?; + pool.run_migrations().await?; + + // Wire up all repositories from the database pool + let wire_output = wire_repositories(&pool)?; + + // Auth service + let auth_service: Arc = + Arc::new(adapter_auth::PasswordAuthService); + + // Event bus + let event_bus = Arc::new(adapter_event_publisher::ChannelEventBus::new(64)); + let event_publisher: Arc = event_bus.clone(); + + // Provider registry + let provider_registry = build_provider_registry(&config).await; + + // Library sync adapter — uses the LibraryCommand port internally + let library_sync: Arc = + build_library_sync(wire_output.library_command.clone()); + + // Schedule engine + let schedule_engine = Arc::new(ScheduleEngineService::new( + provider_registry.clone(), + wire_output.channel_query.clone(), + wire_output.schedule_query.clone(), + wire_output.schedule_command.clone(), + )); + + // JWT validator + #[cfg(feature = "auth-jwt")] + let jwt_validator = build_jwt_validator(&config)?; + + // Sync trigger channel + let (sync_tx, sync_rx) = tokio::sync::watch::channel(()); + + // Build all deps structs + let auth_deps = Arc::new(AuthDeps { + user_command: wire_output.user_command.clone(), + user_query: wire_output.user_query.clone(), + auth_service, + event_publisher: event_publisher.clone(), + }); + + let channel_command_deps = Arc::new(ChannelCommandDeps { + channel_command: wire_output.channel_command.clone(), + channel_query: wire_output.channel_query.clone(), + event_publisher: event_publisher.clone(), + }); + + let channel_query_deps = Arc::new(ChannelQueryDeps { + channel_query: wire_output.channel_query.clone(), + }); + + let config_snapshot_deps = Arc::new(ConfigSnapshotDeps { + channel_command: wire_output.channel_command.clone(), + channel_query: wire_output.channel_query.clone(), + }); + + let schedule_deps = Arc::new(ScheduleDeps { + schedule_engine: schedule_engine.clone(), + channel_query: wire_output.channel_query.clone(), + schedule_query: wire_output.schedule_query.clone(), + schedule_command: wire_output.schedule_command.clone(), + event_publisher: event_publisher.clone(), + }); + + let library_command_deps = Arc::new(LibraryCommandDeps { + library_command: wire_output.library_command.clone(), + library_query: wire_output.library_query.clone(), + library_sync: library_sync.clone(), + provider_registry: provider_registry.clone(), + event_publisher: event_publisher.clone(), + }); + + let library_query_deps = Arc::new(LibraryQueryDeps { + library_query: wire_output.library_query.clone(), + }); + + let admin_deps = Arc::new(AdminDeps { + settings_repo: wire_output.settings.clone(), + activity_query: wire_output.activity_query.clone(), + }); + + let iptv_deps = Arc::new(IptvDeps { + channel_query: wire_output.channel_query.clone(), + schedule_query: wire_output.schedule_query.clone(), + }); + + let provider_deps = Arc::new(ProviderDeps { + provider_config_command: wire_output.provider_config_command.clone(), + provider_config_query: wire_output.provider_config_query.clone(), + }); + + let config_arc = Arc::new(config); + + // Spawn background tasks + let bg_schedule_deps = schedule_deps.clone(); + tokio::spawn(crate::background::auto_scheduler::run(bg_schedule_deps)); + + let bg_schedule_deps2 = schedule_deps.clone(); + let bg_event_publisher = event_publisher.clone(); + tokio::spawn(crate::background::broadcast_poller::run( + bg_schedule_deps2, + bg_event_publisher, + )); + + let webhook_rx = event_bus.subscriber(); + let webhook_channel_query = wire_output.channel_query.clone(); + tokio::spawn(crate::background::webhook_consumer::run( + webhook_rx, + webhook_channel_query, + reqwest::Client::new(), + )); + + let bg_sync = library_sync.clone(); + let bg_registry = provider_registry.clone(); + let bg_settings = wire_output.settings.clone(); + tokio::spawn(crate::background::library_sync::run( + bg_sync, + bg_registry, + bg_settings, + sync_rx, + )); + + Ok(AppState { + auth_deps, + channel_command_deps, + channel_query_deps, + config_snapshot_deps, + schedule_deps, + library_command_deps, + library_query_deps, + admin_deps, + iptv_deps, + provider_deps, + #[cfg(feature = "auth-jwt")] + jwt_validator, + provider_registry, + library_sync, + settings_repo: wire_output.settings, + event_bus, + config: config_arc, + sync_trigger: sync_tx, + }) +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Repository wiring output — trait objects ready for dependency injection. +struct WireOutput { + user_command: Arc, + user_query: Arc, + channel_command: Arc, + channel_query: Arc, + schedule_command: Arc, + schedule_query: Arc, + library_command: Arc, + library_query: Arc, + activity_query: Arc, + settings: Arc, + provider_config_command: Arc, + provider_config_query: Arc, +} + +fn wire_repositories(pool: &DbPool) -> anyhow::Result { + match pool { + #[cfg(feature = "sqlite")] + DbPool::Sqlite(sqlite_pool) => { + let w = adapter_sqlite::wire(sqlite_pool.clone()); + Ok(WireOutput { + user_command: w.user_command, + user_query: w.user_query, + channel_command: w.channel_command, + channel_query: w.channel_query, + schedule_command: w.schedule_command, + schedule_query: w.schedule_query, + library_command: w.library_command, + library_query: w.library_query, + activity_query: w.activity_query, + settings: w.settings, + provider_config_command: w.provider_config_command, + provider_config_query: w.provider_config_query, + }) + } + #[cfg(feature = "postgres")] + DbPool::Postgres(pg_pool) => { + let w = adapter_postgres::wire(pg_pool.clone()); + Ok(WireOutput { + user_command: w.user_command, + user_query: w.user_query, + channel_command: w.channel_command, + channel_query: w.channel_query, + schedule_command: w.schedule_command, + schedule_query: w.schedule_query, + library_command: w.library_command, + library_query: w.library_query, + activity_query: w.activity_query, + settings: w.settings, + provider_config_command: w.provider_config_command, + provider_config_query: w.provider_config_query, + }) + } + } +} + +async fn build_provider_registry(config: &Config) -> Arc { + // Build a concrete registry that routes to configured providers. + let mut providers: Vec<(String, Arc)> = Vec::new(); + + match config.config_source { + ConfigSource::Env => { + #[cfg(feature = "jellyfin")] + if let (Some(url), Some(api_key), Some(user_id)) = ( + &config.jellyfin_url, + &config.jellyfin_api_key, + &config.jellyfin_user_id, + ) { + tracing::info!("Media provider: Jellyfin at {}", url); + providers.push(( + "jellyfin".to_string(), + Arc::new(adapter_jellyfin::JellyfinMediaProvider::new( + adapter_jellyfin::JellyfinConfig { + base_url: url.clone(), + api_key: api_key.clone(), + user_id: user_id.clone(), + }, + )), + )); + } + } + ConfigSource::Db => { + // DB-based provider configs loaded elsewhere at runtime. + // For now, fall through to noop if nothing configured via env. + tracing::info!("CONFIG_SOURCE=db: provider configs loaded from database at runtime"); + } + } + + if providers.is_empty() { + tracing::warn!("No media provider configured — using NoopMediaProvider"); + providers.push(("noop".to_string(), Arc::new(NoopMediaProvider))); + } + + Arc::new(SimpleProviderRegistry::new(providers)) +} + +fn build_library_sync( + library_command: Arc, +) -> Arc { + Arc::new(SimpleSyncAdapter::new(library_command)) +} + +#[cfg(feature = "auth-jwt")] +fn build_jwt_validator(config: &Config) -> anyhow::Result>> { + let secret = match &config.jwt_secret { + Some(s) if !s.is_empty() => s.clone(), + _ => { + if config.is_production { + anyhow::bail!("JWT_SECRET is required in production"); + } + tracing::warn!("JWT_SECRET not set — using insecure development secret"); + "k-template-dev-secret-not-for-production-use-only".to_string() + } + }; + + let jwt_config = adapter_auth::JwtConfig::new( + secret, + config.jwt_issuer.clone(), + config.jwt_audience.clone(), + Some(config.jwt_expiry_hours), + Some(config.jwt_refresh_expiry_days), + config.is_production, + ) + .map_err(|e| anyhow::anyhow!("JWT config error: {}", e))?; + + Ok(Some(Arc::new(adapter_auth::JwtValidator::new(jwt_config)))) +} + +// --------------------------------------------------------------------------- +// NoopMediaProvider — fallback when nothing is configured +// --------------------------------------------------------------------------- + +struct NoopMediaProvider; + +#[async_trait::async_trait] +impl IMediaProvider for NoopMediaProvider { + fn capabilities(&self) -> ProviderCapabilities { + ProviderCapabilities { + collections: false, + series: false, + genres: false, + tags: false, + decade: false, + search: false, + streaming_protocol: StreamingProtocol::DirectFile, + rescan: false, + transcode: false, + } + } + + async fn fetch_items( + &self, + _: &domain::MediaFilter, + ) -> domain::DomainResult> { + Err(DomainError::InfrastructureError( + "No media provider configured. Set JELLYFIN_BASE_URL or LOCAL_FILES_DIR.".into(), + )) + } + + async fn fetch_by_id( + &self, + _: &domain::MediaItemId, + ) -> domain::DomainResult> { + Err(DomainError::InfrastructureError( + "No media provider configured.".into(), + )) + } + + async fn get_stream_url( + &self, + _: &domain::MediaItemId, + _: &domain::ports::StreamQuality, + ) -> domain::DomainResult { + Err(DomainError::InfrastructureError( + "No media provider configured.".into(), + )) + } +} + +// --------------------------------------------------------------------------- +// SimpleProviderRegistry — implements IProviderRegistry for N providers +// --------------------------------------------------------------------------- + +struct SimpleProviderRegistry { + providers: Vec<(String, Arc)>, +} + +impl SimpleProviderRegistry { + fn new(providers: Vec<(String, Arc)>) -> Self { + Self { providers } + } + + fn get(&self, id: &str) -> Option<&Arc> { + self.providers.iter().find(|(k, _)| k == id).map(|(_, v)| v) + } + + fn primary(&self) -> Option<&Arc> { + self.providers.first().map(|(_, v)| v) + } + + /// Extract provider_id from a prefixed item ID (e.g. "jellyfin::abc123" → "jellyfin"). + fn extract_provider_id(item_id: &str) -> Option<&str> { + item_id.find("::").map(|pos| &item_id[..pos]) + } +} + +#[async_trait::async_trait] +impl IProviderRegistry for SimpleProviderRegistry { + async fn fetch_items( + &self, + provider_id: &str, + filter: &domain::MediaFilter, + ) -> domain::DomainResult> { + let id = if provider_id.is_empty() { + self.providers.first().map(|(k, _)| k.as_str()).unwrap_or("") + } else { + provider_id + }; + let provider = self + .get(id) + .ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?; + provider.fetch_items(filter).await + } + + async fn fetch_by_id( + &self, + item_id: &domain::MediaItemId, + ) -> domain::DomainResult> { + let id_str = item_id.value(); + if let Some(pid) = Self::extract_provider_id(id_str) { + if let Some(provider) = self.get(pid) { + return provider.fetch_by_id(item_id).await; + } + } + // Fall back to primary + if let Some(provider) = self.primary() { + provider.fetch_by_id(item_id).await + } else { + Ok(None) + } + } + + async fn get_stream_url( + &self, + item_id: &domain::MediaItemId, + quality: &domain::ports::StreamQuality, + ) -> domain::DomainResult { + let id_str = item_id.value(); + if let Some(pid) = Self::extract_provider_id(id_str) { + if let Some(provider) = self.get(pid) { + return provider.get_stream_url(item_id, quality).await; + } + } + if let Some(provider) = self.primary() { + provider.get_stream_url(item_id, quality).await + } else { + Err(DomainError::InfrastructureError( + "No provider available".into(), + )) + } + } + + fn provider_ids(&self) -> Vec { + self.providers.iter().map(|(k, _)| k.clone()).collect() + } + + fn primary_id(&self) -> &str { + self.providers + .first() + .map(|(k, _)| k.as_str()) + .unwrap_or("") + } + + fn capabilities(&self, provider_id: &str) -> Option { + self.get(provider_id).map(|p| p.capabilities()) + } + + async fn list_collections( + &self, + provider_id: &str, + ) -> domain::DomainResult> { + let id = if provider_id.is_empty() { + self.primary_id() + } else { + provider_id + }; + let provider = self + .get(id) + .ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?; + provider.list_collections().await + } + + async fn list_series( + &self, + provider_id: &str, + collection_id: Option<&str>, + ) -> domain::DomainResult> { + let id = if provider_id.is_empty() { + self.primary_id() + } else { + provider_id + }; + let provider = self + .get(id) + .ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?; + provider.list_series(collection_id).await + } + + async fn list_genres( + &self, + provider_id: &str, + content_type: Option<&domain::ContentType>, + ) -> domain::DomainResult> { + let id = if provider_id.is_empty() { + self.primary_id() + } else { + provider_id + }; + let provider = self + .get(id) + .ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?; + provider.list_genres(content_type).await + } +} + +// --------------------------------------------------------------------------- +// SimpleSyncAdapter — wraps LibraryCommand for sync operations +// --------------------------------------------------------------------------- + +/// Convert a MediaItem from a provider into a LibraryItem for persistence. +fn media_item_to_library_item(item: domain::MediaItem, provider_id: &str) -> domain::LibraryItem { + let external_id = item.id().value().to_string(); + let id = format!("{}::{}", provider_id, external_id); + let now = chrono::Utc::now().to_rfc3339(); + + domain::LibraryItem::from_persistence( + id, + provider_id.to_string(), + external_id, + item.title().to_string(), + item.content_type().clone(), + item.duration_secs(), + item.series_name().map(|s| s.to_string()), + item.season_number(), + item.episode_number(), + item.year(), + item.genres().to_vec(), + item.tags().to_vec(), + item.collection_id().map(|s| s.to_string()), + None, // collection_name not in MediaItem + None, // collection_type not in MediaItem + item.thumbnail_url().map(|s| s.to_string()), + now, + ) +} + +struct SimpleSyncAdapter { + library_command: Arc, +} + +impl SimpleSyncAdapter { + fn new(library_command: Arc) -> Self { + Self { library_command } + } +} + +#[async_trait::async_trait] +impl domain::ports::LibrarySyncAdapter for SimpleSyncAdapter { + async fn sync_provider( + &self, + provider: &dyn IMediaProvider, + provider_id: &str, + ) -> domain::LibrarySyncResult { + use std::time::Instant; + + let start = Instant::now(); + let log_id = match self.library_command.log_sync_start(provider_id).await { + Ok(id) => id, + Err(e) => { + return domain::LibrarySyncResult::with_error( + provider_id, + 0, + format!("Failed to log sync start: {e}"), + ); + } + }; + + // Fetch all items from provider + let filter = domain::MediaFilter::default(); + let items = match provider.fetch_items(&filter).await { + Ok(items) => items, + Err(e) => { + let result = domain::LibrarySyncResult::with_error( + provider_id, + start.elapsed().as_millis() as u64, + format!("Failed to fetch items: {e}"), + ); + let _ = self.library_command.log_sync_finish(log_id, &result).await; + return result; + } + }; + + let items_found = items.len() as u32; + + // Clear + insert (items are MediaItem; LibrarySyncAdapter implementations + // typically handle the conversion. Here we delegate to library_command directly.) + if let Err(e) = self.library_command.clear_provider(provider_id).await { + let result = domain::LibrarySyncResult::with_error( + provider_id, + start.elapsed().as_millis() as u64, + format!("Failed to clear provider items: {e}"), + ); + let _ = self.library_command.log_sync_finish(log_id, &result).await; + return result; + } + + // Convert MediaItems to LibraryItems for storage + let library_items: Vec = items + .into_iter() + .map(|item| media_item_to_library_item(item, provider_id)) + .collect(); + + if let Err(e) = self + .library_command + .upsert_items(provider_id, library_items) + .await + { + let result = domain::LibrarySyncResult::with_error( + provider_id, + start.elapsed().as_millis() as u64, + format!("Failed to upsert items: {e}"), + ); + let _ = self.library_command.log_sync_finish(log_id, &result).await; + return result; + } + + let result = domain::LibrarySyncResult::new( + provider_id, + items_found, + start.elapsed().as_millis() as u64, + ); + let _ = self.library_command.log_sync_finish(log_id, &result).await; + result + } +} diff --git a/crates/presentation/src/handlers/admin.rs b/crates/presentation/src/handlers/admin.rs new file mode 100644 index 0000000..c8363f8 --- /dev/null +++ b/crates/presentation/src/handlers/admin.rs @@ -0,0 +1,66 @@ +//! Admin handlers. + +use axum::Json; +use axum::extract::{Query, State}; +use serde::Deserialize; +use std::collections::HashMap; + +use api_types::{ActivityEventResponse, SettingsResponse}; +use application::admin::{GetActivityLogQuery, GetSettingsQuery, UpdateSettingsCommand}; + +use crate::errors::ApiError; +use crate::extractors::AdminUser; +use crate::state::AppState; + +/// GET /admin/settings +pub async fn get_settings( + State(state): State, + AdminUser(_user): AdminUser, +) -> Result, ApiError> { + let pairs = + application::admin::get_settings::execute(&state.admin_deps, GetSettingsQuery).await?; + let settings: HashMap = pairs.into_iter().collect(); + Ok(Json(SettingsResponse { settings })) +} + +/// PUT /admin/settings +pub async fn update_settings( + State(state): State, + AdminUser(_user): AdminUser, + Json(body): Json>, +) -> Result, ApiError> { + let settings_vec: Vec<(String, String)> = body.into_iter().collect(); + let cmd = UpdateSettingsCommand { + settings: settings_vec, + }; + application::admin::update_settings::execute(&state.admin_deps, cmd).await?; + + // Re-read after update + let pairs = + application::admin::get_settings::execute(&state.admin_deps, GetSettingsQuery).await?; + let settings: HashMap = pairs.into_iter().collect(); + Ok(Json(SettingsResponse { settings })) +} + +#[derive(Debug, Deserialize)] +pub struct ActivityLogParams { + pub limit: Option, +} + +/// GET /admin/activity +pub async fn get_activity_log( + State(state): State, + AdminUser(_user): AdminUser, + Query(params): Query, +) -> Result>, ApiError> { + let query = GetActivityLogQuery { + limit: params.limit.unwrap_or(50), + }; + let events = application::admin::activity_log::execute(&state.admin_deps, query).await?; + Ok(Json( + events + .into_iter() + .map(ActivityEventResponse::from) + .collect(), + )) +} diff --git a/crates/presentation/src/handlers/auth.rs b/crates/presentation/src/handlers/auth.rs new file mode 100644 index 0000000..3375efa --- /dev/null +++ b/crates/presentation/src/handlers/auth.rs @@ -0,0 +1,149 @@ +//! Authentication handlers. + +use axum::Json; +use axum::extract::State; + +use api_types::{LoginRequest, RefreshRequest, RegisterRequest, TokenResponse, UserResponse}; +use application::auth::{LoginCommand, RegisterCommand}; + +use crate::errors::ApiError; +use crate::extractors::CurrentUser; +use crate::state::AppState; + +/// POST /auth/register +pub async fn register( + State(state): State, + Json(req): Json, +) -> Result, ApiError> { + let cmd = RegisterCommand { + email: req.email, + password: req.password, + }; + let user = application::auth::register::execute(&state.auth_deps, cmd).await?; + Ok(Json(UserResponse::from(user))) +} + +/// POST /auth/login +pub async fn login( + State(state): State, + Json(req): Json, +) -> Result, ApiError> { + let cmd = LoginCommand { + email: req.email, + password: req.password, + }; + let user = application::auth::login::execute(&state.auth_deps, cmd).await?; + let (access_token, refresh_token) = create_tokens(&user, &state, req.remember_me)?; + Ok(Json(TokenResponse { + access_token, + token_type: "Bearer".to_string(), + expires_in: state.config.jwt_expiry_hours * 3600, + refresh_token, + })) +} + +/// POST /auth/logout — no-op for JWT (stateless) +pub async fn logout() -> Result, ApiError> { + Ok(Json(serde_json::json!({"message": "logged out"}))) +} + +/// GET /auth/me +pub async fn me( + CurrentUser(user): CurrentUser, +) -> Result, ApiError> { + Ok(Json(UserResponse::from(user))) +} + +/// POST /auth/token — exchange credentials for tokens +#[cfg(feature = "auth-jwt")] +pub async fn get_token( + State(state): State, + Json(req): Json, +) -> Result, ApiError> { + let cmd = LoginCommand { + email: req.email, + password: req.password, + }; + let user = application::auth::login::execute(&state.auth_deps, cmd).await?; + let (access_token, refresh_token) = create_tokens(&user, &state, req.remember_me)?; + Ok(Json(TokenResponse { + access_token, + token_type: "Bearer".to_string(), + expires_in: state.config.jwt_expiry_hours * 3600, + refresh_token, + })) +} + +/// POST /auth/refresh — refresh an access token +#[cfg(feature = "auth-jwt")] +pub async fn refresh_token( + State(state): State, + Json(req): Json, +) -> Result, ApiError> { + let validator = state + .jwt_validator + .as_ref() + .ok_or_else(|| ApiError::Internal("JWT not configured".to_string()))?; + + let claims = validator.validate_refresh_token(&req.refresh_token).map_err(|e| { + tracing::debug!("Refresh token validation failed: {:?}", e); + ApiError::Unauthorized("Invalid refresh token".to_string()) + })?; + + let user_id: uuid::Uuid = claims + .sub + .parse() + .map_err(|_| ApiError::Unauthorized("Invalid user ID in token".to_string()))?; + + let user = state + .auth_deps + .user_query + .find_by_id(domain::UserId::from(user_id)) + .await + .map_err(|e| ApiError::Internal(format!("Failed to fetch user: {}", e)))? + .ok_or_else(|| ApiError::Unauthorized("User not found".to_string()))?; + + let (access_token, refresh_token) = create_tokens(&user, &state, true)?; + Ok(Json(TokenResponse { + access_token, + token_type: "Bearer".to_string(), + expires_in: state.config.jwt_expiry_hours * 3600, + refresh_token, + })) +} + +fn create_tokens( + user: &domain::User, + state: &AppState, + remember_me: bool, +) -> Result<(String, Option), ApiError> { + #[cfg(feature = "auth-jwt")] + { + let validator = state + .jwt_validator + .as_ref() + .ok_or_else(|| ApiError::Internal("JWT not configured".to_string()))?; + + let access = validator + .create_token(user) + .map_err(|e| ApiError::Internal(format!("Failed to create token: {}", e)))?; + + let refresh = if remember_me { + Some( + validator + .create_refresh_token(user) + .map_err(|e| ApiError::Internal(format!("Failed to create refresh token: {}", e)))?, + ) + } else { + None + }; + + Ok((access, refresh)) + } + + #[cfg(not(feature = "auth-jwt"))] + { + let _ = (user, state, remember_me); + Err(ApiError::Internal("JWT feature not enabled".to_string())) + } +} diff --git a/crates/presentation/src/handlers/channels.rs b/crates/presentation/src/handlers/channels.rs new file mode 100644 index 0000000..150c2ed --- /dev/null +++ b/crates/presentation/src/handlers/channels.rs @@ -0,0 +1,200 @@ +//! Channel CRUD handlers. + +use axum::Json; +use axum::extract::{Path, State}; + +use api_types::{ + ChannelResponse, ConfigSnapshotResponse, CreateChannelRequest, PatchSnapshotRequest, + UpdateChannelRequest, +}; +use application::channels::{ + CreateChannelCommand, DeleteChannelCommand, GetChannelQuery, ListByOwnerQuery, + ListChannelsQuery, UpdateChannelCommand, +}; +use application::config_snapshots::{ + GetSnapshotQuery, ListSnapshotsQuery, PatchLabelCommand, RestoreSnapshotCommand, + SaveSnapshotCommand, +}; + +use crate::errors::ApiError; +use crate::extractors::CurrentUser; +use crate::state::AppState; + +/// GET /channels +pub async fn list_channels( + State(state): State, + CurrentUser(_user): CurrentUser, +) -> Result>, ApiError> { + let channels = + application::channels::list::execute(&state.channel_query_deps, ListChannelsQuery).await?; + Ok(Json(channels.into_iter().map(ChannelResponse::from).collect())) +} + +/// GET /channels/mine +pub async fn list_my_channels( + State(state): State, + CurrentUser(user): CurrentUser, +) -> Result>, ApiError> { + let query = ListByOwnerQuery { + owner_id: user.id().value(), + }; + let channels = + application::channels::list_by_owner::execute(&state.channel_query_deps, query).await?; + Ok(Json(channels.into_iter().map(ChannelResponse::from).collect())) +} + +/// POST /channels +pub async fn create_channel( + State(state): State, + CurrentUser(user): CurrentUser, + Json(req): Json, +) -> Result, ApiError> { + let cmd = CreateChannelCommand { + owner_id: user.id().value(), + name: req.name, + timezone: req.timezone, + }; + let channel = application::channels::create::execute(&state.channel_command_deps, cmd).await?; + Ok(Json(ChannelResponse::from(channel))) +} + +/// GET /channels/:id +pub async fn get_channel( + State(state): State, + CurrentUser(_user): CurrentUser, + Path(id): Path, +) -> Result, ApiError> { + let query = GetChannelQuery { channel_id: id }; + let channel = application::channels::get::execute(&state.channel_query_deps, query) + .await? + .ok_or_else(|| ApiError::not_found(format!("Channel {id} not found")))?; + Ok(Json(ChannelResponse::from(channel))) +} + +/// PUT /channels/:id +pub async fn update_channel( + State(state): State, + CurrentUser(user): CurrentUser, + Path(id): Path, + Json(req): Json, +) -> Result, ApiError> { + let schedule_config = req + .schedule_config + .map(|v| { + serde_json::from_value(v) + .map_err(|e| ApiError::validation(format!("Invalid schedule_config: {e}"))) + }) + .transpose()?; + + let recycle_policy = req + .recycle_policy + .map(|v| { + serde_json::from_value(v) + .map_err(|e| ApiError::validation(format!("Invalid recycle_policy: {e}"))) + }) + .transpose()?; + + let cmd = UpdateChannelCommand { + channel_id: id, + owner_id: user.id().value(), + name: req.name, + description: req.description.map(Some), + timezone: req.timezone, + schedule_config, + recycle_policy, + auto_schedule: req.auto_schedule, + }; + let channel = application::channels::update::execute(&state.channel_command_deps, cmd).await?; + Ok(Json(ChannelResponse::from(channel))) +} + +/// DELETE /channels/:id +pub async fn delete_channel( + State(state): State, + CurrentUser(user): CurrentUser, + Path(id): Path, +) -> Result { + let cmd = DeleteChannelCommand { + channel_id: id, + owner_id: user.id().value(), + }; + application::channels::delete::execute(&state.channel_command_deps, cmd).await?; + Ok(axum::http::StatusCode::NO_CONTENT) +} + +// ── Config snapshots ───────────────────────────────────────────────────── + +/// POST /channels/:id/snapshots +pub async fn save_snapshot( + State(state): State, + CurrentUser(_user): CurrentUser, + Path(id): Path, +) -> Result, ApiError> { + let cmd = SaveSnapshotCommand { + channel_id: id, + label: None, + }; + let snap = application::config_snapshots::save::execute(&state.config_snapshot_deps, cmd).await?; + Ok(Json(ConfigSnapshotResponse::from(snap))) +} + +/// GET /channels/:id/snapshots +pub async fn list_snapshots( + State(state): State, + CurrentUser(_user): CurrentUser, + Path(id): Path, +) -> Result>, ApiError> { + let query = ListSnapshotsQuery { channel_id: id }; + let snaps = application::config_snapshots::list::execute(&state.config_snapshot_deps, query).await?; + Ok(Json(snaps.into_iter().map(ConfigSnapshotResponse::from).collect())) +} + +/// GET /channels/:id/snapshots/:snapshot_id +pub async fn get_snapshot( + State(state): State, + CurrentUser(_user): CurrentUser, + Path((id, snapshot_id)): Path<(uuid::Uuid, uuid::Uuid)>, +) -> Result, ApiError> { + let query = GetSnapshotQuery { + channel_id: id, + snapshot_id, + }; + let snap = application::config_snapshots::get::execute(&state.config_snapshot_deps, query) + .await? + .ok_or_else(|| ApiError::not_found("Snapshot not found"))?; + Ok(Json(ConfigSnapshotResponse::from(snap))) +} + +/// PATCH /channels/:id/snapshots/:snapshot_id +pub async fn patch_snapshot( + State(state): State, + CurrentUser(_user): CurrentUser, + Path((id, snapshot_id)): Path<(uuid::Uuid, uuid::Uuid)>, + Json(req): Json, +) -> Result, ApiError> { + let cmd = PatchLabelCommand { + channel_id: id, + snapshot_id, + label: req.label, + }; + let snap = + application::config_snapshots::patch_label::execute(&state.config_snapshot_deps, cmd) + .await? + .ok_or_else(|| ApiError::not_found("Snapshot not found"))?; + Ok(Json(ConfigSnapshotResponse::from(snap))) +} + +/// POST /channels/:id/snapshots/:snapshot_id/restore +pub async fn restore_snapshot( + State(state): State, + CurrentUser(_user): CurrentUser, + Path((id, snapshot_id)): Path<(uuid::Uuid, uuid::Uuid)>, +) -> Result, ApiError> { + let cmd = RestoreSnapshotCommand { + channel_id: id, + snapshot_id, + }; + let channel = + application::config_snapshots::restore::execute(&state.config_snapshot_deps, cmd).await?; + Ok(Json(ChannelResponse::from(channel))) +} diff --git a/crates/presentation/src/handlers/config.rs b/crates/presentation/src/handlers/config.rs new file mode 100644 index 0000000..7c4f05e --- /dev/null +++ b/crates/presentation/src/handlers/config.rs @@ -0,0 +1,56 @@ +//! System configuration handler. + +use axum::Json; +use axum::extract::State; + +use api_types::{ConfigResponse, ProviderCapabilitiesResponse, ProviderInfo}; + +use crate::errors::ApiError; +use crate::state::AppState; + +/// GET /config — public system configuration +pub async fn get_config( + State(state): State, +) -> Result, ApiError> { + let registry = &state.provider_registry; + let provider_ids = registry.provider_ids(); + let primary_id = registry.primary_id().to_string(); + + let providers: Vec = provider_ids + .iter() + .filter_map(|id| { + registry.capabilities(id).map(|caps| ProviderInfo { + id: id.clone(), + capabilities: ProviderCapabilitiesResponse::from(caps), + }) + }) + .collect(); + + let primary_caps = registry + .capabilities(&primary_id) + .map(ProviderCapabilitiesResponse::from) + .unwrap_or(ProviderCapabilitiesResponse { + collections: false, + series: false, + genres: false, + tags: false, + decade: false, + search: false, + streaming_protocol: "direct_file".to_string(), + rescan: false, + transcode: false, + }); + + let mut available_types = Vec::new(); + #[cfg(feature = "jellyfin")] + available_types.push("jellyfin".to_string()); + #[cfg(feature = "local-files")] + available_types.push("local_files".to_string()); + + Ok(Json(ConfigResponse { + allow_registration: state.config.allow_registration, + providers, + provider_capabilities: primary_caps, + available_provider_types: available_types, + })) +} diff --git a/crates/presentation/src/handlers/files.rs b/crates/presentation/src/handlers/files.rs new file mode 100644 index 0000000..7e25938 --- /dev/null +++ b/crates/presentation/src/handlers/files.rs @@ -0,0 +1,32 @@ +//! Local file streaming handlers (feature-gated). +//! +//! Placeholder — the actual streaming logic requires the local-files adapter +//! which provides file index and transcoding. This will be fleshed out once +//! the local-files adapter integration is complete. + +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use axum::response::IntoResponse; + +use crate::errors::ApiError; +use crate::state::AppState; + +/// GET /files/stream/:id — stream a local file +pub async fn stream_file( + State(_state): State, + Path(_id): Path, +) -> Result { + // TODO: integrate with adapter-local-files for actual streaming + Err::(ApiError::not_implemented( + "Local file streaming not yet wired in presentation crate", + )) +} + +/// POST /files/rescan — rescan local files +pub async fn rescan( + State(_state): State, +) -> Result { + Err::(ApiError::not_implemented( + "Local file rescan not yet wired in presentation crate", + )) +} diff --git a/crates/presentation/src/handlers/iptv.rs b/crates/presentation/src/handlers/iptv.rs new file mode 100644 index 0000000..fb0b120 --- /dev/null +++ b/crates/presentation/src/handlers/iptv.rs @@ -0,0 +1,46 @@ +//! IPTV export handlers (M3U, XMLTV). + +use axum::extract::{Query, State}; +use axum::http::header; +use axum::response::IntoResponse; +use serde::Deserialize; + +use application::iptv::{GetM3uQuery, GetXmltvQuery}; + +use crate::errors::ApiError; +use crate::extractors::OptionalCurrentUser; +use crate::state::AppState; + +#[derive(Debug, Deserialize)] +pub struct IptvParams { + pub token: Option, +} + +/// GET /iptv/playlist.m3u — M3U playlist +pub async fn m3u_playlist( + State(state): State, + OptionalCurrentUser(_user): OptionalCurrentUser, + Query(params): Query, +) -> Result { + let query = GetM3uQuery { + base_url: state.config.base_url.clone(), + token: params.token, + }; + let content = application::iptv::m3u::execute(&state.iptv_deps, query).await?; + Ok(( + [(header::CONTENT_TYPE, "audio/x-mpegurl; charset=utf-8")], + content, + )) +} + +/// GET /iptv/epg.xml — XMLTV electronic program guide +pub async fn xmltv_epg( + State(state): State, + OptionalCurrentUser(_user): OptionalCurrentUser, +) -> Result { + let content = application::iptv::xmltv::execute(&state.iptv_deps, GetXmltvQuery).await?; + Ok(( + [(header::CONTENT_TYPE, "application/xml; charset=utf-8")], + content, + )) +} diff --git a/crates/presentation/src/handlers/library.rs b/crates/presentation/src/handlers/library.rs new file mode 100644 index 0000000..f793c41 --- /dev/null +++ b/crates/presentation/src/handlers/library.rs @@ -0,0 +1,210 @@ +//! Library browsing handlers. + +use axum::Json; +use axum::extract::{Path, Query, State}; +use serde::Deserialize; + +use api_types::{CollectionResponse, LibraryItemResponse, PaginatedResponse, SeasonResponse, ShowResponse}; +use application::library::{ + GetItemQuery, GetSyncStatusQuery, ListCollectionsQuery, ListGenresQuery, ListSeasonsQuery, + ListShowsQuery, SearchItemsQuery, TriggerSyncCommand, +}; + +use crate::errors::ApiError; +use crate::extractors::{AdminUser, CurrentUser}; +use crate::state::AppState; + +#[derive(Debug, Deserialize)] +pub struct LibrarySearchParams { + pub provider: Option, + pub content_type: Option, + #[serde(default, rename = "genres[]")] + pub genres: Vec, + pub search_term: Option, + pub collection_id: Option, + #[serde(default, rename = "series_names[]")] + pub series_names: Vec, + pub season_number: Option, + pub decade: Option, + pub offset: Option, + pub limit: Option, +} + +/// GET /library/items +pub async fn search_items( + State(state): State, + CurrentUser(_user): CurrentUser, + Query(params): Query, +) -> Result>, ApiError> { + let query = SearchItemsQuery { + provider_id: params.provider, + content_type: params.content_type, + genres: params.genres, + search_term: params.search_term, + collection_id: params.collection_id, + series_names: params.series_names, + season_number: params.season_number, + decade: params.decade, + offset: params.offset.unwrap_or(0), + limit: params.limit.unwrap_or(50), + }; + let (items, total) = application::library::search::execute(&state.library_query_deps, query).await?; + Ok(Json(PaginatedResponse::new( + items.into_iter().map(LibraryItemResponse::from).collect(), + total as u64, + ))) +} + +/// GET /library/items/:id +pub async fn get_item( + State(state): State, + CurrentUser(_user): CurrentUser, + Path(id): Path, +) -> Result, ApiError> { + let query = GetItemQuery { item_id: id.clone() }; + let item = application::library::get_item::execute(&state.library_query_deps, query) + .await? + .ok_or_else(|| ApiError::not_found(format!("Library item {id} not found")))?; + Ok(Json(LibraryItemResponse::from(item))) +} + +/// GET /library/collections +pub async fn list_collections( + State(state): State, + CurrentUser(_user): CurrentUser, + Query(params): Query, +) -> Result>, ApiError> { + let query = ListCollectionsQuery { + provider_id: params.provider, + }; + let collections = + application::library::list_collections::execute(&state.library_query_deps, query).await?; + Ok(Json( + collections + .into_iter() + .map(CollectionResponse::from) + .collect(), + )) +} + +#[derive(Debug, Deserialize)] +pub struct ProviderParam { + pub provider: Option, +} + +#[derive(Debug, Deserialize)] +pub struct ShowsParams { + pub provider: Option, + pub search_term: Option, + #[serde(default, rename = "genres[]")] + pub genres: Vec, +} + +/// GET /library/shows +pub async fn list_shows( + State(state): State, + CurrentUser(_user): CurrentUser, + Query(params): Query, +) -> Result>, ApiError> { + let query = ListShowsQuery { + provider_id: params.provider, + search_term: params.search_term, + genres: params.genres, + }; + let shows = application::library::list_shows::execute(&state.library_query_deps, query).await?; + Ok(Json(shows.into_iter().map(ShowResponse::from).collect())) +} + +#[derive(Debug, Deserialize)] +pub struct SeasonsParams { + pub series_name: String, + pub provider: Option, +} + +/// GET /library/seasons +pub async fn list_seasons( + State(state): State, + CurrentUser(_user): CurrentUser, + Query(params): Query, +) -> Result>, ApiError> { + let query = ListSeasonsQuery { + series_name: params.series_name, + provider_id: params.provider, + }; + let seasons = + application::library::list_seasons::execute(&state.library_query_deps, query).await?; + Ok(Json( + seasons.into_iter().map(SeasonResponse::from).collect(), + )) +} + +#[derive(Debug, Deserialize)] +pub struct GenresParams { + pub content_type: Option, + pub provider: Option, +} + +/// GET /library/genres +pub async fn list_genres( + State(state): State, + CurrentUser(_user): CurrentUser, + Query(params): Query, +) -> Result>, ApiError> { + let query = ListGenresQuery { + content_type: params.content_type, + provider_id: params.provider, + }; + let genres = + application::library::list_genres::execute(&state.library_query_deps, query).await?; + Ok(Json(genres)) +} + +/// GET /library/sync/status +pub async fn sync_status( + State(state): State, + CurrentUser(_user): CurrentUser, +) -> Result, ApiError> { + let entries = + application::library::get_sync_status::execute(&state.library_query_deps, GetSyncStatusQuery) + .await?; + let result: Vec = entries + .into_iter() + .map(|e| { + serde_json::json!({ + "provider_id": e.provider_id(), + "started_at": e.started_at(), + "finished_at": e.finished_at().unwrap_or(""), + "items_found": e.items_found(), + "status": e.status(), + "error_msg": e.error_msg().unwrap_or(""), + }) + }) + .collect(); + Ok(Json(serde_json::Value::Array(result))) +} + +/// POST /library/sync — trigger sync (admin only) +/// +/// Validates that no sync is already running, then sends a signal to the +/// background sync task to start a sync cycle immediately. +pub async fn trigger_sync( + State(state): State, + AdminUser(_user): AdminUser, +) -> Result { + let cmd = TriggerSyncCommand { provider_id: None }; + let _provider_ids = + application::library::sync::execute(&state.library_command_deps, cmd) + .await + .map_err(|e| { + if e.to_string().contains("already running") { + ApiError::conflict(e.to_string()) + } else { + ApiError::from(e) + } + })?; + + // Signal the background sync task to run immediately + let _ = state.sync_trigger.send(()); + + Ok(axum::http::StatusCode::ACCEPTED) +} diff --git a/crates/presentation/src/handlers/mod.rs b/crates/presentation/src/handlers/mod.rs new file mode 100644 index 0000000..ddd8abc --- /dev/null +++ b/crates/presentation/src/handlers/mod.rs @@ -0,0 +1,10 @@ +pub mod admin; +pub mod auth; +pub mod channels; +pub mod config; +#[cfg(feature = "local-files")] +pub mod files; +pub mod iptv; +pub mod library; +pub mod providers; +pub mod schedule; diff --git a/crates/presentation/src/handlers/providers.rs b/crates/presentation/src/handlers/providers.rs new file mode 100644 index 0000000..05670b4 --- /dev/null +++ b/crates/presentation/src/handlers/providers.rs @@ -0,0 +1,72 @@ +//! Provider configuration CRUD handlers. + +use axum::Json; +use axum::extract::{Path, State}; + +use api_types::{ProviderConfigRequest, ProviderConfigResponse}; +use application::providers::{ + DeleteProviderCommand, GetProviderQuery, ListProvidersQuery, UpsertProviderCommand, +}; + +use crate::errors::ApiError; +use crate::extractors::AdminUser; +use crate::state::AppState; + +/// GET /admin/providers +pub async fn list_providers( + State(state): State, + AdminUser(_user): AdminUser, +) -> Result>, ApiError> { + let providers = + application::providers::list::execute(&state.provider_deps, ListProvidersQuery).await?; + Ok(Json( + providers + .into_iter() + .map(ProviderConfigResponse::from) + .collect(), + )) +} + +/// GET /admin/providers/:id +pub async fn get_provider( + State(state): State, + AdminUser(_user): AdminUser, + Path(id): Path, +) -> Result, ApiError> { + let query = GetProviderQuery { id: id.clone() }; + let provider = application::providers::get::execute(&state.provider_deps, query) + .await? + .ok_or_else(|| ApiError::not_found(format!("Provider {id} not found")))?; + Ok(Json(ProviderConfigResponse::from(provider))) +} + +/// PUT /admin/providers/:id +pub async fn upsert_provider( + State(state): State, + AdminUser(_user): AdminUser, + Path(id): Path, + Json(req): Json, +) -> Result, ApiError> { + let config_json = serde_json::to_string(&req.config) + .map_err(|e| ApiError::validation(format!("Invalid config JSON: {e}")))?; + + let cmd = UpsertProviderCommand { + id, + provider_type: req.provider_type, + config_json, + enabled: req.enabled, + }; + application::providers::upsert::execute(&state.provider_deps, cmd).await?; + Ok(Json(serde_json::json!({"status": "ok"}))) +} + +/// DELETE /admin/providers/:id +pub async fn delete_provider( + State(state): State, + AdminUser(_user): AdminUser, + Path(id): Path, +) -> Result { + let cmd = DeleteProviderCommand { id }; + application::providers::delete::execute(&state.provider_deps, cmd).await?; + Ok(axum::http::StatusCode::NO_CONTENT) +} diff --git a/crates/presentation/src/handlers/schedule.rs b/crates/presentation/src/handlers/schedule.rs new file mode 100644 index 0000000..d4de03b --- /dev/null +++ b/crates/presentation/src/handlers/schedule.rs @@ -0,0 +1,131 @@ +//! Schedule, broadcast, and stream handlers. + +use axum::Json; +use axum::extract::{Path, State}; +use axum::http::StatusCode; + +use api_types::{ + CurrentBroadcastResponse, ScheduleHistoryEntry, ScheduleResponse, SlotResponse, +}; +use application::schedule::{ + GenerateScheduleCommand, GetActiveScheduleQuery, GetCurrentBroadcastQuery, GetEpgQuery, + GetStreamUrlQuery, ListHistoryQuery, +}; + +use crate::errors::ApiError; +use crate::extractors::CurrentUser; +use crate::state::AppState; + +/// POST /channels/:id/schedule — generate a new schedule +pub async fn generate_schedule( + State(state): State, + CurrentUser(_user): CurrentUser, + Path(id): Path, +) -> Result, ApiError> { + let cmd = GenerateScheduleCommand { channel_id: id }; + let schedule = application::schedule::generate::execute(&state.schedule_deps, cmd).await?; + Ok(Json(ScheduleResponse::from(schedule))) +} + +/// GET /channels/:id/schedule — get the active schedule +pub async fn get_active_schedule( + State(state): State, + CurrentUser(_user): CurrentUser, + Path(id): Path, +) -> Result { + let query = GetActiveScheduleQuery { channel_id: id }; + match application::schedule::get_active::execute(&state.schedule_deps, query).await? { + Some(schedule) => Ok(Json(ScheduleResponse::from(schedule)).into_response()), + None => Ok(StatusCode::NO_CONTENT.into_response()), + } +} + +use axum::response::IntoResponse; + +/// GET /channels/:id/now — what's currently playing +pub async fn get_current_broadcast( + State(state): State, + Path(id): Path, +) -> Result { + let query = GetCurrentBroadcastQuery { channel_id: id }; + match application::schedule::get_current_broadcast::execute(&state.schedule_deps, query).await? + { + Some(broadcast) => { + // Look up the channel to resolve block access mode + let channel_query = application::channels::GetChannelQuery { channel_id: id }; + let channel = application::channels::get::execute(&state.channel_query_deps, channel_query).await?; + + let slot_response = match &channel { + Some(ch) => SlotResponse::with_block_access(broadcast.slot().clone(), ch), + None => SlotResponse::from(broadcast.slot().clone()), + }; + + let block_access_mode = slot_response.block_access_mode.clone(); + + Ok(Json(CurrentBroadcastResponse { + slot: slot_response, + offset_secs: broadcast.offset_secs(), + block_access_mode, + }) + .into_response()) + } + None => Ok(StatusCode::NO_CONTENT.into_response()), + } +} + +/// GET /channels/:id/epg — electronic program guide +pub async fn get_epg( + State(state): State, + Path(id): Path, +) -> Result>, ApiError> { + let query = GetEpgQuery { channel_id: id }; + let slots = application::schedule::get_epg::execute(&state.schedule_deps, query).await?; + Ok(Json(slots.into_iter().map(SlotResponse::from).collect())) +} + +/// GET /channels/:id/stream — redirect to stream URL (307) +pub async fn get_stream( + State(state): State, + Path(id): Path, +) -> Result { + // Find the current broadcast first to get the item ID + let broadcast_query = GetCurrentBroadcastQuery { channel_id: id }; + let broadcast = + application::schedule::get_current_broadcast::execute(&state.schedule_deps, broadcast_query) + .await?; + + match broadcast { + Some(b) => { + let stream_query = GetStreamUrlQuery { + channel_id: id, + item_id: b.slot().item().id().value().to_string(), + }; + let url = + application::schedule::get_stream_url::execute(&state.schedule_deps, stream_query) + .await?; + Ok(( + StatusCode::TEMPORARY_REDIRECT, + [("Location", url.as_str())], + ) + .into_response()) + } + None => Ok(StatusCode::NO_CONTENT.into_response()), + } +} + +/// GET /channels/:id/schedule/history — list schedule generations +pub async fn list_schedule_history( + State(state): State, + CurrentUser(_user): CurrentUser, + Path(id): Path, +) -> Result>, ApiError> { + let query = ListHistoryQuery { channel_id: id }; + let history = + application::schedule::list_history::execute(&state.schedule_deps, query).await?; + Ok(Json( + history + .into_iter() + .map(ScheduleHistoryEntry::from) + .collect(), + )) +} diff --git a/crates/presentation/src/main.rs b/crates/presentation/src/main.rs new file mode 100644 index 0000000..866aaf5 --- /dev/null +++ b/crates/presentation/src/main.rs @@ -0,0 +1,76 @@ +//! k-tv server entry point. + +use std::net::SocketAddr; + +use tower_http::cors::{Any, CorsLayer}; +use tower_http::trace::TraceLayer; +use tracing::info; + +mod background; +mod errors; +mod extractors; +mod factory; +mod handlers; +mod mappers; +mod routes; +mod state; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + // Load .env file if present + let _ = dotenvy::dotenv(); + + // Initialize tracing + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "info".into()), + ) + .init(); + + // Load config + let config = infra_wiring::Config::from_env() + .map_err(|e| anyhow::anyhow!("Config error: {}", e))?; + + let host = config.host.clone(); + let port = config.port; + let cors_origins = config.cors_origins.clone(); + + info!("Starting k-tv server on {}:{}", host, port); + + // Build the application state (connects DB, creates adapters, spawns background tasks) + let app_state = factory::build_app_state(config).await?; + + // Build CORS layer + let cors = if cors_origins.iter().any(|o| o == "*") { + CorsLayer::new() + .allow_origin(Any) + .allow_methods(Any) + .allow_headers(Any) + } else { + let origins: Vec<_> = cors_origins + .iter() + .filter_map(|o| o.parse().ok()) + .collect(); + CorsLayer::new() + .allow_origin(origins) + .allow_methods(Any) + .allow_headers(Any) + }; + + // Build the router + let app = axum::Router::new() + .nest("/api/v1", routes::api_v1_router()) + .layer(cors) + .layer(TraceLayer::new_for_http()) + .with_state(app_state); + + // Start serving + let addr: SocketAddr = format!("{}:{}", host, port).parse()?; + let listener = tokio::net::TcpListener::bind(addr).await?; + info!("Listening on {}", addr); + + axum::serve(listener, app).await?; + + Ok(()) +} diff --git a/crates/presentation/src/mappers/mod.rs b/crates/presentation/src/mappers/mod.rs new file mode 100644 index 0000000..52b68f9 --- /dev/null +++ b/crates/presentation/src/mappers/mod.rs @@ -0,0 +1,6 @@ +//! Domain → DTO mappings. +//! +//! Most conversions are already handled by `From` impls in the `api-types` crate. +//! This module is reserved for any presentation-layer-specific mappings that +//! don't belong in `api-types` (e.g., combining multiple domain objects into a +//! single response). diff --git a/crates/presentation/src/routes.rs b/crates/presentation/src/routes.rs new file mode 100644 index 0000000..6f6a6af --- /dev/null +++ b/crates/presentation/src/routes.rs @@ -0,0 +1,109 @@ +//! Router construction. + +use axum::{Router, routing::{delete, get, post, put}}; + +use crate::handlers; +use crate::state::AppState; + +/// Construct the API v1 router. +pub fn api_v1_router() -> Router { + Router::new() + .nest("/auth", auth_router()) + .nest("/channels", channel_router()) + .nest("/admin", admin_router()) + .nest("/admin/providers", provider_router()) + .nest("/config", config_router()) + .nest("/iptv", iptv_router()) + .nest("/library", library_router()) + .merge(local_files_router()) +} + +fn auth_router() -> Router { + let r = Router::new() + .route("/register", post(handlers::auth::register)) + .route("/login", post(handlers::auth::login)) + .route("/logout", post(handlers::auth::logout)) + .route("/me", get(handlers::auth::me)); + + #[cfg(feature = "auth-jwt")] + let r = r + .route("/token", post(handlers::auth::get_token)) + .route("/refresh", post(handlers::auth::refresh_token)); + + r +} + +fn channel_router() -> Router { + Router::new() + .route("/", get(handlers::channels::list_channels)) + .route("/", post(handlers::channels::create_channel)) + .route("/mine", get(handlers::channels::list_my_channels)) + .route("/{id}", get(handlers::channels::get_channel)) + .route("/{id}", put(handlers::channels::update_channel)) + .route("/{id}", delete(handlers::channels::delete_channel)) + // Schedule + .route("/{id}/schedule", post(handlers::schedule::generate_schedule)) + .route("/{id}/schedule", get(handlers::schedule::get_active_schedule)) + .route("/{id}/schedule/history", get(handlers::schedule::list_schedule_history)) + // Broadcast + .route("/{id}/now", get(handlers::schedule::get_current_broadcast)) + .route("/{id}/epg", get(handlers::schedule::get_epg)) + .route("/{id}/stream", get(handlers::schedule::get_stream)) + // Config snapshots + .route("/{id}/snapshots", post(handlers::channels::save_snapshot)) + .route("/{id}/snapshots", get(handlers::channels::list_snapshots)) + .route("/{id}/snapshots/{snapshot_id}", get(handlers::channels::get_snapshot)) + .route("/{id}/snapshots/{snapshot_id}", axum::routing::patch(handlers::channels::patch_snapshot)) + .route("/{id}/snapshots/{snapshot_id}/restore", post(handlers::channels::restore_snapshot)) +} + +fn admin_router() -> Router { + Router::new() + .route("/settings", get(handlers::admin::get_settings)) + .route("/settings", put(handlers::admin::update_settings)) + .route("/activity", get(handlers::admin::get_activity_log)) +} + +fn provider_router() -> Router { + Router::new() + .route("/", get(handlers::providers::list_providers)) + .route("/{id}", get(handlers::providers::get_provider)) + .route("/{id}", put(handlers::providers::upsert_provider)) + .route("/{id}", delete(handlers::providers::delete_provider)) +} + +fn config_router() -> Router { + Router::new().route("/", get(handlers::config::get_config)) +} + +fn iptv_router() -> Router { + Router::new() + .route("/playlist.m3u", get(handlers::iptv::m3u_playlist)) + .route("/epg.xml", get(handlers::iptv::xmltv_epg)) +} + +fn library_router() -> Router { + Router::new() + .route("/items", get(handlers::library::search_items)) + .route("/items/{id}", get(handlers::library::get_item)) + .route("/collections", get(handlers::library::list_collections)) + .route("/shows", get(handlers::library::list_shows)) + .route("/seasons", get(handlers::library::list_seasons)) + .route("/genres", get(handlers::library::list_genres)) + .route("/sync/status", get(handlers::library::sync_status)) + .route("/sync", post(handlers::library::trigger_sync)) +} + +fn local_files_router() -> Router { + #[cfg(feature = "local-files")] + { + Router::new() + .route("/files/stream/{id}", get(handlers::files::stream_file)) + .route("/files/rescan", post(handlers::files::rescan)) + } + + #[cfg(not(feature = "local-files"))] + { + Router::new() + } +} diff --git a/crates/presentation/src/state.rs b/crates/presentation/src/state.rs new file mode 100644 index 0000000..525cea9 --- /dev/null +++ b/crates/presentation/src/state.rs @@ -0,0 +1,51 @@ +//! Application state — holds pre-built Deps structs from the application layer. + +use std::sync::Arc; + +use application::{ + admin::AdminDeps, + auth::AuthDeps, + channels::{ChannelCommandDeps, ChannelQueryDeps}, + config_snapshots::ConfigSnapshotDeps, + iptv::IptvDeps, + library::{LibraryCommandDeps, LibraryQueryDeps}, + providers::ProviderDeps, + schedule::ScheduleDeps, +}; + +/// Shared application state, passed to all handlers via `State`. +#[derive(Clone)] +pub struct AppState { + pub auth_deps: Arc, + pub channel_command_deps: Arc, + pub channel_query_deps: Arc, + pub config_snapshot_deps: Arc, + pub schedule_deps: Arc, + pub library_command_deps: Arc, + pub library_query_deps: Arc, + pub admin_deps: Arc, + pub iptv_deps: Arc, + pub provider_deps: Arc, + + /// JWT validator for token creation/validation in auth handlers. + #[cfg(feature = "auth-jwt")] + pub jwt_validator: Option>, + + /// Provider registry for config/capabilities endpoints. + pub provider_registry: Arc, + + /// Library sync adapter — needed for spawning background sync tasks. + pub library_sync: Arc, + + /// App settings — read by library sync background task. + pub settings_repo: Arc, + + /// Event bus for domain events. + pub event_bus: Arc, + + /// Application config. + pub config: Arc, + + /// Trigger for on-demand library sync (sends () to wake the background task). + pub sync_trigger: tokio::sync::watch::Sender<()>, +}