576 lines
20 KiB
Rust
576 lines
20 KiB
Rust
//! 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())
|
|
}
|
|
}
|