Server gap pass: invite-partner subcommand, CLI and drain tests, push info log (D-113, D-114)

- kvitt-server invite-partner: a partner invite by the admin as E-12 does (cancels the pending one, 72 h, link with
  ?from=<admin>), exit 2 household full, 3 no admin; README lines in server/ and infra/.
- Tests: invite-partner, migrate (schema at SCHEMA_VERSION, idempotent), healthcheck (200, 503, nothing listening),
  serve SIGTERM drain (in-flight request finishes, new connections refused, exit 0; gives up after SHUTDOWN_DRAIN).
- Push sender: one info line "push sent" per message with push_type, devices_sent, devices_targeted and 8-character
  device-id tails (SPEC §8.7 aggregates); per-device line stays at debug.
- tools/contracts-check.py: stale TODO(spec) replaced by the §9.11 CS-5 citation; shapes updated for invite_kind
  (D-086), app_version_name (D-090) and fid (D-096), 134 checks pass.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TQd3pEP9UeN2uifbHknQ7B
This commit is contained in:
2026-10-18 22:12:21 +02:00
parent e70fe0d9fe
commit 8660b7fa16
9 changed files with 528 additions and 18 deletions
@@ -0,0 +1,93 @@
//! `kvitt-server invite-partner` (D-113): what E-12 `POST /invites` with `kind: partner` does for the admin, run on the
//! server host, so a partner can be invited when the admin has no phone at hand to create the invite in the app.
//!
//! Cancels any pending `partner` invite and creates a new one by the admin (72 hours, same code format), then prints
//! its code, the landing link naming the admin (`?from=`, D-087) and the expiry, like `bootstrap`. The household
//! already has two users: `household full`, exit 2. No admin yet: `no admin, run bootstrap first`, exit 3. Nothing
//! is created in either case. E-12 writes no audit entry (the audit log holds expenses and settlements only, §14.1),
//! so neither does this. The code goes to stdout only, never to the log (SPEC §8.7).
use std::process::ExitCode;
use chrono::{DateTime, Utc};
use crate::api::invites::{self, MAX_USERS, NewInvite};
use crate::api::rows;
use crate::config::Config;
use crate::db::{self, Db};
use crate::error::AppError;
use crate::wire::{InviteKind, Role};
/// Exit code when the household already has two users (D-113).
pub const EXIT_HOUSEHOLD_FULL: u8 = 2;
/// Exit code when no admin exists yet (D-113).
pub const EXIT_NO_ADMIN: u8 = 3;
/// What `invite-partner` did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Outcome {
/// The new invite and the admin's display name, for the link's `from`.
Created {
invite: NewInvite,
admin_name: String,
},
HouseholdFull,
NoAdmin,
}
/// Creates a `partner` invite by the admin, as E-12 does.
pub async fn invite_partner(db: &Db, now: DateTime<Utc>) -> Result<Outcome, AppError> {
let mut tx = db::begin_immediate(db).await?;
let users = rows::household(&mut tx).await?;
let Some(admin) = users.iter().find(|u| u.role == Role::Admin) else {
return Ok(Outcome::NoAdmin);
};
if users.len() >= MAX_USERS {
return Ok(Outcome::HouseholdFull);
}
let (admin_id, admin_name) = (admin.id, admin.display_name.clone());
invites::cancel_pending(&mut tx, InviteKind::Partner, now).await?;
let invite =
invites::insert_invite(&mut tx, InviteKind::Partner, None, Some(admin_id), now).await?;
tx.commit().await?;
Ok(Outcome::Created { invite, admin_name })
}
/// The printed text: as [`super::bootstrap::render`], with the admin named in the link (D-087).
pub fn render(invite: &NewInvite, public_url: &str, admin_name: &str) -> String {
format!(
"Code: {}\nLink: {}\nExpires: {}\n",
invites::display_code(&invite.code),
invites::invite_link(public_url, &invite.code, Some(admin_name)),
invite.expires_at
)
}
/// Opens the database and runs [`invite_partner`] with the system clock.
pub async fn run(config: &Config) -> ExitCode {
let result = async {
let pool = super::open_database(config).await?;
let outcome = invite_partner(&pool, Utc::now()).await;
pool.close().await;
outcome
}
.await;
match result {
Ok(Outcome::Created { invite, admin_name }) => {
print!("{}", render(&invite, &config.public_url, &admin_name));
ExitCode::SUCCESS
}
Ok(Outcome::HouseholdFull) => {
eprintln!("household full");
ExitCode::from(EXIT_HOUSEHOLD_FULL)
}
Ok(Outcome::NoAdmin) => {
eprintln!("no admin, run bootstrap first");
ExitCode::from(EXIT_NO_ADMIN)
}
Err(error) => {
eprintln!("invite-partner: {error}");
ExitCode::FAILURE
}
}
}
+5 -1
View File
@@ -1,8 +1,9 @@
//! `kvitt-server` subcommands (SPEC §10.4, exhaustive), all implemented: `serve`, `migrate` and `healthcheck` here,
//! `bootstrap` and `pair-device` (M3) and `rotate-epoch` (M4) in their own files.
//! `bootstrap` and `pair-device` (M3), `rotate-epoch` (M4) and `invite-partner` (D-113) in their own files.
pub mod bootstrap;
pub mod healthcheck;
pub mod invite_partner;
pub mod pair_device;
pub mod rotate_epoch;
@@ -47,6 +48,8 @@ pub enum Command {
#[arg(long = "user", value_name = "USER_ID")]
user: String,
},
/// Create a partner invite by the admin, as E-12 does (D-113). Exit 0 ok, 2 household full, 3 no admin.
InvitePartner,
/// Set a new random sync_epoch (SPEC §10.9).
RotateEpoch,
/// GET /health on this host; exit 0 if 200, 1 otherwise.
@@ -63,6 +66,7 @@ pub async fn run(command: Command, config: Config) -> ExitCode {
Command::Healthcheck => return healthcheck::run(&config).await,
Command::Bootstrap => return bootstrap::run(&config).await,
Command::PairDevice { user } => return pair_device::run(&config, &user).await,
Command::InvitePartner => return invite_partner::run(&config).await,
Command::RotateEpoch => return rotate_epoch::run(&config).await,
};
match result {
+52 -9
View File
@@ -46,6 +46,19 @@ pub enum Delivery {
GaveUp { attempts: usize, detail: String },
}
/// Characters of a device id in the `push sent` info line.
pub const SHORT_DEVICE_ID_LEN: usize = 8;
/// The last [`SHORT_DEVICE_ID_LEN`] characters of a device id, enough to tell the household's devices apart in the
/// info log. The tail, not the head: device ids are UUIDv7, whose leading characters are the creation time and are
/// shared by devices paired within about a minute of each other.
pub fn short_device_id(device_id: &str) -> String {
let chars: Vec<char> = device_id.chars().collect();
chars[chars.len().saturating_sub(SHORT_DEVICE_ID_LEN)..]
.iter()
.collect()
}
/// The background sender.
pub struct Sender {
db: Db,
@@ -108,18 +121,25 @@ impl Sender {
.collect()
}
/// Sends `message` to every recipient. Returns one delivery per recipient, in recipient order.
/// Sends `message` to every recipient. Returns one delivery per recipient, in recipient order. When at least one
/// device got it, logs one info line `push sent` with aggregates only (SPEC §8.7): the push type, how many devices
/// got it out of how many were targeted, and the short device ids of [`short_device_id`].
pub async fn deliver(&self, message: &PushMessage) -> Result<Vec<Delivery>, AppError> {
let recipients = self.recipients(message).await?;
let mut deliveries = Vec::new();
for recipient in self.recipients(message).await? {
let delivery = self.send_to(&recipient, message).await?;
let mut sent = Vec::new();
for recipient in &recipients {
let delivery = self.send_to(recipient, message).await?;
match &delivery {
Delivery::Sent { attempts } => tracing::debug!(
push_type = message.push_type.as_str(),
device_id = %recipient.device_id,
attempts,
"push sent"
),
Delivery::Sent { attempts } => {
sent.push(short_device_id(&recipient.device_id));
tracing::debug!(
push_type = message.push_type.as_str(),
device_id = %recipient.device_id,
attempts,
"push sent to device"
);
}
Delivery::TokenCleared => tracing::info!(
device_id = %recipient.device_id,
"FCM token unregistered: cleared"
@@ -134,6 +154,15 @@ impl Sender {
}
deliveries.push(delivery);
}
if !sent.is_empty() {
tracing::info!(
push_type = message.push_type.as_str(),
devices_sent = sent.len(),
devices_targeted = recipients.len(),
devices = %sent.join(","),
"push sent"
);
}
Ok(deliveries)
}
@@ -174,3 +203,17 @@ impl Sender {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::short_device_id;
#[test]
fn short_device_id_is_the_tail() {
assert_eq!(
short_device_id("0192a001-0000-7000-8000-00000000abcd"),
"0000abcd"
);
assert_eq!(short_device_id("abc"), "abc");
}
}
@@ -1,4 +1,5 @@
//! `kvitt-server bootstrap` and `pair-device` (FR-1, AC-1.1 to AC-1.6, SPEC §10.4), run as the real binary.
//! `kvitt-server bootstrap`, `pair-device` (FR-1, AC-1.1 to AC-1.6, SPEC §10.4) and `invite-partner` (D-113), run as
//! the real binary.
use std::process::{Command, Output};
@@ -97,3 +98,90 @@ async fn bootstrap_prints_an_admin_setup_invite_cancels_the_previous_one_then_re
);
}
}
/// The code and link of `invite-partner`'s output, checked like [`printed_code`] but with the `?from=` of D-087.
fn printed_partner_code(output: &Output, from: &str) -> String {
let stdout = String::from_utf8(output.stdout.clone()).unwrap();
let link = stdout
.lines()
.find_map(|l| l.strip_prefix("Link:"))
.unwrap()
.trim();
let rest = link.strip_prefix("https://kvitt.example:8443/i/").unwrap();
let (code, query) = rest.split_once('?').unwrap();
assert_eq!(query, format!("from={from}"));
assert_eq!(code.len(), 12);
let display = stdout
.lines()
.find_map(|l| l.strip_prefix("Code:"))
.unwrap()
.trim();
assert_eq!(display.replace('-', ""), code);
assert!(stdout.contains("Expires: "));
code.to_owned()
}
#[tokio::test]
async fn invite_partner_needs_an_admin_cancels_the_previous_invite_and_refuses_a_full_household() {
// No users: exit 3, nothing created.
let dir = TempDir::new();
let none = run(&dir.0, &["invite-partner"]);
assert_eq!(none.status.code(), Some(3), "{none:?}");
assert_eq!(
String::from_utf8_lossy(&none.stderr).trim(),
"no admin, run bootstrap first"
);
assert!(none.stdout.is_empty());
// Only the admin: as E-12 with kind partner.
let boot = run(&dir.0, &["bootstrap"]);
let app = TestApp::at(dir).await;
let admin = app
.public(
Method::POST,
"/invites/accept",
Some(json!({"code": printed_code(&boot), "device_name": "Galaxy S26", "display_name": "Jacob Å"})),
)
.await;
assert_eq!(admin.status, 200, "{}", admin.text());
let admin_id = admin.json["user"]["id"].as_str().unwrap().to_owned();
let first = run(app.data_dir(), &["invite-partner"]);
assert_eq!(first.status.code(), Some(0), "{first:?}");
let first_code = printed_partner_code(&first, "Jacob%20%C3%85");
let second = run(app.data_dir(), &["invite-partner"]);
assert_eq!(second.status.code(), Some(0), "{second:?}");
let second_code = printed_partner_code(&second, "Jacob%20%C3%85");
assert_ne!(first_code, second_code);
let by_admin: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM invites WHERE kind = 'partner' AND target_user_id IS NULL AND created_by_user_id = ?",
)
.bind(&admin_id)
.fetch_one(&app.state.db)
.await
.unwrap();
assert_eq!(by_admin, 2);
let accept = |code: String| {
app.public(
Method::POST,
"/invites/accept",
Some(json!({"code": code, "device_name": "Galaxy S26+", "display_name": "Anna"})),
)
};
// A new partner invite cancels the pending one, as E-12 does.
assert_error(&accept(first_code).await, 400, "invite_invalid");
let partner = accept(second_code).await;
assert_eq!(partner.status, 200, "{}", partner.text());
assert_eq!(partner.json["user"]["role"], "member");
assert_eq!(partner.json["invite_kind"], "partner");
// Two users: exit 2, nothing created.
let full = run(app.data_dir(), &["invite-partner"]);
assert_eq!(full.status.code(), Some(2), "{full:?}");
assert_eq!(
String::from_utf8_lossy(&full.stderr).trim(),
"household full"
);
assert!(full.stdout.is_empty());
assert_eq!(app.scalar_i64("SELECT COUNT(*) FROM invites").await, 3);
}
@@ -465,3 +465,59 @@ async fn sync_push_reaches_fcm_through_the_queue_and_the_sender_task() {
assert_eq!(body["message"]["data"]["entity_id"], id);
assert_eq!(body["message"]["data"]["actor_user_id"], MEMBER_ID);
}
#[derive(Clone, Default)]
struct Captured(Arc<std::sync::Mutex<Vec<u8>>>);
impl std::io::Write for Captured {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
/// The only test of this binary that installs a subscriber. Global, not thread-local, because tracing caches callsite
/// interest process-wide; the other tests' lines are captured too, so only lines of this test's push type are read.
#[tokio::test]
async fn a_successful_send_logs_one_info_line_with_aggregates_only() {
let captured = Captured::default();
let writer = captured.clone();
let subscriber = tracing_subscriber::fmt()
.json()
.with_max_level(tracing::Level::INFO)
.with_writer(move || writer.clone())
.finish();
tracing::subscriber::set_global_default(subscriber).unwrap();
let s = Setup::new().await;
let deliveries = s
.sender()
.deliver(&message(PushType::SettlementDeleted, MEMBER_ID))
.await
.unwrap();
assert_eq!(deliveries, [Delivery::Sent { attempts: 1 }]);
let logs = String::from_utf8(captured.0.lock().unwrap().clone()).unwrap();
let lines: Vec<Value> = logs
.lines()
.map(|l| serde_json::from_str::<Value>(l).unwrap())
.filter(|l| {
l["fields"]["message"] == "push sent"
&& l["fields"]["push_type"] == "settlement_deleted"
})
.collect();
assert_eq!(lines.len(), 1, "{logs}");
let line = &lines[0];
assert_eq!(line["level"], "INFO");
assert_eq!(line["fields"]["devices_sent"], 1);
assert_eq!(line["fields"]["devices_targeted"], 1);
// The member acted, so the admin's one live device got it: its id's last 8 characters only.
assert_eq!(line["fields"]["devices"], "00000001");
let text = line.to_string();
for secret in ["tok-admin", ADMIN_DEVICE_ID, MEMBER_ID, EXPENSE] {
assert!(!text.contains(secret), "{text}");
}
}
@@ -1,5 +1,6 @@
//! `serve` as deployed: the router with `ConnectInfo<SocketAddr>` (so the rate limiter sees each client's address,
//! SPEC §8.6), `/i/` with an empty code as the HTML 404 page, and `rotate-epoch` as the real binary (SPEC §10.4).
//! SPEC §8.6), `/i/` with an empty code as the HTML 404 page, and `rotate-epoch`, `migrate`, `healthcheck` and the
//! SIGTERM drain of `serve` as the real binary (SPEC §10.4).
use std::net::SocketAddr;
use std::process::Command;
@@ -88,3 +89,225 @@ async fn rotate_epoch_sets_a_new_random_epoch_and_exits_0() {
assert!(after.chars().all(|c| matches!(c, '0'..='9' | 'a'..='f')));
assert!(String::from_utf8_lossy(&output.stdout).contains(&after));
}
/// The real binary with a data directory and listen address of the test's own; push disabled (no credentials).
fn binary(dir: &std::path::Path, bind: SocketAddr) -> Command {
let mut command = Command::new(env!("CARGO_BIN_EXE_kvitt-server"));
command
.env("KVITT_DATA_DIR", dir)
.env("KVITT_BIND", bind.to_string())
.env("KVITT_FCM_CREDENTIALS", dir.join("no-such-key.json"))
.env("RUST_LOG", "warn");
command
}
/// A loopback address with a port nothing listens on (bound once, then released).
fn free_addr() -> SocketAddr {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.local_addr().unwrap()
}
#[tokio::test]
async fn migrate_creates_the_database_at_the_current_schema_and_is_idempotent() {
let dir = TempDir::new();
let data = dir.0.join("data");
for _ in 0..2 {
let output = binary(&data, free_addr()).arg("migrate").output().unwrap();
assert_eq!(output.status.code(), Some(0), "{output:?}");
}
// Opened without migrating, so what is read is what `migrate` wrote.
let pool = kvitt_server::db::connect(&data.join("kvitt.sqlite"))
.await
.unwrap();
let version: String = sqlx::query_scalar("SELECT value FROM meta WHERE key = 'schema_version'")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(version, kvitt_server::db::SCHEMA_VERSION.to_string());
let users: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM users")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(users, 0);
pool.close().await;
}
#[tokio::test]
async fn healthcheck_exits_0_on_200_and_1_on_another_status_or_no_server() {
let dir = TempDir::new();
let healthcheck = |addr: SocketAddr| binary(&dir.0, addr).arg("healthcheck").output().unwrap();
// The real router: 200.
let app = TestApp::empty().await;
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let up = listener.local_addr().unwrap();
let service = make_service(app.state.clone());
tokio::spawn(async move { axum::serve(listener, service).await.unwrap() });
let ok = tokio::task::spawn_blocking({
let dir = dir.0.clone();
move || binary(&dir, up).arg("healthcheck").output().unwrap()
})
.await
.unwrap();
assert_eq!(ok.status.code(), Some(0), "{ok:?}");
// A server answering 503 (as E-1 does when the database is unreachable).
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let unhealthy = listener.local_addr().unwrap();
tokio::spawn(async move {
loop {
let (mut stream, _) = listener.accept().await.unwrap();
let mut buf = [0u8; 1024];
let _ = tokio::io::AsyncReadExt::read(&mut stream, &mut buf).await;
let _ = tokio::io::AsyncWriteExt::write_all(
&mut stream,
b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
)
.await;
}
});
let failed = tokio::task::spawn_blocking({
let dir = dir.0.clone();
move || binary(&dir, unhealthy).arg("healthcheck").output().unwrap()
})
.await
.unwrap();
assert_eq!(failed.status.code(), Some(1), "{failed:?}");
assert!(String::from_utf8_lossy(&failed.stderr).contains("returned 503"));
// Nothing listening.
let down = healthcheck(free_addr());
assert_eq!(down.status.code(), Some(1), "{down:?}");
}
#[cfg(unix)]
mod drain {
use std::net::SocketAddr;
use std::process::{Child, ExitStatus, Stdio};
use std::time::{Duration, Instant};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use super::{binary, free_addr};
use crate::common::TempDir;
use kvitt_server::cli::SHUTDOWN_DRAIN;
/// The body of an E-3 request whose last byte the test holds back.
const BODY: &[u8] = br#"{"code":"0000-0000-0000","device_name":"x"}"#;
/// A running `serve`, killed and reaped on drop if the test did not see it exit.
struct Serve(Child);
impl Drop for Serve {
fn drop(&mut self) {
if self.0.try_wait().ok().flatten().is_none() {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
}
/// `serve` on a free port, once `/health` answers 200.
async fn start(dir: &TempDir) -> (Serve, SocketAddr) {
let addr = free_addr();
let child = Serve(
binary(&dir.0, addr)
.arg("serve")
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.unwrap(),
);
let deadline = Instant::now() + Duration::from_secs(30);
loop {
if let Ok(reply) = reqwest::get(format!("http://{addr}/health")).await
&& reply.status() == 200
{
return (child, addr);
}
assert!(Instant::now() < deadline, "serve did not come up");
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
/// An E-3 request in flight: the head and all but the last byte of the body sent.
async fn in_flight(addr: SocketAddr) -> TcpStream {
let mut stream = TcpStream::connect(addr).await.unwrap();
let head = format!(
"POST /invites/accept HTTP/1.1\r\nHost: {addr}\r\nContent-Type: application/json\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n",
BODY.len()
);
stream.write_all(head.as_bytes()).await.unwrap();
stream.write_all(&BODY[..BODY.len() - 1]).await.unwrap();
tokio::time::sleep(Duration::from_millis(200)).await;
stream
}
fn sigterm(child: &Serve) {
let status = std::process::Command::new("kill")
.args(["-TERM", &child.0.id().to_string()])
.status()
.unwrap();
assert!(status.success());
}
async fn wait(child: &mut Serve, within: Duration) -> ExitStatus {
let deadline = Instant::now() + within;
loop {
if let Some(status) = child.0.try_wait().unwrap() {
return status;
}
assert!(Instant::now() < deadline, "serve did not exit");
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
#[tokio::test]
async fn sigterm_lets_an_in_flight_request_finish_then_exits_0() {
let dir = TempDir::new();
let (mut child, addr) = start(&dir).await;
let mut stream = in_flight(addr).await;
sigterm(&child);
tokio::time::sleep(Duration::from_millis(500)).await;
assert!(
child.0.try_wait().unwrap().is_none(),
"exited with a request in flight"
);
assert!(
TcpStream::connect(addr).await.is_err(),
"new connections are no longer accepted"
);
stream.write_all(&BODY[BODY.len() - 1..]).await.unwrap();
let mut reply = Vec::new();
stream.read_to_end(&mut reply).await.unwrap();
let reply = String::from_utf8_lossy(&reply);
assert!(reply.starts_with("HTTP/1.1 400"), "{reply}");
assert!(reply.contains("invite_invalid"), "{reply}");
assert_eq!(
wait(&mut child, Duration::from_secs(5)).await.code(),
Some(0)
);
}
#[tokio::test]
async fn sigterm_stops_waiting_after_the_drain_period_and_exits_0() {
let dir = TempDir::new();
let (mut child, addr) = start(&dir).await;
let _stuck = in_flight(addr).await;
let signalled = Instant::now();
sigterm(&child);
let status = wait(&mut child, SHUTDOWN_DRAIN + Duration::from_secs(5)).await;
assert_eq!(status.code(), Some(0));
assert!(
signalled.elapsed() >= SHUTDOWN_DRAIN - Duration::from_millis(500),
"exited after {:?}, before the drain period",
signalled.elapsed()
);
}
}