From 80cee7d74205c43b06cf0762fcbb81ae829a1cda Mon Sep 17 00:00:00 2001 From: "ilblackdragon@gmail.com" Date: Tue, 24 Mar 2026 13:47:38 -0700 Subject: [PATCH] feat(db): add UserStore trait with users, api_tokens, invitations tables Foundation for DB-backed user management (#1605): - UserRecord, ApiTokenRecord, InvitationRecord types in db/mod.rs - UserStore sub-trait (17 methods) added to Database supertrait - PostgreSQL migration V14__users.sql (users, api_tokens, invitations) - libSQL schema + incremental migration V14 - Full implementations for both PgBackend (via Store delegation) and LibSqlBackend (direct SQL in libsql/users.rs) - authenticate_token JOINs api_tokens+users with active/non-revoked checks; has_any_users for bootstrap detection Co-Authored-By: Claude Opus 4.6 (1M context) --- migrations/V14__users.sql | 42 +++ src/db/libsql/mod.rs | 1 + src/db/libsql/users.rs | 500 ++++++++++++++++++++++++++++++++++++ src/db/libsql_migrations.rs | 83 ++++++ src/db/mod.rs | 122 +++++++++ src/db/postgres.rs | 106 +++++++- src/history/store.rs | 396 ++++++++++++++++++++++++++++ 7 files changed, 1248 insertions(+), 2 deletions(-) create mode 100644 migrations/V14__users.sql create mode 100644 src/db/libsql/users.rs diff --git a/migrations/V14__users.sql b/migrations/V14__users.sql new file mode 100644 index 00000000..1dbf284e --- /dev/null +++ b/migrations/V14__users.sql @@ -0,0 +1,42 @@ +-- User management tables for multi-tenant deployments. +-- +-- Replaces the static GATEWAY_USER_TOKENS env var with DB-backed +-- user registration, API token management, and invitation flow. + +CREATE TABLE users ( + id TEXT PRIMARY KEY, -- matches existing user_id pattern (string, not UUID) + email TEXT UNIQUE, -- nullable for token-only users + display_name TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', -- active | suspended | deactivated + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_login_at TIMESTAMPTZ, + created_by TEXT REFERENCES users(id), -- who invited this user (nullable for bootstrap) + metadata JSONB NOT NULL DEFAULT '{}' -- extensible profile data +); + +CREATE TABLE api_tokens ( + id UUID PRIMARY KEY, + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + token_hash BYTEA NOT NULL, -- SHA-256 hash (never store plaintext) + token_prefix TEXT NOT NULL, -- first 8 hex chars for display + name TEXT NOT NULL, -- human label ("my-laptop", "ci-bot") + expires_at TIMESTAMPTZ, -- nullable = never expires + last_used_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + revoked_at TIMESTAMPTZ -- soft-revoke: set this instead of deleting +); +CREATE INDEX idx_api_tokens_user ON api_tokens(user_id); +CREATE INDEX idx_api_tokens_hash ON api_tokens(token_hash); + +CREATE TABLE invitations ( + id UUID PRIMARY KEY, + email TEXT, -- nullable for link-based invites + invite_token_hash BYTEA NOT NULL, -- SHA-256 hash of the invite link token + invited_by TEXT NOT NULL REFERENCES users(id), + status TEXT NOT NULL DEFAULT 'pending', -- pending | accepted | expired | revoked + expires_at TIMESTAMPTZ NOT NULL, + accepted_at TIMESTAMPTZ, + accepted_by TEXT REFERENCES users(id), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); diff --git a/src/db/libsql/mod.rs b/src/db/libsql/mod.rs index 890aea0c..f3a323f3 100644 --- a/src/db/libsql/mod.rs +++ b/src/db/libsql/mod.rs @@ -12,6 +12,7 @@ mod routines; mod sandbox; mod settings; mod tool_failures; +mod users; mod workspace; use std::path::Path; diff --git a/src/db/libsql/users.rs b/src/db/libsql/users.rs new file mode 100644 index 00000000..715eb300 --- /dev/null +++ b/src/db/libsql/users.rs @@ -0,0 +1,500 @@ +//! UserStore implementation for LibSqlBackend. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use libsql::params; +use uuid::Uuid; + +use super::{fmt_opt_ts, fmt_ts, get_opt_text, get_opt_ts, get_text, get_ts, opt_text}; +use crate::db::libsql::LibSqlBackend; +use crate::db::{ApiTokenRecord, DatabaseError, InvitationRecord, UserRecord, UserStore}; + +fn row_to_user(row: &libsql::Row) -> Result { + let metadata_str = get_text(row, 8); + let metadata: serde_json::Value = serde_json::from_str(&metadata_str) + .map_err(|e| DatabaseError::Serialization(e.to_string()))?; + Ok(UserRecord { + id: get_text(row, 0), + email: get_opt_text(row, 1), + display_name: get_text(row, 2), + status: get_text(row, 3), + created_at: get_ts(row, 4), + updated_at: get_ts(row, 5), + last_login_at: get_opt_ts(row, 6), + created_by: get_opt_text(row, 7), + metadata, + }) +} + +fn row_to_api_token(row: &libsql::Row) -> ApiTokenRecord { + ApiTokenRecord { + id: get_text(row, 0).parse().unwrap_or_default(), + user_id: get_text(row, 1), + name: get_text(row, 2), + token_prefix: get_text(row, 3), + expires_at: get_opt_ts(row, 4), + last_used_at: get_opt_ts(row, 5), + created_at: get_ts(row, 6), + revoked_at: get_opt_ts(row, 7), + } +} + +fn row_to_invitation(row: &libsql::Row) -> InvitationRecord { + InvitationRecord { + id: get_text(row, 0).parse().unwrap_or_default(), + email: get_opt_text(row, 1), + invited_by: get_text(row, 2), + status: get_text(row, 3), + expires_at: get_ts(row, 4), + accepted_at: get_opt_ts(row, 5), + accepted_by: get_opt_text(row, 6), + created_at: get_ts(row, 7), + } +} + +#[async_trait] +impl UserStore for LibSqlBackend { + async fn create_user(&self, user: &UserRecord) -> Result<(), DatabaseError> { + let conn = self.connect().await?; + let metadata_json = serde_json::to_string(&user.metadata) + .map_err(|e| DatabaseError::Serialization(e.to_string()))?; + + conn.execute( + r#" + INSERT INTO users (id, email, display_name, status, created_at, updated_at, last_login_at, created_by, metadata) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) + "#, + params![ + user.id.as_str(), + opt_text(user.email.as_deref()), + user.display_name.as_str(), + user.status.as_str(), + fmt_ts(&user.created_at), + fmt_ts(&user.updated_at), + fmt_opt_ts(&user.last_login_at), + opt_text(user.created_by.as_deref()), + metadata_json, + ], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(()) + } + + async fn get_user(&self, id: &str) -> Result, DatabaseError> { + let conn = self.connect().await?; + let mut rows = conn + .query( + r#" + SELECT id, email, display_name, status, created_at, updated_at, + last_login_at, created_by, metadata + FROM users WHERE id = ?1 + "#, + params![id], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + Some(row) => Ok(Some(row_to_user(&row)?)), + None => Ok(None), + } + } + + async fn get_user_by_email(&self, email: &str) -> Result, DatabaseError> { + let conn = self.connect().await?; + let mut rows = conn + .query( + r#" + SELECT id, email, display_name, status, created_at, updated_at, + last_login_at, created_by, metadata + FROM users WHERE email = ?1 + "#, + params![email], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + Some(row) => Ok(Some(row_to_user(&row)?)), + None => Ok(None), + } + } + + async fn list_users(&self, status: Option<&str>) -> Result, DatabaseError> { + let conn = self.connect().await?; + let mut users = Vec::new(); + + let mut rows = if let Some(status) = status { + conn.query( + r#" + SELECT id, email, display_name, status, created_at, updated_at, + last_login_at, created_by, metadata + FROM users WHERE status = ?1 + ORDER BY created_at DESC + "#, + params![status], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + } else { + conn.query( + r#" + SELECT id, email, display_name, status, created_at, updated_at, + last_login_at, created_by, metadata + FROM users + ORDER BY created_at DESC + "#, + (), + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + }; + + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + users.push(row_to_user(&row)?); + } + Ok(users) + } + + async fn update_user_status(&self, id: &str, status: &str) -> Result<(), DatabaseError> { + let conn = self.connect().await?; + let now = fmt_ts(&Utc::now()); + conn.execute( + "UPDATE users SET status = ?2, updated_at = ?3 WHERE id = ?1", + params![id, status, now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(()) + } + + async fn update_user_profile( + &self, + id: &str, + display_name: &str, + metadata: &serde_json::Value, + ) -> Result<(), DatabaseError> { + let conn = self.connect().await?; + let now = fmt_ts(&Utc::now()); + let metadata_json = serde_json::to_string(metadata) + .map_err(|e| DatabaseError::Serialization(e.to_string()))?; + conn.execute( + "UPDATE users SET display_name = ?2, metadata = ?3, updated_at = ?4 WHERE id = ?1", + params![id, display_name, metadata_json, now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(()) + } + + async fn record_login(&self, id: &str) -> Result<(), DatabaseError> { + let conn = self.connect().await?; + let now = fmt_ts(&Utc::now()); + conn.execute( + "UPDATE users SET last_login_at = ?2, updated_at = ?2 WHERE id = ?1", + params![id, now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(()) + } + + async fn create_api_token( + &self, + user_id: &str, + name: &str, + token_hash: &[u8; 32], + token_prefix: &str, + expires_at: Option>, + ) -> Result { + let conn = self.connect().await?; + let id = Uuid::new_v4(); + let now = Utc::now(); + + conn.execute( + r#" + INSERT INTO api_tokens (id, user_id, token_hash, token_prefix, name, expires_at, created_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) + "#, + params![ + id.to_string(), + user_id, + libsql::Value::Blob(token_hash.to_vec()), + token_prefix, + name, + fmt_opt_ts(&expires_at), + fmt_ts(&now), + ], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + Ok(ApiTokenRecord { + id, + user_id: user_id.to_string(), + name: name.to_string(), + token_prefix: token_prefix.to_string(), + expires_at, + last_used_at: None, + created_at: now, + revoked_at: None, + }) + } + + async fn list_api_tokens(&self, user_id: &str) -> Result, DatabaseError> { + let conn = self.connect().await?; + let mut rows = conn + .query( + r#" + SELECT id, user_id, name, token_prefix, expires_at, last_used_at, created_at, revoked_at + FROM api_tokens WHERE user_id = ?1 + ORDER BY created_at DESC + "#, + params![user_id], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + let mut tokens = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + tokens.push(row_to_api_token(&row)); + } + Ok(tokens) + } + + async fn revoke_api_token(&self, token_id: Uuid, user_id: &str) -> Result { + let conn = self.connect().await?; + let now = fmt_ts(&Utc::now()); + let rows_affected = conn + .execute( + r#" + UPDATE api_tokens SET revoked_at = ?3 + WHERE id = ?1 AND user_id = ?2 AND revoked_at IS NULL + "#, + params![token_id.to_string(), user_id, now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(rows_affected > 0) + } + + async fn authenticate_token( + &self, + token_hash: &[u8; 32], + ) -> Result, DatabaseError> { + let conn = self.connect().await?; + let now = fmt_ts(&Utc::now()); + + let mut rows = conn + .query( + r#" + SELECT + t.id, t.user_id, t.name, t.token_prefix, t.expires_at, + t.last_used_at, t.created_at, t.revoked_at, + u.id, u.email, u.display_name, u.status, u.created_at, + u.updated_at, u.last_login_at, u.created_by, u.metadata + FROM api_tokens t + JOIN users u ON u.id = t.user_id + WHERE t.token_hash = ?1 + AND t.revoked_at IS NULL + AND (t.expires_at IS NULL OR t.expires_at > ?2) + AND u.status = 'active' + "#, + params![libsql::Value::Blob(token_hash.to_vec()), now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + Some(row) => { + let token = ApiTokenRecord { + id: get_text(&row, 0).parse().unwrap_or_default(), + user_id: get_text(&row, 1), + name: get_text(&row, 2), + token_prefix: get_text(&row, 3), + expires_at: get_opt_ts(&row, 4), + last_used_at: get_opt_ts(&row, 5), + created_at: get_ts(&row, 6), + revoked_at: get_opt_ts(&row, 7), + }; + + let metadata_str = get_text(&row, 16); + let metadata: serde_json::Value = serde_json::from_str(&metadata_str) + .map_err(|e| DatabaseError::Serialization(e.to_string()))?; + + let user = UserRecord { + id: get_text(&row, 8), + email: get_opt_text(&row, 9), + display_name: get_text(&row, 10), + status: get_text(&row, 11), + created_at: get_ts(&row, 12), + updated_at: get_ts(&row, 13), + last_login_at: get_opt_ts(&row, 14), + created_by: get_opt_text(&row, 15), + metadata, + }; + + Ok(Some((token, user))) + } + None => Ok(None), + } + } + + async fn record_token_usage(&self, token_id: Uuid) -> Result<(), DatabaseError> { + let conn = self.connect().await?; + let now = fmt_ts(&Utc::now()); + conn.execute( + "UPDATE api_tokens SET last_used_at = ?2 WHERE id = ?1", + params![token_id.to_string(), now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(()) + } + + async fn create_invitation( + &self, + invitation: &InvitationRecord, + invite_hash: &[u8; 32], + ) -> Result<(), DatabaseError> { + let conn = self.connect().await?; + conn.execute( + r#" + INSERT INTO invitations (id, email, invite_token_hash, invited_by, status, expires_at, accepted_at, accepted_by, created_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) + "#, + params![ + invitation.id.to_string(), + opt_text(invitation.email.as_deref()), + libsql::Value::Blob(invite_hash.to_vec()), + invitation.invited_by.as_str(), + invitation.status.as_str(), + fmt_ts(&invitation.expires_at), + fmt_opt_ts(&invitation.accepted_at), + opt_text(invitation.accepted_by.as_deref()), + fmt_ts(&invitation.created_at), + ], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(()) + } + + async fn get_invitation_by_hash( + &self, + invite_hash: &[u8; 32], + ) -> Result, DatabaseError> { + let conn = self.connect().await?; + let mut rows = conn + .query( + r#" + SELECT id, email, invited_by, status, expires_at, accepted_at, accepted_by, created_at + FROM invitations WHERE invite_token_hash = ?1 + "#, + params![libsql::Value::Blob(invite_hash.to_vec())], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + Some(row) => Ok(Some(row_to_invitation(&row))), + None => Ok(None), + } + } + + async fn accept_invitation(&self, id: Uuid, accepted_by: &str) -> Result<(), DatabaseError> { + let conn = self.connect().await?; + let now = fmt_ts(&Utc::now()); + conn.execute( + r#" + UPDATE invitations SET status = 'accepted', accepted_at = ?2, accepted_by = ?3 + WHERE id = ?1 + "#, + params![id.to_string(), now, accepted_by], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(()) + } + + async fn list_invitations( + &self, + invited_by: Option<&str>, + ) -> Result, DatabaseError> { + let conn = self.connect().await?; + let mut invitations = Vec::new(); + + let mut rows = if let Some(invited_by) = invited_by { + conn.query( + r#" + SELECT id, email, invited_by, status, expires_at, accepted_at, accepted_by, created_at + FROM invitations WHERE invited_by = ?1 + ORDER BY created_at DESC + "#, + params![invited_by], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + } else { + conn.query( + r#" + SELECT id, email, invited_by, status, expires_at, accepted_at, accepted_by, created_at + FROM invitations + ORDER BY created_at DESC + "#, + (), + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + }; + + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + invitations.push(row_to_invitation(&row)); + } + Ok(invitations) + } + + async fn has_any_users(&self) -> Result { + let conn = self.connect().await?; + let mut rows = conn + .query("SELECT 1 FROM users LIMIT 1", ()) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + let has_users = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + .is_some(); + Ok(has_users) + } +} diff --git a/src/db/libsql_migrations.rs b/src/db/libsql_migrations.rs index d0ec20ef..af614494 100644 --- a/src/db/libsql_migrations.rs +++ b/src/db/libsql_migrations.rs @@ -579,6 +579,46 @@ INSERT OR IGNORE INTO leak_detection_patterns (id, name, pattern, severity, acti ('550e8400-e29b-41d4-a716-446655440011', 'mailchimp_api_key', '[a-f0-9]{32}-us[0-9]{1,2}', 'medium', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), ('550e8400-e29b-41d4-a716-446655440012', 'high_entropy_hex', '(? Result<(), Databas Ok(()) } +// ==================== User management record types ==================== + +/// A registered user. +#[derive(Debug, Clone)] +pub struct UserRecord { + /// User identifier (string, matches existing `user_id` throughout the codebase). + pub id: String, + pub email: Option, + pub display_name: String, + /// `active`, `suspended`, or `deactivated`. + pub status: String, + pub created_at: DateTime, + pub updated_at: DateTime, + pub last_login_at: Option>, + /// Who created/invited this user (nullable for bootstrap users). + pub created_by: Option, + pub metadata: serde_json::Value, +} + +/// An API token for authenticating requests (hash stored, never plaintext). +#[derive(Debug, Clone)] +pub struct ApiTokenRecord { + pub id: Uuid, + pub user_id: String, + /// Human label (e.g. "my-laptop", "ci-bot"). + pub name: String, + /// First 8 hex chars of the plaintext token for display/identification. + pub token_prefix: String, + pub expires_at: Option>, + pub last_used_at: Option>, + pub created_at: DateTime, + /// Soft-revoke timestamp. Non-null means revoked. + pub revoked_at: Option>, +} + +/// A pending invitation to create an account. +#[derive(Debug, Clone)] +pub struct InvitationRecord { + pub id: Uuid, + pub email: Option, + pub invited_by: String, + /// `pending`, `accepted`, `expired`, or `revoked`. + pub status: String, + pub expires_at: DateTime, + pub accepted_at: Option>, + pub accepted_by: Option, + pub created_at: DateTime, +} + // ==================== Sub-traits ==================== // // Each sub-trait groups related persistence methods. The `Database` supertrait @@ -761,6 +810,78 @@ pub trait WorkspaceStore: Send + Sync { } } +#[async_trait] +pub trait UserStore: Send + Sync { + // ---- Users ---- + + /// Create a new user record. + async fn create_user(&self, user: &UserRecord) -> Result<(), DatabaseError>; + /// Get a user by their string id. + async fn get_user(&self, id: &str) -> Result, DatabaseError>; + /// Get a user by email address. + async fn get_user_by_email(&self, email: &str) -> Result, DatabaseError>; + /// List users, optionally filtered by status. + async fn list_users(&self, status: Option<&str>) -> Result, DatabaseError>; + /// Update a user's status (active/suspended/deactivated). + async fn update_user_status(&self, id: &str, status: &str) -> Result<(), DatabaseError>; + /// Update a user's display name and metadata. + async fn update_user_profile( + &self, + id: &str, + display_name: &str, + metadata: &serde_json::Value, + ) -> Result<(), DatabaseError>; + /// Record a login timestamp. + async fn record_login(&self, id: &str) -> Result<(), DatabaseError>; + + // ---- API Tokens ---- + + /// Create a new API token. The `token_hash` is SHA-256 of the plaintext. + async fn create_api_token( + &self, + user_id: &str, + name: &str, + token_hash: &[u8; 32], + token_prefix: &str, + expires_at: Option>, + ) -> Result; + /// List tokens for a user (never includes the hash). + async fn list_api_tokens(&self, user_id: &str) -> Result, DatabaseError>; + /// Soft-revoke a token. Returns false if the token doesn't exist or doesn't belong to the user. + async fn revoke_api_token(&self, token_id: Uuid, user_id: &str) -> Result; + /// Look up a token by hash, returning the token record and its owning user. + /// Only returns active (non-revoked, non-expired) tokens for active users. + async fn authenticate_token( + &self, + token_hash: &[u8; 32], + ) -> Result, DatabaseError>; + /// Update `last_used_at` for a token. + async fn record_token_usage(&self, token_id: Uuid) -> Result<(), DatabaseError>; + + // ---- Invitations ---- + + /// Create a new invitation. + async fn create_invitation( + &self, + invitation: &InvitationRecord, + invite_hash: &[u8; 32], + ) -> Result<(), DatabaseError>; + /// Look up a pending invitation by its hashed token. + async fn get_invitation_by_hash( + &self, + invite_hash: &[u8; 32], + ) -> Result, DatabaseError>; + /// Accept an invitation (sets status=accepted, accepted_by, accepted_at). + async fn accept_invitation(&self, id: Uuid, accepted_by: &str) -> Result<(), DatabaseError>; + /// List invitations, optionally filtered by inviter. + async fn list_invitations( + &self, + invited_by: Option<&str>, + ) -> Result, DatabaseError>; + /// Check whether any user records exist (for first-run bootstrap detection). + async fn has_any_users(&self) -> Result; +} + /// Backend-agnostic database supertrait. /// /// Combines all sub-traits into one. Existing `Arc` consumers @@ -774,6 +895,7 @@ pub trait Database: + ToolFailureStore + SettingsStore + WorkspaceStore + + UserStore + Send + Sync { diff --git a/src/db/postgres.rs b/src/db/postgres.rs index 9e5ea9ce..35059ef5 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -16,8 +16,8 @@ use crate::agent::routine::{Routine, RoutineRun, RunStatus}; use crate::config::DatabaseConfig; use crate::context::{ActionRecord, JobContext, JobState}; use crate::db::{ - ConversationStore, Database, JobStore, RoutineStore, SandboxStore, SettingsStore, - ToolFailureStore, WorkspaceStore, + ApiTokenRecord, ConversationStore, Database, InvitationRecord, JobStore, RoutineStore, + SandboxStore, SettingsStore, ToolFailureStore, UserRecord, UserStore, WorkspaceStore, }; use crate::error::{DatabaseError, WorkspaceError}; use crate::history::{ @@ -786,3 +786,105 @@ impl WorkspaceStore for PgBackend { .await } } + +// ==================== UserStore ==================== + +#[async_trait] +impl UserStore for PgBackend { + async fn create_user(&self, user: &UserRecord) -> Result<(), DatabaseError> { + self.store.create_user(user).await + } + + async fn get_user(&self, id: &str) -> Result, DatabaseError> { + self.store.get_user(id).await + } + + async fn get_user_by_email(&self, email: &str) -> Result, DatabaseError> { + self.store.get_user_by_email(email).await + } + + async fn list_users(&self, status: Option<&str>) -> Result, DatabaseError> { + self.store.list_users(status).await + } + + async fn update_user_status(&self, id: &str, status: &str) -> Result<(), DatabaseError> { + self.store.update_user_status(id, status).await + } + + async fn update_user_profile( + &self, + id: &str, + display_name: &str, + metadata: &serde_json::Value, + ) -> Result<(), DatabaseError> { + self.store + .update_user_profile(id, display_name, metadata) + .await + } + + async fn record_login(&self, id: &str) -> Result<(), DatabaseError> { + self.store.record_login(id).await + } + + async fn create_api_token( + &self, + user_id: &str, + name: &str, + token_hash: &[u8; 32], + token_prefix: &str, + expires_at: Option>, + ) -> Result { + self.store + .create_api_token(user_id, name, token_hash, token_prefix, expires_at) + .await + } + + async fn list_api_tokens(&self, user_id: &str) -> Result, DatabaseError> { + self.store.list_api_tokens(user_id).await + } + + async fn revoke_api_token(&self, token_id: Uuid, user_id: &str) -> Result { + self.store.revoke_api_token(token_id, user_id).await + } + + async fn authenticate_token( + &self, + token_hash: &[u8; 32], + ) -> Result, DatabaseError> { + self.store.authenticate_token(token_hash).await + } + + async fn record_token_usage(&self, token_id: Uuid) -> Result<(), DatabaseError> { + self.store.record_token_usage(token_id).await + } + + async fn create_invitation( + &self, + invitation: &InvitationRecord, + invite_hash: &[u8; 32], + ) -> Result<(), DatabaseError> { + self.store.create_invitation(invitation, invite_hash).await + } + + async fn get_invitation_by_hash( + &self, + invite_hash: &[u8; 32], + ) -> Result, DatabaseError> { + self.store.get_invitation_by_hash(invite_hash).await + } + + async fn accept_invitation(&self, id: Uuid, accepted_by: &str) -> Result<(), DatabaseError> { + self.store.accept_invitation(id, accepted_by).await + } + + async fn list_invitations( + &self, + invited_by: Option<&str>, + ) -> Result, DatabaseError> { + self.store.list_invitations(invited_by).await + } + + async fn has_any_users(&self) -> Result { + self.store.has_any_users().await + } +} diff --git a/src/history/store.rs b/src/history/store.rs index 625e8b1e..81398ddf 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -2279,6 +2279,402 @@ impl Store { } } +// ==================== Users / API Tokens / Invitations ==================== + +#[cfg(feature = "postgres")] +use crate::db::{ApiTokenRecord, InvitationRecord, UserRecord}; + +#[cfg(feature = "postgres")] +impl Store { + /// Create a new user record. + pub async fn create_user(&self, user: &UserRecord) -> Result<(), DatabaseError> { + let conn = self.conn().await?; + conn.execute( + r#" + INSERT INTO users (id, email, display_name, status, created_at, updated_at, last_login_at, created_by, metadata) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + "#, + &[ + &user.id, + &user.email, + &user.display_name, + &user.status, + &user.created_at, + &user.updated_at, + &user.last_login_at, + &user.created_by, + &user.metadata, + ], + ) + .await?; + Ok(()) + } + + /// Get a user by their string id. + pub async fn get_user(&self, id: &str) -> Result, DatabaseError> { + let conn = self.conn().await?; + let row = conn + .query_opt("SELECT * FROM users WHERE id = $1", &[&id]) + .await?; + Ok(row.map(|r| row_to_user(&r))) + } + + /// Get a user by email address. + pub async fn get_user_by_email( + &self, + email: &str, + ) -> Result, DatabaseError> { + let conn = self.conn().await?; + let row = conn + .query_opt("SELECT * FROM users WHERE email = $1", &[&email]) + .await?; + Ok(row.map(|r| row_to_user(&r))) + } + + /// List users, optionally filtered by status. + pub async fn list_users(&self, status: Option<&str>) -> Result, DatabaseError> { + let conn = self.conn().await?; + let rows = match status { + Some(s) => { + conn.query( + "SELECT * FROM users WHERE status = $1 ORDER BY created_at", + &[&s], + ) + .await? + } + None => { + conn.query("SELECT * FROM users ORDER BY created_at", &[]) + .await? + } + }; + Ok(rows.iter().map(row_to_user).collect()) + } + + /// Update a user's status. + pub async fn update_user_status(&self, id: &str, status: &str) -> Result<(), DatabaseError> { + let conn = self.conn().await?; + conn.execute( + "UPDATE users SET status = $1, updated_at = NOW() WHERE id = $2", + &[&status, &id], + ) + .await?; + Ok(()) + } + + /// Update a user's display name and metadata. + pub async fn update_user_profile( + &self, + id: &str, + display_name: &str, + metadata: &serde_json::Value, + ) -> Result<(), DatabaseError> { + let conn = self.conn().await?; + conn.execute( + "UPDATE users SET display_name = $1, metadata = $2, updated_at = NOW() WHERE id = $3", + &[&display_name, metadata, &id], + ) + .await?; + Ok(()) + } + + /// Record a login timestamp for a user. + pub async fn record_login(&self, id: &str) -> Result<(), DatabaseError> { + let conn = self.conn().await?; + conn.execute( + "UPDATE users SET last_login_at = NOW(), updated_at = NOW() WHERE id = $1", + &[&id], + ) + .await?; + Ok(()) + } + + /// Create a new API token. + pub async fn create_api_token( + &self, + user_id: &str, + name: &str, + token_hash: &[u8; 32], + token_prefix: &str, + expires_at: Option>, + ) -> Result { + let conn = self.conn().await?; + let id = Uuid::new_v4(); + let now = Utc::now(); + conn.execute( + r#" + INSERT INTO api_tokens (id, user_id, token_hash, token_prefix, name, expires_at, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7) + "#, + &[ + &id, + &user_id, + &token_hash.to_vec(), + &token_prefix, + &name, + &expires_at, + &now, + ], + ) + .await?; + Ok(ApiTokenRecord { + id, + user_id: user_id.to_string(), + name: name.to_string(), + token_prefix: token_prefix.to_string(), + expires_at, + last_used_at: None, + created_at: now, + revoked_at: None, + }) + } + + /// List tokens for a user. + pub async fn list_api_tokens( + &self, + user_id: &str, + ) -> Result, DatabaseError> { + let conn = self.conn().await?; + let rows = conn + .query( + r#" + SELECT id, user_id, name, token_prefix, expires_at, last_used_at, created_at, revoked_at + FROM api_tokens + WHERE user_id = $1 + ORDER BY created_at DESC + "#, + &[&user_id], + ) + .await?; + Ok(rows.iter().map(row_to_api_token).collect()) + } + + /// Soft-revoke a token. Returns false if the token doesn't exist or doesn't belong to the user. + pub async fn revoke_api_token( + &self, + token_id: Uuid, + user_id: &str, + ) -> Result { + let conn = self.conn().await?; + let count = conn + .execute( + "UPDATE api_tokens SET revoked_at = NOW() WHERE id = $1 AND user_id = $2 AND revoked_at IS NULL", + &[&token_id, &user_id], + ) + .await?; + Ok(count > 0) + } + + /// Authenticate a token by hash. Returns the token record and its owning user + /// if the token is active (non-revoked, non-expired) and the user is active. + pub async fn authenticate_token( + &self, + token_hash: &[u8; 32], + ) -> Result, DatabaseError> { + let conn = self.conn().await?; + let row = conn + .query_opt( + r#" + SELECT t.id, t.user_id, t.name, t.token_prefix, t.expires_at, t.last_used_at, t.created_at, t.revoked_at, + u.id as u_id, u.email, u.display_name, u.status, u.created_at as u_created_at, u.updated_at, u.last_login_at, u.created_by, u.metadata + FROM api_tokens t + JOIN users u ON t.user_id = u.id + WHERE t.token_hash = $1 + AND t.revoked_at IS NULL + AND (t.expires_at IS NULL OR t.expires_at > NOW()) + AND u.status = 'active' + "#, + &[&token_hash.to_vec()], + ) + .await?; + Ok(row.map(|r| { + let token = ApiTokenRecord { + id: r.get("id"), + user_id: r.get("user_id"), + name: r.get("name"), + token_prefix: r.get("token_prefix"), + expires_at: r.get("expires_at"), + last_used_at: r.get("last_used_at"), + created_at: r.get("created_at"), + revoked_at: r.get("revoked_at"), + }; + let user = UserRecord { + id: r.get("u_id"), + email: r.get("email"), + display_name: r.get("display_name"), + status: r.get("status"), + created_at: r.get("u_created_at"), + updated_at: r.get("updated_at"), + last_login_at: r.get("last_login_at"), + created_by: r.get("created_by"), + metadata: r.get("metadata"), + }; + (token, user) + })) + } + + /// Update `last_used_at` for a token. + pub async fn record_token_usage(&self, token_id: Uuid) -> Result<(), DatabaseError> { + let conn = self.conn().await?; + conn.execute( + "UPDATE api_tokens SET last_used_at = NOW() WHERE id = $1", + &[&token_id], + ) + .await?; + Ok(()) + } + + /// Create a new invitation. + pub async fn create_invitation( + &self, + invitation: &InvitationRecord, + invite_hash: &[u8; 32], + ) -> Result<(), DatabaseError> { + let conn = self.conn().await?; + conn.execute( + r#" + INSERT INTO invitations (id, email, invite_token_hash, invited_by, status, expires_at, accepted_at, accepted_by, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + "#, + &[ + &invitation.id, + &invitation.email, + &invite_hash.to_vec(), + &invitation.invited_by, + &invitation.status, + &invitation.expires_at, + &invitation.accepted_at, + &invitation.accepted_by, + &invitation.created_at, + ], + ) + .await?; + Ok(()) + } + + /// Look up a pending invitation by its hashed token. + pub async fn get_invitation_by_hash( + &self, + invite_hash: &[u8; 32], + ) -> Result, DatabaseError> { + let conn = self.conn().await?; + let row = conn + .query_opt( + r#" + SELECT id, email, invited_by, status, expires_at, accepted_at, accepted_by, created_at + FROM invitations + WHERE invite_token_hash = $1 AND status = 'pending' AND expires_at > NOW() + "#, + &[&invite_hash.to_vec()], + ) + .await?; + Ok(row.map(|r| row_to_invitation(&r))) + } + + /// Accept an invitation. + pub async fn accept_invitation( + &self, + id: Uuid, + accepted_by: &str, + ) -> Result<(), DatabaseError> { + let conn = self.conn().await?; + conn.execute( + "UPDATE invitations SET status = 'accepted', accepted_at = NOW(), accepted_by = $1 WHERE id = $2", + &[&accepted_by, &id], + ) + .await?; + Ok(()) + } + + /// List invitations, optionally filtered by inviter. + pub async fn list_invitations( + &self, + invited_by: Option<&str>, + ) -> Result, DatabaseError> { + let conn = self.conn().await?; + let rows = match invited_by { + Some(user) => { + conn.query( + r#" + SELECT id, email, invited_by, status, expires_at, accepted_at, accepted_by, created_at + FROM invitations + WHERE invited_by = $1 + ORDER BY created_at DESC + "#, + &[&user], + ) + .await? + } + None => { + conn.query( + r#" + SELECT id, email, invited_by, status, expires_at, accepted_at, accepted_by, created_at + FROM invitations + ORDER BY created_at DESC + "#, + &[], + ) + .await? + } + }; + Ok(rows.iter().map(row_to_invitation).collect()) + } + + /// Check whether any user records exist. + pub async fn has_any_users(&self) -> Result { + let conn = self.conn().await?; + let row = conn + .query_one( + "SELECT EXISTS(SELECT 1 FROM users LIMIT 1) as has_users", + &[], + ) + .await?; + Ok(row.get("has_users")) + } +} + +#[cfg(feature = "postgres")] +fn row_to_user(row: &tokio_postgres::Row) -> UserRecord { + UserRecord { + id: row.get("id"), + email: row.get("email"), + display_name: row.get("display_name"), + status: row.get("status"), + created_at: row.get("created_at"), + updated_at: row.get("updated_at"), + last_login_at: row.get("last_login_at"), + created_by: row.get("created_by"), + metadata: row.get("metadata"), + } +} + +#[cfg(feature = "postgres")] +fn row_to_api_token(row: &tokio_postgres::Row) -> ApiTokenRecord { + ApiTokenRecord { + id: row.get("id"), + user_id: row.get("user_id"), + name: row.get("name"), + token_prefix: row.get("token_prefix"), + expires_at: row.get("expires_at"), + last_used_at: row.get("last_used_at"), + created_at: row.get("created_at"), + revoked_at: row.get("revoked_at"), + } +} + +#[cfg(feature = "postgres")] +fn row_to_invitation(row: &tokio_postgres::Row) -> InvitationRecord { + InvitationRecord { + id: row.get("id"), + email: row.get("email"), + invited_by: row.get("invited_by"), + status: row.get("status"), + expires_at: row.get("expires_at"), + accepted_at: row.get("accepted_at"), + accepted_by: row.get("accepted_by"), + created_at: row.get("created_at"), + } +} + #[cfg(test)] mod tests { use super::*;