entertainarr_adapter_sqlite/

podcast_episode.rs

use anyhow::Context;
use entertainarr_domain::podcast::entity::{
    ListPodcastEpisodeParams, PodcastEpisode, PodcastEpisodeField, PodcastEpisodeProgress,
};
use entertainarr_domain::prelude::SortOrder;

use crate::Wrapper;
use crate::prelude::HasAnyOf;

const FIND_BY_ID_QUERY: &str = r#"select id, podcast_id, guid, published_at, title, description, link, duration, file_url, file_size, file_type, created_at, updated_at
from podcast_episodes
where id = ?
limit 1"#;

impl super::Pool {
    #[tracing::instrument(
        skip_all,
        fields(
            otel.kind = "client",
            db.system = "sqlite",
            db.name = "podcast",
            db.operation = "SELECT",
            db.sql.table = "podcast_episodes",
            db.query.text = tracing::field::Empty,
            db.response.returned_rows = tracing::field::Empty,
            error.type = tracing::field::Empty,
            error.message = tracing::field::Empty,
            error.stacktrace = tracing::field::Empty,
        ),
        err(Debug),
    )]
    pub(crate) async fn list_podcast_episodes<'a>(
        &self,
        params: ListPodcastEpisodeParams<'a>,
    ) -> anyhow::Result<Vec<PodcastEpisode>> {
        let mut qb: sqlx::QueryBuilder<'_, sqlx::Sqlite> = sqlx::QueryBuilder::new(
            r#"select
    podcast_episodes.id,
    podcast_episodes.podcast_id,
    podcast_episodes.guid,
    podcast_episodes.published_at,
    podcast_episodes.title,
    podcast_episodes.description,
    podcast_episodes.link,
    podcast_episodes.duration,
    podcast_episodes.file_url,
    podcast_episodes.file_size,
    podcast_episodes.file_type,
    podcast_episodes.created_at,
    podcast_episodes.updated_at
from podcast_episodes"#,
        );

        if let Some(subscribed) = params.filter.subscribed {
            if subscribed {
                qb.push(" join user_podcasts on user_podcasts.podcast_id = podcast_episodes.podcast_id and user_podcasts.user_id = ").push_bind(params.user_id as i64);
            } else {
                // TODO handle filtered those where the user is not subscribed
            }
        } else {
            // nothing to do here
        }

        if params.filter.filtered.is_some() {
            qb.push(" left outer join user_podcasts filters on filters.podcast_id = podcast_episodes.podcast_id and filters.user_id = ").push_bind(params.user_id as i64);
        }

        if params.filter.watched.is_some() {
            qb.push(" left outer join user_podcast_episodes on user_podcast_episodes.podcast_episode_id = podcast_episodes.id ");
            qb.push(" and user_podcast_episodes.user_id = ")
                .push_bind(params.user_id as i64);
        }

        // WHERE

        qb.push(" where true");

        if let Some(watched) = params.filter.watched {
            if watched {
                qb.push(" and user_podcast_episodes.completed");
            } else {
                qb.push(" and (user_podcast_episodes.completed is null or not user_podcast_episodes.completed)");
            }
        }

        if let Some(filtered) = params.filter.filtered {
            if filtered {
                // maybe keep only the filtered
            } else {
                qb.push(" and (filters.user_id is null or podcast_episodes.duration is null or filters.min_duration is null or filters.min_duration < podcast_episodes.duration)");
                qb.push(" and (filters.user_id is null or podcast_episodes.duration is null or filters.max_duration is null or filters.max_duration > podcast_episodes.duration)");
            }
        }

        if !params.filter.podcast_ids.is_empty() {
            qb.push(" and ")
                .push_any("podcast_episodes.podcast_id", params.filter.podcast_ids);
        }

        match params.sort.field {
            PodcastEpisodeField::PublishedAt => qb.push(" order by podcast_episodes.published_at"),
        };
        match params.sort.order {
            SortOrder::Asc => qb.push(" asc"),
            SortOrder::Desc => qb.push(" desc"),
        };

        qb.push(" limit ")
            .push_bind(params.page.limit)
            .push(" offset ")
            .push_bind(params.page.offset);

        let span = tracing::Span::current();
        span.record("db.query.text", qb.sql());

        qb.build_query_as()
            .fetch_all(&self.0)
            .await
            .inspect(super::record_all)
            .inspect_err(super::record_error)
            .map(Wrapper::list)
            .context("unable to query podcast episodes")
    }

    #[tracing::instrument(
        skip_all,
        fields(
            otel.kind = "client",
            db.system = "sqlite",
            db.name = "podcast",
            db.operation = "SELECT",
            db.sql.table = "podcast_episodes",
            db.query.text = FIND_BY_ID_QUERY,
            db.response.returned_rows = tracing::field::Empty,
            error.type = tracing::field::Empty,
            error.message = tracing::field::Empty,
            error.stacktrace = tracing::field::Empty,
        ),
        err(Debug),
    )]
    pub(crate) async fn find_podcast_episode_by_id(
        &self,
        episode_id: u64,
    ) -> anyhow::Result<Option<PodcastEpisode>> {
        sqlx::query_as(FIND_BY_ID_QUERY)
            .bind(episode_id as i64)
            .fetch_optional(self.as_ref())
            .await
            .inspect(super::record_optional)
            .inspect_err(super::record_error)
            .map(super::Wrapper::maybe_inner)
            .context("unable to query podcast episodes")
    }

    #[tracing::instrument(
        skip_all,
        fields(
            otel.kind = "client",
            db.system = "sqlite",
            db.name = "podcast",
            db.operation = "DELETE",
            db.sql.table = "user_podcast_episodes",
            db.query.text = tracing::field::Empty,
            db.response.returned_rows = tracing::field::Empty,
            error.type = tracing::field::Empty,
            error.message = tracing::field::Empty,
            error.stacktrace = tracing::field::Empty,
        ),
        err(Debug),
    )]
    pub(crate) async fn delete_podcast_episode_progressions(
        &self,
        user_id: u64,
        episode_ids: &[u64],
    ) -> anyhow::Result<()> {
        if episode_ids.is_empty() {
            return Ok(());
        }

        let mut qb = sqlx::QueryBuilder::new("delete from user_podcast_episodes");
        qb.push(" where user_id = ").push_bind(user_id as i64);
        qb.push(" and ( ");
        for (index, id) in episode_ids.iter().enumerate() {
            if index > 0 {
                qb.push(" or");
            }
            qb.push(" podcast_episode_id = ").push_bind(*id as i64);
        }
        qb.push(")");

        let span = tracing::Span::current();
        span.record("db.query.text", qb.sql());

        qb.build()
            .execute(self.as_ref())
            .await
            .map(|_| ())
            .inspect_err(super::record_error)
            .context("unable to delete podcast episode progresses")
    }

    #[tracing::instrument(
        skip_all,
        fields(
            otel.kind = "client",
            db.system = "sqlite",
            db.name = "podcast",
            db.operation = "SELECT",
            db.sql.table = "user_podcast_episodes",
            db.query.text = tracing::field::Empty,
            db.response.returned_rows = tracing::field::Empty,
            error.type = tracing::field::Empty,
            error.message = tracing::field::Empty,
            error.stacktrace = tracing::field::Empty,
        ),
        err(Debug),
    )]
    pub(crate) async fn list_podcast_episode_progressions(
        &self,
        user_id: u64,
        episode_ids: &[u64],
    ) -> anyhow::Result<Vec<entertainarr_domain::podcast::entity::PodcastEpisodeProgress>> {
        if episode_ids.is_empty() {
            return Ok(Default::default());
        }

        let mut qb = sqlx::QueryBuilder::new(
            "select user_id, podcast_episode_id, progress, completed, created_at, updated_at from user_podcast_episodes",
        );
        qb.push(" where user_id = ").push_bind(user_id as i64);
        qb.push(" and ( ");
        for (index, id) in episode_ids.iter().enumerate() {
            if index > 0 {
                qb.push(" or");
            }
            qb.push(" podcast_episode_id = ").push_bind(*id as i64);
        }
        qb.push(")");

        let span = tracing::Span::current();
        span.record("db.query.text", qb.sql());

        qb.build_query_as::<'_, super::Wrapper<PodcastEpisodeProgress>>()
            .fetch_all(self.as_ref())
            .await
            .inspect(super::record_all)
            .inspect_err(super::record_error)
            .map(super::Wrapper::list)
            .context("unable to query podcast episode progresses")
    }

    #[tracing::instrument(
        skip_all,
        fields(
            otel.kind = "client",
            db.system = "sqlite",
            db.name = "podcast",
            db.operation = "UPSERT",
            db.sql.table = "user_podcast_episodes",
            db.query.text = tracing::field::Empty,
            db.response.returned_rows = tracing::field::Empty,
            error.type = tracing::field::Empty,
            error.message = tracing::field::Empty,
            error.stacktrace = tracing::field::Empty,
        ),
        err(Debug),
    )]
    pub(crate) async fn upsert_podcast_episode_progressions(
        &self,
        inputs: &[entertainarr_domain::podcast::entity::PodcastEpisodeProgressInput],
    ) -> anyhow::Result<Vec<PodcastEpisodeProgress>> {
        if inputs.is_empty() {
            return Ok(Default::default());
        }

        let mut qb = sqlx::QueryBuilder::new(
            "insert into user_podcast_episodes (user_id, podcast_episode_id, progress, completed) ",
        );
        qb.push_values(inputs, |mut q, item| {
            q.push_bind(item.user_id as i64)
                .push_bind(item.podcast_episode_id as i64)
                .push_bind(item.progress as i64)
                .push_bind(item.completed);
        });
        qb.push(" on conflict (user_id, podcast_episode_id) do update set progress = excluded.progress, completed = excluded.completed, updated_at = CURRENT_TIMESTAMP");
        qb.push(
            " returning user_id, podcast_episode_id, progress, completed, created_at, updated_at",
        );

        let span = tracing::Span::current();
        span.record("db.query.text", qb.sql());

        qb.build_query_as()
            .fetch_all(self.as_ref())
            .await
            .inspect(super::record_all)
            .inspect_err(super::record_error)
            .map(super::Wrapper::list)
            .context("unable to upsert podcast episode progresses")
    }
}

impl entertainarr_domain::podcast::prelude::PodcastEpisodeRepository for super::Pool {
    async fn find_by_id(&self, episode_id: u64) -> anyhow::Result<Option<PodcastEpisode>> {
        self.find_podcast_episode_by_id(episode_id).await
    }

    async fn list<'a>(
        &self,
        params: ListPodcastEpisodeParams<'a>,
    ) -> anyhow::Result<Vec<PodcastEpisode>> {
        self.list_podcast_episodes(params).await
    }

    async fn delete_progressions(&self, user_id: u64, episode_ids: &[u64]) -> anyhow::Result<()> {
        self.delete_podcast_episode_progressions(user_id, episode_ids)
            .await
    }

    async fn list_progressions(
        &self,
        user_id: u64,
        episode_ids: &[u64],
    ) -> anyhow::Result<Vec<entertainarr_domain::podcast::entity::PodcastEpisodeProgress>> {
        self.list_podcast_episode_progressions(user_id, episode_ids)
            .await
    }

    async fn upsert_progressions(
        &self,
        inputs: &[entertainarr_domain::podcast::entity::PodcastEpisodeProgressInput],
    ) -> anyhow::Result<Vec<PodcastEpisodeProgress>> {
        self.upsert_podcast_episode_progressions(inputs).await
    }
}

impl<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow> for super::Wrapper<PodcastEpisode> {
    fn from_row(row: &'r sqlx::sqlite::SqliteRow) -> Result<Self, sqlx::Error> {
        use sqlx::Row;

        Ok(Self(PodcastEpisode {
            id: row.try_get(0)?,
            podcast_id: row.try_get(1)?,
            guid: row.try_get(2)?,
            published_at: row.try_get(3)?,
            title: row.try_get(4)?,
            description: row.try_get(5)?,
            link: row.try_get(6)?,
            duration: row
                .try_get(7)
                .map(|value: Option<u64>| value.map(std::time::Duration::from_secs))?,
            file_url: row.try_get(8)?,
            file_size: row.try_get(9)?,
            file_type: row.try_get(10)?,
            created_at: row.try_get(11)?,
            updated_at: row.try_get(12)?,
        }))
    }
}

impl<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow> for super::Wrapper<PodcastEpisodeProgress> {
    fn from_row(row: &'r sqlx::sqlite::SqliteRow) -> Result<Self, sqlx::Error> {
        use sqlx::Row;

        Ok(Self(PodcastEpisodeProgress {
            user_id: row.try_get(0)?,
            podcast_episode_id: row.try_get(1)?,
            progress: row.try_get(2)?,
            completed: row.try_get(3)?,
            created_at: row.try_get(4)?,
            updated_at: row.try_get(5)?,
        }))
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use entertainarr_domain::podcast::entity::{
        ListPodcastEpisodeFilter, ListPodcastEpisodeParams, PodcastEpisodeField,
        PodcastEpisodeInput, PodcastEpisodeProgressInput, PodcastInput, PodcastSubscriptionUpdate,
    };
    use entertainarr_domain::podcast::prelude::PodcastEpisodeRepository;
    use entertainarr_domain::prelude::{Page, Sort, SortOrder};

    use crate::Pool;

    async fn seed(pool: &Pool) {
        let _: Vec<u64> = sqlx::query_scalar("insert into users (id, email, password) values (1, 'user1@example.com', 'password'), (2, 'user2@example.com', 'password') returning id").fetch_all(pool.as_ref()).await.unwrap();
        let _: Vec<u64> = sqlx::query_scalar("insert into podcasts (id, feed_url, title) values (1, 'first', 'first'), (2, 'second', 'second'), (3, 'third', 'third') returning id").fetch_all(pool.as_ref()).await.unwrap();
        let _: Vec<u64> = sqlx::query_scalar("insert into podcast_episodes (id, podcast_id, title, file_url, duration) values (1, 1, 'title 1', 'url 1', 100), (2, 1, 'title 2', 'url 2', 30), (3, 1, 'title 3', 'url 3', 90), (4, 2, 'title 4', 'url 4', 120), (5, 2, 'title 5', 'url 5', 100), (6, 3, 'title 6', 'url 6', 90) returning id").fetch_all(pool.as_ref()).await.unwrap();
        let _: Vec<(u64, u64)> = sqlx::query_as(
            "insert into user_podcasts (user_id, podcast_id, min_duration) values (1, 1, 80), (1, 2, null), (2, 2, null), (2, 3, 80) returning user_id, podcast_id",
        )
        .fetch_all(pool.as_ref())
        .await
        .unwrap();
        let _: Vec<(u64, u64)> = sqlx::query_as("insert into user_podcast_episodes (user_id, podcast_episode_id, progress, completed) values (1, 1, 123, true), (1, 4, 10, false), (2, 4, 153, true), (2, 5, 10, false) returning user_id, podcast_episode_id").fetch_all(pool.as_ref()).await.unwrap();
    }

    #[tokio::test]
    async fn should_list_all_subscribed_episodes() {
        let _ = tracing_subscriber::fmt::try_init();

        let tmpdir = tempfile::tempdir().unwrap();
        let pool = crate::Pool::test(&tmpdir.path().join("db")).await;

        seed(&pool).await;
        let list = pool
            .list(ListPodcastEpisodeParams {
                user_id: 1,
                filter: ListPodcastEpisodeFilter {
                    podcast_ids: &[],
                    filtered: None,
                    subscribed: Some(true),
                    watched: None,
                },
                sort: Sort {
                    field: PodcastEpisodeField::PublishedAt,
                    order: SortOrder::Asc,
                },
                page: Page {
                    limit: 10,
                    offset: 0,
                },
            })
            .await
            .unwrap();
        assert_eq!(list.len(), 5);
    }

    #[tokio::test]
    async fn should_list_all_watched_episodes() {
        let _ = tracing_subscriber::fmt::try_init();

        let tmpdir = tempfile::tempdir().unwrap();
        let pool = crate::Pool::test(&tmpdir.path().join("db")).await;

        seed(&pool).await;
        let list = pool
            .list(ListPodcastEpisodeParams {
                user_id: 1,
                filter: ListPodcastEpisodeFilter {
                    podcast_ids: &[],
                    filtered: None,
                    subscribed: Some(true),
                    watched: Some(true),
                },
                sort: Sort {
                    field: PodcastEpisodeField::PublishedAt,
                    order: SortOrder::Asc,
                },
                page: Page {
                    limit: 10,
                    offset: 0,
                },
            })
            .await
            .unwrap();
        assert_eq!(list.len(), 1);
    }

    #[tokio::test]
    async fn should_list_all_unwatched_episodes() {
        let _ = tracing_subscriber::fmt::try_init();

        let tmpdir = tempfile::tempdir().unwrap();
        let pool = crate::Pool::test(&tmpdir.path().join("db")).await;

        seed(&pool).await;
        let list = pool
            .list(ListPodcastEpisodeParams {
                user_id: 1,
                filter: ListPodcastEpisodeFilter {
                    podcast_ids: &[],
                    filtered: None,
                    subscribed: Some(true),
                    watched: Some(false),
                },
                sort: Sort {
                    field: PodcastEpisodeField::PublishedAt,
                    order: SortOrder::Asc,
                },
                page: Page {
                    limit: 10,
                    offset: 0,
                },
            })
            .await
            .unwrap();
        assert_eq!(list.len(), 4);
    }

    #[tokio::test]
    async fn should_upsert_progress() {
        let _ = tracing_subscriber::fmt::try_init();

        let tmpdir = tempfile::tempdir().unwrap();
        let pool = crate::Pool::test(&tmpdir.path().join("db")).await;

        seed(&pool).await;
        let updated = pool
            .upsert_progressions(&[
                PodcastEpisodeProgressInput {
                    user_id: 1,
                    podcast_episode_id: 4,
                    progress: 100,
                    completed: true,
                },
                PodcastEpisodeProgressInput {
                    user_id: 1,
                    podcast_episode_id: 2,
                    progress: 60,
                    completed: false,
                },
            ])
            .await
            .unwrap();
        assert_eq!(updated.len(), 2);

        let list = pool
            .list(ListPodcastEpisodeParams {
                user_id: 1,
                filter: ListPodcastEpisodeFilter {
                    podcast_ids: &[],
                    filtered: None,
                    subscribed: Some(true),
                    watched: Some(false),
                },
                sort: Sort {
                    field: PodcastEpisodeField::PublishedAt,
                    order: SortOrder::Asc,
                },
                page: Page {
                    limit: 10,
                    offset: 0,
                },
            })
            .await
            .unwrap();
        assert_eq!(list.len(), 3);
    }

    #[tokio::test]
    async fn should_list_only_filtered_episodes() {
        let _ = tracing_subscriber::fmt::try_init();

        let tmpdir = tempfile::tempdir().unwrap();
        let pool = crate::Pool::test(&tmpdir.path().join("db")).await;

        let user = pool
            .create_user("user@example.com", "password")
            .await
            .unwrap();

        let input = PodcastInput::builder()
            .title("title")
            .feed_url("http://feed_url")
            .episodes(vec![
                PodcastEpisodeInput::builder()
                    .title("first title")
                    .file_url("http://file.mp3")
                    .duration(Duration::from_secs(4000))
                    .build(),
                PodcastEpisodeInput::builder()
                    .title("second title")
                    .file_url("http://file.mp3")
                    .duration(Duration::from_secs(1200))
                    .build(),
                PodcastEpisodeInput::builder()
                    .title("third title")
                    .file_url("http://file.mp3")
                    .build(),
            ])
            .build();
        let podcast = pool.upsert_podcast(pool.as_ref(), &input).await.unwrap();
        pool.upsert_podcast_episodes(pool.as_ref(), podcast.id, &input.episodes)
            .await
            .unwrap();

        pool.create_podcast_subscription(user.id, podcast.id)
            .await
            .unwrap();

        pool.update_podcast_subscription(
            user.id,
            podcast.id,
            PodcastSubscriptionUpdate {
                min_duration: Some(2000),
                max_duration: None,
            },
        )
        .await
        .unwrap();

        // should list all episodes that are not filtered, even not subscribed
        let list = pool
            .list_podcast_episodes(ListPodcastEpisodeParams {
                user_id: user.id,
                filter: ListPodcastEpisodeFilter {
                    podcast_ids: &[],
                    filtered: Some(false),
                    subscribed: None,
                    watched: Some(false),
                },
                sort: Sort {
                    field: PodcastEpisodeField::PublishedAt,
                    order: SortOrder::Asc,
                },
                page: Page {
                    limit: 100,
                    offset: 0,
                },
            })
            .await
            .unwrap();
        assert_eq!(list.len(), 2);

        // should list all episodes that are not filtered, even not subscribed
        let list = pool
            .list_podcast_episodes(ListPodcastEpisodeParams {
                user_id: 0,
                filter: ListPodcastEpisodeFilter {
                    podcast_ids: &[],
                    filtered: Some(false),
                    subscribed: None,
                    watched: Some(false),
                },
                sort: Sort {
                    field: PodcastEpisodeField::PublishedAt,
                    order: SortOrder::Asc,
                },
                page: Page {
                    limit: 100,
                    offset: 0,
                },
            })
            .await
            .unwrap();
        assert_eq!(list.len(), 3);

        // should list all episodes that are not filtered
        let list = pool
            .list_podcast_episodes(ListPodcastEpisodeParams {
                user_id: 0,
                filter: ListPodcastEpisodeFilter {
                    podcast_ids: &[],
                    filtered: Some(false),
                    subscribed: Some(true),
                    watched: Some(false),
                },
                sort: Sort {
                    field: PodcastEpisodeField::PublishedAt,
                    order: SortOrder::Asc,
                },
                page: Page {
                    limit: 100,
                    offset: 0,
                },
            })
            .await
            .unwrap();
        assert!(list.is_empty());
    }
}