GKaszewski/movies-diary
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: remaining MEDIUM — CQRS splits, DI Deps, profile dedup, event Value, response enum

Browse Source 
M1: MovieRepository→MovieCommand/MovieQuery, WatchEventRepository→
WatchEventCommand/WatchEventQuery
M2: goals/ and import/ use Deps structs
M7: extract upload_image helper in update_profile
M8: FederationDeliveryRequested activity_json String→serde_json::Value
M11: UserProfileResponse uses ProfileViewData enum
This commit is contained in:
Gabriel Kaszewski
2026-07-10 03:50:43 +02:00
parent 12da356a40
commit dee013c7eb
99 changed files with 1262 additions and 896 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -79,7 +79,8 @@ pub fn create_profile_fields_repo(
pub struct PostgresWireOutput {
pub pool: PgPool,
pub movie: std::sync::Arc<dyn domain::ports::MovieRepository>,
pub movie_command: std::sync::Arc<dyn domain::ports::MovieCommand>,
pub movie_query: std::sync::Arc<dyn domain::ports::MovieQuery>,
pub review: std::sync::Arc<dyn domain::ports::ReviewRepository>,
pub diary: std::sync::Arc<dyn domain::ports::DiaryRepository>,
pub stats: std::sync::Arc<dyn domain::ports::StatsRepository>,
@@ -114,9 +115,12 @@ pub async fn wire(database_url: &str) -> anyhow::Result<PostgresWireOutput> {
user_settings::PostgresUserSettingsRepository::new(pool.clone()),
);
let movie_repo = std::sync::Arc::new(PostgresMovieRepository::new(pool.clone()));
Ok(PostgresWireOutput {
pool: pool.clone(),
movie: std::sync::Arc::new(PostgresMovieRepository::new(pool.clone())) as _,
movie_command: movie_repo.clone() as _,
movie_query: movie_repo as _,
review: std::sync::Arc::new(PostgresReviewRepository::new(pool.clone())) as _,
diary: std::sync::Arc::new(PostgresDiaryRepository::new(pool.clone())) as _,
stats: std::sync::Arc::new(PostgresStatsRepository::new(pool.clone())) as _,

89
crates/adapters/postgres/src/movie.rs
View File

@@ -5,7 +5,7 @@ use domain::{
Movie, MovieFilter, MovieSummary,
collections::{PageParams, Paginated},
},
ports::MovieRepository,
ports::{MovieCommand, MovieQuery},
value_objects::{ExternalMetadataId, MovieId, MovieTitle, ReleaseYear},
};
use sqlx::PgPool;
@@ -28,7 +28,51 @@ impl PostgresMovieRepository {
}
#[async_trait]
impl MovieRepository for PostgresMovieRepository {
impl MovieCommand for PostgresMovieRepository {
async fn upsert_movie(&self, movie: &Movie) -> Result<(), DomainError> {
let id = movie.id().value().to_string();
let external_metadata_id = movie.external_metadata_id().map(|e| e.value().to_string());
let title = movie.title().value();
let release_year = movie.release_year().value() as i64;
let director = movie.director();
let poster_path = movie.poster_path().map(|p| p.value().to_string());
sqlx::query(
"INSERT INTO movies (id, external_metadata_id, title, release_year, director, poster_path)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT(id) DO UPDATE SET
external_metadata_id = excluded.external_metadata_id,
title = excluded.title,
release_year = excluded.release_year,
director = excluded.director,
poster_path = excluded.poster_path",
)
.bind(&id)
.bind(&external_metadata_id)
.bind(title)
.bind(release_year)
.bind(director)
.bind(&poster_path)
.execute(&self.pool)
.await
.map_err(Self::map_err)?;
Ok(())
}
async fn delete_movie(&self, movie_id: &MovieId) -> Result<(), DomainError> {
let id = movie_id.value().to_string();
sqlx::query("DELETE FROM movies WHERE id = $1")
.bind(&id)
.execute(&self.pool)
.await
.map_err(Self::map_err)?;
Ok(())
}
}
#[async_trait]
impl MovieQuery for PostgresMovieRepository {
async fn get_movie_by_external_id(
&self,
external_metadata_id: &ExternalMetadataId,
@@ -81,47 +125,6 @@ impl MovieRepository for PostgresMovieRepository {
.collect()
}
async fn upsert_movie(&self, movie: &Movie) -> Result<(), DomainError> {
let id = movie.id().value().to_string();
let external_metadata_id = movie.external_metadata_id().map(|e| e.value().to_string());
let title = movie.title().value();
let release_year = movie.release_year().value() as i64;
let director = movie.director();
let poster_path = movie.poster_path().map(|p| p.value().to_string());
sqlx::query(
"INSERT INTO movies (id, external_metadata_id, title, release_year, director, poster_path)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT(id) DO UPDATE SET
external_metadata_id = excluded.external_metadata_id,
title = excluded.title,
release_year = excluded.release_year,
director = excluded.director,
poster_path = excluded.poster_path",
)
.bind(&id)
.bind(&external_metadata_id)
.bind(title)
.bind(release_year)
.bind(director)
.bind(&poster_path)
.execute(&self.pool)
.await
.map_err(Self::map_err)?;
Ok(())
}
async fn delete_movie(&self, movie_id: &MovieId) -> Result<(), DomainError> {
let id = movie_id.value().to_string();
sqlx::query("DELETE FROM movies WHERE id = $1")
.bind(&id)
.execute(&self.pool)
.await
.map_err(Self::map_err)?;
Ok(())
}
async fn existing_external_ids(
&self,
ids: &[ExternalMetadataId],

71
crates/adapters/postgres/src/watch_event.rs
View File

@@ -2,7 +2,7 @@ use async_trait::async_trait;
use domain::{
errors::DomainError,
models::{PersistedWatchEvent, WatchEvent, WatchEventSource, WatchEventStatus, WebhookToken},
ports::{WatchEventRepository, WebhookTokenRepository},
ports::{WatchEventCommand, WatchEventQuery, WebhookTokenRepository},
value_objects::{MovieId, UserId, WatchEventId, WebhookTokenId},
};
use sqlx::{PgPool, Row};
@@ -27,7 +27,7 @@ impl PostgresWatchEventRepository {
}
#[async_trait]
impl WatchEventRepository for PostgresWatchEventRepository {
impl WatchEventCommand for PostgresWatchEventRepository {
async fn save(&self, event: &WatchEvent) -> Result<(), DomainError> {
let id = event.id().value().to_string();
let user_id = event.user_id().value().to_string();
@@ -75,6 +75,41 @@ impl WatchEventRepository for PostgresWatchEventRepository {
Ok(())
}
async fn update_status_batch(
&self,
ids: &[WatchEventId],
status: WatchEventStatus,
) -> Result<u64, DomainError> {
if ids.is_empty() {
return Ok(0);
}
let id_strs: Vec<String> = ids.iter().map(|id| id.value().to_string()).collect();
let status_str = status.to_string();
let result = sqlx::query("UPDATE watch_events SET status = $1 WHERE id = ANY($2)")
.bind(&status_str)
.bind(&id_strs)
.execute(&self.pool)
.await
.map_err(map_err)?;
Ok(result.rows_affected())
}
async fn delete_non_pending_older_than(
&self,
before: chrono::NaiveDateTime,
) -> Result<u64, DomainError> {
let result =
sqlx::query("DELETE FROM watch_events WHERE status != 'pending' AND created_at < $1")
.bind(before)
.execute(&self.pool)
.await
.map_err(map_err)?;
Ok(result.rows_affected())
}
}
#[async_trait]
impl WatchEventQuery for PostgresWatchEventRepository {
async fn list_pending(&self, user_id: &UserId) -> Result<Vec<WatchEvent>, DomainError> {
let uid = user_id.value().to_string();
@@ -135,25 +170,6 @@ impl WatchEventRepository for PostgresWatchEventRepository {
rows.iter().map(row_to_watch_event).collect()
}
async fn update_status_batch(
&self,
ids: &[WatchEventId],
status: WatchEventStatus,
) -> Result<u64, DomainError> {
if ids.is_empty() {
return Ok(0);
}
let id_strs: Vec<String> = ids.iter().map(|id| id.value().to_string()).collect();
let status_str = status.to_string();
let result = sqlx::query("UPDATE watch_events SET status = $1 WHERE id = ANY($2)")
.bind(&status_str)
.bind(&id_strs)
.execute(&self.pool)
.await
.map_err(map_err)?;
Ok(result.rows_affected())
}
async fn find_duplicate(
&self,
user_id: &UserId,
@@ -175,19 +191,6 @@ impl WatchEventRepository for PostgresWatchEventRepository {
Ok(count > 0)
}
async fn delete_non_pending_older_than(
&self,
before: chrono::NaiveDateTime,
) -> Result<u64, DomainError> {
let result =
sqlx::query("DELETE FROM watch_events WHERE status != 'pending' AND created_at < $1")
.bind(before)
.execute(&self.pool)
.await
.map_err(map_err)?;
Ok(result.rows_affected())
}
}
fn row_to_watch_event(row: &sqlx::postgres::PgRow) -> Result<WatchEvent, DomainError> {