From 3caa678dea1ce62b7e1265d2073dd34cee38baaa Mon Sep 17 00:00:00 2001 From: jwinterm Date: Mon, 22 Jun 2026 14:32:17 -0400 Subject: [PATCH] feat: db query layer (sessions, keys, audit, grin relay, login, restore) Six entity query modules on Database, all explicit-column (no SELECT *), legacy methods dropped: - sessions: create + active-lookup by (already-peppered) refresh_token_hash, revoke/revoke-all (race-aware bool), touch, cleanup-expired - user_keys: upsert (ON CONFLICT user_id+asset+key_type), list, get-by-asset - audit: log_action + recent-by-user/action - grin_slatepacks: relay create/fetch/pending/respond/finalize/cancel/expire - login_events: salted IP hash, retention sweep, erasure purge (LoginStats only) - restore_attempts: peppered fingerprint + salted IP (never raw), rate-limit counts, 7-day sweep migrations/...02: login_events + restore_attempts tables (keyed-hash columns, covering indexes for the aggregations). Shared unique_violation_as + hash_ip helpers on Database. --- .../20260622000002_login_and_restore.sql | 31 ++++ src/infra/db/audit.rs | 89 ++++++++++ src/infra/db/grin_slatepacks.rs | 164 ++++++++++++++++++ src/infra/db/login_events.rs | 88 ++++++++++ src/infra/db/mod.rs | 25 ++- src/infra/db/restore_attempts.rs | 76 ++++++++ src/infra/db/sessions.rs | 103 +++++++++++ src/infra/db/user_keys.rs | 87 ++++++++++ src/infra/db/users.rs | 13 +- 9 files changed, 661 insertions(+), 15 deletions(-) create mode 100644 migrations/20260622000002_login_and_restore.sql create mode 100644 src/infra/db/audit.rs create mode 100644 src/infra/db/grin_slatepacks.rs create mode 100644 src/infra/db/login_events.rs create mode 100644 src/infra/db/restore_attempts.rs create mode 100644 src/infra/db/sessions.rs create mode 100644 src/infra/db/user_keys.rs diff --git a/migrations/20260622000002_login_and_restore.sql b/migrations/20260622000002_login_and_restore.sql new file mode 100644 index 0000000..e31da14 --- /dev/null +++ b/migrations/20260622000002_login_and_restore.sql @@ -0,0 +1,31 @@ +-- Login analytics + restore-attempt abuse tracking. +-- +-- Both tables store only keyed hashes of sensitive values: login_events.ip_hash +-- and restore_attempts.ip_hash are salted IP hashes; restore_attempts.fingerprint +-- is a peppered HMAC of the seed fingerprint. No raw IPs or reproducible +-- fingerprints are stored. + +CREATE TABLE login_events ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID REFERENCES users(id) ON DELETE SET NULL, + asset TEXT NOT NULL, -- login method/asset (btc, ..., nostr, extension) + platform TEXT NOT NULL, -- extension | web | nostr + origin TEXT, + ip_hash TEXT, -- salted hash of the client IP + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE INDEX idx_login_events_created_at ON login_events (created_at); +CREATE INDEX idx_login_events_user_id ON login_events (user_id); +-- Covering index for the (asset, platform) GROUP BY aggregations. +CREATE INDEX idx_login_events_asset_platform ON login_events (asset, platform, created_at); + +CREATE TABLE restore_attempts ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + fingerprint TEXT NOT NULL, -- peppered HMAC of the seed fingerprint + ip_hash TEXT, -- salted hash of the client IP + success BOOLEAN NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE INDEX idx_restore_attempts_fingerprint ON restore_attempts (fingerprint, created_at); +CREATE INDEX idx_restore_attempts_ip_hash ON restore_attempts (ip_hash, created_at); +CREATE INDEX idx_restore_attempts_created_at ON restore_attempts (created_at); diff --git a/src/infra/db/audit.rs b/src/infra/db/audit.rs new file mode 100644 index 0000000..3ffae85 --- /dev/null +++ b/src/infra/db/audit.rs @@ -0,0 +1,89 @@ +//! Audit log queries. +//! +//! Append-only record of security-relevant actions (logins, session lifecycle, +//! wallet registration, broadcasts). Explicit column lists (no `SELECT *`) keep +//! the row shape pinned to the `AuditLog` struct. + +use ipnetwork::IpNetwork; +use tracing::instrument; +use uuid::Uuid; + +use crate::error::AppError; +use crate::models::db::{AuditAction, AuditLog, NewAuditLog}; + +use super::Database; + +/// Explicit `audit_logs` columns (matches `AuditLog`; FromRow maps by name). +const AUDIT_LOG_COLS: &str = "id, user_id, action, resource_type, resource_id, \ + details, ip_address, user_agent, created_at"; + +impl Database { + /// Insert an audit log entry. + #[instrument(skip(self, input), fields(action = ?input.action))] + pub async fn create_audit_log(&self, input: NewAuditLog) -> Result { + let sql = format!( + "INSERT INTO audit_logs \ + (user_id, action, resource_type, resource_id, details, ip_address, user_agent) \ + VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING {AUDIT_LOG_COLS}" + ); + let log = sqlx::query_as::<_, AuditLog>(&sql) + .bind(input.user_id) + .bind(input.action) + .bind(&input.resource_type) + .bind(input.resource_id) + .bind(&input.details) + .bind(input.ip_address) + .bind(&input.user_agent) + .fetch_one(self.pool()) + .await?; + Ok(log) + } + + /// Record an action without returning the inserted row (convenience wrapper). + #[allow(clippy::too_many_arguments)] + pub async fn log_action( + &self, + user_id: Option, + action: AuditAction, + resource_type: Option<&str>, + resource_id: Option, + details: Option, + ip_address: Option, + user_agent: Option<&str>, + ) -> Result<(), AppError> { + self.create_audit_log(NewAuditLog { + user_id, + action, + resource_type: resource_type.map(String::from), + resource_id, + details, + ip_address, + user_agent: user_agent.map(String::from), + }) + .await?; + Ok(()) + } + + /// Fetch a user's most recent audit entries (newest first). When `action` is + /// `Some`, restricts to that action; otherwise returns all actions. + #[instrument(skip(self))] + pub async fn get_user_audit_logs( + &self, + user_id: Uuid, + action: Option, + limit: i64, + ) -> Result, AppError> { + let sql = format!( + "SELECT {AUDIT_LOG_COLS} FROM audit_logs \ + WHERE user_id = $1 AND ($2::audit_action IS NULL OR action = $2) \ + ORDER BY created_at DESC LIMIT $3" + ); + let logs = sqlx::query_as::<_, AuditLog>(&sql) + .bind(user_id) + .bind(action) + .bind(limit) + .fetch_all(self.pool()) + .await?; + Ok(logs) + } +} diff --git a/src/infra/db/grin_slatepacks.rs b/src/infra/db/grin_slatepacks.rs new file mode 100644 index 0000000..0cd8961 --- /dev/null +++ b/src/infra/db/grin_slatepacks.rs @@ -0,0 +1,164 @@ +//! Grin slatepack relay queries. +//! +//! Backs an interactive two-party Grin transaction by relaying slatepacks +//! between a sender and a recipient. The relay never holds key material; it +//! stores the opaque slatepack payloads and the lifecycle status. Explicit +//! column lists (no `SELECT *`) keep `FromRow` mapping stable as columns evolve. + +use tracing::instrument; +use uuid::Uuid; + +use crate::error::AppError; +use crate::models::db::{GrinSlatepack, NewGrinSlatepack, SlatepackStatus}; + +use super::Database; + +/// Explicit `grin_slatepacks` columns (matches `GrinSlatepack` field names; +/// FromRow maps by name). +const GRIN_SLATEPACK_COLS: &str = "id, slate_id, sender_user_id, recipient_user_id, \ + recipient_address, slatepack_content, amount_nanogrin, status, response_slatepack, \ + created_at, updated_at, expires_at, finalized_at, tx_hash"; + +impl Database { + /// Create a new slatepack relay entry. Starts in `PendingRecipient`: the + /// recipient must add their response before the sender can finalize. + #[instrument(skip(self, input), fields(slate_id = %input.slate_id))] + pub async fn create_grin_slatepack( + &self, + input: NewGrinSlatepack, + ) -> Result { + let sql = format!( + "INSERT INTO grin_slatepacks \ + (slate_id, sender_user_id, recipient_user_id, recipient_address, \ + slatepack_content, amount_nanogrin, expires_at) \ + VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING {GRIN_SLATEPACK_COLS}" + ); + let slatepack = sqlx::query_as::<_, GrinSlatepack>(&sql) + .bind(&input.slate_id) + .bind(input.sender_user_id) + .bind(input.recipient_user_id) + .bind(&input.recipient_address) + .bind(&input.slatepack_content) + .bind(input.amount_nanogrin) + .bind(input.expires_at) + .fetch_one(self.pool()) + .await?; + Ok(slatepack) + } + + /// Fetch a slatepack by its slate ID. + #[instrument(skip(self))] + pub async fn get_slatepack_by_slate_id( + &self, + slate_id: &str, + ) -> Result, AppError> { + let sql = format!("SELECT {GRIN_SLATEPACK_COLS} FROM grin_slatepacks WHERE slate_id = $1"); + Ok(sqlx::query_as::<_, GrinSlatepack>(&sql) + .bind(slate_id) + .fetch_optional(self.pool()) + .await?) + } + + /// Fetch a slatepack by its row ID. + #[instrument(skip(self))] + pub async fn get_slatepack_by_id(&self, id: Uuid) -> Result, AppError> { + let sql = format!("SELECT {GRIN_SLATEPACK_COLS} FROM grin_slatepacks WHERE id = $1"); + Ok(sqlx::query_as::<_, GrinSlatepack>(&sql) + .bind(id) + .fetch_optional(self.pool()) + .await?) + } + + /// List unexpired slatepacks awaiting a response from the given recipient. + #[instrument(skip(self))] + pub async fn get_pending_slatepacks_for_user( + &self, + user_id: Uuid, + ) -> Result, AppError> { + let sql = format!( + "SELECT {GRIN_SLATEPACK_COLS} FROM grin_slatepacks \ + WHERE recipient_user_id = $1 AND status = $2 AND expires_at > NOW() \ + ORDER BY created_at DESC" + ); + Ok(sqlx::query_as::<_, GrinSlatepack>(&sql) + .bind(user_id) + .bind(SlatepackStatus::PendingRecipient) + .fetch_all(self.pool()) + .await?) + } + + /// Attach the recipient's response slatepack, advancing the relay to + /// `PendingSender` so the sender can finalize. + #[instrument(skip(self, response_slatepack))] + pub async fn add_slatepack_response( + &self, + slate_id: &str, + response_slatepack: &str, + ) -> Result { + let sql = format!( + "UPDATE grin_slatepacks \ + SET response_slatepack = $2, status = $3, updated_at = NOW() \ + WHERE slate_id = $1 RETURNING {GRIN_SLATEPACK_COLS}" + ); + let slatepack = sqlx::query_as::<_, GrinSlatepack>(&sql) + .bind(slate_id) + .bind(response_slatepack) + .bind(SlatepackStatus::PendingSender) + .fetch_one(self.pool()) + .await?; + Ok(slatepack) + } + + /// Mark a slatepack as finalized once the transaction is broadcast. + #[instrument(skip(self))] + pub async fn finalize_slatepack( + &self, + slate_id: &str, + tx_hash: &str, + ) -> Result { + let sql = format!( + "UPDATE grin_slatepacks \ + SET status = $2, tx_hash = $3, finalized_at = NOW(), updated_at = NOW() \ + WHERE slate_id = $1 RETURNING {GRIN_SLATEPACK_COLS}" + ); + let slatepack = sqlx::query_as::<_, GrinSlatepack>(&sql) + .bind(slate_id) + .bind(SlatepackStatus::Finalized) + .bind(tx_hash) + .fetch_one(self.pool()) + .await?; + Ok(slatepack) + } + + /// Cancel a slatepack relay. + #[instrument(skip(self))] + pub async fn cancel_slatepack(&self, slate_id: &str) -> Result { + let sql = format!( + "UPDATE grin_slatepacks \ + SET status = $2, updated_at = NOW() \ + WHERE slate_id = $1 RETURNING {GRIN_SLATEPACK_COLS}" + ); + let slatepack = sqlx::query_as::<_, GrinSlatepack>(&sql) + .bind(slate_id) + .bind(SlatepackStatus::Cancelled) + .fetch_one(self.pool()) + .await?; + Ok(slatepack) + } + + /// Expire pending slatepacks whose `expires_at` has passed. Returns the + /// number of rows affected. + #[instrument(skip(self))] + pub async fn expire_old_slatepacks(&self) -> Result { + let result = sqlx::query( + "UPDATE grin_slatepacks SET status = $1, updated_at = NOW() \ + WHERE status IN ($2, $3) AND expires_at < NOW()", + ) + .bind(SlatepackStatus::Expired) + .bind(SlatepackStatus::PendingRecipient) + .bind(SlatepackStatus::PendingSender) + .execute(self.pool()) + .await?; + Ok(result.rows_affected()) + } +} diff --git a/src/infra/db/login_events.rs b/src/infra/db/login_events.rs new file mode 100644 index 0000000..2e051c8 --- /dev/null +++ b/src/infra/db/login_events.rs @@ -0,0 +1,88 @@ +//! Login event analytics. +//! +//! Privacy-minded: the client IP is stored only as a salted hash (via +//! [`Database::hash_ip`]), never raw. Rows carry a soft FK to the user so they +//! can be purged on erasure; a retention sweep removes old rows. + +use sqlx::FromRow; +use tracing::instrument; +use uuid::Uuid; + +use crate::error::AppError; + +use super::Database; + +/// Aggregated login stats (analytics / optional public landing). +#[derive(Debug, FromRow)] +pub struct LoginStats { + pub asset: String, + pub platform: String, + pub login_count: i64, + pub unique_users: i64, +} + +impl Database { + /// Record a login event. `ip` is the raw client IP; it is salted-hashed here. + #[instrument(skip(self, origin, ip))] + pub async fn record_login_event( + &self, + user_id: Option, + asset: &str, + platform: &str, + origin: Option<&str>, + ip: Option<&str>, + ) -> Result<(), AppError> { + let ip_hash = ip.map(|v| self.hash_ip(v)); + sqlx::query( + "INSERT INTO login_events (user_id, asset, platform, origin, ip_hash) \ + VALUES ($1, $2, $3, $4, $5)", + ) + .bind(user_id) + .bind(asset) + .bind(platform) + .bind(origin) + .bind(ip_hash) + .execute(self.pool()) + .await?; + Ok(()) + } + + /// Login counts grouped by asset + platform over the last `days`. + #[instrument(skip(self))] + pub async fn get_login_counts_recent(&self, days: i32) -> Result, AppError> { + let stats = sqlx::query_as::<_, LoginStats>( + "SELECT asset, platform, COUNT(*) AS login_count, \ + COUNT(DISTINCT user_id) AS unique_users \ + FROM login_events \ + WHERE created_at > NOW() - INTERVAL '1 day' * $1 \ + GROUP BY asset, platform \ + ORDER BY login_count DESC", + ) + .bind(days) + .fetch_all(self.pool()) + .await?; + Ok(stats) + } + + /// Erasure: purge a user's login events (default per the erasure policy). + #[instrument(skip(self))] + pub async fn delete_login_events_for_user(&self, user_id: Uuid) -> Result { + let result = sqlx::query("DELETE FROM login_events WHERE user_id = $1") + .bind(user_id) + .execute(self.pool()) + .await?; + Ok(result.rows_affected()) + } + + /// Retention sweep: delete events older than `days`. + #[instrument(skip(self))] + pub async fn cleanup_old_login_events(&self, days: i32) -> Result { + let result = sqlx::query( + "DELETE FROM login_events WHERE created_at < NOW() - INTERVAL '1 day' * $1", + ) + .bind(days) + .execute(self.pool()) + .await?; + Ok(result.rows_affected()) + } +} diff --git a/src/infra/db/mod.rs b/src/infra/db/mod.rs index 328acce..1ba0c43 100644 --- a/src/infra/db/mod.rs +++ b/src/infra/db/mod.rs @@ -6,8 +6,16 @@ //! `seed_fingerprint`) are peppered inside these methods so the plaintext never //! reaches a column and the peppering cannot drift between call sites. +mod audit; +mod grin_slatepacks; +mod login_events; +mod restore_attempts; +mod sessions; +mod user_keys; mod users; +pub use login_events::LoginStats; + use sqlx::PgPool; use crate::core::crypto::pepper::peppered_hex; @@ -36,9 +44,9 @@ impl Database { &self.pool } - /// Salt for IP hashing (used by login_events / restore_attempts). - pub fn ip_salt(&self) -> &str { - &self.ip_salt + /// Salted hash of a client IP for analytics/abuse tables (never store raw). + pub(crate) fn hash_ip(&self, ip: &str) -> String { + peppered_hex(&self.ip_salt, "ip", ip) } /// Pepper an identity value for storage/lookup (see [`peppered_hex`]). @@ -54,3 +62,14 @@ impl Database { Ok(true) } } + +/// Map a unique-constraint violation to a 409 CONFLICT with `msg`; pass other +/// errors through. Shared by the entity query modules. +pub(crate) fn unique_violation_as(msg: &'static str) -> impl Fn(sqlx::Error) -> AppError { + move |e| match &e { + sqlx::Error::Database(db_err) if db_err.is_unique_violation() => { + AppError::Conflict(msg.to_string()) + } + _ => AppError::from(e), + } +} diff --git a/src/infra/db/restore_attempts.rs b/src/infra/db/restore_attempts.rs new file mode 100644 index 0000000..1af658d --- /dev/null +++ b/src/infra/db/restore_attempts.rs @@ -0,0 +1,76 @@ +//! Restore-attempt tracking for abuse prevention. +//! +//! Both the seed fingerprint and the client IP are stored only as keyed hashes +//! (peppered fingerprint, salted IP), never raw — so this abuse-evidence table +//! is not itself a seed-existence oracle or an IP log. A retention sweep ages +//! rows out. + +use tracing::instrument; + +use crate::error::AppError; + +use super::Database; + +impl Database { + /// Record a restore attempt. `fingerprint` and `ip` are raw; both are + /// keyed-hashed here before storage. + #[instrument(skip(self, fingerprint, ip))] + pub async fn record_restore_attempt( + &self, + fingerprint: &str, + ip: Option<&str>, + success: bool, + ) -> Result<(), AppError> { + let fp = self.pepper("restore_fingerprint", fingerprint); + let ip_hash = ip.map(|v| self.hash_ip(v)); + sqlx::query( + "INSERT INTO restore_attempts (fingerprint, ip_hash, success) VALUES ($1, $2, $3)", + ) + .bind(fp) + .bind(ip_hash) + .bind(success) + .execute(self.pool()) + .await?; + Ok(()) + } + + /// Failed restore attempts for a fingerprint in the last hour. + #[instrument(skip(self, fingerprint))] + pub async fn count_failed_restore_attempts(&self, fingerprint: &str) -> Result { + let fp = self.pepper("restore_fingerprint", fingerprint); + let count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM restore_attempts \ + WHERE fingerprint = $1 AND success = FALSE \ + AND created_at > NOW() - INTERVAL '1 hour'", + ) + .bind(fp) + .fetch_one(self.pool()) + .await?; + Ok(count) + } + + /// All restore attempts from an IP in the last hour. + #[instrument(skip(self, ip))] + pub async fn count_ip_restore_attempts(&self, ip: &str) -> Result { + let ip_hash = self.hash_ip(ip); + let count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM restore_attempts \ + WHERE ip_hash = $1 AND created_at > NOW() - INTERVAL '1 hour'", + ) + .bind(ip_hash) + .fetch_one(self.pool()) + .await?; + Ok(count) + } + + /// Retention sweep: delete attempts older than 7 days. + #[instrument(skip(self))] + pub async fn cleanup_old_restore_attempts(&self) -> Result { + let result = sqlx::query( + "DELETE FROM restore_attempts WHERE created_at < NOW() - INTERVAL '7 days'", + ) + .execute(self.pool()) + .await?; + Ok(result.rows_affected()) + } +} diff --git a/src/infra/db/sessions.rs b/src/infra/db/sessions.rs new file mode 100644 index 0000000..0612cbb --- /dev/null +++ b/src/infra/db/sessions.rs @@ -0,0 +1,103 @@ +//! Session queries. +//! +//! Sessions back JWT refresh tokens. `refresh_token_hash` arrives already +//! peppered from `core::session::hash_refresh_token`, so it is stored and +//! looked up exactly as given. Active lookups filter out revoked and expired +//! rows. Explicit column lists (no `SELECT *`) keep the refresh path lean. + +use tracing::instrument; +use uuid::Uuid; + +use crate::error::AppError; +use crate::models::db::{NewSession, Session}; + +use super::Database; + +/// Explicit `sessions` columns (matches `Session` field names; FromRow maps by name). +const SESSION_COLS: &str = "id, user_id, refresh_token_hash, platform, device_info, \ + ip_address, created_at, expires_at, revoked_at, last_used_at"; + +impl Database { + /// Create a new session. `refresh_token_hash` is already peppered by the caller. + #[instrument(skip(self, input), fields(user_id = %input.user_id, platform = %input.platform))] + pub async fn create_session(&self, input: NewSession) -> Result { + let sql = format!( + "INSERT INTO sessions \ + (user_id, refresh_token_hash, platform, device_info, ip_address, expires_at) \ + VALUES ($1, $2, $3, $4, $5, $6) RETURNING {SESSION_COLS}" + ); + let session = sqlx::query_as::<_, Session>(&sql) + .bind(input.user_id) + .bind(&input.refresh_token_hash) + .bind(&input.platform) + .bind(&input.device_info) + .bind(input.ip_address) + .bind(input.expires_at) + .fetch_one(self.pool()) + .await?; + Ok(session) + } + + /// Look up an active (non-revoked, non-expired) session by refresh token hash. + /// The hash is already peppered by the caller; do not pepper it here. + #[instrument(skip(self, token_hash))] + pub async fn get_session_by_token_hash( + &self, + token_hash: &str, + ) -> Result, AppError> { + let sql = format!( + "SELECT {SESSION_COLS} FROM sessions \ + WHERE refresh_token_hash = $1 \ + AND revoked_at IS NULL \ + AND expires_at > NOW()" + ); + Ok(sqlx::query_as::<_, Session>(&sql) + .bind(token_hash) + .fetch_optional(self.pool()) + .await?) + } + + /// Update a session's `last_used_at` timestamp. + #[instrument(skip(self))] + pub async fn touch_session(&self, session_id: Uuid) -> Result<(), AppError> { + sqlx::query("UPDATE sessions SET last_used_at = NOW() WHERE id = $1") + .bind(session_id) + .execute(self.pool()) + .await?; + Ok(()) + } + + /// Revoke a session. Returns `true` if it was active (and is now revoked), + /// `false` if it was already revoked — disambiguating refresh-token races. + #[instrument(skip(self))] + pub async fn revoke_session(&self, session_id: Uuid) -> Result { + let result = sqlx::query( + "UPDATE sessions SET revoked_at = NOW() WHERE id = $1 AND revoked_at IS NULL", + ) + .bind(session_id) + .execute(self.pool()) + .await?; + Ok(result.rows_affected() > 0) + } + + /// Revoke all active sessions for a user. Returns the number revoked. + #[instrument(skip(self))] + pub async fn revoke_all_user_sessions(&self, user_id: Uuid) -> Result { + let result = sqlx::query( + "UPDATE sessions SET revoked_at = NOW() WHERE user_id = $1 AND revoked_at IS NULL", + ) + .bind(user_id) + .execute(self.pool()) + .await?; + Ok(result.rows_affected()) + } + + /// Delete expired sessions. Returns the number removed. + #[instrument(skip(self))] + pub async fn cleanup_expired_sessions(&self) -> Result { + let result = sqlx::query("DELETE FROM sessions WHERE expires_at < NOW()") + .execute(self.pool()) + .await?; + Ok(result.rows_affected()) + } +} diff --git a/src/infra/db/user_keys.rs b/src/infra/db/user_keys.rs new file mode 100644 index 0000000..2d6d68e --- /dev/null +++ b/src/infra/db/user_keys.rs @@ -0,0 +1,87 @@ +//! User key queries. +//! +//! Per-asset public keys let others send to a user. Each `(user_id, asset, +//! key_type)` is unique; upserts replace the public material in place. +//! Explicit column lists (no `SELECT *`) keep the row shape stable. + +use tracing::instrument; +use uuid::Uuid; + +use crate::error::AppError; +use crate::models::db::{AssetType, NewUserKey, UserKey}; + +use super::Database; + +/// Explicit `user_keys` columns (matches `UserKey` field names; FromRow maps by name). +const USER_KEY_COLS: &str = + "id, user_id, asset, public_key, public_spend_key, key_type, created_at, updated_at"; + +impl Database { + /// Create or update a user's public key for an asset. The UNIQUE + /// `(user_id, asset, key_type)` constraint drives the upsert. + #[instrument(skip(self, input), fields(user_id = %input.user_id, asset = %input.asset))] + pub async fn upsert_user_key(&self, input: NewUserKey) -> Result { + let sql = format!( + "INSERT INTO user_keys (user_id, asset, public_key, public_spend_key, key_type) \ + VALUES ($1, $2, $3, $4, $5) \ + ON CONFLICT (user_id, asset, key_type) DO UPDATE SET \ + public_key = EXCLUDED.public_key, \ + public_spend_key = EXCLUDED.public_spend_key, \ + updated_at = NOW() \ + RETURNING {USER_KEY_COLS}" + ); + let key = sqlx::query_as::<_, UserKey>(&sql) + .bind(input.user_id) + .bind(input.asset) + .bind(&input.public_key) + .bind(&input.public_spend_key) + .bind(&input.key_type) + .fetch_one(self.pool()) + .await?; + Ok(key) + } + + /// List all of a user's keys, ordered by asset. + #[instrument(skip(self))] + pub async fn get_user_keys(&self, user_id: Uuid) -> Result, AppError> { + let sql = + format!("SELECT {USER_KEY_COLS} FROM user_keys WHERE user_id = $1 ORDER BY asset"); + let keys = sqlx::query_as::<_, UserKey>(&sql) + .bind(user_id) + .fetch_all(self.pool()) + .await?; + Ok(keys) + } + + /// Fetch a user's primary key for an asset. + #[instrument(skip(self))] + pub async fn get_user_key( + &self, + user_id: Uuid, + asset: AssetType, + ) -> Result, AppError> { + let sql = format!( + "SELECT {USER_KEY_COLS} FROM user_keys \ + WHERE user_id = $1 AND asset = $2 AND key_type = 'primary'" + ); + let key = sqlx::query_as::<_, UserKey>(&sql) + .bind(user_id) + .bind(asset) + .fetch_optional(self.pool()) + .await?; + Ok(key) + } + + /// Delete a user's primary key for an asset. Returns `true` if a row was removed. + #[instrument(skip(self))] + pub async fn delete_user_key(&self, user_id: Uuid, asset: AssetType) -> Result { + let result = sqlx::query( + "DELETE FROM user_keys WHERE user_id = $1 AND asset = $2 AND key_type = 'primary'", + ) + .bind(user_id) + .bind(asset) + .execute(self.pool()) + .await?; + Ok(result.rows_affected() > 0) + } +} diff --git a/src/infra/db/users.rs b/src/infra/db/users.rs index 5cb61d5..e0ff9ee 100644 --- a/src/infra/db/users.rs +++ b/src/infra/db/users.rs @@ -11,7 +11,7 @@ use uuid::Uuid; use crate::error::AppError; use crate::models::db::{NewUser, User}; -use super::Database; +use super::{unique_violation_as, Database}; /// Explicit `users` columns (matches `User` field names; FromRow maps by name). const USER_COLS: &str = "id, username, pubkey_hash, nostr_pubkey, wallet_birthday, \ @@ -249,14 +249,3 @@ impl Database { ) } } - -/// Map a unique-constraint violation to a 409 CONFLICT with `msg`; pass other -/// errors through. -fn unique_violation_as(msg: &'static str) -> impl Fn(sqlx::Error) -> AppError { - move |e| match &e { - sqlx::Error::Database(db_err) if db_err.is_unique_violation() => { - AppError::Conflict(msg.to_string()) - } - _ => AppError::from(e), - } -}