syn2mas: Migrate threepids to MAS (#3878)

* Add a table to hold unsupported threepids

* Migrate threepids from Synapse to MAS

Author: reivilibre

crates/storage-pg/migrations/20250124151529_unsupported_threepids_table.sql

@@ -0,0 +1,30 @@
+-- Copyright 2025 New Vector Ltd.
+--
+-- SPDX-License-Identifier: AGPL-3.0-only
+-- Please see LICENSE in the repository root for full details.
+
+
+
+-- Tracks third-party ID associations that have been verified but are
+-- not currently supported by MAS.
+-- This is currently used when importing third-party IDs from Synapse,
+-- which historically could verify at least phone numbers.
+-- E-mail associations will not be stored in this table because those are natively
+-- supported by MAS; see the `user_emails` table.
+
+CREATE TABLE user_unsupported_third_party_ids(
+    -- The owner of the third-party ID assocation
+    user_id UUID NOT NULL
+      REFERENCES users(user_id) ON DELETE CASCADE,
+
+    -- What type of association is this?
+    medium TEXT NOT NULL,
+
+    -- The address of the associated ID, e.g. a phone number or other identifier.
+    address TEXT NOT NULL,
+
+    -- When the association was created
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
+
+    PRIMARY KEY (user_id, medium, address)
+);

crates/syn2mas/.sqlx/query-b11590549fdd4cdcd36c937a353b5b37ab50db3505712c35610b822cda322b5b.json

@@ -27,14 +27,14 @@ rand.workspace = true
 uuid = "1.10.0"
 ulid = { workspace = true, features = ["uuid"] }
 
+mas-config.workspace = true
+
 [dev-dependencies]
 mas-storage-pg.workspace = true
 
 anyhow.workspace = true
 insta.workspace = true
 serde.workspace = true
 
-mas-config.workspace = true
-
 [lints]
 workspace = true

crates/syn2mas/.sqlx/query-dfbd462f7874d3dae551f2a0328a853a8a7efccdc20b968d99d8c18deda8dd00.json

@@ -207,14 +207,33 @@ pub struct MasNewUserPassword {
     pub created_at: DateTime<Utc>,
 }
 
+pub struct MasNewEmailThreepid {
+    pub user_email_id: Uuid,
+    pub user_id: Uuid,
+    pub email: String,
+    pub created_at: DateTime<Utc>,
+}
+
+pub struct MasNewUnsupportedThreepid {
+    pub user_id: Uuid,
+    pub medium: String,
+    pub address: String,
+    pub created_at: DateTime<Utc>,
+}
+
 /// The 'version' of the password hashing scheme used for passwords when they are
 /// migrated from Synapse to MAS.
 /// This is version 1, as in the previous syn2mas script.
 // TODO hardcoding version to `1` may not be correct long-term?
 pub const MIGRATED_PASSWORD_VERSION: u16 = 1;
 
 /// List of all MAS tables that are written to by syn2mas.
-pub const MAS_TABLES_AFFECTED_BY_MIGRATION: &[&str] = &["users", "user_passwords"];
+pub const MAS_TABLES_AFFECTED_BY_MIGRATION: &[&str] = &[
+    "users",
+    "user_passwords",
+    "user_emails",
+    "user_unsupported_third_party_ids",
+];
 
 /// Detect whether a syn2mas migration has started on the given database.
 ///
@@ -563,11 +582,11 @@ impl<'conn> MasWriter<'conn> {
         &mut self,
         passwords: Vec<MasNewUserPassword>,
     ) -> Result<(), Error> {
-        self.writer_pool.spawn_with_connection(move |conn| Box::pin(async move {
-            if passwords.is_empty() {
-                return Ok(());
-            }
+        if passwords.is_empty() {
+            return Ok(());
+        }
 
+        self.writer_pool.spawn_with_connection(move |conn| Box::pin(async move {
             let mut user_password_ids: Vec<Uuid> = Vec::with_capacity(passwords.len());
             let mut user_ids: Vec<Uuid> = Vec::with_capacity(passwords.len());
             let mut hashed_passwords: Vec<String> = Vec::with_capacity(passwords.len());
@@ -603,6 +622,100 @@ impl<'conn> MasWriter<'conn> {
             Ok(())
         })).await
     }
+
+    #[tracing::instrument(skip_all, level = Level::DEBUG)]
+    pub async fn write_email_threepids(
+        &mut self,
+        threepids: Vec<MasNewEmailThreepid>,
+    ) -> Result<(), Error> {
+        if threepids.is_empty() {
+            return Ok(());
+        }
+        self.writer_pool.spawn_with_connection(move |conn| {
+            Box::pin(async move {
+                let mut user_email_ids: Vec<Uuid> = Vec::with_capacity(threepids.len());
+                let mut user_ids: Vec<Uuid> = Vec::with_capacity(threepids.len());
+                let mut emails: Vec<String> = Vec::with_capacity(threepids.len());
+                let mut created_ats: Vec<DateTime<Utc>> = Vec::with_capacity(threepids.len());
+
+                for MasNewEmailThreepid {
+                    user_email_id,
+                    user_id,
+                    email,
+                    created_at,
+                } in threepids
+                {
+                    user_email_ids.push(user_email_id);
+                    user_ids.push(user_id);
+                    emails.push(email);
+                    created_ats.push(created_at);
+                }
+
+                // `confirmed_at` is going to get removed in a future MAS release,
+                // so just populate with `created_at`
+                sqlx::query!(
+                    r#"
+                    INSERT INTO syn2mas__user_emails
+                    (user_email_id, user_id, email, created_at, confirmed_at)
+                    SELECT * FROM UNNEST($1::UUID[], $2::UUID[], $3::TEXT[], $4::TIMESTAMP WITH TIME ZONE[], $4::TIMESTAMP WITH TIME ZONE[])
+                    "#,
+                    &user_email_ids[..],
+                    &user_ids[..],
+                    &emails[..],
+                    &created_ats[..],
+                ).execute(&mut *conn).await.into_database("writing emails to MAS")?;
+
+                Ok(())
+            })
+        }).await
+    }
+
+    #[tracing::instrument(skip_all, level = Level::DEBUG)]
+    pub async fn write_unsupported_threepids(
+        &mut self,
+        threepids: Vec<MasNewUnsupportedThreepid>,
+    ) -> Result<(), Error> {
+        if threepids.is_empty() {
+            return Ok(());
+        }
+        self.writer_pool.spawn_with_connection(move |conn| {
+            Box::pin(async move {
+                let mut user_ids: Vec<Uuid> = Vec::with_capacity(threepids.len());
+                let mut mediums: Vec<String> = Vec::with_capacity(threepids.len());
+                let mut addresses: Vec<String> = Vec::with_capacity(threepids.len());
+                let mut created_ats: Vec<DateTime<Utc>> = Vec::with_capacity(threepids.len());
+
+                for MasNewUnsupportedThreepid {
+                    user_id,
+                    medium,
+                    address,
+                    created_at,
+                } in threepids
+                {
+                    user_ids.push(user_id);
+                    mediums.push(medium);
+                    addresses.push(address);
+                    created_ats.push(created_at);
+                }
+
+                // `confirmed_at` is going to get removed in a future MAS release,
+                // so just populate with `created_at`
+                sqlx::query!(
+                    r#"
+                    INSERT INTO syn2mas__user_unsupported_third_party_ids
+                    (user_id, medium, address, created_at)
+                    SELECT * FROM UNNEST($1::UUID[], $2::TEXT[], $3::TEXT[], $4::TIMESTAMP WITH TIME ZONE[])
+                    "#,
+                    &user_ids[..],
+                    &mediums[..],
+                    &addresses[..],
+                    &created_ats[..],
+                ).execute(&mut *conn).await.into_database("writing unsupported threepids to MAS")?;
+
+                Ok(())
+            })
+        }).await
+    }
 }
 
 // How many entries to buffer at once, before writing a batch of rows to the database.
@@ -670,6 +783,63 @@ impl<'writer, 'conn> MasUserWriteBuffer<'writer, 'conn> {
     }
 }
 
+pub struct MasThreepidWriteBuffer<'writer, 'conn> {
+    email: Vec<MasNewEmailThreepid>,
+    unsupported: Vec<MasNewUnsupportedThreepid>,
+    writer: &'writer mut MasWriter<'conn>,
+}
+
+impl<'writer, 'conn> MasThreepidWriteBuffer<'writer, 'conn> {
+    pub fn new(writer: &'writer mut MasWriter<'conn>) -> Self {
+        MasThreepidWriteBuffer {
+            email: Vec::with_capacity(WRITE_BUFFER_BATCH_SIZE),
+            unsupported: Vec::with_capacity(WRITE_BUFFER_BATCH_SIZE),
+            writer,
+        }
+    }
+
+    pub async fn finish(mut self) -> Result<(), Error> {
+        self.flush_emails().await?;
+        self.flush_unsupported().await?;
+        Ok(())
+    }
+
+    pub async fn flush_emails(&mut self) -> Result<(), Error> {
+        self.writer
+            .write_email_threepids(std::mem::take(&mut self.email))
+            .await?;
+        self.email.reserve_exact(WRITE_BUFFER_BATCH_SIZE);
+        Ok(())
+    }
+
+    pub async fn flush_unsupported(&mut self) -> Result<(), Error> {
+        self.writer
+            .write_unsupported_threepids(std::mem::take(&mut self.unsupported))
+            .await?;
+        self.unsupported.reserve_exact(WRITE_BUFFER_BATCH_SIZE);
+        Ok(())
+    }
+
+    pub async fn write_email(&mut self, user: MasNewEmailThreepid) -> Result<(), Error> {
+        self.email.push(user);
+        if self.email.len() >= WRITE_BUFFER_BATCH_SIZE {
+            self.flush_emails().await?;
+        }
+        Ok(())
+    }
+
+    pub async fn write_password(
+        &mut self,
+        unsupported: MasNewUnsupportedThreepid,
+    ) -> Result<(), Error> {
+        self.unsupported.push(unsupported);
+        if self.unsupported.len() >= WRITE_BUFFER_BATCH_SIZE {
+            self.flush_unsupported().await?;
+        }
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 mod test {
     use std::collections::{BTreeMap, BTreeSet};

crates/syn2mas/Cargo.toml

@@ -10,3 +10,5 @@ DROP TABLE syn2mas_restore_indices;
 
 ALTER TABLE syn2mas__users RENAME TO users;
 ALTER TABLE syn2mas__user_passwords RENAME TO user_passwords;
+ALTER TABLE syn2mas__user_emails RENAME TO user_emails;
+ALTER TABLE syn2mas__user_unsupported_third_party_ids RENAME TO user_unsupported_third_party_ids;

crates/syn2mas/src/mas_writer/mod.rs

@@ -39,3 +39,5 @@ CREATE TABLE syn2mas_restore_indices (
 -- Now we rename all tables that we touch during the migration.
 ALTER TABLE users RENAME TO syn2mas__users;
 ALTER TABLE user_passwords RENAME TO syn2mas__user_passwords;
+ALTER TABLE user_emails RENAME TO syn2mas__user_emails;
+ALTER TABLE user_unsupported_third_party_ids RENAME TO syn2mas__user_unsupported_third_party_ids;