feat: implement advanced subscription changes POST request

bruno/Subscriptions API/Upload subscription changes for device.bru

@@ -0,0 +1,16 @@
+meta {
+  name: Upload subscription changes for device
+  type: http
+  seq: 4
+}
+
+get {
+  url: http://localhost:8080/api/2/subscriptions/:username/:device_id
+  body: none
+  auth: none
+}
+
+params:path {
+  username: 
+  device_id: 
+}

bruno/Subscriptions API/Upload subscriptions for device.bru

@@ -21,5 +21,5 @@ auth:basic {
 }
 
 body:json {
-  ["https://example1.com", "https://example2.com"]
+  ["https://example2.com", "testing"]
 }

src/db/models/subscription.rs

@@ -120,4 +120,84 @@ impl Subscription {
             Ok(())
         })
     }
+
+    pub fn update_for_device(
+        pool: &DbPool,
+        device_id: i64,
+        added: Vec<String>,
+        removed: Vec<String>,
+        timestamp: i64,
+    ) -> DbResult<()> {
+        // TODO URLs that are in both the added and removed lists will currently get "re-added",
+        // meaning their change timestamp will be updated even though they haven't really changed.
+        let added: HashSet<_> = added.into_iter().collect();
+        let removed: HashSet<_> = removed.into_iter().collect();
+
+        pool.get()?.transaction(|conn| {
+            let urls_in_db: HashSet<String> = subscriptions::table
+                .select(subscriptions::url)
+                .filter(subscriptions::device_id.eq(device_id))
+                .get_results(&mut pool.get()?)?
+                .into_iter()
+                .collect();
+
+            // Subscriptions to remove are those that were already in the database and are now part
+            // of the removed list. Subscriptions that were never added in the first place don't
+            // need to be marked as deleted. We also only update those that aren't already marked
+            // as deleted.
+            let urls_to_delete = removed.intersection(&urls_in_db);
+
+            diesel::update(
+                subscriptions::table.filter(
+                    subscriptions::device_id
+                        .eq(device_id)
+                        .and(subscriptions::url.eq_any(urls_to_delete))
+                        .and(subscriptions::deleted.eq(false)),
+                ),
+            )
+            .set((
+                subscriptions::deleted.eq(true),
+                subscriptions::time_changed.eq(timestamp),
+            ))
+            .execute(conn)?;
+
+            // Subscriptions to update are those that are already in the database, but are also in
+            // the added list. Only those who were originally marked as deleted get updated.
+            let urls_to_update = added.intersection(&urls_in_db);
+
+            diesel::update(
+                subscriptions::table.filter(
+                    subscriptions::device_id
+                        .eq(device_id)
+                        .and(subscriptions::url.eq_any(urls_to_update))
+                        .and(subscriptions::deleted.eq(true)),
+                ),
+            )
+            .set((
+                subscriptions::deleted.eq(false),
+                subscriptions::time_changed.eq(timestamp),
+            ))
+            .execute(conn)?;
+
+            // Subscriptions to insert are those that aren't in the database and are part of the
+            // added list
+            let urls_to_insert = added.difference(&urls_in_db);
+
+            diesel::insert_into(subscriptions::table)
+                .values(
+                    urls_to_insert
+                        .into_iter()
+                        .map(|url| NewSubscription {
+                            device_id,
+                            url: url.to_string(),
+                            deleted: false,
+                            time_changed: timestamp,
+                        })
+                        .collect::<Vec<_>>(),
+                )
+                .execute(conn)?;
+
+            Ok(())
+        })
+    }
 }

src/server/gpodder/advanced/mod.rs

@@ -1,12 +1,14 @@
+mod auth;
+mod devices;
+mod subscriptions;
+
 use axum::Router;
 
 use crate::server::Context;
 
-mod auth;
-mod devices;
-
 pub fn router(ctx: Context) -> Router<Context> {
     Router::new()
         .nest("/auth", auth::router())
         .nest("/devices", devices::router(ctx.clone()))
+        .nest("/subscriptions", subscriptions::router(ctx.clone()))
 }

src/server/gpodder/advanced/subscriptions.rs

@@ -0,0 +1,72 @@
+use axum::{
+    extract::{Path, State},
+    middleware,
+    routing::post,
+    Extension, Json, Router,
+};
+
+use crate::{
+    db,
+    server::{
+        error::{AppError, AppResult},
+        gpodder::{
+            auth_middleware,
+            format::{Format, StringWithFormat},
+            models::{DeviceType, SubscriptionChangeResponse, SubscriptionDelta},
+        },
+        Context,
+    },
+};
+
+pub fn router(ctx: Context) -> Router<Context> {
+    Router::new()
+        .route("/{username}/{id}", post(post_subscription_changes))
+        .layer(middleware::from_fn_with_state(ctx.clone(), auth_middleware))
+}
+
+pub async fn post_subscription_changes(
+    State(ctx): State<Context>,
+    Path((username, id)): Path<(String, StringWithFormat)>,
+    Extension(user): Extension<db::User>,
+    Json(delta): Json<SubscriptionDelta>,
+) -> AppResult<Json<SubscriptionChangeResponse>> {
+    if id.format != Format::Json {
+        return Err(AppError::NotFound);
+    }
+
+    if username != user.username {
+        return Err(AppError::BadRequest);
+    }
+
+    let timestamp = chrono::Utc::now().timestamp_millis();
+
+    tokio::task::spawn_blocking(move || {
+        let device = if let Some(device) = db::Device::by_device_id(&ctx.pool, user.id, &id)? {
+            device
+        } else {
+            db::NewDevice::new(
+                user.id,
+                id.to_string(),
+                String::new(),
+                DeviceType::Other.into(),
+            )
+            .insert(&ctx.pool)?
+        };
+
+        db::Subscription::update_for_device(
+            &ctx.pool,
+            device.id,
+            delta.add,
+            delta.remove,
+            timestamp,
+        )
+    })
+    .await
+    .unwrap()?;
+
+    Ok(Json(SubscriptionChangeResponse {
+        timestamp: timestamp + 1,
+        // TODO implement URL sanitization
+        update_urls: vec![],
+    }))
+}

src/server/gpodder/models.rs

@@ -49,3 +49,15 @@ pub struct DevicePatch {
     pub caption: Option<String>,
     pub r#type: Option<DeviceType>,
 }
+
+#[derive(Deserialize)]
+pub struct SubscriptionDelta {
+    pub add: Vec<String>,
+    pub remove: Vec<String>,
+}
+
+#[derive(Serialize)]
+pub struct SubscriptionChangeResponse {
+    pub timestamp: i64,
+    pub update_urls: Vec<(String, String)>,
+}