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

remove OIDC/Postgres, replace ApiError w/ AppError, move params to api-types

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-07-12 05:14:17 +02:00
parent c0e685a4ee
commit 9b18d3ff6d
42 changed files with 243 additions and 3367 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
crates/adapters/auth/Cargo.toml
View File

@@ -6,7 +6,6 @@ edition = "2024"
[features]
default = ["jwt"]
jwt = ["dep:jsonwebtoken"]
oidc = ["dep:openidconnect", "dep:reqwest", "dep:url"]
[dependencies]
domain = { workspace = true }
@@ -18,10 +17,5 @@ serde_json = { workspace = true }
# JWT deps
jsonwebtoken = { workspace = true, optional = true }
# OIDC deps (optional)
openidconnect = { version = "4", optional = true }
reqwest = { workspace = true, optional = true }
url = { workspace = true, optional = true }
# Password hashing
password-auth = "1"

6
crates/adapters/auth/src/lib.rs
View File

@@ -3,13 +3,7 @@ pub mod password;
#[cfg(feature = "jwt")]
pub mod jwt;
#[cfg(feature = "oidc")]
pub mod oidc;
pub use password::PasswordAuthService;
#[cfg(feature = "jwt")]
pub use jwt::{JwtClaims, JwtConfig, JwtError, JwtValidator};
#[cfg(feature = "oidc")]
pub use oidc::{OidcService, OidcState, OidcUser};

245
crates/adapters/auth/src/oidc.rs
View File

@@ -1,245 +0,0 @@
use domain::{
AuthorizationCode, AuthorizationUrlData, ClientId, ClientSecret, CsrfToken, IssuerUrl,
OidcNonce, PkceVerifier, RedirectUrl, ResourceId,
};
use openidconnect::{
AccessTokenHash, Client, EmptyAdditionalClaims, EndpointMaybeSet, EndpointNotSet, EndpointSet,
OAuth2TokenResponse, PkceCodeChallenge, Scope, StandardErrorResponse, TokenResponse,
UserInfoClaims,
core::{
CoreAuthDisplay, CoreAuthPrompt, CoreAuthenticationFlow, CoreClient, CoreErrorResponseType,
CoreGenderClaim, CoreJsonWebKey, CoreJweContentEncryptionAlgorithm, CoreProviderMetadata,
CoreRevocableToken, CoreRevocationErrorResponse, CoreTokenIntrospectionResponse,
CoreTokenResponse,
},
reqwest,
};
use serde::{Deserialize, Serialize};
pub type OidcClient = Client<
EmptyAdditionalClaims,
CoreAuthDisplay,
CoreGenderClaim,
CoreJweContentEncryptionAlgorithm,
CoreJsonWebKey,
CoreAuthPrompt,
StandardErrorResponse<CoreErrorResponseType>,
CoreTokenResponse,
CoreTokenIntrospectionResponse,
CoreRevocableToken,
CoreRevocationErrorResponse,
EndpointSet,
EndpointNotSet,
EndpointNotSet,
EndpointNotSet,
EndpointMaybeSet,
EndpointMaybeSet,
>;
#[derive(Debug, thiserror::Error)]
pub enum OidcError {
#[error("OIDC discovery failed: {0}")]
Discovery(String),
#[error("Token exchange failed: {0}")]
TokenExchange(String),
#[error("ID token verification failed: {0}")]
IdTokenVerification(String),
#[error("Missing ID token in response")]
MissingIdToken,
#[error("Invalid access token hash")]
InvalidAccessTokenHash,
#[error("User has no email address")]
MissingEmail,
#[error("HTTP client error: {0}")]
Http(String),
}
#[derive(Debug, Serialize, Deserialize)]
pub struct OidcState {
pub csrf_token: CsrfToken,
pub nonce: OidcNonce,
pub pkce_verifier: PkceVerifier,
}
#[derive(Debug)]
pub struct OidcUser {
pub subject: String,
pub email: String,
}
#[derive(Clone)]
pub struct OidcService {
client: OidcClient,
http_client: reqwest::Client,
resource_id: Option<ResourceId>,
}
impl OidcService {
pub async fn new(
issuer: IssuerUrl,
client_id: ClientId,
client_secret: Option<ClientSecret>,
redirect_url: RedirectUrl,
resource_id: Option<ResourceId>,
) -> Result<Self, OidcError> {
tracing::debug!("OIDC setup: client_id={client_id}, redirect={redirect_url}");
tracing::debug!(
"OIDC setup: secret={}",
if client_secret.is_some() {
"SET"
} else {
"NONE"
}
);
let http_client = reqwest::ClientBuilder::new()
.redirect(reqwest::redirect::Policy::none())
.build()
.map_err(|e| OidcError::Http(e.to_string()))?;
let provider_metadata = CoreProviderMetadata::discover_async(
openidconnect::IssuerUrl::new(issuer.as_ref().to_string())
.map_err(|e| OidcError::Discovery(e.to_string()))?,
&http_client,
)
.await
.map_err(|e| OidcError::Discovery(e.to_string()))?;
let oidc_client_id = openidconnect::ClientId::new(client_id.as_ref().to_string());
let oidc_client_secret = client_secret
.as_ref()
.filter(|s| !s.is_empty())
.map(|s| openidconnect::ClientSecret::new(s.as_ref().to_string()));
let oidc_redirect_url =
openidconnect::RedirectUrl::new(redirect_url.as_ref().to_string())
.map_err(|e| OidcError::Discovery(e.to_string()))?;
let client = CoreClient::from_provider_metadata(
provider_metadata,
oidc_client_id,
oidc_client_secret,
)
.set_redirect_uri(oidc_redirect_url);
Ok(Self {
client,
http_client,
resource_id,
})
}
pub fn get_authorization_url(&self) -> (AuthorizationUrlData, OidcState) {
let (pkce_challenge, pkce_verifier) = PkceCodeChallenge::new_random_sha256();
let (auth_url, csrf_token, nonce) = self
.client
.authorize_url(
CoreAuthenticationFlow::AuthorizationCode,
openidconnect::CsrfToken::new_random,
openidconnect::Nonce::new_random,
)
.add_scope(Scope::new("profile".to_string()))
.add_scope(Scope::new("email".to_string()))
.set_pkce_challenge(pkce_challenge)
.url();
let oidc_state = OidcState {
csrf_token: CsrfToken::new(csrf_token.secret().to_string()),
nonce: OidcNonce::new(nonce.secret().to_string()),
pkce_verifier: PkceVerifier::new(pkce_verifier.secret().to_string()),
};
let auth_data = AuthorizationUrlData {
url: auth_url.into(),
csrf_token: oidc_state.csrf_token.clone(),
nonce: oidc_state.nonce.clone(),
pkce_verifier: oidc_state.pkce_verifier.clone(),
};
(auth_data, oidc_state)
}
pub async fn resolve_callback(
&self,
code: AuthorizationCode,
nonce: OidcNonce,
pkce_verifier: PkceVerifier,
) -> Result<OidcUser, OidcError> {
let oidc_pkce_verifier =
openidconnect::PkceCodeVerifier::new(pkce_verifier.as_ref().to_string());
let oidc_nonce = openidconnect::Nonce::new(nonce.as_ref().to_string());
let token_response = self
.client
.exchange_code(openidconnect::AuthorizationCode::new(
code.as_ref().to_string(),
))
.map_err(|e| OidcError::TokenExchange(e.to_string()))?
.set_pkce_verifier(oidc_pkce_verifier)
.request_async(&self.http_client)
.await
.map_err(|e| OidcError::TokenExchange(e.to_string()))?;
let id_token = token_response
.id_token()
.ok_or(OidcError::MissingIdToken)?;
let mut id_token_verifier = self.client.id_token_verifier().clone();
if let Some(resource_id) = &self.resource_id {
let trusted = resource_id.as_ref().to_string();
id_token_verifier =
id_token_verifier.set_other_audience_verifier_fn(move |aud| aud.as_str() == trusted);
}
let claims = id_token
.claims(&id_token_verifier, &oidc_nonce)
.map_err(|e| OidcError::IdTokenVerification(e.to_string()))?;
if let Some(expected_hash) = claims.access_token_hash() {
let actual_hash = AccessTokenHash::from_token(
token_response.access_token(),
id_token
.signing_alg()
.map_err(|e| OidcError::IdTokenVerification(e.to_string()))?,
id_token
.signing_key(&id_token_verifier)
.map_err(|e| OidcError::IdTokenVerification(e.to_string()))?,
)
.map_err(|e| OidcError::IdTokenVerification(e.to_string()))?;
if actual_hash != *expected_hash {
return Err(OidcError::InvalidAccessTokenHash);
}
}
let email = if let Some(email) = claims.email() {
Some(email.as_str().to_string())
} else {
tracing::debug!("Email missing in ID token, fetching UserInfo");
let user_info: UserInfoClaims<EmptyAdditionalClaims, CoreGenderClaim> = self
.client
.user_info(token_response.access_token().clone(), None)
.map_err(|e| OidcError::TokenExchange(e.to_string()))?
.request_async(&self.http_client)
.await
.map_err(|e| OidcError::TokenExchange(e.to_string()))?;
user_info.email().map(|e| e.as_str().to_string())
};
let email = email.ok_or(OidcError::MissingEmail)?;
Ok(OidcUser {
subject: claims.subject().to_string(),
email,
})
}
}

16
crates/adapters/postgres/Cargo.toml
View File

@@ -1,16 +0,0 @@
[package]
name = "adapter-postgres"
version = "0.1.0"
edition = "2024"
[dependencies]
domain = { workspace = true }
adapter-common = { workspace = true }
infra-wiring = { workspace = true, features = ["postgres"] }
async-trait = { workspace = true }
sqlx = { workspace = true, features = ["postgres"] }
chrono = { workspace = true }
uuid = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
tracing = { workspace = true }

83
crates/adapters/postgres/src/activity.rs
View File

@@ -1,83 +0,0 @@
use async_trait::async_trait;
use chrono::Utc;
use sqlx::PgPool;
use uuid::Uuid;
use adapter_common::{map_sqlx_error, parse_dt, parse_uuid};
use domain::{
ports::activity::{ActivityLogCommand, ActivityLogQuery},
ActivityEvent, ActivityEventId, ChannelId, DomainResult,
};
pub struct PgActivityLog {
pool: PgPool,
}
impl PgActivityLog {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
#[async_trait]
impl ActivityLogCommand for PgActivityLog {
async fn log(
&self,
event_type: &str,
detail: &str,
channel_id: Option<ChannelId>,
) -> DomainResult<()> {
let id = Uuid::new_v4().to_string();
let timestamp = Utc::now().to_rfc3339();
let channel_id_str = channel_id.map(|id| id.value().to_string());
sqlx::query(
"INSERT INTO activity_log (id, timestamp, event_type, detail, channel_id) VALUES ($1, $2, $3, $4, $5)",
)
.bind(&id)
.bind(&timestamp)
.bind(event_type)
.bind(detail)
.bind(&channel_id_str)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
}
#[async_trait]
impl ActivityLogQuery for PgActivityLog {
async fn recent(&self, limit: u32) -> DomainResult<Vec<ActivityEvent>> {
let rows: Vec<(String, String, String, String, Option<String>)> = sqlx::query_as(
"SELECT id, timestamp, event_type, detail, channel_id FROM activity_log ORDER BY timestamp DESC LIMIT $1",
)
.bind(limit as i64)
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
let mut events = Vec::with_capacity(rows.len());
for (id_str, ts_str, event_type, detail, channel_id_str) in rows {
let Ok(id) = parse_uuid(&id_str, "activity id") else {
continue;
};
let Ok(timestamp) = parse_dt(&ts_str) else {
continue;
};
let channel_id = channel_id_str
.and_then(|s| Uuid::parse_str(&s).ok())
.map(ChannelId::from_uuid);
events.push(ActivityEvent::from_persistence(
ActivityEventId::from_uuid(id),
timestamp,
event_type,
detail,
channel_id,
));
}
Ok(events)
}
}

329
crates/adapters/postgres/src/channel.rs
View File

@@ -1,329 +0,0 @@
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use sqlx::{PgPool, Row};
use uuid::Uuid;
use adapter_common::{
map_sqlx_error, parse_dt, parse_enum_or_default, parse_recycle_policy, parse_schedule_config,
parse_uuid, serialize_enum_as_string,
};
use domain::{
ports::channel::{ChannelCommand, ChannelQuery},
Channel, ChannelConfigSnapshot, ChannelId, ChannelRow as DomainChannelRow, DomainError,
DomainResult, ScheduleConfig, SnapshotId, UserId,
};
pub struct PgChannelRepository {
pool: PgPool,
}
impl PgChannelRepository {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
const SELECT_COLS: &str = "id, owner_id, name, description, timezone, schedule_config, recycle_policy, auto_schedule, access_mode, access_password_hash, logo, logo_position, logo_opacity, webhook_url, webhook_poll_interval_secs, webhook_body_template, webhook_headers, created_at, updated_at";
#[derive(Debug, sqlx::FromRow)]
struct ChannelRow {
id: String,
owner_id: String,
name: String,
description: Option<String>,
timezone: String,
schedule_config: String,
recycle_policy: String,
auto_schedule: bool,
access_mode: String,
access_password_hash: Option<String>,
logo: Option<String>,
logo_position: String,
logo_opacity: f32,
webhook_url: Option<String>,
webhook_poll_interval_secs: i64,
webhook_body_template: Option<String>,
webhook_headers: Option<String>,
created_at: String,
updated_at: String,
}
impl ChannelRow {
fn into_channel(self) -> DomainResult<Channel> {
Ok(Channel::from_persistence(DomainChannelRow {
id: ChannelId::from_uuid(parse_uuid(&self.id, "channel id")?),
owner_id: UserId::from_uuid(parse_uuid(&self.owner_id, "owner id")?),
name: self.name,
description: self.description,
timezone: self.timezone,
schedule_config: parse_schedule_config(&self.schedule_config)?,
recycle_policy: parse_recycle_policy(&self.recycle_policy)?,
auto_schedule: self.auto_schedule,
access_mode: parse_enum_or_default(self.access_mode),
access_password_hash: self.access_password_hash,
logo: self.logo,
logo_position: parse_enum_or_default(self.logo_position),
logo_opacity: self.logo_opacity,
webhook_url: self.webhook_url,
webhook_poll_interval_secs: self.webhook_poll_interval_secs as u32,
webhook_body_template: self.webhook_body_template,
webhook_headers: self.webhook_headers,
created_at: parse_dt(&self.created_at)?,
updated_at: parse_dt(&self.updated_at)?,
}))
}
}
fn map_snapshot_row(
row: &sqlx::postgres::PgRow,
channel_id: ChannelId,
) -> DomainResult<ChannelConfigSnapshot> {
let id_str: String = row.get("id");
let id = SnapshotId::from_uuid(parse_uuid(&id_str, "snapshot id")?);
let config_json: String = row.get("config_json");
let config = parse_schedule_config(&config_json)?;
let version_num: i64 = row.get("version_num");
let label: Option<String> = row.get("label");
let created_at_str: String = row.get("created_at");
let created_at: DateTime<Utc> = parse_dt(&created_at_str)?;
Ok(ChannelConfigSnapshot::from_persistence(
id,
channel_id,
config,
version_num,
label,
created_at,
))
}
#[async_trait]
impl ChannelCommand for PgChannelRepository {
async fn save(&self, channel: &Channel) -> DomainResult<()> {
let schedule_config = serde_json::to_string(channel.schedule_config())
.map_err(|e| DomainError::RepositoryError(format!("serialize schedule_config: {e}")))?;
let recycle_policy = serde_json::to_string(channel.recycle_policy())
.map_err(|e| DomainError::RepositoryError(format!("serialize recycle_policy: {e}")))?;
let access_mode = serialize_enum_as_string(channel.access_mode(), "public");
let logo_position = serialize_enum_as_string(channel.logo_position(), "top_right");
sqlx::query(
r#"
INSERT INTO channels
(id, owner_id, name, description, timezone, schedule_config, recycle_policy,
auto_schedule, access_mode, access_password_hash, logo, logo_position,
logo_opacity, webhook_url, webhook_poll_interval_secs, webhook_body_template,
webhook_headers, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)
ON CONFLICT(id) DO UPDATE SET
name = EXCLUDED.name,
description = EXCLUDED.description,
timezone = EXCLUDED.timezone,
schedule_config = EXCLUDED.schedule_config,
recycle_policy = EXCLUDED.recycle_policy,
auto_schedule = EXCLUDED.auto_schedule,
access_mode = EXCLUDED.access_mode,
access_password_hash = EXCLUDED.access_password_hash,
logo = EXCLUDED.logo,
logo_position = EXCLUDED.logo_position,
logo_opacity = EXCLUDED.logo_opacity,
webhook_url = EXCLUDED.webhook_url,
webhook_poll_interval_secs = EXCLUDED.webhook_poll_interval_secs,
webhook_body_template = EXCLUDED.webhook_body_template,
webhook_headers = EXCLUDED.webhook_headers,
updated_at = EXCLUDED.updated_at
"#,
)
.bind(channel.id().value().to_string())
.bind(channel.owner_id().value().to_string())
.bind(channel.name())
.bind(channel.description())
.bind(channel.timezone())
.bind(&schedule_config)
.bind(&recycle_policy)
.bind(channel.auto_schedule())
.bind(&access_mode)
.bind(channel.access_password_hash())
.bind(channel.logo())
.bind(&logo_position)
.bind(channel.logo_opacity())
.bind(channel.webhook_url())
.bind(channel.webhook_poll_interval_secs() as i64)
.bind(channel.webhook_body_template())
.bind(channel.webhook_headers())
.bind(channel.created_at().to_rfc3339())
.bind(channel.updated_at().to_rfc3339())
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
async fn delete(&self, id: ChannelId) -> DomainResult<()> {
sqlx::query("DELETE FROM channels WHERE id = $1")
.bind(id.value().to_string())
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
async fn save_config_snapshot(
&self,
channel_id: ChannelId,
config: &ScheduleConfig,
label: Option<String>,
) -> DomainResult<ChannelConfigSnapshot> {
let id = Uuid::new_v4();
let now = Utc::now();
let config_json = serde_json::to_string(config)
.map_err(|e| DomainError::RepositoryError(e.to_string()))?;
let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
let version_num: i64 = sqlx::query_scalar(
"SELECT COALESCE(MAX(version_num), 0) + 1 FROM channel_config_snapshots WHERE channel_id = $1",
)
.bind(channel_id.value().to_string())
.fetch_one(&mut *tx)
.await
.map_err(map_sqlx_error)?;
sqlx::query(
"INSERT INTO channel_config_snapshots (id, channel_id, config_json, version_num, label, created_at)
VALUES ($1, $2, $3, $4, $5, $6)",
)
.bind(id.to_string())
.bind(channel_id.value().to_string())
.bind(&config_json)
.bind(version_num)
.bind(&label)
.bind(now.to_rfc3339())
.execute(&mut *tx)
.await
.map_err(map_sqlx_error)?;
tx.commit().await.map_err(map_sqlx_error)?;
Ok(ChannelConfigSnapshot::from_persistence(
SnapshotId::from_uuid(id),
channel_id,
config.clone(),
version_num,
label,
now,
))
}
async fn patch_config_snapshot_label(
&self,
channel_id: ChannelId,
snapshot_id: SnapshotId,
label: Option<String>,
) -> DomainResult<Option<ChannelConfigSnapshot>> {
let updated = sqlx::query(
"UPDATE channel_config_snapshots SET label = $1 WHERE id = $2 AND channel_id = $3 RETURNING id",
)
.bind(&label)
.bind(snapshot_id.value().to_string())
.bind(channel_id.value().to_string())
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
if updated.is_none() {
return Ok(None);
}
self.get_config_snapshot(channel_id, snapshot_id).await
}
}
#[async_trait]
impl ChannelQuery for PgChannelRepository {
async fn find_by_id(&self, id: ChannelId) -> DomainResult<Option<Channel>> {
let sql = format!("SELECT {SELECT_COLS} FROM channels WHERE id = $1");
let row: Option<ChannelRow> = sqlx::query_as(&sql)
.bind(id.value().to_string())
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
row.map(ChannelRow::into_channel).transpose()
}
async fn find_by_owner(&self, owner_id: UserId) -> DomainResult<Vec<Channel>> {
let sql = format!(
"SELECT {SELECT_COLS} FROM channels WHERE owner_id = $1 ORDER BY created_at ASC"
);
let rows: Vec<ChannelRow> = sqlx::query_as(&sql)
.bind(owner_id.value().to_string())
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
rows.into_iter().map(ChannelRow::into_channel).collect()
}
async fn find_all(&self) -> DomainResult<Vec<Channel>> {
let sql = format!("SELECT {SELECT_COLS} FROM channels ORDER BY created_at ASC");
let rows: Vec<ChannelRow> = sqlx::query_as(&sql)
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
rows.into_iter().map(ChannelRow::into_channel).collect()
}
async fn find_auto_schedule_enabled(&self) -> DomainResult<Vec<Channel>> {
let sql = format!(
"SELECT {SELECT_COLS} FROM channels WHERE auto_schedule = TRUE ORDER BY created_at ASC"
);
let rows: Vec<ChannelRow> = sqlx::query_as(&sql)
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
rows.into_iter().map(ChannelRow::into_channel).collect()
}
async fn list_config_snapshots(
&self,
channel_id: ChannelId,
) -> DomainResult<Vec<ChannelConfigSnapshot>> {
let rows = sqlx::query(
"SELECT id, config_json, version_num, label, created_at
FROM channel_config_snapshots WHERE channel_id = $1
ORDER BY version_num DESC",
)
.bind(channel_id.value().to_string())
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
rows.iter()
.map(|row| map_snapshot_row(row, channel_id))
.collect()
}
async fn get_config_snapshot(
&self,
channel_id: ChannelId,
snapshot_id: SnapshotId,
) -> DomainResult<Option<ChannelConfigSnapshot>> {
let row = sqlx::query(
"SELECT id, config_json, version_num, label, created_at
FROM channel_config_snapshots WHERE id = $1 AND channel_id = $2",
)
.bind(snapshot_id.value().to_string())
.bind(channel_id.value().to_string())
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
match row {
None => Ok(None),
Some(row) => Ok(Some(map_snapshot_row(&row, channel_id)?)),
}
}
}

11
crates/adapters/postgres/src/lib.rs
View File

@@ -1,11 +0,0 @@
pub mod activity;
pub mod channel;
pub mod library;
pub mod provider_config;
pub mod schedule;
pub mod settings;
pub mod transcode;
pub mod user;
pub mod wire;
pub use wire::{wire, PostgresWireOutput};

530
crates/adapters/postgres/src/library.rs
View File

@@ -1,530 +0,0 @@
use async_trait::async_trait;
use sqlx::PgPool;
use adapter_common::{content_type_str, parse_content_type, parse_genres_blob};
use domain::{
ports::library::{LibraryCommand, LibraryQuery},
ContentType, DomainError, DomainResult, LibraryCollection, LibraryItem,
LibraryItemRow as DomainLibraryItemRow, LibrarySearchFilter, LibrarySyncLogEntry,
LibrarySyncResult, SeasonSummary, ShowSummary,
};
pub struct PgLibraryRepository {
pool: PgPool,
}
impl PgLibraryRepository {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
#[derive(sqlx::FromRow)]
struct LibraryItemRow {
id: String,
provider_id: String,
external_id: String,
title: String,
content_type: String,
duration_secs: i64,
series_name: Option<String>,
season_number: Option<i64>,
episode_number: Option<i64>,
year: Option<i64>,
genres: String,
tags: String,
collection_id: Option<String>,
collection_name: Option<String>,
collection_type: Option<String>,
thumbnail_url: Option<String>,
synced_at: String,
}
impl LibraryItemRow {
fn into_library_item(self) -> LibraryItem {
LibraryItem::from_persistence(DomainLibraryItemRow {
id: self.id,
provider_id: self.provider_id,
external_id: self.external_id,
title: self.title,
content_type: parse_content_type(&self.content_type),
duration_secs: self.duration_secs as u32,
series_name: self.series_name,
season_number: self.season_number.map(|n| n as u32),
episode_number: self.episode_number.map(|n| n as u32),
year: self.year.map(|n| n as u16),
genres: serde_json::from_str(&self.genres).unwrap_or_default(),
tags: serde_json::from_str(&self.tags).unwrap_or_default(),
collection_id: self.collection_id,
collection_name: self.collection_name,
collection_type: self.collection_type,
thumbnail_url: self.thumbnail_url,
synced_at: self.synced_at,
})
}
}
#[derive(sqlx::FromRow)]
struct SyncLogRow {
id: i64,
provider_id: String,
started_at: String,
finished_at: Option<String>,
items_found: i64,
status: String,
error_msg: Option<String>,
}
#[derive(sqlx::FromRow)]
struct ShowSummaryRow {
series_name: String,
episode_count: i64,
season_count: i64,
thumbnail_url: Option<String>,
genres_blob: String,
}
#[derive(sqlx::FromRow)]
struct SeasonSummaryRow {
season_number: i64,
episode_count: i64,
thumbnail_url: Option<String>,
}
#[async_trait]
impl LibraryCommand for PgLibraryRepository {
async fn upsert_items(&self, _provider_id: &str, items: Vec<LibraryItem>) -> DomainResult<()> {
let mut tx = self
.pool
.begin()
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
for item in items {
sqlx::query(
"INSERT INTO library_items
(id, provider_id, external_id, title, content_type, duration_secs,
series_name, season_number, episode_number, year, genres, tags,
collection_id, collection_name, collection_type, thumbnail_url, synced_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)
ON CONFLICT(id) DO UPDATE SET
provider_id = EXCLUDED.provider_id,
external_id = EXCLUDED.external_id,
title = EXCLUDED.title,
content_type = EXCLUDED.content_type,
duration_secs = EXCLUDED.duration_secs,
series_name = EXCLUDED.series_name,
season_number = EXCLUDED.season_number,
episode_number = EXCLUDED.episode_number,
year = EXCLUDED.year,
genres = EXCLUDED.genres,
tags = EXCLUDED.tags,
collection_id = EXCLUDED.collection_id,
collection_name = EXCLUDED.collection_name,
collection_type = EXCLUDED.collection_type,
thumbnail_url = EXCLUDED.thumbnail_url,
synced_at = EXCLUDED.synced_at",
)
.bind(item.id())
.bind(item.provider_id())
.bind(item.external_id())
.bind(item.title())
.bind(content_type_str(item.content_type()))
.bind(item.duration_secs() as i64)
.bind(item.series_name())
.bind(item.season_number().map(|n| n as i64))
.bind(item.episode_number().map(|n| n as i64))
.bind(item.year().map(|n| n as i64))
.bind(serde_json::to_string(item.genres()).unwrap_or_default())
.bind(serde_json::to_string(item.tags()).unwrap_or_default())
.bind(item.collection_id())
.bind(item.collection_name())
.bind(item.collection_type())
.bind(item.thumbnail_url())
.bind(item.synced_at())
.execute(&mut *tx)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
}
tx.commit()
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
async fn clear_provider(&self, provider_id: &str) -> DomainResult<()> {
sqlx::query("DELETE FROM library_items WHERE provider_id = $1")
.bind(provider_id)
.execute(&self.pool)
.await
.map(|_| ())
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
async fn log_sync_start(&self, provider_id: &str) -> DomainResult<i64> {
let now = chrono::Utc::now().to_rfc3339();
let id = sqlx::query_scalar::<_, i64>(
"INSERT INTO library_sync_log (provider_id, started_at, status)
VALUES ($1, $2, 'running') RETURNING id",
)
.bind(provider_id)
.bind(&now)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(id)
}
async fn log_sync_finish(&self, log_id: i64, result: &LibrarySyncResult) -> DomainResult<()> {
let now = chrono::Utc::now().to_rfc3339();
let status = if result.error().is_none() {
"done"
} else {
"error"
};
sqlx::query(
"UPDATE library_sync_log
SET finished_at = $1, items_found = $2, status = $3, error_msg = $4
WHERE id = $5",
)
.bind(&now)
.bind(result.items_found() as i64)
.bind(status)
.bind(result.error())
.bind(log_id)
.execute(&self.pool)
.await
.map(|_| ())
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
}
#[async_trait]
impl LibraryQuery for PgLibraryRepository {
async fn search(
&self,
filter: &LibrarySearchFilter,
) -> DomainResult<(Vec<LibraryItem>, u32)> {
let mut conditions: Vec<String> = vec![];
if let Some(p) = filter.provider_id() {
conditions.push(format!("provider_id = '{}'", p.replace('\'', "''")));
}
if let Some(ct) = filter.content_type() {
conditions.push(format!("content_type = '{}'", content_type_str(ct)));
}
if let Some(st) = filter.search_term() {
conditions.push(format!("title ILIKE '%{}%'", st.replace('\'', "''")));
}
if let Some(cid) = filter.collection_id() {
conditions.push(format!("collection_id = '{}'", cid.replace('\'', "''")));
}
if let Some(decade) = filter.decade() {
let end = decade + 10;
conditions.push(format!("year >= {} AND year < {}", decade, end));
}
if let Some(min) = filter.min_duration_secs() {
conditions.push(format!("duration_secs >= {}", min));
}
if let Some(max) = filter.max_duration_secs() {
conditions.push(format!("duration_secs <= {}", max));
}
if !filter.series_names().is_empty() {
let quoted: Vec<String> = filter
.series_names()
.iter()
.map(|s| format!("'{}'", s.replace('\'', "''")))
.collect();
conditions.push(format!("series_name IN ({})", quoted.join(",")));
}
if !filter.genres().is_empty() {
let genre_conditions: Vec<String> = filter
.genres()
.iter()
.map(|g| {
format!(
"EXISTS (SELECT 1 FROM jsonb_array_elements_text(library_items.genres::jsonb) je WHERE je = '{}')",
g.replace('\'', "''")
)
})
.collect();
conditions.push(format!("({})", genre_conditions.join(" OR ")));
}
if let Some(sn) = filter.season_number() {
conditions.push(format!("season_number = {}", sn));
}
let where_clause = if conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", conditions.join(" AND "))
};
let count_sql = format!("SELECT COUNT(*) FROM library_items {}", where_clause);
let total: i64 = sqlx::query_scalar(&count_sql)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
let items_sql = format!(
"SELECT * FROM library_items {} ORDER BY title ASC LIMIT {} OFFSET {}",
where_clause,
filter.limit(),
filter.offset()
);
let rows = sqlx::query_as::<_, LibraryItemRow>(&items_sql)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok((
rows.into_iter()
.map(LibraryItemRow::into_library_item)
.collect(),
total as u32,
))
}
async fn get_by_id(&self, id: &str) -> DomainResult<Option<LibraryItem>> {
let row = sqlx::query_as::<_, LibraryItemRow>(
"SELECT * FROM library_items WHERE id = $1",
)
.bind(id)
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(row.map(LibraryItemRow::into_library_item))
}
async fn list_collections(
&self,
provider_id: Option<&str>,
) -> DomainResult<Vec<LibraryCollection>> {
let rows: Vec<(String, Option<String>, Option<String>)> = if let Some(p) = provider_id {
sqlx::query_as(
"SELECT DISTINCT collection_id, collection_name, collection_type
FROM library_items WHERE collection_id IS NOT NULL AND provider_id = $1
ORDER BY collection_name ASC",
)
.bind(p)
.fetch_all(&self.pool)
.await
} else {
sqlx::query_as(
"SELECT DISTINCT collection_id, collection_name, collection_type
FROM library_items WHERE collection_id IS NOT NULL
ORDER BY collection_name ASC",
)
.fetch_all(&self.pool)
.await
}
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(rows
.into_iter()
.map(|(id, name, ct)| {
LibraryCollection::from_persistence(id, name.unwrap_or_default(), ct)
})
.collect())
}
async fn list_series(&self, provider_id: Option<&str>) -> DomainResult<Vec<String>> {
let rows: Vec<(String,)> = if let Some(p) = provider_id {
sqlx::query_as(
"SELECT DISTINCT series_name FROM library_items
WHERE series_name IS NOT NULL AND provider_id = $1 ORDER BY series_name ASC",
)
.bind(p)
.fetch_all(&self.pool)
.await
} else {
sqlx::query_as(
"SELECT DISTINCT series_name FROM library_items
WHERE series_name IS NOT NULL ORDER BY series_name ASC",
)
.fetch_all(&self.pool)
.await
}
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(rows.into_iter().map(|(s,)| s).collect())
}
async fn list_genres(
&self,
content_type: Option<&ContentType>,
provider_id: Option<&str>,
) -> DomainResult<Vec<String>> {
let sql = match (content_type, provider_id) {
(Some(ct), Some(p)) => format!(
"SELECT DISTINCT je AS value FROM library_items li, \
LATERAL jsonb_array_elements_text(li.genres::jsonb) je \
WHERE li.content_type = '{}' AND li.provider_id = '{}' ORDER BY value ASC",
content_type_str(ct),
p.replace('\'', "''")
),
(Some(ct), None) => format!(
"SELECT DISTINCT je AS value FROM library_items li, \
LATERAL jsonb_array_elements_text(li.genres::jsonb) je \
WHERE li.content_type = '{}' ORDER BY value ASC",
content_type_str(ct)
),
(None, Some(p)) => format!(
"SELECT DISTINCT je AS value FROM library_items li, \
LATERAL jsonb_array_elements_text(li.genres::jsonb) je \
WHERE li.provider_id = '{}' ORDER BY value ASC",
p.replace('\'', "''")
),
(None, None) => {
"SELECT DISTINCT je AS value FROM library_items li, \
LATERAL jsonb_array_elements_text(li.genres::jsonb) je \
ORDER BY value ASC"
.to_string()
}
};
let rows: Vec<(String,)> = sqlx::query_as(&sql)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(rows.into_iter().map(|(s,)| s).collect())
}
async fn latest_sync_status(&self) -> DomainResult<Vec<LibrarySyncLogEntry>> {
let rows = sqlx::query_as::<_, SyncLogRow>(
"SELECT * FROM library_sync_log
WHERE id IN (
SELECT MAX(id) FROM library_sync_log GROUP BY provider_id
)
ORDER BY started_at DESC",
)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(rows
.into_iter()
.map(|r| {
LibrarySyncLogEntry::from_persistence(
r.id,
r.provider_id,
r.started_at,
r.finished_at,
r.items_found as u32,
r.status,
r.error_msg,
)
})
.collect())
}
async fn is_sync_running(&self, provider_id: &str) -> DomainResult<bool> {
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM library_sync_log WHERE provider_id = $1 AND status = 'running'",
)
.bind(provider_id)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(count > 0)
}
async fn list_shows(
&self,
provider_id: Option<&str>,
search_term: Option<&str>,
genres: &[String],
) -> DomainResult<Vec<ShowSummary>> {
let mut conditions = vec![
"content_type = 'episode'".to_string(),
"series_name IS NOT NULL".to_string(),
];
if let Some(p) = provider_id {
conditions.push(format!("provider_id = '{}'", p.replace('\'', "''")));
}
if let Some(st) = search_term {
let escaped = st.replace('\'', "''");
conditions.push(format!(
"(title ILIKE '%{escaped}%' OR series_name ILIKE '%{escaped}%')"
));
}
if !genres.is_empty() {
let genre_conditions: Vec<String> = genres
.iter()
.map(|g| {
format!(
"EXISTS (SELECT 1 FROM jsonb_array_elements_text(library_items.genres::jsonb) je WHERE je = '{}')",
g.replace('\'', "''")
)
})
.collect();
conditions.push(format!("({})", genre_conditions.join(" OR ")));
}
let where_clause = format!("WHERE {}", conditions.join(" AND "));
let sql = format!(
"SELECT series_name, COUNT(*) AS episode_count, \
COUNT(DISTINCT season_number) AS season_count, \
MAX(thumbnail_url) AS thumbnail_url, \
STRING_AGG(genres, ',') AS genres_blob \
FROM library_items {} GROUP BY series_name ORDER BY series_name ASC",
where_clause
);
let rows = sqlx::query_as::<_, ShowSummaryRow>(&sql)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(rows
.into_iter()
.map(|r| {
let genres = parse_genres_blob(&r.genres_blob);
ShowSummary::from_persistence(
r.series_name,
r.episode_count as u32,
r.season_count as u32,
r.thumbnail_url,
genres,
)
})
.collect())
}
async fn list_seasons(
&self,
series_name: &str,
provider_id: Option<&str>,
) -> DomainResult<Vec<SeasonSummary>> {
let mut conditions = vec![
format!("series_name = '{}'", series_name.replace('\'', "''")),
"content_type = 'episode'".to_string(),
];
if let Some(p) = provider_id {
conditions.push(format!("provider_id = '{}'", p.replace('\'', "''")));
}
let where_clause = format!("WHERE {}", conditions.join(" AND "));
let sql = format!(
"SELECT season_number, COUNT(*) AS episode_count, \
MAX(thumbnail_url) AS thumbnail_url \
FROM library_items {} GROUP BY season_number ORDER BY season_number ASC",
where_clause
);
let rows = sqlx::query_as::<_, SeasonSummaryRow>(&sql)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(rows
.into_iter()
.map(|r| {
SeasonSummary::from_persistence(
r.season_number as u32,
r.episode_count as u32,
r.thumbnail_url,
)
})
.collect())
}
}

90
crates/adapters/postgres/src/provider_config.rs
View File

@@ -1,90 +0,0 @@
use async_trait::async_trait;
use sqlx::PgPool;
use adapter_common::map_sqlx_error;
use domain::{
ports::provider_config::{ProviderConfigCommand, ProviderConfigQuery},
DomainResult, ProviderConfigRow,
};
pub struct PgProviderConfig {
pool: PgPool,
}
impl PgProviderConfig {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
#[async_trait]
impl ProviderConfigCommand for PgProviderConfig {
async fn upsert(&self, row: &ProviderConfigRow) -> DomainResult<()> {
sqlx::query(
r#"INSERT INTO provider_configs (id, provider_type, config_json, enabled, updated_at)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT(id) DO UPDATE SET
provider_type = EXCLUDED.provider_type,
config_json = EXCLUDED.config_json,
enabled = EXCLUDED.enabled,
updated_at = EXCLUDED.updated_at"#,
)
.bind(row.id())
.bind(row.provider_type())
.bind(row.config_json())
.bind(row.enabled())
.bind(row.updated_at())
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
async fn delete(&self, id: &str) -> DomainResult<()> {
sqlx::query("DELETE FROM provider_configs WHERE id = $1")
.bind(id)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
}
#[async_trait]
impl ProviderConfigQuery for PgProviderConfig {
async fn get_all(&self) -> DomainResult<Vec<ProviderConfigRow>> {
let rows: Vec<(String, String, String, bool, String)> = sqlx::query_as(
"SELECT id, provider_type, config_json, enabled, updated_at FROM provider_configs",
)
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(rows
.into_iter()
.map(|(id, provider_type, config_json, enabled, updated_at)| {
ProviderConfigRow::from_persistence(
id,
provider_type,
config_json,
enabled,
updated_at,
)
})
.collect())
}
async fn get_by_id(&self, id: &str) -> DomainResult<Option<ProviderConfigRow>> {
let row: Option<(String, String, String, bool, String)> = sqlx::query_as(
"SELECT id, provider_type, config_json, enabled, updated_at FROM provider_configs WHERE id = $1",
)
.bind(id)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(row.map(|(id, provider_type, config_json, enabled, updated_at)| {
ProviderConfigRow::from_persistence(id, provider_type, config_json, enabled, updated_at)
}))
}
}

349
crates/adapters/postgres/src/schedule.rs
View File

@@ -1,349 +0,0 @@
use std::collections::HashMap;
use async_trait::async_trait;
use sqlx::PgPool;
use adapter_common::{map_sqlx_error, parse_dt, parse_json, parse_uuid};
use domain::{
ports::schedule::{ScheduleCommand, ScheduleQuery},
BlockId, ChannelId, DomainError, DomainResult, GeneratedSchedule, MediaItem, MediaItemId,
PlaybackRecord, PlaybackRecordId, ScheduleId, ScheduledSlot, SlotId,
};
pub struct PgScheduleRepository {
pool: PgPool,
}
impl PgScheduleRepository {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
#[derive(Debug, sqlx::FromRow)]
struct ScheduleRow {
id: String,
channel_id: String,
valid_from: String,
valid_until: String,
generation: i64,
}
#[derive(Debug, sqlx::FromRow)]
struct SlotRow {
id: String,
#[sqlx(rename = "schedule_id")]
_schedule_id: String,
start_at: String,
end_at: String,
item: String,
source_block_id: String,
}
#[derive(Debug, sqlx::FromRow)]
struct LastSlotRow {
source_block_id: String,
item: String,
}
#[derive(Debug, sqlx::FromRow)]
struct PlaybackRecordRow {
id: String,
channel_id: String,
item_id: String,
played_at: String,
generation: i64,
}
fn map_slot_row(row: SlotRow) -> DomainResult<ScheduledSlot> {
let id = SlotId::from_uuid(parse_uuid(&row.id, "slot id")?);
let source_block_id = BlockId::from_uuid(parse_uuid(&row.source_block_id, "block id")?);
let item: MediaItem = parse_json(&row.item, "slot item")?;
Ok(ScheduledSlot::from_persistence(
id,
parse_dt(&row.start_at)?,
parse_dt(&row.end_at)?,
item,
source_block_id,
))
}
fn map_schedule(row: ScheduleRow, slot_rows: Vec<SlotRow>) -> DomainResult<GeneratedSchedule> {
let id = ScheduleId::from_uuid(parse_uuid(&row.id, "schedule id")?);
let channel_id = ChannelId::from_uuid(parse_uuid(&row.channel_id, "channel id")?);
let slots: Result<Vec<ScheduledSlot>, _> = slot_rows.into_iter().map(map_slot_row).collect();
Ok(GeneratedSchedule::from_persistence(
id,
channel_id,
parse_dt(&row.valid_from)?,
parse_dt(&row.valid_until)?,
row.generation as u32,
slots?,
))
}
fn map_playback_row(row: PlaybackRecordRow) -> DomainResult<PlaybackRecord> {
let id = PlaybackRecordId::from_uuid(parse_uuid(&row.id, "playback record id")?);
let channel_id = ChannelId::from_uuid(parse_uuid(&row.channel_id, "channel id")?);
Ok(PlaybackRecord::from_persistence(
id,
channel_id,
MediaItemId::new(row.item_id),
parse_dt(&row.played_at)?,
row.generation as u32,
))
}
impl PgScheduleRepository {
async fn fetch_slots(&self, schedule_id: &str) -> DomainResult<Vec<SlotRow>> {
sqlx::query_as(
"SELECT id, schedule_id, start_at, end_at, item, source_block_id \
FROM scheduled_slots WHERE schedule_id = $1 ORDER BY start_at",
)
.bind(schedule_id)
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)
}
}
#[async_trait]
impl ScheduleCommand for PgScheduleRepository {
async fn save(&self, schedule: &GeneratedSchedule) -> DomainResult<()> {
sqlx::query(
r#"
INSERT INTO generated_schedules (id, channel_id, valid_from, valid_until, generation)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT(id) DO UPDATE SET
valid_from = EXCLUDED.valid_from,
valid_until = EXCLUDED.valid_until,
generation = EXCLUDED.generation
"#,
)
.bind(schedule.id().value().to_string())
.bind(schedule.channel_id().value().to_string())
.bind(schedule.valid_from().to_rfc3339())
.bind(schedule.valid_until().to_rfc3339())
.bind(schedule.generation() as i64)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
sqlx::query("DELETE FROM scheduled_slots WHERE schedule_id = $1")
.bind(schedule.id().value().to_string())
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
for slot in schedule.slots() {
let item_json = serde_json::to_string(slot.item())
.map_err(|e| DomainError::RepositoryError(format!("serialize slot item: {e}")))?;
sqlx::query(
"INSERT INTO scheduled_slots (id, schedule_id, start_at, end_at, item, source_block_id)
VALUES ($1, $2, $3, $4, $5, $6)",
)
.bind(slot.id().value().to_string())
.bind(schedule.id().value().to_string())
.bind(slot.start_at().to_rfc3339())
.bind(slot.end_at().to_rfc3339())
.bind(&item_json)
.bind(slot.source_block_id().value().to_string())
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
}
Ok(())
}
async fn save_playback_record(&self, record: &PlaybackRecord) -> DomainResult<()> {
sqlx::query(
r#"
INSERT INTO playback_records (id, channel_id, item_id, played_at, generation)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT(id) DO NOTHING
"#,
)
.bind(record.id().to_string())
.bind(record.channel_id().value().to_string())
.bind(record.item_id().value())
.bind(record.played_at().to_rfc3339())
.bind(record.generation() as i64)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
async fn delete_schedules_after(
&self,
channel_id: ChannelId,
target_generation: u32,
) -> DomainResult<()> {
let ch = channel_id.value().to_string();
let target_gen = target_generation as i64;
sqlx::query("DELETE FROM playback_records WHERE channel_id = $1 AND generation > $2")
.bind(&ch)
.bind(target_gen)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
sqlx::query("DELETE FROM generated_schedules WHERE channel_id = $1 AND generation > $2")
.bind(&ch)
.bind(target_gen)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
}
#[async_trait]
impl ScheduleQuery for PgScheduleRepository {
async fn find_active(
&self,
channel_id: ChannelId,
at: chrono::DateTime<chrono::Utc>,
) -> DomainResult<Option<GeneratedSchedule>> {
let at_str = at.to_rfc3339();
let row: Option<ScheduleRow> = sqlx::query_as(
"SELECT id, channel_id, valid_from, valid_until, generation \
FROM generated_schedules \
WHERE channel_id = $1 AND valid_from <= $2 AND valid_until > $3 \
LIMIT 1",
)
.bind(channel_id.value().to_string())
.bind(&at_str)
.bind(&at_str)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
match row {
None => Ok(None),
Some(r) => {
let slots = self.fetch_slots(&r.id).await?;
Some(map_schedule(r, slots)).transpose()
}
}
}
async fn find_latest(&self, channel_id: ChannelId) -> DomainResult<Option<GeneratedSchedule>> {
let row: Option<ScheduleRow> = sqlx::query_as(
"SELECT id, channel_id, valid_from, valid_until, generation \
FROM generated_schedules \
WHERE channel_id = $1 ORDER BY valid_from DESC LIMIT 1",
)
.bind(channel_id.value().to_string())
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
match row {
None => Ok(None),
Some(r) => {
let slots = self.fetch_slots(&r.id).await?;
Some(map_schedule(r, slots)).transpose()
}
}
}
async fn find_playback_history(
&self,
channel_id: ChannelId,
) -> DomainResult<Vec<PlaybackRecord>> {
let rows: Vec<PlaybackRecordRow> = sqlx::query_as(
"SELECT id, channel_id, item_id, played_at, generation \
FROM playback_records WHERE channel_id = $1 ORDER BY played_at DESC",
)
.bind(channel_id.value().to_string())
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
rows.into_iter().map(map_playback_row).collect()
}
async fn find_last_slot_per_block(
&self,
channel_id: ChannelId,
) -> DomainResult<HashMap<BlockId, MediaItemId>> {
let channel_id_str = channel_id.value().to_string();
let rows: Vec<LastSlotRow> = sqlx::query_as(
"SELECT ss.source_block_id, ss.item \
FROM scheduled_slots ss \
INNER JOIN generated_schedules gs ON gs.id = ss.schedule_id \
WHERE gs.channel_id = $1 \
AND ss.start_at = ( \
SELECT MAX(ss2.start_at) \
FROM scheduled_slots ss2 \
INNER JOIN generated_schedules gs2 ON gs2.id = ss2.schedule_id \
WHERE ss2.source_block_id = ss.source_block_id \
AND gs2.channel_id = $2 \
)",
)
.bind(&channel_id_str)
.bind(&channel_id_str)
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
let mut map = HashMap::new();
for row in rows {
let block_id =
BlockId::from_uuid(parse_uuid(&row.source_block_id, "block id")?);
let item: MediaItem = parse_json(&row.item, "slot item")?;
map.insert(block_id, item.id().clone());
}
Ok(map)
}
async fn list_schedule_history(
&self,
channel_id: ChannelId,
) -> DomainResult<Vec<GeneratedSchedule>> {
let rows: Vec<ScheduleRow> = sqlx::query_as(
"SELECT id, channel_id, valid_from, valid_until, generation \
FROM generated_schedules WHERE channel_id = $1 ORDER BY generation DESC",
)
.bind(channel_id.value().to_string())
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
rows.into_iter()
.map(|r| map_schedule(r, vec![]))
.collect()
}
async fn get_schedule_by_id(
&self,
channel_id: ChannelId,
schedule_id: ScheduleId,
) -> DomainResult<Option<GeneratedSchedule>> {
let row: Option<ScheduleRow> = sqlx::query_as(
"SELECT id, channel_id, valid_from, valid_until, generation \
FROM generated_schedules WHERE id = $1 AND channel_id = $2",
)
.bind(schedule_id.value().to_string())
.bind(channel_id.value().to_string())
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
match row {
None => Ok(None),
Some(r) => {
let slots = self.fetch_slots(&r.id).await?;
Some(map_schedule(r, slots)).transpose()
}
}
}
}

48
crates/adapters/postgres/src/settings.rs
View File

@@ -1,48 +0,0 @@
use async_trait::async_trait;
use sqlx::PgPool;
use domain::{
ports::settings::AppSettingsRepository,
DomainError, DomainResult,
};
pub struct PgAppSettings {
pool: PgPool,
}
impl PgAppSettings {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
#[async_trait]
impl AppSettingsRepository for PgAppSettings {
async fn get(&self, key: &str) -> DomainResult<Option<String>> {
sqlx::query_scalar::<_, String>("SELECT value FROM app_settings WHERE key = $1")
.bind(key)
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
async fn set(&self, key: &str, value: &str) -> DomainResult<()> {
sqlx::query(
"INSERT INTO app_settings (key, value) VALUES ($1, $2)
ON CONFLICT(key) DO UPDATE SET value = EXCLUDED.value",
)
.bind(key)
.bind(value)
.execute(&self.pool)
.await
.map(|_| ())
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
async fn get_all(&self) -> DomainResult<Vec<(String, String)>> {
sqlx::query_as::<_, (String, String)>("SELECT key, value FROM app_settings ORDER BY key")
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
}

38
crates/adapters/postgres/src/transcode.rs
View File

@@ -1,38 +0,0 @@
use async_trait::async_trait;
use sqlx::PgPool;
use domain::{
ports::transcode::TranscodeSettingsRepository,
DomainError, DomainResult,
};
pub struct PgTranscodeSettings {
pool: PgPool,
}
impl PgTranscodeSettings {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
#[async_trait]
impl TranscodeSettingsRepository for PgTranscodeSettings {
async fn load_cleanup_ttl(&self) -> DomainResult<Option<u32>> {
let row: Option<(i64,)> =
sqlx::query_as("SELECT cleanup_ttl_hours FROM transcode_settings WHERE id = 1")
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(row.map(|(h,)| h as u32))
}
async fn save_cleanup_ttl(&self, hours: u32) -> DomainResult<()> {
sqlx::query("UPDATE transcode_settings SET cleanup_ttl_hours = $1 WHERE id = 1")
.bind(hours as i64)
.execute(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(())
}
}

140
crates/adapters/postgres/src/user.rs
View File

@@ -1,140 +0,0 @@
use async_trait::async_trait;
use sqlx::PgPool;
use adapter_common::{map_sqlx_error, parse_dt, parse_uuid};
use domain::{
ports::user::{UserCommand, UserQuery},
DomainError, DomainResult, Email, User, UserId,
};
pub struct PgUserRepository {
pool: PgPool,
}
impl PgUserRepository {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
#[derive(Debug, sqlx::FromRow)]
struct UserRow {
id: String,
subject: String,
email: String,
password_hash: Option<String>,
is_admin: bool,
created_at: String,
}
impl UserRow {
fn into_user(self) -> DomainResult<User> {
let id = UserId::from_uuid(parse_uuid(&self.id, "user id")?);
let email = Email::new(&self.email)
.map_err(|e| DomainError::RepositoryError(format!("Invalid email: {e}")))?;
let created_at = parse_dt(&self.created_at)?;
Ok(User::from_persistence(
id,
self.subject,
email,
self.password_hash,
self.is_admin,
created_at,
))
}
}
#[async_trait]
impl UserCommand for PgUserRepository {
async fn save(&self, user: &User) -> DomainResult<()> {
let id = user.id().value().to_string();
let created_at = user.created_at().to_rfc3339();
sqlx::query(
r#"
INSERT INTO users (id, subject, email, password_hash, is_admin, created_at)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT(id) DO UPDATE SET
subject = EXCLUDED.subject,
email = EXCLUDED.email,
password_hash = EXCLUDED.password_hash,
is_admin = EXCLUDED.is_admin
"#,
)
.bind(&id)
.bind(user.subject())
.bind(user.email().as_ref())
.bind(user.password_hash())
.bind(user.is_admin())
.bind(&created_at)
.execute(&self.pool)
.await
.map_err(|e| {
let msg = e.to_string();
if msg.contains("unique constraint") || msg.contains("duplicate key") {
DomainError::UserAlreadyExists(user.email().as_ref().to_string())
} else {
map_sqlx_error(e)
}
})?;
Ok(())
}
async fn delete(&self, id: UserId) -> DomainResult<()> {
sqlx::query("DELETE FROM users WHERE id = $1")
.bind(id.value().to_string())
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
}
#[async_trait]
impl UserQuery for PgUserRepository {
async fn find_by_id(&self, id: UserId) -> DomainResult<Option<User>> {
let row: Option<UserRow> = sqlx::query_as(
"SELECT id, subject, email, password_hash, is_admin, created_at FROM users WHERE id = $1",
)
.bind(id.value().to_string())
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
row.map(UserRow::into_user).transpose()
}
async fn find_by_subject(&self, subject: &str) -> DomainResult<Option<User>> {
let row: Option<UserRow> = sqlx::query_as(
"SELECT id, subject, email, password_hash, is_admin, created_at FROM users WHERE subject = $1",
)
.bind(subject)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
row.map(UserRow::into_user).transpose()
}
async fn find_by_email(&self, email: &str) -> DomainResult<Option<User>> {
let row: Option<UserRow> = sqlx::query_as(
"SELECT id, subject, email, password_hash, is_admin, created_at FROM users WHERE email = $1",
)
.bind(email)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
row.map(UserRow::into_user).transpose()
}
async fn count_users(&self) -> DomainResult<u64> {
let (count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM users")
.fetch_one(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(count as u64)
}
}

70
crates/adapters/postgres/src/wire.rs
View File

@@ -1,70 +0,0 @@
use std::sync::Arc;
use sqlx::PgPool;
use domain::ports::{
activity::{ActivityLogCommand, ActivityLogQuery},
channel::{ChannelCommand, ChannelQuery},
library::{LibraryCommand, LibraryQuery},
provider_config::{ProviderConfigCommand, ProviderConfigQuery},
schedule::{ScheduleCommand, ScheduleQuery},
settings::AppSettingsRepository,
transcode::TranscodeSettingsRepository,
user::{UserCommand, UserQuery},
};
use crate::{
activity::PgActivityLog,
channel::PgChannelRepository,
library::PgLibraryRepository,
provider_config::PgProviderConfig,
schedule::PgScheduleRepository,
settings::PgAppSettings,
transcode::PgTranscodeSettings,
user::PgUserRepository,
};
pub struct PostgresWireOutput {
pub user_command: Arc<dyn UserCommand>,
pub user_query: Arc<dyn UserQuery>,
pub channel_command: Arc<dyn ChannelCommand>,
pub channel_query: Arc<dyn ChannelQuery>,
pub schedule_command: Arc<dyn ScheduleCommand>,
pub schedule_query: Arc<dyn ScheduleQuery>,
pub library_command: Arc<dyn LibraryCommand>,
pub library_query: Arc<dyn LibraryQuery>,
pub activity_command: Arc<dyn ActivityLogCommand>,
pub activity_query: Arc<dyn ActivityLogQuery>,
pub settings: Arc<dyn AppSettingsRepository>,
pub provider_config_command: Arc<dyn ProviderConfigCommand>,
pub provider_config_query: Arc<dyn ProviderConfigQuery>,
pub transcode_settings: Arc<dyn TranscodeSettingsRepository>,
}
pub fn wire(pool: PgPool) -> PostgresWireOutput {
let user = Arc::new(PgUserRepository::new(pool.clone()));
let channel = Arc::new(PgChannelRepository::new(pool.clone()));
let schedule = Arc::new(PgScheduleRepository::new(pool.clone()));
let library = Arc::new(PgLibraryRepository::new(pool.clone()));
let activity = Arc::new(PgActivityLog::new(pool.clone()));
let settings = Arc::new(PgAppSettings::new(pool.clone()));
let provider_config = Arc::new(PgProviderConfig::new(pool.clone()));
let transcode_settings = Arc::new(PgTranscodeSettings::new(pool));
PostgresWireOutput {
user_command: user.clone(),
user_query: user,
channel_command: channel.clone(),
channel_query: channel,
schedule_command: schedule.clone(),
schedule_query: schedule,
library_command: library.clone(),
library_query: library,
activity_command: activity.clone(),
activity_query: activity,
settings,
provider_config_command: provider_config.clone(),
provider_config_query: provider_config,
transcode_settings,
}
}