k-tv/crates/adapters/sqlite/src/schedule.rs

use std::collections::HashMap;

use async_trait::async_trait;
use sqlx::SqlitePool;

use adapter_common::{map_sqlx_error, parse_dt, parse_json, parse_uuid};
use domain::{
    ports::schedule::{ScheduleCommand, ScheduleQuery},
    BlockId, ChannelId, DomainError, DomainResult, GeneratedSchedule, MediaItem, MediaItemId,
    PlaybackRecord, ScheduleId, ScheduledSlot, SlotId,
};

pub struct SqliteScheduleRepository {
    pool: SqlitePool,
}

impl SqliteScheduleRepository {
    pub fn new(pool: SqlitePool) -> Self {
        Self { pool }
    }
}

#[derive(Debug, sqlx::FromRow)]
struct ScheduleRow {
    id: String,
    channel_id: String,
    valid_from: String,
    valid_until: String,
    generation: i64,
}

#[derive(Debug, sqlx::FromRow)]
struct SlotRow {
    id: String,
    _schedule_id: String,
    start_at: String,
    end_at: String,
    item: String,
    source_block_id: String,
}

#[derive(Debug, sqlx::FromRow)]
struct LastSlotRow {
    source_block_id: String,
    item: String,
}

#[derive(Debug, sqlx::FromRow)]
struct PlaybackRecordRow {
    id: String,
    channel_id: String,
    item_id: String,
    played_at: String,
    generation: i64,
}

fn map_slot_row(row: SlotRow) -> DomainResult<ScheduledSlot> {
    let id = SlotId::from_uuid(parse_uuid(&row.id, "slot id")?);
    let source_block_id = BlockId::from_uuid(parse_uuid(&row.source_block_id, "block id")?);
    let item: MediaItem = parse_json(&row.item, "slot item")?;

    Ok(ScheduledSlot::from_persistence(
        id,
        parse_dt(&row.start_at)?,
        parse_dt(&row.end_at)?,
        item,
        source_block_id,
    ))
}

fn map_schedule(row: ScheduleRow, slot_rows: Vec<SlotRow>) -> DomainResult<GeneratedSchedule> {
    let id = ScheduleId::from_uuid(parse_uuid(&row.id, "schedule id")?);
    let channel_id = ChannelId::from_uuid(parse_uuid(&row.channel_id, "channel id")?);
    let slots: Result<Vec<ScheduledSlot>, _> = slot_rows.into_iter().map(map_slot_row).collect();

    Ok(GeneratedSchedule::from_persistence(
        id,
        channel_id,
        parse_dt(&row.valid_from)?,
        parse_dt(&row.valid_until)?,
        row.generation as u32,
        slots?,
    ))
}

fn map_playback_row(row: PlaybackRecordRow) -> DomainResult<PlaybackRecord> {
    let id = parse_uuid(&row.id, "playback record id")?;
    let channel_id = ChannelId::from_uuid(parse_uuid(&row.channel_id, "channel id")?);

    Ok(PlaybackRecord::from_persistence(
        id,
        channel_id,
        MediaItemId::new(row.item_id),
        parse_dt(&row.played_at)?,
        row.generation as u32,
    ))
}

impl SqliteScheduleRepository {
    async fn fetch_slots(&self, schedule_id: &str) -> DomainResult<Vec<SlotRow>> {
        sqlx::query_as(
            "SELECT id, schedule_id, start_at, end_at, item, source_block_id \
             FROM scheduled_slots WHERE schedule_id = ? ORDER BY start_at",
        )
        .bind(schedule_id)
        .fetch_all(&self.pool)
        .await
        .map_err(map_sqlx_error)
    }
}

#[async_trait]
impl ScheduleCommand for SqliteScheduleRepository {
    async fn save(&self, schedule: &GeneratedSchedule) -> DomainResult<()> {
        sqlx::query(
            r#"
            INSERT INTO generated_schedules (id, channel_id, valid_from, valid_until, generation)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                valid_from  = excluded.valid_from,
                valid_until = excluded.valid_until,
                generation  = excluded.generation
            "#,
        )
        .bind(schedule.id().value().to_string())
        .bind(schedule.channel_id().value().to_string())
        .bind(schedule.valid_from().to_rfc3339())
        .bind(schedule.valid_until().to_rfc3339())
        .bind(schedule.generation() as i64)
        .execute(&self.pool)
        .await
        .map_err(map_sqlx_error)?;

        sqlx::query("DELETE FROM scheduled_slots WHERE schedule_id = ?")
            .bind(schedule.id().value().to_string())
            .execute(&self.pool)
            .await
            .map_err(map_sqlx_error)?;

        for slot in schedule.slots() {
            let item_json = serde_json::to_string(slot.item())
                .map_err(|e| DomainError::RepositoryError(format!("serialize slot item: {e}")))?;

            sqlx::query(
                "INSERT INTO scheduled_slots (id, schedule_id, start_at, end_at, item, source_block_id)
                 VALUES (?, ?, ?, ?, ?, ?)",
            )
            .bind(slot.id().value().to_string())
            .bind(schedule.id().value().to_string())
            .bind(slot.start_at().to_rfc3339())
            .bind(slot.end_at().to_rfc3339())
            .bind(&item_json)
            .bind(slot.source_block_id().value().to_string())
            .execute(&self.pool)
            .await
            .map_err(map_sqlx_error)?;
        }

        Ok(())
    }

    async fn save_playback_record(&self, record: &PlaybackRecord) -> DomainResult<()> {
        sqlx::query(
            r#"
            INSERT INTO playback_records (id, channel_id, item_id, played_at, generation)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(id) DO NOTHING
            "#,
        )
        .bind(record.id().to_string())
        .bind(record.channel_id().value().to_string())
        .bind(record.item_id().value())
        .bind(record.played_at().to_rfc3339())
        .bind(record.generation() as i64)
        .execute(&self.pool)
        .await
        .map_err(map_sqlx_error)?;
        Ok(())
    }

    async fn delete_schedules_after(
        &self,
        channel_id: ChannelId,
        target_generation: u32,
    ) -> DomainResult<()> {
        let ch = channel_id.value().to_string();
        let target_gen = target_generation as i64;

        sqlx::query("DELETE FROM playback_records WHERE channel_id = ? AND generation > ?")
            .bind(&ch)
            .bind(target_gen)
            .execute(&self.pool)
            .await
            .map_err(map_sqlx_error)?;

        sqlx::query("DELETE FROM generated_schedules WHERE channel_id = ? AND generation > ?")
            .bind(&ch)
            .bind(target_gen)
            .execute(&self.pool)
            .await
            .map_err(map_sqlx_error)?;

        Ok(())
    }
}

#[async_trait]
impl ScheduleQuery for SqliteScheduleRepository {
    async fn find_active(
        &self,
        channel_id: ChannelId,
        at: chrono::DateTime<chrono::Utc>,
    ) -> DomainResult<Option<GeneratedSchedule>> {
        let at_str = at.to_rfc3339();
        let row: Option<ScheduleRow> = sqlx::query_as(
            "SELECT id, channel_id, valid_from, valid_until, generation \
             FROM generated_schedules \
             WHERE channel_id = ? AND valid_from <= ? AND valid_until > ? \
             LIMIT 1",
        )
        .bind(channel_id.value().to_string())
        .bind(&at_str)
        .bind(&at_str)
        .fetch_optional(&self.pool)
        .await
        .map_err(map_sqlx_error)?;

        match row {
            None => Ok(None),
            Some(r) => {
                let slots = self.fetch_slots(&r.id).await?;
                Some(map_schedule(r, slots)).transpose()
            }
        }
    }

    async fn find_latest(&self, channel_id: ChannelId) -> DomainResult<Option<GeneratedSchedule>> {
        let row: Option<ScheduleRow> = sqlx::query_as(
            "SELECT id, channel_id, valid_from, valid_until, generation \
             FROM generated_schedules \
             WHERE channel_id = ? ORDER BY valid_from DESC LIMIT 1",
        )
        .bind(channel_id.value().to_string())
        .fetch_optional(&self.pool)
        .await
        .map_err(map_sqlx_error)?;

        match row {
            None => Ok(None),
            Some(r) => {
                let slots = self.fetch_slots(&r.id).await?;
                Some(map_schedule(r, slots)).transpose()
            }
        }
    }

    async fn find_playback_history(
        &self,
        channel_id: ChannelId,
    ) -> DomainResult<Vec<PlaybackRecord>> {
        let rows: Vec<PlaybackRecordRow> = sqlx::query_as(
            "SELECT id, channel_id, item_id, played_at, generation \
             FROM playback_records WHERE channel_id = ? ORDER BY played_at DESC",
        )
        .bind(channel_id.value().to_string())
        .fetch_all(&self.pool)
        .await
        .map_err(map_sqlx_error)?;

        rows.into_iter().map(map_playback_row).collect()
    }

    async fn find_last_slot_per_block(
        &self,
        channel_id: ChannelId,
    ) -> DomainResult<HashMap<BlockId, MediaItemId>> {
        let channel_id_str = channel_id.value().to_string();
        let rows: Vec<LastSlotRow> = sqlx::query_as(
            "SELECT ss.source_block_id, ss.item \
             FROM scheduled_slots ss \
             INNER JOIN generated_schedules gs ON gs.id = ss.schedule_id \
             WHERE gs.channel_id = ? \
               AND ss.start_at = ( \
                   SELECT MAX(ss2.start_at) \
                   FROM scheduled_slots ss2 \
                   INNER JOIN generated_schedules gs2 ON gs2.id = ss2.schedule_id \
                   WHERE ss2.source_block_id = ss.source_block_id \
                     AND gs2.channel_id = ? \
               )",
        )
        .bind(&channel_id_str)
        .bind(&channel_id_str)
        .fetch_all(&self.pool)
        .await
        .map_err(map_sqlx_error)?;

        let mut map = HashMap::new();
        for row in rows {
            let block_id =
                BlockId::from_uuid(parse_uuid(&row.source_block_id, "block id")?);
            let item: MediaItem = parse_json(&row.item, "slot item")?;
            map.insert(block_id, item.id().clone());
        }
        Ok(map)
    }

    async fn list_schedule_history(
        &self,
        channel_id: ChannelId,
    ) -> DomainResult<Vec<GeneratedSchedule>> {
        let rows: Vec<ScheduleRow> = sqlx::query_as(
            "SELECT id, channel_id, valid_from, valid_until, generation \
             FROM generated_schedules WHERE channel_id = ? ORDER BY generation DESC",
        )
        .bind(channel_id.value().to_string())
        .fetch_all(&self.pool)
        .await
        .map_err(map_sqlx_error)?;

        rows.into_iter()
            .map(|r| map_schedule(r, vec![]))
            .collect()
    }

    async fn get_schedule_by_id(
        &self,
        channel_id: ChannelId,
        schedule_id: ScheduleId,
    ) -> DomainResult<Option<GeneratedSchedule>> {
        let row: Option<ScheduleRow> = sqlx::query_as(
            "SELECT id, channel_id, valid_from, valid_until, generation \
             FROM generated_schedules WHERE id = ? AND channel_id = ?",
        )
        .bind(schedule_id.value().to_string())
        .bind(channel_id.value().to_string())
        .fetch_optional(&self.pool)
        .await
        .map_err(map_sqlx_error)?;

        match row {
            None => Ok(None),
            Some(r) => {
                let slots = self.fetch_slots(&r.id).await?;
                Some(map_schedule(r, slots)).transpose()
            }
        }
    }
}