kvitt/server/crates/kvitt-server/tests/common/mod.rs

//! The integration-test harness shared by every test binary of this crate (included with `#[path]`, so it is not a
//! test target itself): a migrated temporary database, a fixed clock, recording fakes of `AuditWriter` and
//! `PushQueue`, a seeded household with the ids of `contracts/examples/`, one router per app (so the rate limiter
//! keeps its state across requests) and helpers to compare responses with the contract examples.

use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

use axum::Router;
use axum::body::Body;
use axum::extract::ConnectInfo;
use axum::http::{HeaderMap, Method, Request, StatusCode};
use http_body_util::BodyExt;
use kvitt_server::audit::{AuditEvent, AuditWriter, BoxFuture};
use kvitt_server::config::Config;
use kvitt_server::push::{PushMessage, PushQueue};
use kvitt_server::time::{FixedClock, parse_ts};
use kvitt_server::{AppState, auth, db, router};
use serde_json::{Value, json};
use sqlx::SqliteConnection;
use tower::ServiceExt;
use uuid::Uuid;

pub const ADMIN_ID: &str = "0192a000-0000-7000-8000-000000000001";
pub const MEMBER_ID: &str = "0192a000-0000-7000-8000-000000000002";
pub const ADMIN_DEVICE_ID: &str = "0192a001-0000-7000-8000-000000000001";
pub const MEMBER_DEVICE_ID: &str = "0192a001-0000-7000-8000-000000000002";
pub const ADMIN_OLD_DEVICE_ID: &str = "0192a001-0000-7000-8000-000000000003";
pub const ACCOUNT_1: &str = "0192a003-0000-7000-8000-000000000001";
pub const ACCOUNT_2: &str = "0192a003-0000-7000-8000-000000000002";
pub const GROCERIES: &str = "01920000-0000-7000-8000-000000000001";
pub const HOME: &str = "01920000-0000-7000-8000-000000000003";
pub const UNKNOWN_ID: &str = "0192ffff-0000-7000-8000-000000000001";
/// Saturday 17 October 2026, 14:00 in Stockholm (CEST).
pub const NOW: &str = "2026-10-17T12:00:00.000Z";
/// The default client address of test requests.
pub const CLIENT: &str = "192.0.2.10:50000";

pub fn token(tag: char) -> String {
    format!("kvt_{}", tag.to_string().repeat(43))
}
pub fn admin_token() -> String {
    token('A')
}
pub fn member_token() -> String {
    token('M')
}
pub fn old_device_token() -> String {
    token('R')
}

/// Records every audit event (`sync_fixtures::app` installs the real writer instead).
#[derive(Default)]
pub struct RecordingAudit(pub Mutex<Vec<AuditEvent>>);

impl AuditWriter for RecordingAudit {
    fn write<'a>(
        &'a self,
        _conn: &'a mut SqliteConnection,
        event: AuditEvent,
    ) -> BoxFuture<'a, Result<Uuid, kvitt_server::error::AppError>> {
        self.0.lock().unwrap().push(event);
        Box::pin(async { Ok(Uuid::now_v7()) })
    }
}

/// Records every enqueued push.
#[derive(Default)]
pub struct RecordingPush(pub Mutex<Vec<PushMessage>>);

impl PushQueue for RecordingPush {
    fn enqueue(&self, message: PushMessage) {
        self.0.lock().unwrap().push(message);
    }
}

/// A data directory removed on drop.
pub struct TempDir(pub PathBuf);

impl TempDir {
    pub fn new() -> Self {
        let dir = std::env::temp_dir().join(format!("kvitt-core-{}", Uuid::now_v7()));
        std::fs::create_dir_all(&dir).unwrap();
        Self(dir)
    }
}

impl Default for TempDir {
    fn default() -> Self {
        Self::new()
    }
}

impl Drop for TempDir {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.0);
    }
}

pub struct TestApp {
    pub dir: TempDir,
    pub state: AppState,
    pub clock: FixedClock,
    pub audit: Arc<RecordingAudit>,
    pub push: Arc<RecordingPush>,
    pub router: Router,
}

pub struct Reply {
    pub status: StatusCode,
    pub headers: HeaderMap,
    pub bytes: Vec<u8>,
    pub json: Value,
}

impl Reply {
    pub fn code(&self) -> &str {
        self.json["code"].as_str().unwrap_or("")
    }

    pub fn text(&self) -> String {
        String::from_utf8_lossy(&self.bytes).into_owned()
    }
}

/// Asserts the error shape of SPEC §14.3.2 with the given status and code.
#[track_caller]
pub fn assert_error(reply: &Reply, status: u16, code: &str) {
    assert_eq!(reply.status.as_u16(), status, "{}", reply.text());
    assert_eq!(reply.code(), code, "{}", reply.text());
    assert!(reply.json["message"].is_string());
    assert!(reply.json.as_object().unwrap().contains_key("field"));
}

/// Like [`assert_error`] and also checks `field`.
#[track_caller]
pub fn assert_field_error(reply: &Reply, status: u16, code: &str, field: &str) {
    assert_error(reply, status, code);
    assert_eq!(reply.json["field"], json!(field), "{}", reply.text());
}

impl TestApp {
    /// A migrated, empty database (no users) at a new temporary directory.
    pub async fn empty() -> Self {
        Self::at(TempDir::new()).await
    }

    /// A migrated database in `dir` (which may already hold one).
    pub async fn at(dir: TempDir) -> Self {
        let path = dir.0.to_string_lossy().into_owned();
        let config = Config::from_lookup(|k| match k {
            "KVITT_DATA_DIR" => Some(path.clone()),
            // No proxy is trusted unless a test says so.
            "KVITT_TRUSTED_PROXY" => Some("127.0.0.9".to_owned()),
            _ => None,
        })
        .unwrap();
        let pool = db::connect(&config.database_path()).await.unwrap();
        db::migrate(&pool).await.unwrap();
        let clock = FixedClock::new(parse_ts(NOW).unwrap());
        let audit = Arc::new(RecordingAudit::default());
        let push = Arc::new(RecordingPush::default());
        let mut state = AppState::new(config, pool).with_clock(Arc::new(clock.clone()));
        state.audit = audit.clone();
        state.push = push.clone();
        let router = router::router(state.clone());
        Self {
            dir,
            state,
            clock,
            audit,
            push,
            router,
        }
    }

    /// Only the admin (Jacob) with one device: the household is incomplete.
    pub async fn admin_only() -> Self {
        let app = Self::empty().await;
        app.insert_user(ADMIN_ID, "admin", "Jacob", 15).await;
        app.insert_device(
            ADMIN_DEVICE_ID,
            ADMIN_ID,
            "Galaxy S26",
            &admin_token(),
            None,
        )
        .await;
        app
    }

    /// Admin Jacob and member Anna, one device each, plus a revoked old device of the admin.
    pub async fn household() -> Self {
        let app = Self::admin_only().await;
        app.insert_user(MEMBER_ID, "member", "Anna", 16).await;
        app.insert_device(
            MEMBER_DEVICE_ID,
            MEMBER_ID,
            "Galaxy S26+",
            &member_token(),
            None,
        )
        .await;
        app.insert_device(
            ADMIN_OLD_DEVICE_ID,
            ADMIN_ID,
            "Old phone",
            &old_device_token(),
            Some("2026-10-10T00:00:00.000Z"),
        )
        .await;
        app
    }

    pub async fn insert_user(&self, id: &str, role: &str, name: &str, version: i64) {
        let created = if role == "admin" {
            "2026-10-01T00:00:00.000Z"
        } else {
            "2026-10-02T00:00:00.000Z"
        };
        sqlx::query(
            "INSERT INTO users (id, role, display_name, picture_version, created_at, updated_at, server_version) \
             VALUES (?, ?, ?, NULL, ?, ?, ?)",
        )
        .bind(id)
        .bind(role)
        .bind(name)
        .bind(created)
        .bind(created)
        .bind(version)
        .execute(&self.state.db)
        .await
        .unwrap();
        sqlx::query("UPDATE sync_sequence SET value = MAX(value, ?)")
            .bind(version)
            .execute(&self.state.db)
            .await
            .unwrap();
    }

    pub async fn insert_device(
        &self,
        id: &str,
        user: &str,
        name: &str,
        tok: &str,
        revoked: Option<&str>,
    ) {
        let created = match id {
            ADMIN_DEVICE_ID => "2026-10-01T00:00:00.000Z",
            ADMIN_OLD_DEVICE_ID => "2026-10-01T01:00:00.000Z",
            _ => "2026-10-02T00:00:00.000Z",
        };
        sqlx::query(
            "INSERT INTO devices (id, user_id, name, token_hash, created_at, last_seen_at, revoked_at) \
             VALUES (?, ?, ?, ?, ?, ?, ?)",
        )
        .bind(id)
        .bind(user)
        .bind(name)
        .bind(auth::sha256_hex(tok))
        .bind(created)
        .bind(created)
        .bind(revoked)
        .execute(&self.state.db)
        .await
        .unwrap();
    }

    /// An account inserted directly (fixed id, so responses can be compared with the examples).
    pub async fn insert_account(&self, id: &str, name: &str, created_at: &str, bp_admin: i64) {
        let version = self.next_version().await;
        sqlx::query(
            "INSERT INTO accounts (id, name, name_key, created_by_user_id, created_at, updated_at, server_version) \
             VALUES (?, ?, ?, ?, ?, ?, ?)",
        )
        .bind(id)
        .bind(name)
        .bind(name.to_lowercase())
        .bind(ADMIN_ID)
        .bind(created_at)
        .bind(created_at)
        .bind(version)
        .execute(&self.state.db)
        .await
        .unwrap();
        for (user, bp) in [(ADMIN_ID, bp_admin), (MEMBER_ID, 10_000 - bp_admin)] {
            sqlx::query(
                "INSERT INTO account_default_shares (account_id, user_id, share_bp) VALUES (?, ?, ?)",
            )
            .bind(id)
            .bind(user)
            .bind(bp)
            .execute(&self.state.db)
            .await
            .unwrap();
        }
    }

    /// An expense inserted directly (not through sync): `shares` are (admin öre, member öre).
    pub async fn insert_expense(
        &self,
        account: &str,
        category: &str,
        amount: i64,
        payer: &str,
        shares: (i64, i64),
        deleted: bool,
    ) -> String {
        let id = Uuid::now_v7().to_string();
        let version = self.next_version().await;
        sqlx::query(
            "INSERT INTO expenses (id, account_id, category_id, amount_ore, description, notes, expense_date, \
             payer_user_id, split_mode, created_by_user_id, updated_by_user_id, created_at, updated_at, deleted_at, \
             server_received_at, server_version) \
             VALUES (?, ?, ?, ?, 'x', '', '2026-10-16', ?, 'exact', ?, ?, ?, ?, ?, ?, ?)",
        )
        .bind(&id)
        .bind(account)
        .bind(category)
        .bind(amount)
        .bind(payer)
        .bind(payer)
        .bind(payer)
        .bind(NOW)
        .bind(NOW)
        .bind(deleted.then_some(NOW))
        .bind(NOW)
        .bind(version)
        .execute(&self.state.db)
        .await
        .unwrap();
        for (user, ore) in [(ADMIN_ID, shares.0), (MEMBER_ID, shares.1)] {
            sqlx::query(
                "INSERT INTO expense_splits (expense_id, user_id, share_ore, share_bp) VALUES (?, ?, ?, NULL)",
            )
            .bind(&id)
            .bind(user)
            .bind(ore)
            .execute(&self.state.db)
            .await
            .unwrap();
        }
        id
    }

    /// A settlement inserted directly, from `from` to the other user, with allocations.
    pub async fn insert_settlement(
        &self,
        from: &str,
        allocations: &[(&str, i64)],
        deleted: bool,
    ) -> String {
        let id = Uuid::now_v7().to_string();
        let to = if from == ADMIN_ID {
            MEMBER_ID
        } else {
            ADMIN_ID
        };
        let amount: i64 = allocations.iter().map(|(_, a)| a).sum();
        let version = self.next_version().await;
        sqlx::query(
            "INSERT INTO settlements (id, from_user_id, to_user_id, amount_ore, settlement_date, note, \
             created_by_user_id, updated_by_user_id, created_at, updated_at, deleted_at, server_received_at, \
             server_version) VALUES (?, ?, ?, ?, '2026-10-16', '', ?, ?, ?, ?, ?, ?, ?)",
        )
        .bind(&id)
        .bind(from)
        .bind(to)
        .bind(amount)
        .bind(from)
        .bind(from)
        .bind(NOW)
        .bind(NOW)
        .bind(deleted.then_some(NOW))
        .bind(NOW)
        .bind(version)
        .execute(&self.state.db)
        .await
        .unwrap();
        for (account, a) in allocations {
            sqlx::query(
                "INSERT INTO settlement_allocations (settlement_id, account_id, amount_ore) VALUES (?, ?, ?)",
            )
            .bind(&id)
            .bind(account)
            .bind(a)
            .execute(&self.state.db)
            .await
            .unwrap();
        }
        id
    }

    pub async fn next_version(&self) -> i64 {
        sqlx::query_scalar(
            "UPDATE sync_sequence SET value = value + 1 WHERE id = 1 RETURNING value",
        )
        .fetch_one(&self.state.db)
        .await
        .unwrap()
    }

    pub async fn scalar_i64(&self, sql: &'static str) -> i64 {
        sqlx::query_scalar(sql)
            .fetch_one(&self.state.db)
            .await
            .unwrap()
    }

    pub fn set_now(&self, ts: &str) {
        self.clock.set(parse_ts(ts).unwrap());
    }

    pub fn data_dir(&self) -> &Path {
        &self.dir.0
    }

    pub async fn send(&self, request: Request<Body>) -> Reply {
        let mut request = request;
        if request
            .extensions()
            .get::<ConnectInfo<SocketAddr>>()
            .is_none()
        {
            request
                .extensions_mut()
                .insert(ConnectInfo(CLIENT.parse::<SocketAddr>().unwrap()));
        }
        let response = self.router.clone().oneshot(request).await.unwrap();
        let status = response.status();
        let headers = response.headers().clone();
        let bytes = response
            .into_body()
            .collect()
            .await
            .unwrap()
            .to_bytes()
            .to_vec();
        let json = serde_json::from_slice(&bytes).unwrap_or(Value::Null);
        Reply {
            status,
            headers,
            bytes,
            json,
        }
    }

    /// An authenticated JSON request (`body` `None` sends no body). The future borrows only the app.
    pub fn call<'a>(
        &'a self,
        method: Method,
        path: &str,
        tok: &str,
        body: Option<Value>,
    ) -> impl std::future::Future<Output = Reply> + use<'a> {
        self.send(req(method, path, Some(tok), body))
    }

    /// An unauthenticated JSON request.
    pub fn public<'a>(
        &'a self,
        method: Method,
        path: &str,
        body: Option<Value>,
    ) -> impl std::future::Future<Output = Reply> + use<'a> {
        self.send(req(method, path, None, body))
    }
}

/// A request with the app's usual headers (`X-Kvitt-Protocol: 1`, JSON content type when there is a body).
pub fn req(method: Method, path: &str, tok: Option<&str>, body: Option<Value>) -> Request<Body> {
    let mut builder = Request::builder()
        .method(method)
        .uri(path)
        .header("x-kvitt-protocol", "1");
    if let Some(tok) = tok {
        builder = builder.header("authorization", format!("Bearer {tok}"));
    }
    match body {
        Some(body) => builder
            .header("content-type", "application/json; charset=utf-8")
            .body(Body::from(body.to_string()))
            .unwrap(),
        None => builder.body(Body::empty()).unwrap(),
    }
}

/// A raw request with a body and content type.
pub fn raw(
    method: Method,
    path: &str,
    tok: &str,
    content_type: Option<&str>,
    body: Vec<u8>,
) -> Request<Body> {
    let mut builder = Request::builder()
        .method(method)
        .uri(path)
        .header("x-kvitt-protocol", "1")
        .header("authorization", format!("Bearer {tok}"));
    if let Some(ct) = content_type {
        builder = builder.header("content-type", ct);
    }
    builder.body(Body::from(body)).unwrap()
}

/// `contracts/examples/<name>` parsed.
pub fn example(name: &str) -> Value {
    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("../../../contracts/examples")
        .join(name);
    serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap()
}

/// Compares `actual` with an example by JSON value. Keys in `volatile` (ids, timestamps, versions, codes, which the
/// examples cannot fix) are compared by JSON type only; everything else, the key sets included, must be equal.
#[track_caller]
pub fn assert_like_example(actual: &Value, name: &str, volatile: &[&str]) {
    fn walk(a: &Value, e: &Value, path: &str, volatile: &[&str]) -> Result<(), String> {
        match (a, e) {
            (Value::Object(a), Value::Object(e)) => {
                let ak: Vec<_> = a.keys().collect();
                let ek: Vec<_> = e.keys().collect();
                if ak != ek {
                    return Err(format!("{path}: keys {ak:?} != {ek:?}"));
                }
                for (k, ev) in e {
                    let av = &a[k];
                    let p = format!("{path}.{k}");
                    if volatile.contains(&k.as_str()) {
                        let same_kind = std::mem::discriminant(av) == std::mem::discriminant(ev);
                        if !same_kind {
                            return Err(format!("{p}: type of {av} differs from {ev}"));
                        }
                    } else {
                        walk(av, ev, &p, volatile)?;
                    }
                }
                Ok(())
            }
            (Value::Array(a), Value::Array(e)) => {
                if a.len() != e.len() {
                    return Err(format!("{path}: length {} != {}", a.len(), e.len()));
                }
                for (i, (av, ev)) in a.iter().zip(e).enumerate() {
                    walk(av, ev, &format!("{path}[{i}]"), volatile)?;
                }
                Ok(())
            }
            _ if a == e => Ok(()),
            _ => Err(format!("{path}: {a} != {e}")),
        }
    }
    if let Err(diff) = walk(actual, &example(name), "$", volatile) {
        panic!("{name}: {diff}\nactual: {actual:#}");
    }
}

/// The timestamp and version keys every row example has.
pub const ROW_VOLATILE: &[&str] = &["id", "created_at", "updated_at", "server_version"];