feat: implement episode actions add endpoint

bruno/Device API/Update device data.bru

@@ -12,7 +12,7 @@ post {
 
 params:path {
   user: test
-  device_id: test4.json
+  device_id: test_device.json
 }
 
 body:json {

bruno/Episode Actions API/Upload action changes.bru

@@ -6,10 +6,30 @@ meta {
 
 post {
   url: http://localhost:8080/api/2/episodes/:username
-  body: none
+  body: json
-  auth: none
+  auth: inherit
 }
 
 params:path {
-  username: 
+  username: test.json
+}
+
+body:json {
+  [
+    {
+     "podcast": "http://example.com/feed.rss",
+     "episode": "http://example.com/files/s01e20.mp3",
+     "device": "test_device",
+     "action": "download",
+     "timestamp": "2009-12-12T09:00:00"
+    },
+    {
+     "podcast": "http://example.org/podcast.php",
+     "episode": "http://ftp.example.org/foo.ogg",
+     "action": "play",
+     "started": 15,
+     "position": 120,
+     "total":  500
+    }
+  ]
 }

migrations/2025-02-23-095541_initial/down.sql

@@ -1,6 +1,6 @@
 drop table episode_actions;
 
-drop table subscriptions;
+drop table device_subscriptions;
 
 drop table devices;
 

migrations/2025-02-23-095541_initial/up.sql

@@ -38,7 +38,7 @@ create table device_subscriptions (
 
     podcast_url text not null,
 
-    time_changed bigint not null default 0,
+    time_changed bigint not null,
     deleted boolean not null default false,
 
     unique (device_id, podcast_url)
@@ -55,6 +55,8 @@ create table episode_actions (
     podcast_url text not null,
     episode_url text not null,
 
+    time_changed bigint not null,
+
     timestamp bigint,
     action text not null,
     started integer,
@@ -62,10 +64,10 @@ create table episode_actions (
     total integer,
 
     -- Position should be set if the action type is "Play" and null otherwise
-    check ((action = "Play") = (position is not null)),
+    check ((action = "play") = (position is not null)),
 
     -- Started and position can only be set if the action type is "Play"
-    check (action = "Play" or (started is null and position is null)),
+    check (action = "play" or (started is null and position is null)),
 
     -- Started and position should be provided together
     check ((started is null) = (total is null))

src/db/models/device.rs

@@ -52,16 +52,34 @@ impl Device {
             .get_results(&mut pool.get()?)?)
     }
 
-    pub fn by_device_id(pool: &DbPool, user_id: i64, device_id: &str) -> DbResult<Option<Self>> {
+    pub fn device_id_to_id(
-        Ok(devices::dsl::devices
+        conn: &mut SqliteConnection,
+        user_id: i64,
+        device_id: &str,
+    ) -> diesel::QueryResult<i64> {
+        devices::table
+            .select(devices::id)
+            .filter(
+                devices::user_id
+                    .eq(user_id)
+                    .and(devices::device_id.eq(device_id)),
+            )
+            .get_result(conn)
+    }
+
+    pub fn by_device_id(
+        conn: &mut SqliteConnection,
+        user_id: i64,
+        device_id: &str,
+    ) -> diesel::QueryResult<Self> {
+        devices::dsl::devices
             .select(Self::as_select())
             .filter(
                 devices::user_id
                     .eq(user_id)
                     .and(devices::device_id.eq(device_id)),
             )
-            .get_result(&mut pool.get()?)
+            .get_result(conn)
-            .optional()?)
     }
 
     pub fn update(&self, pool: &DbPool) -> DbResult<()> {

src/db/models/episode_action.rs

@@ -22,6 +22,7 @@ pub struct EpisodeAction {
     device_id: Option<i64>,
     podcast_url: String,
     episode_url: String,
+    time_changed: i64,
     timestamp: Option<i64>,
     action: ActionType,
     started: Option<i32>,
@@ -33,14 +34,15 @@ pub struct EpisodeAction {
 #[diesel(table_name = episode_actions)]
 #[diesel(check_for_backend(diesel::sqlite::Sqlite))]
 pub struct NewEpisodeAction {
-    device_id: Option<i64>,
+    pub device_id: Option<i64>,
-    podcast_url: String,
+    pub podcast_url: String,
-    episode_url: String,
+    pub episode_url: String,
-    timestamp: Option<i64>,
+    pub time_changed: i64,
-    action: ActionType,
+    pub timestamp: Option<i64>,
-    started: Option<i32>,
+    pub action: ActionType,
-    position: Option<i32>,
+    pub started: Option<i32>,
-    total: Option<i32>,
+    pub position: Option<i32>,
+    pub total: Option<i32>,
 }
 
 #[derive(Serialize, Deserialize, FromSqlRow, Debug, AsExpression, Clone)]

src/db/repository/episode_action.rs

@@ -6,18 +6,76 @@ use crate::{
     gpodder,
 };
 
+impl From<gpodder::EpisodeAction> for db::NewEpisodeAction {
+    fn from(value: gpodder::EpisodeAction) -> Self {
+        let (action, started, position, total) = match value.action {
+            gpodder::EpisodeActionType::New => (db::ActionType::New, None, None, None),
+            gpodder::EpisodeActionType::Delete => (db::ActionType::Delete, None, None, None),
+            gpodder::EpisodeActionType::Download => (db::ActionType::Download, None, None, None),
+            gpodder::EpisodeActionType::Play {
+                started,
+                position,
+                total,
+            } => (db::ActionType::Play, started, Some(position), total),
+        };
+
+        db::NewEpisodeAction {
+            device_id: None,
+            podcast_url: value.podcast,
+            episode_url: value.episode,
+            time_changed: 0,
+            timestamp: value.timestamp.map(|t| t.and_utc().timestamp()),
+            action,
+            started,
+            position,
+            total,
+        }
+    }
+}
+
 impl gpodder::EpisodeActionRepository for SqliteRepository {
     fn add_episode_actions(
         &self,
         user: &gpodder::User,
         actions: Vec<gpodder::EpisodeAction>,
     ) -> Result<i64, gpodder::AuthErr> {
+        let time_changed = chrono::Utc::now().timestamp_millis();
+
+        // 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
         self.pool.get()?.transaction(|conn| {
-            for action in actions {}
+            for action in actions {
+                let device_id = if let Some(device) = &action.device {
+                    Some(db::Device::device_id_to_id(conn, user.id, device)?)
+                } else {
+                    None
+                };
+
+                let mut new_action: db::NewEpisodeAction = action.into();
+                new_action.device_id = device_id;
+                new_action.time_changed = time_changed;
+
+                diesel::insert_into(episode_actions::table)
+                    .values(&new_action)
+                    .execute(conn)?;
+            }
 
             Ok::<_, diesel::result::Error>(())
         })?;
 
+        Ok(time_changed + 1)
+    }
+
+    fn episode_actions_for_user(
+        &self,
+        user: &gpodder::User,
+        podcast: Option<String>,
+        device: Option<String>,
+        aggregated: bool,
+    ) -> (i64, Vec<gpodder::EpisodeAction>) {
         todo!()
     }
 }

src/db/schema.rs

@@ -27,6 +27,7 @@ diesel::table! {
         device_id -> Nullable<BigInt>,
         podcast_url -> Text,
         episode_url -> Text,
+        time_changed -> BigInt,
         timestamp -> Nullable<BigInt>,
         action -> Text,
         started -> Nullable<Integer>,

src/gpodder/mod.rs

@@ -78,4 +78,13 @@ pub trait EpisodeActionRepository {
     /// Insert the given episode actions into the datastore.
     fn add_episode_actions(&self, user: &User, actions: Vec<EpisodeAction>)
         -> Result<i64, AuthErr>;
+
+    /// Retrieve the list of episode actions for the given user.
+    fn episode_actions_for_user(
+        &self,
+        user: &User,
+        podcast: Option<String>,
+        device: Option<String>,
+        aggregated: bool,
+    ) -> (i64, Vec<EpisodeAction>);
 }

src/gpodder/models.rs

@@ -49,11 +49,11 @@ pub enum EpisodeActionType {
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct EpisodeAction {
-    podcast: String,
+    pub podcast: String,
-    episode: String,
+    pub episode: String,
-    timestamp: Option<NaiveDateTime>,
+    pub timestamp: Option<NaiveDateTime>,
     #[serde(default)]
-    device: Option<String>,
+    pub device: Option<String>,
     #[serde(flatten)]
-    action: EpisodeActionType,
+    pub action: EpisodeActionType,
 }

src/server/gpodder/advanced/episodes.rs

@@ -6,13 +6,13 @@ use axum::{
 };
 
 use crate::{
-    db,
+    gpodder::{self, EpisodeActionRepository},
     server::{
         error::{AppError, AppResult},
         gpodder::{
             auth_middleware,
             format::{Format, StringWithFormat},
-            models::{EpisodeAction, UpdatedUrlsResponse},
+            models::UpdatedUrlsResponse,
         },
         Context,
     },
@@ -27,8 +27,8 @@ pub fn router(ctx: Context) -> Router<Context> {
 async fn post_episode_actions(
     State(ctx): State<Context>,
     Path(username): Path<StringWithFormat>,
-    Extension(user): Extension<db::User>,
+    Extension(user): Extension<gpodder::User>,
-    Json(actions): Json<Vec<EpisodeAction>>,
+    Json(actions): Json<Vec<gpodder::EpisodeAction>>,
 ) -> AppResult<Json<UpdatedUrlsResponse>> {
     if username.format != Format::Json {
         return Err(AppError::NotFound);
@@ -38,7 +38,15 @@ async fn post_episode_actions(
         return Err(AppError::BadRequest);
     }
 
-    tracing::debug!("{:?}", actions);
+    Ok(
-
+        tokio::task::spawn_blocking(move || ctx.repo.add_episode_actions(&user, actions))
-    todo!()
+            .await
+            .unwrap()
+            .map(|timestamp| {
+                Json(UpdatedUrlsResponse {
+                    timestamp,
+                    update_urls: Vec::new(),
+                })
+            })?,
+    )
 }

src/server/gpodder/advanced/subscriptions.rs

@@ -13,9 +13,7 @@ use crate::{
         gpodder::{
             auth_middleware,
             format::{Format, StringWithFormat},
-            models::{
+            models::{SubscriptionDelta, SubscriptionDeltaResponse, UpdatedUrlsResponse},
-                DeviceType, SubscriptionDelta, SubscriptionDeltaResponse, UpdatedUrlsResponse,
-            },
         },
         Context,
     },

src/server/gpodder/models.rs

@@ -69,30 +69,3 @@ pub struct UpdatedUrlsResponse {
     pub timestamp: i64,
     pub update_urls: Vec<(String, String)>,
 }
-
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(rename_all = "lowercase")]
-#[serde(tag = "action")]
-pub enum EpisodeActionType {
-    Download,
-    Play {
-        #[serde(default)]
-        started: Option<i32>,
-        position: i32,
-        #[serde(default)]
-        total: Option<i32>,
-    },
-    Delete,
-    New,
-}
-
-#[derive(Serialize, Deserialize, Debug)]
-pub struct EpisodeAction {
-    podcast: String,
-    episode: String,
-    timestamp: Option<NaiveDateTime>,
-    #[serde(default)]
-    device: Option<String>,
-    #[serde(flatten)]
-    action: EpisodeActionType,
-}