feat(v1.1.8): invitations flow (migration 0030 + accept_invite returns session)

migration 0030: app_user_invitations table — surrogate id PK + unique
token_hash, app_id FK cascading, pre-stages email + display_name +
roles for a user that doesn't exist yet. One-shot via atomic UPDATE
SET accepted_at = NOW() WHERE accepted_at IS NULL.

UsersServiceImpl gains invitations: Arc<dyn AppUserInvitationRepo>
plus a mint_session() helper factored from login() and reused by
accept_invite().

users::invite(email, opts) is gated on AppUsersAdmin (per brief —
the most senior of the three new capabilities). Optional
EmailTemplateOpts inside InviteOpts: omitting the template skips the
email send so an admin can stamp invitations for out-of-band
delivery (mailers, printed onboarding letters, etc.). If the template
is present and the email service isn't configured, surfaces as
NotConfigured; non-NotConfigured failures are logged but kept silent
so the invitation row remains valid for retry.

users::accept_invite(token, password, display_name?) atomically
consumes the invitation, validates the new password, creates the
user (returning () on DuplicateEmail — sign-up beat acceptance,
they'll log in normally), and mints a fresh session via mint_session
so the caller can return both the user and a working session token
in one round trip.

Pre-staged roles are stored on the invitation row but not yet
applied — the app_user_roles table arrives in commit 8 (migration
0031). For commit 7 the staged-but-not-applied case logs an info
record so an operator can audit the gap.

list_invitations + revoke_invitation (admin-mediated, gated on
AppUsersAdmin) ship in this commit and become reachable from the
HTTP surface later in the series.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/manager-core/src/app_user_invitation_repo.rs

@@ -0,0 +1,219 @@
+//! CRUD over `app_user_invitations` (v1.1.8 commit 7).
+//!
+//! Distinct shape from the verification / reset repos: pre-stages
+//! email + display_name + roles for a user that doesn't exist yet.
+//! The admin UI references rows by surrogate `id`; the consumer
+//! references by `token_hash`. Both views are scoped by `app_id`.
+
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use picloud_shared::{AppId, InvitationId};
+use sqlx::PgPool;
+
+#[derive(Debug, thiserror::Error)]
+pub enum AppUserInvitationRepoError {
+    #[error("database error: {0}")]
+    Db(#[from] sqlx::Error),
+}
+
+#[derive(Debug, Clone)]
+pub struct InvitationRow {
+    pub id: InvitationId,
+    pub app_id: AppId,
+    pub email: String,
+    pub display_name: Option<String>,
+    pub roles: Vec<String>,
+    pub created_at: DateTime<Utc>,
+    pub expires_at: DateTime<Utc>,
+    pub accepted_at: Option<DateTime<Utc>>,
+}
+
+/// Snapshot returned by `consume` — just the fields needed to create
+/// the new user. Excludes timestamps and ids since the caller already
+/// has the token hash and the consume happened atomically.
+#[derive(Debug, Clone)]
+pub struct ConsumedInvitation {
+    pub email: String,
+    pub display_name: Option<String>,
+    pub roles: Vec<String>,
+}
+
+#[async_trait]
+pub trait AppUserInvitationRepo: Send + Sync {
+    async fn create(
+        &self,
+        app_id: AppId,
+        email: &str,
+        display_name: Option<&str>,
+        roles: &[String],
+        token_hash: &str,
+        expires_at: DateTime<Utc>,
+    ) -> Result<InvitationRow, AppUserInvitationRepoError>;
+
+    /// Pending invitations (accepted_at IS NULL) for the admin UI.
+    async fn list_pending(
+        &self,
+        app_id: AppId,
+    ) -> Result<Vec<InvitationRow>, AppUserInvitationRepoError>;
+
+    /// Admin revoke (deletes the pending row). Returns whether a row
+    /// was actually removed.
+    async fn revoke(
+        &self,
+        app_id: AppId,
+        invite_id: InvitationId,
+    ) -> Result<bool, AppUserInvitationRepoError>;
+
+    /// Atomic one-shot consume by token. Returns the staged fields
+    /// for the caller to create the user from; None on
+    /// missing / already-accepted / expired / wrong app.
+    async fn consume_by_token(
+        &self,
+        app_id: AppId,
+        token_hash: &str,
+    ) -> Result<Option<ConsumedInvitation>, AppUserInvitationRepoError>;
+
+    /// GC: drop accepted or expired rows.
+    async fn gc(&self, batch_size: i64) -> Result<u64, AppUserInvitationRepoError>;
+}
+
+pub struct PostgresAppUserInvitationRepo {
+    pool: PgPool,
+}
+
+impl PostgresAppUserInvitationRepo {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl AppUserInvitationRepo for PostgresAppUserInvitationRepo {
+    async fn create(
+        &self,
+        app_id: AppId,
+        email: &str,
+        display_name: Option<&str>,
+        roles: &[String],
+        token_hash: &str,
+        expires_at: DateTime<Utc>,
+    ) -> Result<InvitationRow, AppUserInvitationRepoError> {
+        let row: InvitationRecord = sqlx::query_as(
+            "INSERT INTO app_user_invitations \
+                 (token_hash, app_id, email, display_name, roles, expires_at) \
+             VALUES ($1, $2, $3, $4, $5, $6) \
+             RETURNING id, app_id, email, display_name, roles, \
+                       created_at, expires_at, accepted_at",
+        )
+        .bind(token_hash)
+        .bind(app_id.into_inner())
+        .bind(email)
+        .bind(display_name)
+        .bind(roles)
+        .bind(expires_at)
+        .fetch_one(&self.pool)
+        .await?;
+        Ok(row.into())
+    }
+
+    async fn list_pending(
+        &self,
+        app_id: AppId,
+    ) -> Result<Vec<InvitationRow>, AppUserInvitationRepoError> {
+        let rows: Vec<InvitationRecord> = sqlx::query_as(
+            "SELECT id, app_id, email, display_name, roles, \
+                    created_at, expires_at, accepted_at \
+             FROM app_user_invitations \
+             WHERE app_id = $1 AND accepted_at IS NULL \
+             ORDER BY created_at DESC",
+        )
+        .bind(app_id.into_inner())
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows.into_iter().map(Into::into).collect())
+    }
+
+    async fn revoke(
+        &self,
+        app_id: AppId,
+        invite_id: InvitationId,
+    ) -> Result<bool, AppUserInvitationRepoError> {
+        let res = sqlx::query(
+            "DELETE FROM app_user_invitations \
+             WHERE app_id = $1 AND id = $2 AND accepted_at IS NULL",
+        )
+        .bind(app_id.into_inner())
+        .bind(invite_id.into_inner())
+        .execute(&self.pool)
+        .await?;
+        Ok(res.rows_affected() > 0)
+    }
+
+    async fn consume_by_token(
+        &self,
+        app_id: AppId,
+        token_hash: &str,
+    ) -> Result<Option<ConsumedInvitation>, AppUserInvitationRepoError> {
+        let row: Option<(String, Option<String>, Vec<String>)> = sqlx::query_as(
+            "UPDATE app_user_invitations \
+             SET accepted_at = NOW() \
+             WHERE token_hash = $1 \
+               AND app_id = $2 \
+               AND accepted_at IS NULL \
+               AND expires_at > NOW() \
+             RETURNING email, display_name, roles",
+        )
+        .bind(token_hash)
+        .bind(app_id.into_inner())
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.map(|(email, display_name, roles)| ConsumedInvitation {
+            email,
+            display_name,
+            roles,
+        }))
+    }
+
+    async fn gc(&self, batch_size: i64) -> Result<u64, AppUserInvitationRepoError> {
+        let res = sqlx::query(
+            "DELETE FROM app_user_invitations WHERE id IN ( \
+                 SELECT id FROM app_user_invitations \
+                 WHERE expires_at <= NOW() OR accepted_at IS NOT NULL \
+                 LIMIT $1 \
+                 FOR UPDATE SKIP LOCKED \
+             )",
+        )
+        .bind(batch_size)
+        .execute(&self.pool)
+        .await?;
+        Ok(res.rows_affected())
+    }
+}
+
+#[derive(sqlx::FromRow)]
+struct InvitationRecord {
+    id: uuid::Uuid,
+    app_id: uuid::Uuid,
+    email: String,
+    display_name: Option<String>,
+    roles: Vec<String>,
+    created_at: DateTime<Utc>,
+    expires_at: DateTime<Utc>,
+    accepted_at: Option<DateTime<Utc>>,
+}
+
+impl From<InvitationRecord> for InvitationRow {
+    fn from(r: InvitationRecord) -> Self {
+        Self {
+            id: r.id.into(),
+            app_id: r.app_id.into(),
+            email: r.email,
+            display_name: r.display_name,
+            roles: r.roles,
+            created_at: r.created_at,
+            expires_at: r.expires_at,
+            accepted_at: r.accepted_at,
+        }
+    }
+}

crates/manager-core/src/lib.rs

@@ -17,6 +17,7 @@ pub mod app_members_api;
 pub mod app_members_repo;
 pub mod app_repo;
 pub mod app_secrets_repo;
+pub mod app_user_invitation_repo;
 pub mod app_user_password_reset_repo;
 pub mod app_user_repo;
 pub mod app_user_session_repo;
@@ -89,6 +90,10 @@ pub use app_user_session_repo::{
     AppUserSessionLookup, AppUserSessionRepository, AppUserSessionRepositoryError,
     PostgresAppUserSessionRepository,
 };
+pub use app_user_invitation_repo::{
+    AppUserInvitationRepo, AppUserInvitationRepoError, ConsumedInvitation, InvitationRow,
+    PostgresAppUserInvitationRepo,
+};
 pub use app_user_password_reset_repo::{
     AppUserPasswordResetRepo, AppUserPasswordResetRepoError, PostgresAppUserPasswordResetRepo,
 };

crates/manager-core/src/users_service.rs

@@ -33,6 +33,7 @@ use picloud_shared::{
     UsersListOpts, UsersListPage, UsersService,
 };
 
+use crate::app_user_invitation_repo::{AppUserInvitationRepo, AppUserInvitationRepoError};
 use crate::app_user_password_reset_repo::{
     AppUserPasswordResetRepo, AppUserPasswordResetRepoError,
 };
@@ -127,6 +128,7 @@ pub struct UsersServiceImpl {
     sessions: Arc<dyn AppUserSessionRepository>,
     verifications: Arc<dyn AppUserVerificationRepo>,
     password_resets: Arc<dyn AppUserPasswordResetRepo>,
+    invitations: Arc<dyn AppUserInvitationRepo>,
     email: Arc<dyn EmailService>,
     authz: Arc<dyn AuthzRepo>,
     events: Arc<dyn ServiceEventEmitter>,
@@ -141,6 +143,7 @@ impl UsersServiceImpl {
         sessions: Arc<dyn AppUserSessionRepository>,
         verifications: Arc<dyn AppUserVerificationRepo>,
         password_resets: Arc<dyn AppUserPasswordResetRepo>,
+        invitations: Arc<dyn AppUserInvitationRepo>,
         email: Arc<dyn EmailService>,
         authz: Arc<dyn AuthzRepo>,
         events: Arc<dyn ServiceEventEmitter>,
@@ -151,6 +154,7 @@ impl UsersServiceImpl {
             sessions,
             verifications,
             password_resets,
+            invitations,
             email,
             authz,
             events,
@@ -287,6 +291,14 @@ impl From<AppUserPasswordResetRepoError> for UsersError {
     }
 }
 
+impl From<AppUserInvitationRepoError> for UsersError {
+    fn from(e: AppUserInvitationRepoError) -> Self {
+        match e {
+            AppUserInvitationRepoError::Db(err) => UsersError::Backend(err.to_string()),
+        }
+    }
+}
+
 fn map_email_error(e: EmailError) -> UsersError {
     match e {
         EmailError::NotConfigured => UsersError::EmailNotConfigured,
@@ -698,20 +710,125 @@ impl UsersService for UsersServiceImpl {
     }
     async fn invite(
         &self,
-        _cx: &SdkCallCx,
-        _email: &str,
-        _opts: InviteOpts,
+        cx: &SdkCallCx,
+        email: &str,
+        opts: InviteOpts,
     ) -> Result<(), UsersError> {
-        Err(UsersError::Backend(NOT_YET_IMPL.into()))
+        self.require(cx.principal.as_ref(), Capability::AppUsersAdmin(cx.app_id))
+            .await?;
+        let normalized = validate_email(email)?;
+        // Roles are stored verbatim; commit 8 (per-app roles) is what
+        // actually applies them on accept. v1.1.8 ships with empty
+        // role lists or whatever the script supplies — the storage
+        // round-trip is identical either way.
+        let display_name = validate_display_name(opts.display_name.as_deref())?;
+        let token = generate_session_token();
+        let expires_at = Utc::now()
+            + chrono::Duration::from_std(self.config.invitation_ttl)
+                .map_err(|e| UsersError::Backend(format!("invitation ttl: {e}")))?;
+        let row = self
+            .invitations
+            .create(
+                cx.app_id,
+                &normalized,
+                display_name.as_deref(),
+                &opts.roles,
+                &token.hash,
+                expires_at,
+            )
+            .await?;
+
+        // Send the invitation email if a template was supplied.
+        // Inviters can skip the email by omitting the template — useful
+        // when the invitation flow is driven by an external delivery
+        // (e.g., printed onboarding letter).
+        if let Some(template) = opts.template {
+            let link = build_link(&template.link_base, &token.raw);
+            let body = template.body_template.replace("{link}", &link);
+            let outbound = OutboundEmail {
+                to: vec![normalized.clone()],
+                from: template.from,
+                subject: template.subject,
+                text: Some(body),
+                ..Default::default()
+            };
+            let send_cx = internal_cx(cx);
+            if let Err(e) = self.email.send(&send_cx, outbound).await {
+                if matches!(e, EmailError::NotConfigured) {
+                    return Err(UsersError::EmailNotConfigured);
+                }
+                tracing::warn!(
+                    error = %e,
+                    invite_id = %row.id,
+                    app_id = %cx.app_id,
+                    "invite email send failed; invitation row remains valid"
+                );
+            }
+        }
+        Ok(())
     }
+
     async fn accept_invite(
         &self,
-        _cx: &SdkCallCx,
-        _token: &str,
-        _password: &str,
-        _display_name: Option<String>,
+        cx: &SdkCallCx,
+        token: &str,
+        password: &str,
+        display_name: Option<String>,
     ) -> Result<Option<GeneratedAccept>, UsersError> {
-        Err(UsersError::Backend(NOT_YET_IMPL.into()))
+        // No require: the invitation token IS the authorization.
+        validate_password(password)?;
+        let token_hash = hash_token(token);
+        let Some(consumed) = self.invitations.consume_by_token(cx.app_id, &token_hash).await? else {
+            return Ok(None);
+        };
+
+        // Display name on the new user: caller override beats the
+        // invitation's pre-staged value beats None.
+        let final_display = display_name
+            .as_deref()
+            .map(str::trim)
+            .filter(|s| !s.is_empty())
+            .map(str::to_string)
+            .or(consumed.display_name);
+
+        let password_hash = hash_password(password)
+            .map_err(|e| UsersError::Backend(format!("password hashing failed: {e}")))?;
+        let row = match self
+            .users
+            .create(
+                cx.app_id,
+                &consumed.email,
+                &password_hash,
+                final_display.as_deref(),
+            )
+            .await
+        {
+            Ok(row) => row,
+            Err(AppUserRepositoryError::DuplicateEmail(_)) => {
+                // User already exists (sign-up beat acceptance). The
+                // invitation is consumed — they'll need to log in
+                // normally with their existing password.
+                return Ok(None);
+            }
+            Err(e) => return Err(e.into()),
+        };
+        // Commit 8 wires the role-application path; in commit 7 the
+        // staged roles are stored on the invitation row but not
+        // applied (the role table doesn't exist yet). Logged so an
+        // operator can audit the gap.
+        if !consumed.roles.is_empty() {
+            tracing::info!(
+                user_id = %row.id,
+                app_id = %cx.app_id,
+                roles = ?consumed.roles,
+                "accept_invite: pre-staged roles will be applied once migration 0031 lands"
+            );
+        }
+        let user_id = row.id;
+        let user = Self::to_app_user(row, Vec::new());
+        let session = self.mint_session(cx.app_id, user_id, &user).await?;
+        self.emit(cx, "create", user_id).await;
+        Ok(Some(GeneratedAccept { user, session }))
     }
     async fn add_role(
         &self,
@@ -755,22 +872,69 @@ impl UsersService for UsersServiceImpl {
     }
     async fn list_invitations(
         &self,
-        _principal: &Principal,
-        _app_id: AppId,
+        principal: &Principal,
+        app_id: AppId,
     ) -> Result<Vec<Invitation>, UsersError> {
-        Err(UsersError::Backend(NOT_YET_IMPL.into()))
+        self.require(Some(principal), Capability::AppUsersAdmin(app_id))
+            .await?;
+        let rows = self.invitations.list_pending(app_id).await?;
+        Ok(rows
+            .into_iter()
+            .map(|r| Invitation {
+                id: r.id,
+                app_id: r.app_id,
+                email: r.email,
+                display_name: r.display_name,
+                roles: r.roles,
+                created_at: r.created_at,
+                expires_at: r.expires_at,
+            })
+            .collect())
     }
     async fn revoke_invitation(
         &self,
-        _principal: &Principal,
-        _app_id: AppId,
-        _invite_id: InvitationId,
+        principal: &Principal,
+        app_id: AppId,
+        invite_id: InvitationId,
     ) -> Result<bool, UsersError> {
-        Err(UsersError::Backend(NOT_YET_IMPL.into()))
+        self.require(Some(principal), Capability::AppUsersAdmin(app_id))
+            .await?;
+        Ok(self.invitations.revoke(app_id, invite_id).await?)
     }
 }
 
 impl UsersServiceImpl {
+    /// Mint a session for an already-authenticated user. Used by
+    /// `login` (after password verify) and `accept_invite` (after
+    /// invitation consume). Returns the raw token to embed in the
+    /// response; only the SHA-256 hash hits storage.
+    async fn mint_session(
+        &self,
+        app_id: AppId,
+        user_id: AppUserId,
+        user: &AppUser,
+    ) -> Result<GeneratedSession, UsersError> {
+        let now = Utc::now();
+        let absolute = now
+            + chrono::Duration::from_std(self.config.session_absolute_ttl)
+                .map_err(|e| UsersError::Backend(format!("absolute ttl: {e}")))?;
+        let sliding = now
+            + chrono::Duration::from_std(self.config.session_ttl)
+                .map_err(|e| UsersError::Backend(format!("sliding ttl: {e}")))?;
+        let expires_at = sliding.min(absolute);
+
+        let token = generate_session_token();
+        self.sessions
+            .create(app_id, user_id, &token.hash, expires_at, absolute)
+            .await?;
+        self.users.touch_last_login(app_id, user_id).await?;
+        Ok(GeneratedSession {
+            token: token.raw,
+            user: user.clone(),
+            expires_at,
+        })
+    }
+
     /// Shared lookup logic for `verify` and `verify_session_for_realtime`.
     /// Honors the absolute cap and cross-app isolation, then bumps the
     /// sliding window (capped at the absolute cap) on success.