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

adapter-postgres: all repository implementations

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-07-12 02:39:56 +02:00
parent e8179d1f53
commit 0fe80b545e
13 changed files with 1829 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

575
crates/adapters/postgres/src/library.rs Normal file
View File

@@ -0,0 +1,575 @@
//! PostgreSQL adapter for library persistence (LibraryCommand + LibraryQuery).
use std::collections::HashSet;
use async_trait::async_trait;
use sqlx::PgPool;
use domain::{
ports::library::{LibraryCommand, LibraryQuery},
ContentType, DomainError, DomainResult, LibraryCollection, LibraryItem, LibrarySearchFilter,
LibrarySyncLogEntry, LibrarySyncResult, SeasonSummary, ShowSummary,
};
pub struct PgLibraryRepository {
pool: PgPool,
}
impl PgLibraryRepository {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
// -- Helpers -----------------------------------------------------------------
fn content_type_str(ct: &ContentType) -> &'static str {
match ct {
ContentType::Movie => "movie",
ContentType::Episode => "episode",
ContentType::Short => "short",
}
}
fn parse_content_type(s: &str) -> ContentType {
match s {
"episode" => ContentType::Episode,
"short" => ContentType::Short,
_ => ContentType::Movie,
}
}
// -- Row types ---------------------------------------------------------------
#[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(
self.id,
self.provider_id,
self.external_id,
self.title,
parse_content_type(&self.content_type),
self.duration_secs as u32,
self.series_name,
self.season_number.map(|n| n as u32),
self.episode_number.map(|n| n as u32),
self.year.map(|n| n as u16),
serde_json::from_str(&self.genres).unwrap_or_default(),
serde_json::from_str(&self.tags).unwrap_or_default(),
self.collection_id,
self.collection_name,
self.collection_type,
self.thumbnail_url,
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>,
}
// -- Command -----------------------------------------------------------------
#[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()))
}
}
// -- Query -------------------------------------------------------------------
#[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: Vec<String> = r
.genres_blob
.split("],[")
.flat_map(|chunk| {
let cleaned = chunk.trim_start_matches('[').trim_end_matches(']');
cleaned
.split(',')
.filter_map(|s| {
let s = s.trim().trim_matches('"');
if s.is_empty() {
None
} else {
Some(s.to_string())
}
})
.collect::<Vec<_>>()
})
.collect::<HashSet<_>>()
.into_iter()
.collect();
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())
}
}