otter/gpodder_sqlite/src/repository/episode_action.rs

use chrono::{DateTime, Utc};
use diesel::prelude::*;
use gpodder::AuthErr;

use super::SqliteRepository;
use crate::{
    DbError,
    models::{
        device::Device,
        episode_action::{ActionType, EpisodeAction, NewEpisodeAction},
    },
    schema::*,
};

impl From<gpodder::EpisodeAction> for NewEpisodeAction {
    fn from(value: gpodder::EpisodeAction) -> Self {
        let (action, started, position, total) = match value.action {
            gpodder::EpisodeActionType::New => (ActionType::New, None, None, None),
            gpodder::EpisodeActionType::Delete => (ActionType::Delete, None, None, None),
            gpodder::EpisodeActionType::Download => (ActionType::Download, None, None, None),
            gpodder::EpisodeActionType::Play {
                started,
                position,
                total,
            } => (ActionType::Play, started, Some(position), total),
        };

        NewEpisodeAction {
            user_id: 0,
            device_id: None,
            podcast_url: value.podcast,
            episode_url: value.episode,
            time_changed: 0,
            timestamp: value.timestamp.map(|t| t.timestamp()),
            action,
            started,
            position,
            total,
        }
    }
}

fn to_gpodder_action(
    (device_id, db_action): (Option<String>, EpisodeAction),
) -> gpodder::EpisodeAction {
    let action = match db_action.action {
        ActionType::Play => gpodder::EpisodeActionType::Play {
            started: db_action.started,
            // SAFETY: the condition that this isn't null if the action type is "play" is
            // explicitely enforced by the database using a CHECK constraint.
            position: db_action.position.unwrap(),
            total: db_action.total,
        },
        ActionType::New => gpodder::EpisodeActionType::New,
        ActionType::Delete => gpodder::EpisodeActionType::Delete,
        ActionType::Download => gpodder::EpisodeActionType::Download,
    };

    gpodder::EpisodeAction {
        podcast: db_action.podcast_url,
        episode: db_action.episode_url,
        timestamp: db_action
            .timestamp
            // SAFETY the input to the from_timestamp function is always the result of a
            // previous timestamp() function call, which is guaranteed to be each other's
            // reverse
            .map(|ts| DateTime::from_timestamp(ts, 0).unwrap()),
        time_changed: DateTime::from_timestamp(db_action.time_changed, 0).unwrap(),
        device: device_id,
        action,
    }
}

impl gpodder::GpodderEpisodeActionStore for SqliteRepository {
    fn add_episode_actions(
        &self,
        user: &gpodder::User,
        actions: Vec<gpodder::EpisodeAction>,
        time_changed: DateTime<Utc>,
    ) -> Result<(), gpodder::AuthErr> {
        (|| {
            let time_changed = time_changed.timestamp();

            // TODO optimize this query
            // 1. The lookup for a device could be replaced with a subquery, although Diesel seems to
            //    have a problem using an Option<String> to match equality with a String
            // 2. Ideally the for loop would be replaced with a single query inserting multiple values,
            //    although each value would need its own subquery
            //
            // NOTE this function usually gets called from the same device, so optimizing the
            // amount of device lookups required would be useful.
            self.pool.get()?.transaction(|conn| {
                for action in actions {
                    let device_id = if let Some(device) = &action.device {
                        Some(Device::device_id_to_id(conn, user.id, device)?)
                    } else {
                        None
                    };

                    let mut new_action: NewEpisodeAction = action.into();
                    new_action.user_id = user.id;
                    new_action.device_id = device_id;
                    new_action.time_changed = time_changed;

                    diesel::insert_into(episode_actions::table)
                        .values(&new_action)
                        .execute(conn)?;
                }

                Ok::<_, DbError>(())
            })
        })()
        .map_err(AuthErr::from)
    }

    fn episode_actions_for_user(
        &self,
        user: &gpodder::User,
        since: Option<DateTime<Utc>>,
        podcast: Option<String>,
        device: Option<String>,
        aggregated: bool,
    ) -> Result<Vec<gpodder::EpisodeAction>, gpodder::AuthErr> {
        (|| {
            let since = since.map(|ts| ts.timestamp()).unwrap_or(0);
            let conn = &mut self.pool.get()?;

            let mut query = episode_actions::table
                .left_join(devices::table)
                .filter(
                    episode_actions::user_id
                        .eq(user.id)
                        .and(episode_actions::time_changed.ge(since)),
                )
                .select((devices::device_id.nullable(), EpisodeAction::as_select()))
                .into_boxed();

            if let Some(device_id) = device {
                query = query.filter(devices::device_id.eq(device_id));
            }

            if let Some(podcast_url) = podcast {
                query = query.filter(episode_actions::podcast_url.eq(podcast_url));
            }

            let db_actions: Vec<(Option<String>, EpisodeAction)> = if aggregated {
                // https://stackoverflow.com/a/7745635
                // For each episode URL, we want to return the row with the highest `time_changed`
                // value. We achieve this be left joining with self on the URL, as well as whether the
                // left row's time_changed value is less than the right one. Rows with the largest
                // time_changed value for a given URL will join with a NULL value (because of the left
                // join), so we filter those out to retrieve the correct rows.
                let a2 = diesel::alias!(episode_actions as a2);

                query
                    .left_join(
                        a2.on(episode_actions::episode_url
                            .eq(a2.field(episode_actions::episode_url))
                            .and(
                                episode_actions::time_changed
                                    .lt(a2.field(episode_actions::time_changed)),
                            )),
                    )
                    .filter(a2.field(episode_actions::episode_url).is_null())
                    .get_results(conn)?
            } else {
                query.get_results(conn)?
            };

            let actions = db_actions.into_iter().map(to_gpodder_action).collect();

            Ok::<_, DbError>(actions)
        })()
        .map_err(AuthErr::from)
    }
}