feat(v1.1.8): email verification flow (migration 0028 + SDK)

migration 0028: app_user_email_verifications table — token_hash PK,
app_id + user_id FKs cascading, expires_at, consumed_at. Single-use
via atomic UPDATE WHERE consumed_at IS NULL.

UsersServiceImpl gains:
  * verifications: Arc<dyn AppUserVerificationRepo>
  * email: Arc<dyn EmailService>

users::send_verification_email(user_id, opts) mints a 32-byte token,
stores SHA-256(token), and calls EmailService::send with the body
template's {link} placeholder substituted by link_base + ?token=raw.
EmailError::NotConfigured propagates as UsersError::EmailNotConfigured
so scripts already handling email-disabled mode (v1.1.7 email::send)
don't need new branches.

users::verify_email(token) atomically consumes the one-shot token via
the verifications repo, marks the user's email_verified_at = NOW(),
and emits a "users::email_verified" event for future triggers.

Internal email send uses a synthesized SdkCallCx with principal=None
so the email-service AppEmailSend authz check is skipped (the
users::* surface has already gated on AppUsersWrite — the internal
hop isn't the script's direct call). Documented in HANDBACK §7.

EmailTemplateOpts now requires `from` (the v1.1.7 email service needs
an envelope sender). The brief example omitted it; deviation logged
in HANDBACK §7.

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

crates/manager-core/migrations/0028_app_user_email_verifications.sql

@@ -0,0 +1,19 @@
+-- v1.1.8 User Management — email verification tokens.
+--
+-- Created by `users::send_verification_email`; consumed by
+-- `users::verify_email`. Same SHA-256 token-hash shape as
+-- `app_user_sessions`. Single-use: the consume path is an atomic
+-- UPDATE WHERE consumed_at IS NULL so a replayed token returns
+-- "missing" the second time around without spurious side effects.
+
+CREATE TABLE app_user_email_verifications (
+    token_hash    TEXT PRIMARY KEY,
+    app_id        UUID NOT NULL REFERENCES apps(id) ON DELETE CASCADE,
+    user_id       UUID NOT NULL REFERENCES app_users(id) ON DELETE CASCADE,
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    expires_at    TIMESTAMPTZ NOT NULL,
+    consumed_at   TIMESTAMPTZ
+);
+
+CREATE INDEX idx_app_user_email_verifications_user
+    ON app_user_email_verifications (app_id, user_id);

crates/manager-core/src/app_user_verification_repo.rs

@@ -0,0 +1,114 @@
+//! CRUD over `app_user_email_verifications` (v1.1.8 commit 5).
+//!
+//! One-shot tokens. The consume path is an atomic UPDATE WHERE
+//! consumed_at IS NULL: rowcount = 1 means the token was valid and is
+//! now used; rowcount = 0 means missing / already-consumed / wrong
+//! app. The repo never returns the raw token — only the user id the
+//! consumed token belonged to.
+
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use picloud_shared::{AppId, AppUserId};
+use sqlx::PgPool;
+
+#[derive(Debug, thiserror::Error)]
+pub enum AppUserVerificationRepoError {
+    #[error("database error: {0}")]
+    Db(#[from] sqlx::Error),
+}
+
+#[async_trait]
+pub trait AppUserVerificationRepo: Send + Sync {
+    async fn create(
+        &self,
+        app_id: AppId,
+        user_id: AppUserId,
+        token_hash: &str,
+        expires_at: DateTime<Utc>,
+    ) -> Result<(), AppUserVerificationRepoError>;
+
+    /// Atomic consume. Returns `Some(user_id)` on success (rowcount = 1),
+    /// `None` on missing / already-consumed / expired / wrong app.
+    /// Scoped to `app_id` so a token issued for app A cannot be
+    /// presented to app B.
+    async fn consume(
+        &self,
+        app_id: AppId,
+        token_hash: &str,
+    ) -> Result<Option<AppUserId>, AppUserVerificationRepoError>;
+
+    /// GC: drop consumed or expired rows, matching the dead-letter /
+    /// session sweep pattern.
+    async fn gc(&self, batch_size: i64) -> Result<u64, AppUserVerificationRepoError>;
+}
+
+pub struct PostgresAppUserVerificationRepo {
+    pool: PgPool,
+}
+
+impl PostgresAppUserVerificationRepo {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl AppUserVerificationRepo for PostgresAppUserVerificationRepo {
+    async fn create(
+        &self,
+        app_id: AppId,
+        user_id: AppUserId,
+        token_hash: &str,
+        expires_at: DateTime<Utc>,
+    ) -> Result<(), AppUserVerificationRepoError> {
+        sqlx::query(
+            "INSERT INTO app_user_email_verifications \
+                 (token_hash, app_id, user_id, expires_at) \
+             VALUES ($1, $2, $3, $4)",
+        )
+        .bind(token_hash)
+        .bind(app_id.into_inner())
+        .bind(user_id.into_inner())
+        .bind(expires_at)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
+    async fn consume(
+        &self,
+        app_id: AppId,
+        token_hash: &str,
+    ) -> Result<Option<AppUserId>, AppUserVerificationRepoError> {
+        let row: Option<(uuid::Uuid,)> = sqlx::query_as(
+            "UPDATE app_user_email_verifications \
+             SET consumed_at = NOW() \
+             WHERE token_hash = $1 \
+               AND app_id = $2 \
+               AND consumed_at IS NULL \
+               AND expires_at > NOW() \
+             RETURNING user_id",
+        )
+        .bind(token_hash)
+        .bind(app_id.into_inner())
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.map(|(uid,)| uid.into()))
+    }
+
+    async fn gc(&self, batch_size: i64) -> Result<u64, AppUserVerificationRepoError> {
+        let res = sqlx::query(
+            "DELETE FROM app_user_email_verifications WHERE token_hash IN ( \
+                 SELECT token_hash FROM app_user_email_verifications \
+                 WHERE expires_at <= NOW() OR consumed_at IS NOT NULL \
+                 LIMIT $1 \
+                 FOR UPDATE SKIP LOCKED \
+             )",
+        )
+        .bind(batch_size)
+        .execute(&self.pool)
+        .await?;
+        Ok(res.rows_affected())
+    }
+}

crates/manager-core/src/lib.rs

@@ -19,6 +19,7 @@ pub mod app_repo;
 pub mod app_secrets_repo;
 pub mod app_user_repo;
 pub mod app_user_session_repo;
+pub mod app_user_verification_repo;
 pub mod users_service;
 pub mod apps_api;
 pub mod auth;
@@ -87,6 +88,9 @@ pub use app_user_session_repo::{
     AppUserSessionLookup, AppUserSessionRepository, AppUserSessionRepositoryError,
     PostgresAppUserSessionRepository,
 };
+pub use app_user_verification_repo::{
+    AppUserVerificationRepo, AppUserVerificationRepoError, PostgresAppUserVerificationRepo,
+};
 pub use users_service::{UsersServiceConfig, UsersServiceImpl};
 pub use admin_users_api::{admins_router, AdminsState};
 pub use api::{admin_router, AdminState};

crates/manager-core/src/users_service.rs

@@ -27,15 +27,17 @@ use std::time::Duration;
 use async_trait::async_trait;
 use chrono::Utc;
 use picloud_shared::{
-    AppId, AppUser, AppUserId, CreateUserInput, EmailTemplateOpts, GeneratedAccept,
-    GeneratedSession, InviteOpts, Invitation, InvitationId, Principal, SdkCallCx, ServiceEvent,
-    ServiceEventEmitter, UpdateUserInput, UsersError, UsersListOpts, UsersListPage, UsersService,
+    AppId, AppUser, AppUserId, CreateUserInput, EmailError, EmailService, EmailTemplateOpts,
+    GeneratedAccept, GeneratedSession, InviteOpts, Invitation, InvitationId, OutboundEmail,
+    Principal, SdkCallCx, ServiceEvent, ServiceEventEmitter, UpdateUserInput, UsersError,
+    UsersListOpts, UsersListPage, UsersService,
 };
 
 use crate::app_user_repo::{
     AppUserRepository, AppUserRepositoryError, AppUserRow, ListOpts as RepoListOpts,
 };
 use crate::app_user_session_repo::{AppUserSessionRepository, AppUserSessionRepositoryError};
+use crate::app_user_verification_repo::{AppUserVerificationRepo, AppUserVerificationRepoError};
 use crate::auth::{
     self, generate_session_token, hash_password, hash_token, verify_password, TIMING_FLAT_DUMMY_HASH,
 };
@@ -120,6 +122,8 @@ fn env_duration_days(var: &str, default: Duration) -> Duration {
 pub struct UsersServiceImpl {
     users: Arc<dyn AppUserRepository>,
     sessions: Arc<dyn AppUserSessionRepository>,
+    verifications: Arc<dyn AppUserVerificationRepo>,
+    email: Arc<dyn EmailService>,
     authz: Arc<dyn AuthzRepo>,
     events: Arc<dyn ServiceEventEmitter>,
     config: UsersServiceConfig,
@@ -127,9 +131,12 @@ pub struct UsersServiceImpl {
 
 impl UsersServiceImpl {
     #[must_use]
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         users: Arc<dyn AppUserRepository>,
         sessions: Arc<dyn AppUserSessionRepository>,
+        verifications: Arc<dyn AppUserVerificationRepo>,
+        email: Arc<dyn EmailService>,
         authz: Arc<dyn AuthzRepo>,
         events: Arc<dyn ServiceEventEmitter>,
         config: UsersServiceConfig,
@@ -137,6 +144,8 @@ impl UsersServiceImpl {
         Self {
             users,
             sessions,
+            verifications,
+            email,
             authz,
             events,
             config,
@@ -256,6 +265,54 @@ impl From<AppUserSessionRepositoryError> for UsersError {
     }
 }
 
+impl From<AppUserVerificationRepoError> for UsersError {
+    fn from(e: AppUserVerificationRepoError) -> Self {
+        match e {
+            AppUserVerificationRepoError::Db(err) => UsersError::Backend(err.to_string()),
+        }
+    }
+}
+
+fn map_email_error(e: EmailError) -> UsersError {
+    match e {
+        EmailError::NotConfigured => UsersError::EmailNotConfigured,
+        EmailError::Forbidden => UsersError::Forbidden,
+        EmailError::MissingField(_)
+        | EmailError::InvalidAddress(_)
+        | EmailError::TooLarge { .. } => UsersError::EmailTransport(e.to_string()),
+        EmailError::Transport(s) => UsersError::EmailTransport(s),
+    }
+}
+
+/// Build the token-bearing URL: `link_base?token=<raw>` (or `&token=...`
+/// if the base already has a query string). The raw token never lives
+/// in storage; it appears only here, in the email body, and in the
+/// user's inbox.
+fn build_link(link_base: &str, raw_token: &str) -> String {
+    let sep = if link_base.contains('?') { '&' } else { '?' };
+    format!("{link_base}{sep}token={raw_token}")
+}
+
+/// Synthesize an `SdkCallCx` with `principal: None` for the internal
+/// email send. The users::* surface has already done its own authz
+/// check (AppUsersWrite / AppUsersAdmin); the underlying email send is
+/// a system-level action, not the user's direct invocation. Mirrors
+/// how downstream services would internally hop without re-running
+/// the caller's authz.
+fn internal_cx(cx: &SdkCallCx) -> SdkCallCx {
+    SdkCallCx {
+        app_id: cx.app_id,
+        principal: None,
+        script_id: cx.script_id,
+        execution_id: cx.execution_id,
+        request_id: cx.request_id,
+        trigger_depth: cx.trigger_depth,
+        root_execution_id: cx.root_execution_id,
+        is_dead_letter_handler: cx.is_dead_letter_handler,
+        event: cx.event.clone(),
+    }
+}
+
 #[async_trait]
 impl UsersService for UsersServiceImpl {
     async fn create(
@@ -479,18 +536,64 @@ impl UsersService for UsersServiceImpl {
 
     async fn send_verification_email(
         &self,
-        _cx: &SdkCallCx,
-        _user_id: AppUserId,
-        _opts: EmailTemplateOpts,
+        cx: &SdkCallCx,
+        user_id: AppUserId,
+        opts: EmailTemplateOpts,
     ) -> Result<(), UsersError> {
-        Err(UsersError::Backend(NOT_YET_IMPL.into()))
+        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
+            .await?;
+        let user = self
+            .users
+            .get(cx.app_id, user_id)
+            .await?
+            .ok_or(UsersError::NotFound)?;
+
+        // Mint a single-use token. The hash hits the verifications
+        // table; the raw token only ever appears in the email body.
+        let token = generate_session_token();
+        let expires_at = Utc::now()
+            + chrono::Duration::from_std(self.config.verification_ttl)
+                .map_err(|e| UsersError::Backend(format!("verification ttl: {e}")))?;
+        self.verifications
+            .create(cx.app_id, user.id, &token.hash, expires_at)
+            .await?;
+
+        let link = build_link(&opts.link_base, &token.raw);
+        let body = opts.body_template.replace("{link}", &link);
+        let outbound = OutboundEmail {
+            to: vec![user.email.clone()],
+            from: opts.from,
+            subject: opts.subject,
+            text: Some(body),
+            ..Default::default()
+        };
+        let send_cx = internal_cx(cx);
+        self.email
+            .send(&send_cx, outbound)
+            .await
+            .map_err(map_email_error)?;
+        Ok(())
     }
+
     async fn verify_email(
         &self,
-        _cx: &SdkCallCx,
-        _token: &str,
+        cx: &SdkCallCx,
+        token: &str,
     ) -> Result<Option<AppUser>, UsersError> {
-        Err(UsersError::Backend(NOT_YET_IMPL.into()))
+        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
+            .await?;
+        let token_hash = hash_token(token);
+        let Some(user_id) = self.verifications.consume(cx.app_id, &token_hash).await? else {
+            return Ok(None);
+        };
+        let row = self
+            .users
+            .mark_email_verified(cx.app_id, user_id)
+            .await?;
+        let roles = self.fetch_roles(cx.app_id, row.id).await?;
+        let user = Self::to_app_user(row, roles);
+        self.emit(cx, "email_verified", user.id).await;
+        Ok(Some(user))
     }
     async fn request_password_reset(
         &self,