movies-diary/crates/adapters/postgres/src/persons.rs

use async_trait::async_trait;
use domain::{
    errors::DomainError,
    models::{
        CastCredit, CrewCredit, ExternalPersonId, Person, PersonCredits, PersonEnrichmentData,
        PersonId,
    },
    ports::{PersonCommand, PersonQuery},
    value_objects::MovieId,
};
use sqlx::PgPool;
use std::sync::Arc;

pub struct PostgresPersonAdapter {
    pool: PgPool,
}

impl PostgresPersonAdapter {
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }
}

pub fn create_person_adapter(pool: PgPool) -> (Arc<dyn PersonCommand>, Arc<dyn PersonQuery>) {
    let adapter = Arc::new(PostgresPersonAdapter::new(pool));
    (
        Arc::clone(&adapter) as Arc<dyn PersonCommand>,
        adapter as Arc<dyn PersonQuery>,
    )
}

fn map_err(e: sqlx::Error) -> DomainError {
    DomainError::InfrastructureError(e.to_string())
}

#[async_trait]
impl PersonCommand for PostgresPersonAdapter {
    async fn upsert_batch(&self, persons: &[Person]) -> Result<(), DomainError> {
        for person in persons {
            let tmdb_id = person.external_id().tmdb_id();
            sqlx::query(
                "INSERT INTO persons (id, external_id, tmdb_person_id, name, known_for_department, profile_path)
                 VALUES ($1, $2, $3, $4, $5, $6)
                 ON CONFLICT(id) DO UPDATE SET
                   external_id = EXCLUDED.external_id,
                   tmdb_person_id = EXCLUDED.tmdb_person_id,
                   name = EXCLUDED.name,
                   known_for_department = EXCLUDED.known_for_department,
                   profile_path = EXCLUDED.profile_path",
            )
            .bind(person.id().value().to_string())
            .bind(person.external_id().value())
            .bind(tmdb_id)
            .bind(person.name())
            .bind(person.known_for_department())
            .bind(person.profile_path())
            .execute(&self.pool)
            .await
            .map_err(map_err)?;
        }
        Ok(())
    }

    async fn backfill_from_credits_batch(
        &self,
        batch_size: u32,
    ) -> Result<(u64, bool), DomainError> {
        #[derive(sqlx::FromRow)]
        struct MissingPerson {
            tmdb_person_id: i64,
            name: String,
            department: Option<String>,
            profile_path: Option<String>,
        }

        let rows = sqlx::query_as::<_, MissingPerson>(
            "SELECT mc.tmdb_person_id, mc.name, 'Acting' AS department, mc.profile_path
             FROM movie_cast mc
             WHERE NOT EXISTS (SELECT 1 FROM persons WHERE persons.tmdb_person_id = mc.tmdb_person_id)
             GROUP BY mc.tmdb_person_id, mc.name, mc.profile_path
             UNION ALL
             SELECT mc.tmdb_person_id, mc.name, mc.department, mc.profile_path
             FROM movie_crew mc
             WHERE NOT EXISTS (SELECT 1 FROM persons WHERE persons.tmdb_person_id = mc.tmdb_person_id)
               AND NOT EXISTS (SELECT 1 FROM movie_cast c2 WHERE c2.tmdb_person_id = mc.tmdb_person_id)
             GROUP BY mc.tmdb_person_id, mc.name, mc.department, mc.profile_path
             LIMIT $1",
        )
        .bind(batch_size as i64)
        .fetch_all(&self.pool)
        .await
        .map_err(map_err)?;

        let has_more = rows.len() as u32 >= batch_size;
        let mut count = 0u64;
        for row in &rows {
            let ext = ExternalPersonId::new(format!("tmdb:{}", row.tmdb_person_id));
            let pid = PersonId::from_external(&ext);
            sqlx::query(
                "INSERT INTO persons (id, external_id, tmdb_person_id, name, known_for_department, profile_path)
                 VALUES ($1, $2, $3, $4, $5, $6)
                 ON CONFLICT(tmdb_person_id) DO NOTHING",
            )
            .bind(pid.value().to_string())
            .bind(ext.value())
            .bind(row.tmdb_person_id)
            .bind(&row.name)
            .bind(&row.department)
            .bind(&row.profile_path)
            .execute(&self.pool)
            .await
            .map_err(map_err)?;
            count += 1;
        }
        Ok((count, has_more))
    }

    async fn update_enrichment(
        &self,
        _id: &PersonId,
        _data: &PersonEnrichmentData,
    ) -> Result<(), DomainError> {
        todo!("person enrichment persistence")
    }
}

#[async_trait]
impl PersonQuery for PostgresPersonAdapter {
    async fn get_by_id(&self, id: &PersonId) -> Result<Option<Person>, DomainError> {
        #[derive(sqlx::FromRow)]
        struct Row {
            id: String,
            external_id: String,
            name: String,
            known_for_department: Option<String>,
            profile_path: Option<String>,
        }

        let row = sqlx::query_as::<_, Row>(
            "SELECT id, external_id, name, known_for_department, profile_path FROM persons WHERE id = $1",
        )
        .bind(id.value().to_string())
        .fetch_optional(&self.pool)
        .await
        .map_err(map_err)?;

        Ok(row.map(|r| {
            let ext = ExternalPersonId::new(r.external_id);
            Person::basic(
                PersonId::from_uuid(uuid::Uuid::parse_str(&r.id).unwrap_or_default()),
                ext,
                r.name,
                r.known_for_department,
                r.profile_path,
            )
        }))
    }

    async fn get_by_external_id(
        &self,
        id: &ExternalPersonId,
    ) -> Result<Option<Person>, DomainError> {
        #[derive(sqlx::FromRow)]
        struct Row {
            id: String,
            external_id: String,
            name: String,
            known_for_department: Option<String>,
            profile_path: Option<String>,
        }

        let row = sqlx::query_as::<_, Row>(
            "SELECT id, external_id, name, known_for_department, profile_path FROM persons WHERE external_id = $1",
        )
        .bind(id.value())
        .fetch_optional(&self.pool)
        .await
        .map_err(map_err)?;

        Ok(row.map(|r| {
            let ext = ExternalPersonId::new(r.external_id);
            Person::basic(
                PersonId::from_uuid(uuid::Uuid::parse_str(&r.id).unwrap_or_default()),
                ext,
                r.name,
                r.known_for_department,
                r.profile_path,
            )
        }))
    }

    async fn get_credits(&self, id: &PersonId) -> Result<PersonCredits, DomainError> {
        let person = self
            .get_by_id(id)
            .await?
            .ok_or_else(|| DomainError::NotFound(format!("Person {} not found", id.value())))?;

        let tmdb_id: Option<i64> =
            sqlx::query_scalar("SELECT tmdb_person_id FROM persons WHERE id = $1")
                .bind(id.value().to_string())
                .fetch_optional(&self.pool)
                .await
                .map_err(map_err)?
                .flatten();

        let Some(tmdb_id) = tmdb_id else {
            return Ok(PersonCredits {
                person,
                cast: vec![],
                crew: vec![],
            });
        };

        #[derive(sqlx::FromRow)]
        struct CastRow {
            id: String,
            title: String,
            release_year: Option<i32>,
            character: String,
            poster_path: Option<String>,
        }
        #[derive(sqlx::FromRow)]
        struct CrewRow {
            id: String,
            title: String,
            release_year: Option<i32>,
            job: String,
            department: String,
            poster_path: Option<String>,
        }

        let cast = sqlx::query_as::<_, CastRow>(
            "SELECT m.id, m.title, m.release_year, mc.character, m.poster_path
             FROM movie_cast mc JOIN movies m ON m.id = mc.movie_id
             WHERE mc.tmdb_person_id = $1 ORDER BY mc.billing_order",
        )
        .bind(tmdb_id)
        .fetch_all(&self.pool)
        .await
        .map_err(map_err)?
        .into_iter()
        .map(|r| CastCredit {
            movie_id: MovieId::from_uuid(uuid::Uuid::parse_str(&r.id).unwrap_or_default()),
            title: r.title,
            release_year: r.release_year.map(|y| y as u16),
            character: r.character,
            poster_path: r.poster_path,
        })
        .collect();

        let crew = sqlx::query_as::<_, CrewRow>(
            "SELECT m.id, m.title, m.release_year, mc.job, mc.department, m.poster_path
             FROM movie_crew mc JOIN movies m ON m.id = mc.movie_id
             WHERE mc.tmdb_person_id = $1 ORDER BY m.title",
        )
        .bind(tmdb_id)
        .fetch_all(&self.pool)
        .await
        .map_err(map_err)?
        .into_iter()
        .map(|r| CrewCredit {
            movie_id: MovieId::from_uuid(uuid::Uuid::parse_str(&r.id).unwrap_or_default()),
            title: r.title,
            release_year: r.release_year.map(|y| y as u16),
            job: r.job,
            department: r.department,
            poster_path: r.poster_path,
        })
        .collect();

        Ok(PersonCredits { person, cast, crew })
    }

    async fn list_page(&self, limit: u32, offset: u32) -> Result<Vec<Person>, DomainError> {
        #[derive(sqlx::FromRow)]
        struct Row {
            id: String,
            external_id: String,
            name: String,
            known_for_department: Option<String>,
            profile_path: Option<String>,
        }

        let rows = sqlx::query_as::<_, Row>(
            "SELECT id, external_id, name, known_for_department, profile_path FROM persons ORDER BY id LIMIT $1 OFFSET $2",
        )
        .bind(limit as i64)
        .bind(offset as i64)
        .fetch_all(&self.pool)
        .await
        .map_err(map_err)?;

        Ok(rows
            .into_iter()
            .map(|r| {
                let ext = ExternalPersonId::new(r.external_id);
                Person::basic(
                    PersonId::from_uuid(uuid::Uuid::parse_str(&r.id).unwrap_or_default()),
                    ext,
                    r.name,
                    r.known_for_department,
                    r.profile_path,
                )
            })
            .collect())
    }

    async fn list_orphaned_persons(&self) -> Result<Vec<PersonId>, DomainError> {
        let rows: Vec<(String,)> = sqlx::query_as(
            "SELECT id FROM persons
             WHERE NOT EXISTS (
                 SELECT 1 FROM movie_cast WHERE movie_cast.tmdb_person_id = persons.tmdb_person_id
             )
             AND NOT EXISTS (
                 SELECT 1 FROM movie_crew WHERE movie_crew.tmdb_person_id = persons.tmdb_person_id
             )
             LIMIT 500",
        )
        .fetch_all(&self.pool)
        .await
        .map_err(map_err)?;

        Ok(rows
            .into_iter()
            .filter_map(|(id,)| uuid::Uuid::parse_str(&id).ok().map(PersonId::from_uuid))
            .collect())
    }
}