Add stubs for migrating devices, access tokens and refresh tokens

Author: reivilibre

crates/syn2mas/src/migration.rs

@@ -68,6 +68,12 @@ struct UsersMigrated {
     user_localparts_to_uuid: HashMap<CompactString, Uuid>,
 }
 
+struct DevicesMigrated {
+    /// Lookup table from `(user_id, device_id)` pairs to the
+    /// UUID of the `compat_session` in MAS
+    devices_to_uuid: HashMap<(CompactString, CompactString), Uuid>,
+}
+
 /// Performs a migration from Synapse's database to MAS' database.
 ///
 /// # Panics
@@ -96,7 +102,7 @@ pub async fn migrate(
         counts
             .users
             .try_into()
-            .expect("More than usize::MAX users — wow!"),
+            .expect("More than usize::MAX users — unable to handle this many!"),
         server_name,
         rng,
     )
@@ -121,6 +127,29 @@ pub async fn migrate(
     )
     .await?;
 
+    let migrated_devices = migrate_devices(
+        synapse,
+        mas,
+        counts
+            .devices
+            .try_into()
+            .expect("More than usize::MAX devices — unable to handle this many!"),
+        server_name,
+        rng,
+        &migrated_users.user_localparts_to_uuid,
+    )
+    .await?;
+
+    migrate_access_and_refresh_tokens(
+        synapse,
+        mas,
+        server_name,
+        rng,
+        &migrated_users.user_localparts_to_uuid,
+        &migrated_devices.devices_to_uuid,
+    )
+    .await?;
+
     Ok(())
 }
 
@@ -312,6 +341,36 @@ async fn migrate_external_ids(
 
     Ok(())
 }
+#[tracing::instrument(skip_all, level = Level::INFO)]
+async fn migrate_devices(
+    synapse: &mut SynapseReader<'_>,
+    mas: &mut MasWriter<'_>,
+    device_count_hint: usize,
+    server_name: &str,
+    rng: &mut impl RngCore,
+    user_localparts_to_uuid: &HashMap<CompactString, Uuid>,
+) -> Result<DevicesMigrated, Error> {
+    // TODO is 1:1 enough capacity for a HashMap?
+    let mut devices_to_uuid = HashMap::with_capacity(device_count_hint);
+
+    todo!();
+
+    Ok(DevicesMigrated { devices_to_uuid })
+}
+
+#[tracing::instrument(skip_all, level = Level::INFO)]
+async fn migrate_access_and_refresh_tokens(
+    synapse: &mut SynapseReader<'_>,
+    mas: &mut MasWriter<'_>,
+    server_name: &str,
+    rng: &mut impl RngCore,
+    user_localparts_to_uuid: &HashMap<CompactString, Uuid>,
+    devices: &HashMap<(CompactString, CompactString), Uuid>,
+) -> Result<(), Error> {
+    todo!();
+
+    Ok(())
+}
 
 fn transform_user(
     user: &SynapseUser,

crates/syn2mas/src/synapse_reader/mod.rs

@@ -235,13 +235,21 @@ pub struct SynapseRefreshToken {
 /// underneath our feet. It's still not a good idea to run Synapse at the same
 /// time as the migration.
 // TODO not complete!
-const TABLES_TO_LOCK: &[&str] = &["users", "user_threepids", "user_external_ids"];
+const TABLES_TO_LOCK: &[&str] = &[
+    "users",
+    "user_threepids",
+    "user_external_ids",
+    "devices",
+    "access_tokens",
+    "refresh_tokens",
+];
 
 /// Number of migratable rows in various Synapse tables.
 /// Used to estimate progress.
 #[derive(Clone, Debug)]
 pub struct SynapseRowCounts {
     pub users: i64,
+    pub devices: i64,
 }
 
 pub struct SynapseReader<'c> {
@@ -312,19 +320,27 @@ impl<'conn> SynapseReader<'conn> {
     ///
     /// - An underlying database error
     pub async fn count_rows(&mut self) -> Result<SynapseRowCounts, Error> {
-        let users = sqlx::query(
+        let users: i64 = sqlx::query_scalar(
             "
             SELECT COUNT(1) FROM users
             WHERE appservice_id IS NULL AND is_guest = 0
             ",
         )
         .fetch_one(&mut *self.txn)
         .await
-        .into_database("counting Synapse users")?
-        .try_get::<i64, _>(0)
-        .into_database("couldn't decode count of Synapse users table")?;
+        .into_database("counting Synapse users")?;
 
-        Ok(SynapseRowCounts { users })
+        let devices = sqlx::query_scalar(
+            "
+            SELECT COUNT(1) FROM devices
+            WHERE NOT hidden
+            ",
+        )
+        .fetch_one(&mut *self.txn)
+        .await
+        .into_database("counting Synapse devices")?;
+
+        Ok(SynapseRowCounts { users, devices })
     }
 
     /// Reads Synapse users, excluding application service users (which do not