@@ -22,7 +22,6 @@ impl PostgresDiaryRepository {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
|
||||
async fn count_diary_entries(&self, movie_id: Option<&str>) -> Result<i64, DomainError> {
|
||||
match movie_id {
|
||||
None => sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM reviews")
|
||||
@@ -134,7 +133,9 @@ impl PostgresDiaryRepository {
|
||||
if has_search {
|
||||
q = q.bind(search.unwrap());
|
||||
}
|
||||
q.fetch_one(&self.pool).await.map_err(adapter_common::map_sqlx_error)
|
||||
q.fetch_one(&self.pool)
|
||||
.await
|
||||
.map_err(adapter_common::map_sqlx_error)
|
||||
}
|
||||
|
||||
async fn fetch_user_diary_rows(
|
||||
@@ -371,7 +372,10 @@ impl DiaryQuery for PostgresDiaryRepository {
|
||||
}
|
||||
|
||||
let count_q = bind_filter_params!(sqlx::query_scalar::<_, i64>(&count_sql));
|
||||
let total = count_q.fetch_one(&self.pool).await.map_err(adapter_common::map_sqlx_error)?;
|
||||
let total = count_q
|
||||
.fetch_one(&self.pool)
|
||||
.await
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
|
||||
let rows_q = bind_filter_params!(sqlx::query_as::<_, FeedRow>(&select_sql));
|
||||
let rows = rows_q
|
||||
|
||||
@@ -17,7 +17,6 @@ impl PostgresGoalRepository {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
||||
@@ -96,7 +96,6 @@ impl PostgresImportProfileRepository {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
||||
@@ -202,7 +202,6 @@ impl PostgresImportSessionRepository {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
|
||||
fn serialize_session(
|
||||
s: &ImportSession,
|
||||
) -> Result<(String, Option<String>, Option<String>), DomainError> {
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
models::{DiaryEntry, FeedEntry, Movie, MovieSummary, Review},
|
||||
};
|
||||
use adapter_common::{
|
||||
movie_row_to_domain, movie_stats_to_domain, movie_summary_to_domain, review_row_to_domain,
|
||||
user_summary_to_domain,
|
||||
};
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
models::{DiaryEntry, FeedEntry, Movie, MovieSummary, Review},
|
||||
};
|
||||
|
||||
#[derive(sqlx::FromRow)]
|
||||
pub(crate) struct MovieRow {
|
||||
|
||||
@@ -20,7 +20,6 @@ impl PostgresMovieRepository {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
||||
@@ -12,7 +12,6 @@ impl PostgresMovieDeduplicator {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -32,7 +31,11 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
|
||||
let director = canonical.director().map(str::to_string);
|
||||
let poster = canonical.poster_path().map(|p| p.value().to_string());
|
||||
|
||||
let mut tx = self.pool.begin().await.map_err(adapter_common::map_sqlx_error)?;
|
||||
let mut tx = self
|
||||
.pool
|
||||
.begin()
|
||||
.await
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
|
||||
// 1. Upsert canonical movie record
|
||||
sqlx::query(
|
||||
|
||||
@@ -29,7 +29,6 @@ pub fn create_person_adapter(pool: PgPool) -> (Arc<dyn PersonCommand>, Arc<dyn P
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
#[async_trait]
|
||||
impl PersonCommand for PostgresPersonAdapter {
|
||||
async fn upsert_batch(&self, persons: &[Person]) -> Result<(), DomainError> {
|
||||
|
||||
@@ -16,7 +16,6 @@ impl PostgresMovieProfileRepository {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -24,7 +23,11 @@ impl MovieProfileRepository for PostgresMovieProfileRepository {
|
||||
async fn upsert(&self, p: &MovieProfile) -> Result<(), DomainError> {
|
||||
let movie_id = p.movie_id.value().to_string();
|
||||
|
||||
let mut tx = self.pool.begin().await.map_err(adapter_common::map_sqlx_error)?;
|
||||
let mut tx = self
|
||||
.pool
|
||||
.begin()
|
||||
.await
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
|
||||
sqlx::query(
|
||||
r#"INSERT INTO movie_profiles
|
||||
|
||||
@@ -16,7 +16,6 @@ impl PostgresRefreshSessionAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
#[async_trait]
|
||||
impl RefreshSessionRepository for PostgresRefreshSessionAdapter {
|
||||
async fn create(&self, session: &RefreshSession) -> Result<(), DomainError> {
|
||||
|
||||
@@ -7,8 +7,8 @@ use domain::{
|
||||
};
|
||||
use sqlx::PgPool;
|
||||
|
||||
use adapter_common::datetime_to_str;
|
||||
use crate::models::ReviewRow;
|
||||
use adapter_common::datetime_to_str;
|
||||
|
||||
pub struct PostgresReviewRepository {
|
||||
pool: PgPool,
|
||||
@@ -18,7 +18,6 @@ impl PostgresReviewRepository {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
||||
@@ -7,8 +7,8 @@ use domain::{
|
||||
};
|
||||
use sqlx::PgPool;
|
||||
|
||||
use adapter_common::format_year_month;
|
||||
use crate::models::{DirectorCountRow, MonthlyRatingRow, UserTotalsRow};
|
||||
use adapter_common::format_year_month;
|
||||
|
||||
pub struct PostgresStatsRepository {
|
||||
pool: PgPool,
|
||||
@@ -19,7 +19,6 @@ impl PostgresStatsRepository {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
|
||||
async fn fetch_user_totals(&self, user_id: &str) -> Result<UserTotalsRow, DomainError> {
|
||||
sqlx::query_as::<_, UserTotalsRow>(
|
||||
r#"SELECT COUNT(DISTINCT movie_id) AS total,
|
||||
|
||||
@@ -15,7 +15,6 @@ impl PostgresUserSettingsRepository {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
||||
@@ -20,7 +20,6 @@ impl PostgresUserRepository {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
|
||||
fn parse_role(s: &str) -> UserRole {
|
||||
match s {
|
||||
"admin" => UserRole::Admin,
|
||||
|
||||
@@ -9,7 +9,6 @@ use sqlx::{PgPool, Row};
|
||||
|
||||
use adapter_common::{parse_datetime, parse_uuid};
|
||||
|
||||
|
||||
// ── WatchEventRepository ──────────────────────────────────────────────────────
|
||||
|
||||
pub struct PostgresWatchEventRepository {
|
||||
@@ -191,15 +190,33 @@ impl WatchEventQuery for PostgresWatchEventRepository {
|
||||
|
||||
fn row_to_watch_event(row: &sqlx::postgres::PgRow) -> Result<WatchEvent, DomainError> {
|
||||
let id_str: String = row.try_get("id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let user_id_str: String = row.try_get("user_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let movie_id_str: Option<String> = row.try_get("movie_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let title: String = row.try_get("title").map_err(adapter_common::map_sqlx_error)?;
|
||||
let year: Option<i32> = row.try_get("year").map_err(adapter_common::map_sqlx_error)?;
|
||||
let ext_id: Option<String> = row.try_get("external_metadata_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let source_str: String = row.try_get("source").map_err(adapter_common::map_sqlx_error)?;
|
||||
let watched_at_str: String = row.try_get("watched_at").map_err(adapter_common::map_sqlx_error)?;
|
||||
let status_str: String = row.try_get("status").map_err(adapter_common::map_sqlx_error)?;
|
||||
let created_at_str: String = row.try_get("created_at").map_err(adapter_common::map_sqlx_error)?;
|
||||
let user_id_str: String = row
|
||||
.try_get("user_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let movie_id_str: Option<String> = row
|
||||
.try_get("movie_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let title: String = row
|
||||
.try_get("title")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let year: Option<i32> = row
|
||||
.try_get("year")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let ext_id: Option<String> = row
|
||||
.try_get("external_metadata_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let source_str: String = row
|
||||
.try_get("source")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let watched_at_str: String = row
|
||||
.try_get("watched_at")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let status_str: String = row
|
||||
.try_get("status")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let created_at_str: String = row
|
||||
.try_get("created_at")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
|
||||
let source: WatchEventSource = source_str
|
||||
.parse()
|
||||
@@ -330,12 +347,24 @@ impl WebhookTokenRepository for PostgresWebhookTokenRepository {
|
||||
|
||||
fn row_to_webhook_token(row: &sqlx::postgres::PgRow) -> Result<WebhookToken, DomainError> {
|
||||
let id_str: String = row.try_get("id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let user_id_str: String = row.try_get("user_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let token_hash: String = row.try_get("token_hash").map_err(adapter_common::map_sqlx_error)?;
|
||||
let provider_str: String = row.try_get("provider").map_err(adapter_common::map_sqlx_error)?;
|
||||
let label: Option<String> = row.try_get("label").map_err(adapter_common::map_sqlx_error)?;
|
||||
let created_at_str: String = row.try_get("created_at").map_err(adapter_common::map_sqlx_error)?;
|
||||
let last_used_str: Option<String> = row.try_get("last_used_at").map_err(adapter_common::map_sqlx_error)?;
|
||||
let user_id_str: String = row
|
||||
.try_get("user_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let token_hash: String = row
|
||||
.try_get("token_hash")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let provider_str: String = row
|
||||
.try_get("provider")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let label: Option<String> = row
|
||||
.try_get("label")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let created_at_str: String = row
|
||||
.try_get("created_at")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let last_used_str: Option<String> = row
|
||||
.try_get("last_used_at")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
|
||||
let provider: WatchEventSource = provider_str
|
||||
.parse()
|
||||
|
||||
@@ -10,8 +10,8 @@ use domain::{
|
||||
};
|
||||
use sqlx::{PgPool, Row};
|
||||
|
||||
use adapter_common::{parse_datetime, parse_uuid};
|
||||
use crate::models::MovieRow;
|
||||
use adapter_common::{parse_datetime, parse_uuid};
|
||||
|
||||
pub struct PostgresWatchlistRepository {
|
||||
pool: PgPool,
|
||||
@@ -21,7 +21,6 @@ impl PostgresWatchlistRepository {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
||||
@@ -15,7 +15,6 @@ use uuid::Uuid;
|
||||
|
||||
use adapter_common::{parse_datetime, parse_uuid};
|
||||
|
||||
|
||||
fn status_to_str(s: &WrapUpStatus) -> &'static str {
|
||||
match s {
|
||||
WrapUpStatus::Pending => "pending",
|
||||
@@ -216,14 +215,30 @@ impl WrapUpRepository for PostgresWrapUpRepository {
|
||||
|
||||
fn row_to_record(row: &sqlx::postgres::PgRow) -> Result<WrapUpRecord, DomainError> {
|
||||
let id_str: String = row.try_get("id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let user_id_str: Option<String> = row.try_get("user_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let start_date: NaiveDate = row.try_get("start_date").map_err(adapter_common::map_sqlx_error)?;
|
||||
let end_date: NaiveDate = row.try_get("end_date").map_err(adapter_common::map_sqlx_error)?;
|
||||
let status_str: String = row.try_get("status").map_err(adapter_common::map_sqlx_error)?;
|
||||
let report_json: Option<String> = row.try_get("report_json").map_err(adapter_common::map_sqlx_error)?;
|
||||
let error_message: Option<String> = row.try_get("error_message").map_err(adapter_common::map_sqlx_error)?;
|
||||
let created_at_str: String = row.try_get("created_at").map_err(adapter_common::map_sqlx_error)?;
|
||||
let completed_at_str: Option<String> = row.try_get("completed_at").map_err(adapter_common::map_sqlx_error)?;
|
||||
let user_id_str: Option<String> = row
|
||||
.try_get("user_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let start_date: NaiveDate = row
|
||||
.try_get("start_date")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let end_date: NaiveDate = row
|
||||
.try_get("end_date")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let status_str: String = row
|
||||
.try_get("status")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let report_json: Option<String> = row
|
||||
.try_get("report_json")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let error_message: Option<String> = row
|
||||
.try_get("error_message")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let created_at_str: String = row
|
||||
.try_get("created_at")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let completed_at_str: Option<String> = row
|
||||
.try_get("completed_at")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
|
||||
let user_id = user_id_str.as_deref().map(parse_uuid).transpose()?;
|
||||
|
||||
@@ -288,7 +303,10 @@ impl WrapUpStatsQuery for PostgresWrapUpStatsQuery {
|
||||
q = q.bind(uid);
|
||||
}
|
||||
|
||||
let rows = q.fetch_all(&self.pool).await.map_err(adapter_common::map_sqlx_error)?;
|
||||
let rows = q
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
|
||||
if rows.is_empty() {
|
||||
return Ok(vec![]);
|
||||
@@ -298,7 +316,9 @@ impl WrapUpStatsQuery for PostgresWrapUpStatsQuery {
|
||||
let mut movie_ids: Vec<String> = Vec::new();
|
||||
let mut seen = std::collections::HashSet::new();
|
||||
for row in &rows {
|
||||
let mid: String = row.try_get("movie_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let mid: String = row
|
||||
.try_get("movie_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
if seen.insert(mid.clone()) {
|
||||
movie_ids.push(mid);
|
||||
}
|
||||
@@ -314,18 +334,39 @@ impl WrapUpStatsQuery for PostgresWrapUpStatsQuery {
|
||||
// 3) Build result
|
||||
let mut result = Vec::with_capacity(rows.len());
|
||||
for row in &rows {
|
||||
let movie_id_str: String = row.try_get("movie_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let title: String = row.try_get("title").map_err(adapter_common::map_sqlx_error)?;
|
||||
let release_year: i64 = row.try_get("release_year").map_err(adapter_common::map_sqlx_error)?;
|
||||
let director: Option<String> = row.try_get("director").map_err(adapter_common::map_sqlx_error)?;
|
||||
let poster_path: Option<String> = row.try_get("poster_path").map_err(adapter_common::map_sqlx_error)?;
|
||||
let rating: i64 = row.try_get("rating").map_err(adapter_common::map_sqlx_error)?;
|
||||
let watched_at_str: String = row.try_get("watched_at").map_err(adapter_common::map_sqlx_error)?;
|
||||
let user_id_str: String = row.try_get("user_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let runtime_minutes: Option<i32> = row.try_get("runtime_minutes").map_err(adapter_common::map_sqlx_error)?;
|
||||
let budget_usd: Option<i64> = row.try_get("budget_usd").map_err(adapter_common::map_sqlx_error)?;
|
||||
let original_language: Option<String> =
|
||||
row.try_get("original_language").map_err(adapter_common::map_sqlx_error)?;
|
||||
let movie_id_str: String = row
|
||||
.try_get("movie_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let title: String = row
|
||||
.try_get("title")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let release_year: i64 = row
|
||||
.try_get("release_year")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let director: Option<String> = row
|
||||
.try_get("director")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let poster_path: Option<String> = row
|
||||
.try_get("poster_path")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let rating: i64 = row
|
||||
.try_get("rating")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let watched_at_str: String = row
|
||||
.try_get("watched_at")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let user_id_str: String = row
|
||||
.try_get("user_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let runtime_minutes: Option<i32> = row
|
||||
.try_get("runtime_minutes")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let budget_usd: Option<i64> = row
|
||||
.try_get("budget_usd")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let original_language: Option<String> = row
|
||||
.try_get("original_language")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
|
||||
let genres = genres_map.get(&movie_id_str).cloned().unwrap_or_default();
|
||||
let keywords = keywords_map.get(&movie_id_str).cloned().unwrap_or_default();
|
||||
@@ -383,8 +424,12 @@ async fn fetch_genres_pg(
|
||||
|
||||
let mut map: HashMap<String, Vec<String>> = HashMap::new();
|
||||
for row in rows {
|
||||
let mid: String = row.try_get("movie_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let name: String = row.try_get("name").map_err(adapter_common::map_sqlx_error)?;
|
||||
let mid: String = row
|
||||
.try_get("movie_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let name: String = row
|
||||
.try_get("name")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
map.entry(mid).or_default().push(name);
|
||||
}
|
||||
Ok(map)
|
||||
@@ -404,8 +449,12 @@ async fn fetch_keywords_pg(
|
||||
|
||||
let mut map: HashMap<String, Vec<String>> = HashMap::new();
|
||||
for row in rows {
|
||||
let mid: String = row.try_get("movie_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let name: String = row.try_get("name").map_err(adapter_common::map_sqlx_error)?;
|
||||
let mid: String = row
|
||||
.try_get("movie_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let name: String = row
|
||||
.try_get("name")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
map.entry(mid).or_default().push(name);
|
||||
}
|
||||
Ok(map)
|
||||
@@ -428,11 +477,21 @@ async fn fetch_cast_pg(
|
||||
|
||||
let mut map: HashMap<String, Vec<CastEntry>> = HashMap::new();
|
||||
for row in rows {
|
||||
let mid: String = row.try_get("movie_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let name: String = row.try_get("name").map_err(adapter_common::map_sqlx_error)?;
|
||||
let billing_order: i32 = row.try_get("billing_order").map_err(adapter_common::map_sqlx_error)?;
|
||||
let tmdb_person_id: i64 = row.try_get("tmdb_person_id").map_err(adapter_common::map_sqlx_error)?;
|
||||
let profile_path: Option<String> = row.try_get("profile_path").map_err(adapter_common::map_sqlx_error)?;
|
||||
let mid: String = row
|
||||
.try_get("movie_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let name: String = row
|
||||
.try_get("name")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let billing_order: i32 = row
|
||||
.try_get("billing_order")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let tmdb_person_id: i64 = row
|
||||
.try_get("tmdb_person_id")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
let profile_path: Option<String> = row
|
||||
.try_get("profile_path")
|
||||
.map_err(adapter_common::map_sqlx_error)?;
|
||||
map.entry(mid).or_default().push(CastEntry {
|
||||
name,
|
||||
billing_order: billing_order as u32,
|
||||
|
||||
Reference in New Issue
Block a user