feat: implement synchronize sync group in sqlite repository

src/db/repository/device.rs

@@ -1,5 +1,7 @@
+use std::collections::HashSet;
+
 use chrono::{DateTime, Utc};
-use diesel::prelude::*;
+use diesel::{alias, dsl::not, prelude::*};
 
 use super::SqliteRepository;
 use crate::{
@@ -185,7 +187,54 @@ impl gpodder::DeviceRepository for SqliteRepository {
         group_id: i64,
         time_changed: DateTime<Utc>,
     ) -> Result<(), gpodder::AuthErr> {
-        todo!()
+        let time_changed = time_changed.timestamp();
+        let conn = &mut self.pool.get()?;
+
+        conn.transaction(|conn| {
+            let device_ids: Vec<i64> = devices::table
+                .filter(devices::sync_group_id.eq(group_id))
+                .select(devices::id)
+                .get_results(conn)?;
+
+            // For each device in the group, we get the list of subscriptions not yet in its own
+            // non-deleted list, and add it to the database
+            for device_id in device_ids.iter().copied() {
+                let d1 = alias!(device_subscriptions as d1);
+
+                let own_subscriptions = d1
+                    .filter(
+                        d1.field(device_subscriptions::device_id)
+                            .eq(device_id)
+                            .and(d1.field(device_subscriptions::deleted).eq(false)),
+                    )
+                    .select(d1.field(device_subscriptions::podcast_url));
+
+                let urls_to_add = device_subscriptions::table
+                    .select(device_subscriptions::podcast_url)
+                    .filter(
+                        device_subscriptions::device_id
+                            .eq_any(device_ids.iter())
+                            .and(device_subscriptions::deleted.eq(false))
+                            .and(not(
+                                device_subscriptions::podcast_url.eq_any(own_subscriptions)
+                            )),
+                    )
+                    .distinct()
+                    .load_iter(conn)?
+                    .collect::<Result<HashSet<String>, _>>()?;
+
+                super::subscription::insert_subscriptions_for_single_device(
+                    conn,
+                    device_id,
+                    urls_to_add.iter(),
+                    time_changed,
+                )?;
+            }
+
+            Ok::<_, diesel::result::Error>(())
+        })?;
+
+        Ok(())
     }
 
     fn devices_by_sync_group(

src/db/repository/subscription.rs

@@ -93,7 +93,32 @@ fn set_subscriptions_for_single_device(
     Ok(())
 }
 
-fn update_subscriptions_for_single_device(
+/// Add the given URLs to the device's list of subscriptions, meaning the URLs are truly inserted
+/// into the database. This function assumes the list of URLs is already free of URLs that already
+/// have a corresponding row in the database, so no conflict checks are performed.
+pub fn insert_subscriptions_for_single_device<'a>(
+    conn: &mut SqliteConnection,
+    device_id: i64,
+    urls: impl Iterator<Item = &'a String>,
+    time_changed: i64,
+) -> QueryResult<()> {
+    diesel::insert_into(device_subscriptions::table)
+        .values(
+            urls.into_iter()
+                .map(|url| db::NewDeviceSubscription {
+                    device_id,
+                    podcast_url: url.to_string(),
+                    deleted: false,
+                    time_changed,
+                })
+                .collect::<Vec<_>>(),
+        )
+        .execute(conn)?;
+
+    Ok(())
+}
+
+pub fn update_subscriptions_for_single_device(
     conn: &mut SqliteConnection,
     device_id: i64,
     add: &HashSet<String>,
@@ -149,19 +174,7 @@ fn update_subscriptions_for_single_device(
     // added list
     let urls_to_insert = add.difference(&urls_in_db);
 
-    diesel::insert_into(device_subscriptions::table)
+    insert_subscriptions_for_single_device(conn, device_id, urls_to_insert, time_changed)?;
-        .values(
-            urls_to_insert
-                .into_iter()
-                .map(|url| db::NewDeviceSubscription {
-                    device_id,
-                    podcast_url: url.to_string(),
-                    deleted: false,
-                    time_changed,
-                })
-                .collect::<Vec<_>>(),
-        )
-        .execute(conn)?;
 
     Ok(())
 }
@@ -214,7 +227,6 @@ impl gpodder::SubscriptionRepository for SqliteRepository {
         urls: Vec<String>,
         time_changed: chrono::DateTime<chrono::Utc>,
     ) -> Result<(), gpodder::AuthErr> {
-        // TODO use a better timestamp
         let time_changed = time_changed.timestamp();
         let urls: HashSet<String> = urls.into_iter().collect();
 
@@ -257,7 +269,6 @@ impl gpodder::SubscriptionRepository for SqliteRepository {
         remove: Vec<String>,
         time_changed: chrono::DateTime<chrono::Utc>,
     ) -> Result<(), gpodder::AuthErr> {
-        // TODO use a better timestamp
         let time_changed = time_changed.timestamp();
 
         // TODO URLs that are in both the added and removed lists will currently get "re-added",