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.

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);

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<AuditLog, AppError> {
+        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<Uuid>,
+        action: AuditAction,
+        resource_type: Option<&str>,
+        resource_id: Option<Uuid>,
+        details: Option<serde_json::Value>,
+        ip_address: Option<IpNetwork>,
+        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<AuditAction>,
+        limit: i64,
+    ) -> Result<Vec<AuditLog>, 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)
+    }
+}

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<GrinSlatepack, AppError> {
+        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<Option<GrinSlatepack>, 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<Option<GrinSlatepack>, 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<Vec<GrinSlatepack>, 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<GrinSlatepack, AppError> {
+        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<GrinSlatepack, AppError> {
+        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<GrinSlatepack, AppError> {
+        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<u64, AppError> {
+        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())
+    }
+}

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<Uuid>,
+        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<Vec<LoginStats>, 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<u64, AppError> {
+        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<u64, AppError> {
+        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())
+    }
+}

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),
+    }
+}

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<i64, AppError> {
+        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<i64, AppError> {
+        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<u64, AppError> {
+        let result = sqlx::query(
+            "DELETE FROM restore_attempts WHERE created_at < NOW() - INTERVAL '7 days'",
+        )
+        .execute(self.pool())
+        .await?;
+        Ok(result.rows_affected())
+    }
+}

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<Session, AppError> {
+        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<Option<Session>, 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<bool, AppError> {
+        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<u64, AppError> {
+        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<u64, AppError> {
+        let result = sqlx::query("DELETE FROM sessions WHERE expires_at < NOW()")
+            .execute(self.pool())
+            .await?;
+        Ok(result.rows_affected())
+    }
+}

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<UserKey, AppError> {
+        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<Vec<UserKey>, 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<Option<UserKey>, 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<bool, AppError> {
+        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)
+    }
+}

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),
-    }
-}