[Add] Proper User managment

migrations/009_user_invitations.sql

@@ -0,0 +1,29 @@
+CREATE TABLE user_invitations (
+    invitation_id CHAR(36) NOT NULL,
+    token_hash BINARY(32) NOT NULL,
+    iota_id BIGINT NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    expires_at TIMESTAMP NULL,
+    state ENUM('pending', 'redeemed', 'revoked', 'expired') NOT NULL DEFAULT 'pending',
+    redeemed_user_id BIGINT NULL,
+    redeemed_at TIMESTAMP NULL,
+    revoked_at TIMESTAMP NULL,
+    PRIMARY KEY (invitation_id),
+    KEY idx_user_invitations_target_state (iota_id, state),
+    CONSTRAINT fk_user_invitations_iota
+        FOREIGN KEY (iota_id) REFERENCES iotas(id) ON DELETE CASCADE
+);
+
+CREATE TABLE pending_iota_user_provisioning (
+    user_id BIGINT NOT NULL,
+    iota_id BIGINT NOT NULL,
+    invitation_id CHAR(36) NOT NULL,
+    username VARBINARY(255) NOT NULL,
+    public_key BLOB NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY (user_id, iota_id),
+    CONSTRAINT fk_pending_iota_user_provisioning_iota
+        FOREIGN KEY (iota_id) REFERENCES iotas(id) ON DELETE CASCADE,
+    CONSTRAINT fk_pending_iota_user_provisioning_invitation
+        FOREIGN KEY (invitation_id) REFERENCES user_invitations(invitation_id) ON DELETE CASCADE
+);

mtp-type-maps

@@ -1 +1 @@
-Subproject commit f4e45aa3a3ad0e3c3a257f66857b904a1af7901c
+Subproject commit 909b3977cb6a233e74a925eb467412fb384844bc

src/db/user_repo.rs

@@ -11,6 +11,105 @@ use std::collections::HashMap;
 pub const MAX_PROTOCOL_ID: i64 = (1_i64 << 48) - 1;
 const ID_ALLOCATION_ATTEMPTS: usize = 16;
 
+#[derive(Clone, Debug)]
+pub struct PendingIotaUserProvisioning {
+    pub user_id: UserId,
+    pub iota_id: IotaId,
+    pub invitation_id: String,
+    pub username: String,
+    pub public_key: PublicKeyBundle,
+}
+
+pub async fn create_invitation(
+    invitation_id: &str,
+    token_hash: &[u8],
+    iota_id: IotaId,
+) -> Result<()> {
+    if uuid::Uuid::parse_str(invitation_id).is_err()
+        || token_hash.len() != 32
+        || !valid_protocol_id(iota_id.0)
+    {
+        return Err(OmegaError::Validation("invalid invitation".into()));
+    }
+    sqlx::query(
+        "INSERT INTO user_invitations (invitation_id, token_hash, iota_id) VALUES (?, ?, ?)",
+    )
+    .bind(invitation_id)
+    .bind(token_hash)
+    .bind(iota_id.0)
+    .execute(&pool().await?)
+    .await?;
+    Ok(())
+}
+
+pub async fn revoke_invitation(invitation_id: &str, iota_id: IotaId) -> Result<bool> {
+    let result = sqlx::query(
+        "UPDATE user_invitations SET state = 'revoked', revoked_at = UTC_TIMESTAMP() WHERE invitation_id = ? AND iota_id = ? AND state = 'pending'",
+    )
+    .bind(invitation_id)
+    .bind(iota_id.0)
+    .execute(&pool().await?)
+    .await?;
+    Ok(result.rows_affected() == 1)
+}
+
+pub async fn enqueue_iota_user_provisioning(
+    provisioning: &PendingIotaUserProvisioning,
+) -> Result<()> {
+    if !valid_protocol_id(provisioning.user_id.0)
+        || !valid_protocol_id(provisioning.iota_id.0)
+        || !valid_username(&provisioning.username)
+    {
+        return Err(OmegaError::Validation("invalid user provisioning".into()));
+    }
+    sqlx::query(
+        "INSERT INTO pending_iota_user_provisioning (user_id, iota_id, invitation_id, username, public_key) VALUES (?, ?, ?, ?, ?)",
+    )
+    .bind(provisioning.user_id.0)
+    .bind(provisioning.iota_id.0)
+    .bind(&provisioning.invitation_id)
+    .bind(&provisioning.username)
+    .bind(provisioning.public_key.try_as_bytes()?)
+    .execute(&pool().await?)
+    .await?;
+    Ok(())
+}
+
+pub async fn pending_iota_user_provisioning(
+    iota_id: IotaId,
+) -> Result<Vec<PendingIotaUserProvisioning>> {
+    let rows = sqlx::query(
+        "SELECT user_id, iota_id, invitation_id, username, public_key FROM pending_iota_user_provisioning WHERE iota_id = ? ORDER BY created_at",
+    )
+    .bind(iota_id.0)
+    .fetch_all(&pool().await?)
+    .await?;
+    rows.into_iter()
+        .map(|row| {
+            let public_key = PublicKeyBundle::from_bytes(&row.get::<Vec<u8>, _>("public_key"))
+                .map_err(|error| OmegaError::Validation(error.to_string()))?;
+            Ok(PendingIotaUserProvisioning {
+                user_id: UserId::from(row.get::<i64, _>("user_id")),
+                iota_id: IotaId::from(row.get::<i64, _>("iota_id")),
+                invitation_id: row.get("invitation_id"),
+                username: String::from_utf8(row.get::<Vec<u8>, _>("username"))
+                    .map_err(|error| OmegaError::Validation(error.to_string()))?,
+                public_key,
+            })
+        })
+        .collect()
+}
+
+pub async fn acknowledge_iota_user_provisioning(user_id: UserId, iota_id: IotaId) -> Result<bool> {
+    let result =
+        sqlx::query("DELETE FROM pending_iota_user_provisioning WHERE user_id = ? AND iota_id = ?")
+            .bind(user_id.0)
+            .bind(iota_id.0)
+            .execute(&pool().await?)
+            .await?;
+    Ok(result.rows_affected() == 1)
+}
+
 pub fn generate_protocol_id() -> UserId {
     loop {
         let value = rand::random::<u64>() & ((1_u64 << 48) - 1);

src/transport/connection.rs

@@ -79,6 +79,7 @@ pub(crate) fn validate_dispatch_fields(value: &CommunicationValue) -> OmikronRes
             | mtp::codec::CommunicationType::DeleteUserCredentialBegin
             | mtp::codec::CommunicationType::DeleteUserCredentialComplete
             | mtp::codec::CommunicationType::EraseHostedUserDataAck
+            | mtp::codec::CommunicationType::AcknowledgeIotaUserProvision
             | mtp::codec::CommunicationType::ReleaseUserFromIota
             | mtp::codec::CommunicationType::DeleteIota
             | mtp::codec::CommunicationType::GetNotifications

src/transport/handlers/account.rs

@@ -436,3 +436,44 @@ pub async fn erase_hosted_user_data_ack(
         }
     }
 }
+
+pub async fn acknowledge_iota_user_provisioning(
+    connection: Arc<OmikronConnection>,
+    value: CommunicationValue,
+) -> OmikronResult<()> {
+    value.require_id()?;
+    let Some(user_id) = value
+        .get_data(DataType::UserId)
+        .as_signed_number()
+        .and_then(|id| i64::try_from(id).ok())
+        .filter(|id| *id > 0)
+    else {
+        return connection
+            .send_error_response(value.require_id()?, CommunicationType::ErrorInvalidUserId)
+            .await;
+    };
+    let iota_id = IotaId::from(value.require_sender_i64()?);
+    match user_repo::acknowledge_iota_user_provisioning(UserId::from(user_id), iota_id).await {
+        Ok(true) => {
+            connection
+                .send(
+                    &CommunicationValue::new(CommunicationType::Success)
+                        .with_id(value.require_id()?),
+                )
+                .await
+        }
+        Ok(false) => {
+            connection
+                .send_error_response(
+                    value.require_id()?,
+                    CommunicationType::ErrorNotAuthenticated,
+                )
+                .await
+        }
+        Err(_) => {
+            connection
+                .send_error_response(value.require_id()?, CommunicationType::ErrorInternal)
+                .await
+        }
+    }
+}

src/transport/handlers/presence.rs

@@ -337,6 +337,7 @@ pub async fn iota_connected(
         .add_typed_default(DataType::UserIds, DataValue::Array(user_ids));
     connection.clone().send(&response).await?;
     crate::transport::omikron_manager::deliver_pending_erasures(iota_id).await;
+    crate::transport::omikron_manager::deliver_pending_user_provisioning(iota_id).await;
     publish_changed_states(&state, &before, &users).await;
     connection
         .send(&CommunicationValue::new(CommunicationType::Success).with_id(value.require_id()?))

src/transport/omikron_connection.rs

@@ -476,6 +476,10 @@ impl OmikronConnection {
             Some(CommunicationType::EraseHostedUserDataAck) => {
                 crate::transport::handlers::account::erase_hosted_user_data_ack(self, value).await
             }
+            Some(CommunicationType::AcknowledgeIotaUserProvision) => {
+                crate::transport::handlers::account::acknowledge_iota_user_provisioning(self, value)
+                    .await
+            }
             Some(CommunicationType::ReleaseUserFromIota) => {
                 crate::transport::handlers::account::release_from_iota(self, value).await
             }

src/transport/omikron_manager.rs

@@ -213,3 +213,33 @@ pub async fn deliver_pending_erasures(iota_id: i64) {
         let _ = connection.clone().send(&request).await;
     }
 }
+
+/* Provisioning retries until the destination Iota has persisted public
+ * account metadata and acknowledges it. Credentials never enter this path. */
+pub async fn deliver_pending_user_provisioning(iota_id: i64) {
+    let Ok(users) =
+        user_repo::pending_iota_user_provisioning(crate::models::IotaId::from(iota_id)).await
+    else {
+        return;
+    };
+    let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) else {
+        return;
+    };
+    let Some(connection) = get_connected_omikron(omikron_id) else {
+        return;
+    };
+    for user in users {
+        let Ok(public_key) = user.public_key.try_to_base64() else {
+            continue;
+        };
+        let request = CommunicationValue::new(CommunicationType::ProvisionIotaUser)
+            .add_typed_default(
+                DataType::UserId,
+                DataValue::SignedNumber(user.user_id.0.into()),
+            )
+            .add_typed_default(DataType::Username, DataValue::Str(user.username))
+            .add_typed_default(DataType::PublicKey, DataValue::Str(public_key))
+            .add_typed_default(DataType::InvitationId, DataValue::Str(user.invitation_id));
+        let _ = connection.clone().send(&request).await;
+    }
+}