adapter-sqlite: all repo implementations + wire fn + migrations copy
This commit is contained in:
362
crates/adapters/sqlite/src/schedule.rs
Normal file
362
crates/adapters/sqlite/src/schedule.rs
Normal file
@@ -0,0 +1,362 @@
|
||||
//! SQLite adapter for schedule persistence (ScheduleCommand + ScheduleQuery).
|
||||
|
||||
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 }
|
||||
}
|
||||
}
|
||||
|
||||
// -- Row types ---------------------------------------------------------------
|
||||
|
||||
#[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,
|
||||
#[allow(dead_code)]
|
||||
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,
|
||||
}
|
||||
|
||||
// -- Mapping -----------------------------------------------------------------
|
||||
|
||||
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,
|
||||
))
|
||||
}
|
||||
|
||||
// -- Internal helpers --------------------------------------------------------
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// -- Command -----------------------------------------------------------------
|
||||
|
||||
#[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)?;
|
||||
|
||||
// Delete-then-insert all slots
|
||||
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(())
|
||||
}
|
||||
}
|
||||
|
||||
// -- Query -------------------------------------------------------------------
|
||||
|
||||
#[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()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user