Cargo.lock

@@ -1185,17 +1185,6 @@ dependencies = [
 ]
 
 [[package]]
-name = "derivative"
-version = "2.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fcc3dd5e9e9c0b295d6e1e4d811fb6f157d5ffd784b8d202fc62eac8035a770b"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 1.0.109",
-]
-
-[[package]]
 name = "digest"
 version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1324,6 +1313,18 @@ dependencies = [
 ]
 
 [[package]]
+name = "educe"
+version = "0.5.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e4bd92664bf78c4d3dba9b7cdafce6fa15b13ed3ed16175218196942e99168a8"
+dependencies = [
+ "enum-ordinalize",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.66",
+]
+
+[[package]]
 name = "either"
 version = "1.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1333,6 +1334,26 @@ dependencies = [
 ]
 
 [[package]]
+name = "enum-ordinalize"
+version = "4.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fea0dcfa4e54eeb516fe454635a95753ddd39acda650ce703031c6973e315dd5"
+dependencies = [
+ "enum-ordinalize-derive",
+]
+
+[[package]]
+name = "enum-ordinalize-derive"
+version = "4.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0d28318a75d4aead5c4db25382e8ef717932d0346600cacae6357eb5941bc5ff"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.66",
+]
+
+[[package]]
 name = "equivalent"
 version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -3314,9 +3335,9 @@ dependencies = [
 
 [[package]]
 name = "sea-orm"
-version = "0.12.15"
+version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c8814e37dc25de54398ee62228323657520b7f29713b8e238649385dbe473ee0"
+checksum = "ea1fee0cf8528dbe6eda29d5798afc522a63b75e44c5b15721e6e64af9c7cc4b"
 dependencies = [
  "async-stream",
  "async-trait",
@@ -3342,9 +3363,9 @@ dependencies = [
 
 [[package]]
 name = "sea-orm-cli"
-version = "0.12.15"
+version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "620bc560062ae251b1366bde43b3f1508445cab5c2c8cbdb397034638ab1b357"
+checksum = "5f0b8869c75cf3fbb1bd860abb025033cd2e514c5f4fa43e792697cb1fe6c882"
 dependencies = [
  "chrono",
  "clap",
@@ -3359,9 +3380,9 @@ dependencies = [
 
 [[package]]
 name = "sea-orm-macros"
-version = "0.12.15"
+version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5e115c6b078e013aa963cc2d38c196c2c40b05f03d0ac872fe06b6e0d5265603"
+checksum = "8737b566799ed0444f278d13c300c4c6f1a91782f60ff5825a591852d5502030"
 dependencies = [
  "heck 0.4.1",
  "proc-macro2",
@@ -3373,9 +3394,9 @@ dependencies = [
 
 [[package]]
 name = "sea-orm-migration"
-version = "0.12.15"
+version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ee8269bc6ff71afd6b78aa4333ac237a69eebd2cdb439036291e64fb4b8db23c"
+checksum = "216643749e26ce27ab6c51d3475f2692981d4a902d34455bcd322f412900df5c"
 dependencies = [
  "async-trait",
  "clap",
@@ -3390,13 +3411,13 @@ dependencies = [
 
 [[package]]
 name = "sea-query"
-version = "0.30.7"
+version = "0.31.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4166a1e072292d46dc91f31617c2a1cdaf55a8be4b5c9f4bf2ba248e3ac4999b"
+checksum = "7e5073b2cfed767511a57d18115f3b3d8bcb5690bf8c89518caec6cb22c0cd74"
 dependencies = [
  "bigdecimal",
  "chrono",
- "derivative",
+ "educe",
  "inherent",
  "ordered-float",
  "rust_decimal",
@@ -3408,9 +3429,9 @@ dependencies = [
 
 [[package]]
 name = "sea-query-binder"
-version = "0.5.0"
+version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "36bbb68df92e820e4d5aeb17b4acd5cc8b5d18b2c36a4dd6f4626aabfa7ab1b9"
+checksum = "754965d4aee6145bec25d0898e5c931e6c22859789ce62fd85a42a15ed5a8ce3"
 dependencies = [
  "bigdecimal",
  "chrono",
@@ -3437,9 +3458,9 @@ dependencies = [
 
 [[package]]
 name = "sea-schema"
-version = "0.14.2"
+version = "0.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "30d148608012d25222442d1ebbfafd1228dbc5221baf4ec35596494e27a2394e"
+checksum = "ad52149fc81836ea7424c3425d8f6ed8ad448dd16d2e4f6a3907ba46f3f2fd78"
 dependencies = [
  "futures",
  "sea-query",
@@ -3448,14 +3469,14 @@ dependencies = [
 
 [[package]]
 name = "sea-schema-derive"
-version = "0.2.0"
+version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c6f686050f76bffc4f635cda8aea6df5548666b830b52387e8bc7de11056d11e"
+checksum = "debdc8729c37fdbf88472f97fd470393089f997a909e535ff67c544d18cfccf0"
 dependencies = [
  "heck 0.4.1",
  "proc-macro2",
  "quote",
- "syn 1.0.109",
+ "syn 2.0.66",
 ]
 
 [[package]]
@@ -4008,9 +4029,9 @@ checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f"
 
 [[package]]
 name = "strum"
-version = "0.25.0"
+version = "0.26.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "290d54ea6f91c969195bdbcd7442c8c2a2ba87da8bf60a7ee86a235d4bc1e125"
+checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06"
 
 [[package]]
 name = "subtle"

Cargo.toml

@@ -62,8 +62,8 @@ openmls = "0.5"
 password-hash = "0.5"
 pbkdf2 = { version = "0.12", features = ["simple"] }
 rand = "0.8"
-sea-orm = { version = "0.12", features = ["sqlx-postgres", "sqlx-sqlite", "macros"] }
-sea-orm-migration = { version = "0.12", features = ["runtime-tokio-native-tls", "sqlx-postgres", "sqlx-sqlite"] }
+sea-orm = { version = "1.0", features = ["sqlx-postgres", "sqlx-sqlite", "macros"] }
+sea-orm-migration = { version = "1.0", features = ["runtime-tokio-native-tls", "sqlx-postgres", "sqlx-sqlite"] }
 serde = "1"
 serde_bytes = "0.11"
 serde_json = "1"

client/src/device_resolver.rs

@@ -24,7 +24,7 @@ impl DeviceResolver {
         let ident = *dirpkg.last_update().ident();
         info!(%ident, "storing new device entry");
         let dent = DirEntry::Pkg(dirpkg.clone());
-        let res = self.ident_ds.try_insert_dir_entry(&dent, None).await?;
+        let res = self.ident_ds.try_insert_dir_entry(&dent).await?;
         if res.is_some() {
             warn!(%ident, "had apparently better device entry when storing what we thought was a better one");
         }

db/src/device.rs

@@ -1,5 +1,7 @@
 use std::cmp::Ordering;
 
+use aspect_ident::DeviceIdent;
+use chrono::NaiveDateTime;
 use sea_orm::prelude::*;
 use sea_orm::{
     ActiveValue, EntityOrSelect, IntoActiveModel, QueryOrder, QuerySelect, TransactionTrait,
@@ -7,7 +9,7 @@ use sea_orm::{
 use tracing::*;
 
 use aspect_ident::device::{self, EquivocationReport};
-use aspect_sqldb::device as sqldevice;
+use aspect_sqldb::{device as sqldevice, user_device};
 use aspect_util::codec;
 use aspect_util::prelude::*;
 
@@ -60,11 +62,7 @@ impl DeviceDatastore {
     /// If we have a "better" entry in the database (like a device record with
     /// a higher sequence number) or construct one (like an equivocation report
     /// from trying to insert a conflicting DIR, we return that instead).
-    pub async fn try_insert_dir_entry(
-        &self,
-        dir: &DirEntry,
-        user_id: Option<i32>,
-    ) -> Result<Option<DirEntry>, Error> {
+    pub async fn try_insert_dir_entry(&self, dir: &DirEntry) -> Result<Option<DirEntry>, Error> {
         // TODO try to flatten this function out a bit
 
         let ident = dir.ident();
@@ -89,7 +87,6 @@ impl DeviceDatastore {
             device_id: ActiveValue::Set(ident_raw),
             seqno: ActiveValue::Set(seqno),
             is_equivocation: ActiveValue::Set(new_is_equiv),
-            owner_user_id: ActiveValue::Set(user_id),
             // don't set record yet so we can refer to dir_enc later
             ..Default::default()
         };
@@ -218,7 +215,7 @@ impl DeviceDatastore {
         &self,
         id: &device::DeviceIdent,
     ) -> Result<Option<DirEntry>, Error> {
-        let idbuf = id.as_ref().to_vec();
+        let idbuf = id.to_vec();
 
         let m = sqldevice::Entity::find()
             .filter(sqldevice::Column::DeviceId.eq(idbuf))
@@ -232,23 +229,81 @@ impl DeviceDatastore {
         Ok(Some(ent))
     }
 
+    /// Adds a device to a user if not already registered.
+    pub async fn add_device_to_user(
+        &self,
+        id: &device::DeviceIdent,
+        uid: i32,
+    ) -> Result<(), Error> {
+        let idbuf = id.to_vec();
+        let tx = self.dbc.begin().await?;
+
+        let found = user_device::Entity::find()
+            .filter(user_device::Column::DeviceId.eq(idbuf.clone()))
+            .limit(1)
+            .one(&tx)
+            .await?;
+
+        // If it's already registered then we should report who registered them.
+        if let Some(found_m) = found {
+            return Err(Error::DeviceAlreadyClaimed(*id, found_m.user_id));
+        }
+
+        let user_device_am = user_device::ActiveModel {
+            user_id: ActiveValue::Set(uid),
+            device_id: ActiveValue::Set(id.to_vec()),
+            added: ActiveValue::Set(get_now_ndt()),
+            ..Default::default()
+        };
+
+        user_device_am.insert(&tx).await?;
+
+        tx.commit().await?;
+        Ok(())
+    }
+
     /// Reports all of the valid device entries, returning the most recent
-    /// instances of each.  If `local` is set then only reports devices that
-    /// correspond to users in the database, but does not report which users
-    /// they correspond to.
+    /// instances of each.
     pub async fn dump_valid_devices(&self, local: bool) -> Result<Vec<DirEntry>, Error> {
         // TODO add a range parameter
 
-        let clause = if local {
-            sqldevice::Column::OwnerUserId
-                .is_not_null()
-                .and(sqldevice::Column::IsMostRecentKnown.eq(true))
-                .and(sqldevice::Column::IsEquivocation.eq(false))
-        } else {
-            sqldevice::Column::IsMostRecentKnown
-                .eq(true)
-                .and(sqldevice::Column::IsEquivocation.eq(false))
-        };
+        if local {
+            warn!("queried for local devices but we don't support this directly anymore");
+        }
+
+        let clause = sqldevice::Column::IsMostRecentKnown
+            .eq(true)
+            .and(sqldevice::Column::IsEquivocation.eq(false));
+
+        let mm = sqldevice::Entity::find()
+            .filter(clause)
+            .all(&self.dbc)
+            .await?;
+
+        let mut ents = Vec::new();
+        for m in mm {
+            // TODO Skip invalid devices?
+            ents.push(codec::decode::<DirEntry>(&m.record)?);
+        }
+
+        Ok(ents)
+    }
+
+    /// Reports all of the valid device entries, returning the most recent
+    /// instances of each in some range.
+    pub async fn dump_valid_devices_range(
+        &self,
+        start: DeviceIdent,
+        end: DeviceIdent,
+    ) -> Result<Vec<DirEntry>, Error> {
+        let clause = sqldevice::Column::DeviceId
+            .gte(start.to_vec())
+            .add(sqldevice::Column::DeviceId.lt(end.to_vec()))
+            .and(
+                sqldevice::Column::IsMostRecentKnown
+                    .eq(true)
+                    .and(sqldevice::Column::IsEquivocation.eq(false)),
+            );
 
         let mm = sqldevice::Entity::find()
             .filter(clause)
@@ -257,9 +312,14 @@ impl DeviceDatastore {
 
         let mut ents = Vec::new();
         for m in mm {
+            // TODO Skip invalid devices?
             ents.push(codec::decode::<DirEntry>(&m.record)?);
         }
 
         Ok(ents)
     }
 }
+
+fn get_now_ndt() -> NaiveDateTime {
+    chrono::Utc::now().naive_utc()
+}

db/src/errors.rs

@@ -50,6 +50,9 @@ pub enum Error {
     #[error("all workers for job queue '{0}' exited")]
     JobWorkersExited(String),
 
+    #[error("device {0} already claimed by UID {1}")]
+    DeviceAlreadyClaimed(DeviceIdent, i32),
+
     #[error("ident: {0}")]
     Ident(#[from] aspect_ident::errors::Error),
 

db/src/user.rs

@@ -3,11 +3,11 @@ use std::time;
 use chrono::NaiveDateTime;
 use sea_orm::{
     sea_query::OnConflict, ActiveValue, ColumnTrait, DatabaseConnection, EntityOrSelect,
-    EntityTrait, QueryFilter,
+    EntityTrait, QueryFilter, QuerySelect,
 };
 
 use aspect_ident::DeviceIdent;
-use aspect_sqldb::{device, user};
+use aspect_sqldb::{device, user, user_device};
 use tracing::*;
 
 use crate::{errors::*, password};
@@ -78,24 +78,48 @@ impl UserDatastore {
         Ok(password::check_password(password, &m.password_digest).is_ok())
     }
 
+    /// Checks if the user owns some particular device.
+    pub async fn check_user_owns_device(
+        &self,
+        username: &str,
+        device: &DeviceIdent,
+    ) -> Result<bool, Error> {
+        let uid = self.find_user_id_ok(username).await?;
+        let ud_m = user_device::Entity::find()
+            .filter(
+                user_device::Column::UserId
+                    .eq(uid)
+                    .and(user_device::Column::DeviceId.eq(device.to_vec())),
+            )
+            .limit(1)
+            .one(&self.dbc)
+            .await?;
+        Ok(ud_m.is_some())
+    }
+
     /// Gets the IDs of all of the user's devices.
     pub async fn get_user_devices(&self, username: &str) -> Result<Vec<DeviceIdent>, Error> {
         let uid = self.find_user_id_ok(username).await?;
 
-        // We only care about the device idents.  This is a bit of a waste of
-        // bandwidth but I don't really know how to make SeaORM query just that
-        // column using `find_with_related`.  Maybe we should just do two
-        // queries for that?  Or a sub select?
+        // We only care about the user's devices.  This is really inefficient
+        // since we shouldn't have to actually fetch the data, we should be able
+        // to use a sub query or something.
         //
         // See: https://www.sea-ql.org/SeaORM/docs/advanced-query/custom-select/
-        let devices = device::Entity::find()
-            .filter(
-                device::Column::OwnerUserId
-                    .eq(uid)
-                    .and(device::Column::IsMostRecentKnown.eq(true)),
-            )
+        let user_idents = user_device::Entity::find()
+            .filter(user_device::Column::UserId.eq(uid))
             .all(&self.dbc)
             .await?;
+        let devices =
+            device::Entity::find()
+                .filter(device::Column::IsMostRecentKnown.eq(true).and(
+                    device::Column::DeviceId.is_in(user_idents.into_iter().map(|e| {
+                        let v = Box::new(e.device_id);
+                        sea_orm::Value::Bytes(Some(v))
+                    })),
+                ))
+                .all(&self.dbc)
+                .await?;
 
         let mut idents = Vec::new();
         for m in devices {

homeserver/src/storage.rs

@@ -44,7 +44,7 @@ pub enum StorageError {
 
 pub struct HomeserverStorage {
     user_ds: Arc<UserDatastore>,
-    ident_ds: Arc<DeviceDatastore>,
+    device_ds: Arc<DeviceDatastore>,
     dmq_ds: Arc<DeviceMqDatastore>,
 }
 
@@ -92,9 +92,11 @@ impl HomeserverStorage {
     ) -> Result<bool, StorageError> {
         // TODO Make this query directly if the entry is there instead of
         // fetching all the devices.
-        let devs = self.user_ds.get_user_devices(username).await?;
-        trace!(%device, ?devs, "checking device membership"); // TODO remove
-        Ok(devs.contains(device))
+        trace!(%username, %device, "checking device membership"); // TODO remove
+        Ok(self
+            .user_ds
+            .check_user_owns_device(username, device)
+            .await?)
     }
 
     /// Checks if the provided password matches the user and that the device belongs to the user.
@@ -122,18 +124,19 @@ impl HomeserverStorage {
         let ident = *pkg.last_update().ident();
         let user_id = self.user_ds.find_user_id_ok(&username).await?;
 
-        // First insert the device entry.
+        // First add the device to the user so that it's locked in.
+        self.device_ds.add_device_to_user(&ident, user_id).await?;
+
+        // Then we can insert the actual device entry.
         let de = DirEntry::Pkg(pkg);
-        let res = self
-            .ident_ds
-            .try_insert_dir_entry(&de, Some(user_id))
-            .await?;
+        let res = self.device_ds.try_insert_dir_entry(&de).await?;
 
         if res.is_some() {
             return Err(StorageError::DeviceAlreadyRegistered(ident));
         }
 
-        // And then set up the queues the device needs.
+        // And while we're here we can insert the queues the device needs.
+        // TODO Maybe do this elsewhere?
         self.setup_device_queues(&ident).await?;
 
         Ok(())
@@ -184,7 +187,7 @@ impl HomeserverStorage {
         &self,
         id: &DeviceIdent,
     ) -> Result<Option<DirEntry>, StorageError> {
-        Ok(self.ident_ds.query_dir_entry(id).await?)
+        Ok(self.device_ds.query_dir_entry(id).await?)
     }
 
     /// Tries to accept the dir package, returning an error if unsuccessful or a
@@ -194,7 +197,7 @@ impl HomeserverStorage {
         pkg: DirPackage,
     ) -> Result<Option<DirEntry>, StorageError> {
         let ent = DirEntry::Pkg(pkg);
-        let res = self.ident_ds.try_insert_dir_entry(&ent, None).await;
+        let res = self.device_ds.try_insert_dir_entry(&ent).await;
 
         match res {
             Ok(ent) => Ok(ent),
@@ -212,7 +215,7 @@ impl HomeserverStorage {
 
     /// Dumps all known valid devices.
     pub async fn dump_devices(&self) -> Result<Vec<DirPackage>, StorageError> {
-        let ents = self.ident_ds.dump_valid_devices(true).await?;
+        let ents = self.device_ds.dump_valid_devices(true).await?;
         let mut pkgs = Vec::new();
         for ent in ents {
             if let DirEntry::Pkg(pkg) = ent {
@@ -291,11 +294,11 @@ pub fn open_on_primitives(
 
     // The more conventional datastores.
     let user_ds = Arc::new(UserDatastore::new(dbc.clone())?);
-    let ident_ds = Arc::new(DeviceDatastore::new(dbc.clone())?);
+    let device_ds = Arc::new(DeviceDatastore::new(dbc.clone())?);
 
     Ok(HomeserverStorage {
         user_ds,
-        ident_ds,
+        device_ds,
         dmq_ds,
     })
 }

sqldb/migration/src/idents.rs

@@ -44,13 +44,29 @@ pub enum Device {
     /// If the entry is an equivocation.
     IsEquivocation,
 
-    /// User ID of the owner, if they're local to this server.
-    OwnerUserId,
-
     /// Encoded DIR data or equivocation record
     Record,
 }
 
+/// Represents the devices that a user owns, possibly with some other metadata.
+#[derive(DeriveIden)]
+pub enum UserDevice {
+    Table,
+
+    /// ID of the entry, in case we ever need to refer succinctly to the
+    /// relation itself.
+    Id,
+
+    /// Owning user's ID.
+    UserId,
+
+    /// Device ID.
+    DeviceId,
+
+    /// Timestamp added.
+    Added,
+}
+
 /// Stupid-simple key-value store for the client to store run-time state in,
 /// codec-encoded.
 #[derive(DeriveIden)]

sqldb/migration/src/m000002_create_device_tables.rs

@@ -33,20 +33,12 @@ impl MigrationTrait for Migration {
                             .not_null()
                             .default(false),
                     )
-                    .col(ColumnDef::new(Device::OwnerUserId).integer().unsigned())
                     .col(ColumnDef::new(Device::Record).binary().not_null())
-                    .foreign_key(
-                        ForeignKey::create()
-                            .name("FK_device_owner_user_id_to_user")
-                            .from(Device::Table, Device::OwnerUserId)
-                            .to(User::Table, User::Id)
-                            .on_delete(ForeignKeyAction::SetNull)
-                            .on_update(ForeignKeyAction::Cascade),
-                    )
                     .to_owned(),
             )
             .await?;
 
+        // Needs to be split out because of how the syntax gets generated.
         manager
             .create_index(
                 Index::create()
@@ -60,12 +52,41 @@ impl MigrationTrait for Migration {
             .await?;
 
         manager
-            .create_index(
-                Index::create()
-                    .name("IDX_device_on_owner_user_id")
-                    .table(Device::Table)
-                    .col(Device::OwnerUserId)
+            .create_table(
+                Table::create()
+                    .table(UserDevice::Table)
                     .if_not_exists()
+                    .col(
+                        ColumnDef::new(UserDevice::Id)
+                            .big_integer()
+                            .primary_key()
+                            .auto_increment(),
+                    )
+                    .col(ColumnDef::new(UserDevice::UserId).integer().not_null())
+                    .col(ColumnDef::new(UserDevice::DeviceId).binary().not_null())
+                    .col(ColumnDef::new(UserDevice::Added).timestamp().not_null())
+                    .index(
+                        Index::create()
+                            .name("IDX_user_device_on_owner_user_id")
+                            .if_not_exists()
+                            .col(UserDevice::UserId)
+                            .unique(),
+                    )
+                    .index(
+                        Index::create()
+                            .name("IDX_user_device_on_device_id")
+                            .if_not_exists()
+                            .col(UserDevice::DeviceId)
+                            .unique(),
+                    )
+                    .foreign_key(
+                        ForeignKey::create()
+                            .name("FK_user_device_user_id_to_user_id")
+                            .from(UserDevice::Table, UserDevice::UserId)
+                            .to(User::Table, User::Id)
+                            .on_delete(ForeignKeyAction::Cascade)
+                            .on_update(ForeignKeyAction::Cascade),
+                    )
                     .to_owned(),
             )
             .await?;
@@ -78,6 +99,10 @@ impl MigrationTrait for Migration {
             .drop_table(Table::drop().table(Device::Table).to_owned())
             .await?;
 
+        manager
+            .drop_table(Table::drop().table(UserDevice::Table).to_owned())
+            .await?;
+
         Ok(())
     }
 }