initial commit

src/auth.rs

@@ -0,0 +1,432 @@
+use argon2::{Algorithm, Argon2, Params, Version};
+use base64::{
+    Engine, alphabet,
+    engine::{
+        DecodePaddingMode, GeneralPurpose, GeneralPurposeConfig,
+        general_purpose::{STANDARD_NO_PAD, URL_SAFE_NO_PAD},
+    },
+};
+use chrono::{DateTime, TimeDelta, Utc};
+use hmac::{Hmac, Mac};
+use rand::{RngCore, rngs::OsRng};
+use serde::{
+    Deserialize, Deserializer, Serialize,
+    de::{MapAccess, Visitor},
+};
+use sha2::{Digest, Sha256};
+
+fn random_bytes(count: usize) -> Result<Vec<u8>, String> {
+    let mut bytes = vec![0; count];
+    OsRng
+        .try_fill_bytes(&mut bytes)
+        .map_err(|_| "could not generate random bytes".to_string())?;
+    Ok(bytes)
+}
+
+pub fn random_token(count: usize) -> Result<String, String> {
+    Ok(URL_SAFE_NO_PAD.encode(random_bytes(count)?))
+}
+
+pub fn identifier() -> Result<String, String> {
+    Ok(hex(&random_bytes(16)?))
+}
+
+fn hex(bytes: &[u8]) -> String {
+    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
+pub fn token_hash(token: &str) -> String {
+    hex(&Sha256::digest(token.as_bytes()))
+}
+
+// Go's Raw{Std,URL}Encoding ignores CR/LF and permits nonzero unused bits.
+// Keep signatures canonical: this decoder is only for password data/payloads.
+fn decode_go_base64(value: &str, url_safe: bool) -> Result<Vec<u8>, base64::DecodeError> {
+    let config = GeneralPurposeConfig::new()
+        .with_decode_padding_mode(DecodePaddingMode::RequireNone)
+        .with_decode_allow_trailing_bits(true);
+    let engine = GeneralPurpose::new(
+        if url_safe {
+            &alphabet::URL_SAFE
+        } else {
+            &alphabet::STANDARD
+        },
+        config,
+    );
+    let cleaned: Vec<u8> = value
+        .bytes()
+        .filter(|byte| !matches!(byte, b'\r' | b'\n'))
+        .collect();
+    engine.decode(cleaned)
+}
+
+pub fn hash_password(password: &str) -> Result<String, String> {
+    if password.len() < 8 {
+        return Err("password must contain at least 8 characters".into());
+    }
+    if password.len() > 1024 {
+        return Err("password is too long".into());
+    }
+    let salt = random_bytes(16)?;
+    let mut hash = [0; 32];
+    let params = Params::new(64 * 1024, 3, 2, Some(32)).map_err(|error| error.to_string())?;
+    Argon2::new(Algorithm::Argon2id, Version::V0x13, params)
+        .hash_password_into(password.as_bytes(), &salt, &mut hash)
+        .map_err(|error| error.to_string())?;
+    Ok(format!(
+        "$argon2id$v=19$m=65536,t=3,p=2${}${}",
+        STANDARD_NO_PAD.encode(salt),
+        STANDARD_NO_PAD.encode(hash)
+    ))
+}
+
+pub fn verify_password(encoded: &str, password: &str) -> bool {
+    let parts: Vec<_> = encoded.split('$').collect();
+    if parts.len() != 6 || parts[1] != "argon2id" || parts[2] != "v=19" {
+        return false;
+    }
+    let Some((memory, iterations, threads)) = password_parameters(parts[3]) else {
+        return false;
+    };
+    if memory > 256 * 1024 || iterations > 10 || threads > 16 {
+        return false;
+    }
+    let (Ok(salt), Ok(expected)) = (
+        decode_go_base64(parts[4], false),
+        decode_go_base64(parts[5], false),
+    ) else {
+        return false;
+    };
+    if expected.is_empty() {
+        return false;
+    }
+    let Ok(params) = Params::new(memory, iterations, threads, Some(expected.len())) else {
+        return false;
+    };
+    let mut actual = vec![0; expected.len()];
+    if Argon2::new(Algorithm::Argon2id, Version::V0x13, params)
+        .hash_password_into(password.as_bytes(), &salt, &mut actual)
+        .is_err()
+    {
+        return false;
+    }
+    // Use the MAC crate's constant-time verification primitive for equality.
+    let mut expected_mac = Hmac::<Sha256>::new_from_slice(b"balance password comparison")
+        .expect("HMAC accepts any key size");
+    expected_mac.update(&expected);
+    let expected_tag = expected_mac.finalize().into_bytes();
+    let mut actual_mac = Hmac::<Sha256>::new_from_slice(b"balance password comparison")
+        .expect("HMAC accepts any key size");
+    actual_mac.update(&actual);
+    actual_mac.verify_slice(&expected_tag).is_ok()
+}
+
+fn password_parameters(mut text: &str) -> Option<(u32, u32, u32)> {
+    fn scan(text: &mut &str, prefix: &str) -> Option<u32> {
+        *text = text.strip_prefix(prefix)?;
+        *text = text.trim_start_matches(|c: char| c.is_whitespace() && c != '\n');
+        let digits = *text;
+        let end = digits
+            .find(|c: char| !c.is_ascii_digit())
+            .unwrap_or(digits.len());
+        let value = digits[..end].parse().ok()?;
+        *text = &digits[end..];
+        Some(value)
+    }
+    let memory = scan(&mut text, "m=")?;
+    let iterations = scan(&mut text, ",t=")?;
+    let threads = scan(&mut text, ",p=")?;
+    // fmt.Sscanf stops after its final conversion and accepts a trailing suffix.
+    if threads > u8::MAX as u32 {
+        return None;
+    }
+    Some((memory, iterations, threads))
+}
+
+#[derive(Clone, Default, Debug, Serialize)]
+pub struct Claims {
+    #[serde(rename = "sub")]
+    pub subject: String,
+    pub email: String,
+    #[serde(rename = "iss")]
+    pub issuer: String,
+    #[serde(rename = "iat")]
+    pub issued: i64,
+    #[serde(rename = "exp")]
+    pub expires: i64,
+    #[serde(rename = "jti")]
+    pub id: String,
+}
+
+impl<'de> Deserialize<'de> for Claims {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        struct ClaimsVisitor;
+        impl<'de> Visitor<'de> for ClaimsVisitor {
+            type Value = Claims;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                formatter.write_str("a JWT claims object")
+            }
+
+            fn visit_unit<E: serde::de::Error>(self) -> Result<Claims, E> {
+                Ok(Claims::default())
+            }
+
+            fn visit_map<M: MapAccess<'de>>(self, mut map: M) -> Result<Claims, M::Error> {
+                let mut claims = Claims::default();
+                while let Some(key) = map.next_key::<String>()? {
+                    // encoding/json matches tagged struct fields without case,
+                    // and applies duplicate fields in order. Null leaves scalar
+                    // fields unchanged, including a previously decoded value.
+                    match key.to_ascii_lowercase().replace('ſ', "s").as_str() {
+                        "sub" => {
+                            if let Some(value) = map.next_value::<Option<String>>()? {
+                                claims.subject = value;
+                            }
+                        }
+                        "email" => {
+                            if let Some(value) = map.next_value::<Option<String>>()? {
+                                claims.email = value;
+                            }
+                        }
+                        "iss" => {
+                            if let Some(value) = map.next_value::<Option<String>>()? {
+                                claims.issuer = value;
+                            }
+                        }
+                        "iat" => {
+                            if let Some(value) = map.next_value::<Option<i64>>()? {
+                                claims.issued = value;
+                            }
+                        }
+                        "exp" => {
+                            if let Some(value) = map.next_value::<Option<i64>>()? {
+                                claims.expires = value;
+                            }
+                        }
+                        "jti" => {
+                            if let Some(value) = map.next_value::<Option<String>>()? {
+                                claims.id = value;
+                            }
+                        }
+                        _ => {
+                            map.next_value::<serde::de::IgnoredAny>()?;
+                        }
+                    }
+                }
+                Ok(claims)
+            }
+        }
+        deserializer.deserialize_any(ClaimsVisitor)
+    }
+}
+
+#[derive(Clone)]
+pub struct TokenManager {
+    secret: Vec<u8>,
+    access_ttl: TimeDelta,
+}
+
+impl TokenManager {
+    pub fn new(secret: &str, access_ttl: TimeDelta) -> Self {
+        Self {
+            secret: secret.as_bytes().to_vec(),
+            access_ttl,
+        }
+    }
+
+    pub fn access_token(
+        &self,
+        user_id: &str,
+        email: &str,
+    ) -> Result<(String, DateTime<Utc>), String> {
+        let now = Utc::now();
+        let expires = now + self.access_ttl;
+        let claims = Claims {
+            subject: user_id.into(),
+            email: email.into(),
+            issuer: "balance-server".into(),
+            issued: now.timestamp(),
+            expires: expires.timestamp(),
+            id: random_token(16)?,
+        };
+        let payload = serde_json::to_vec(&claims).map_err(|error| error.to_string())?;
+        let unsigned = format!(
+            "{}.{}",
+            URL_SAFE_NO_PAD.encode(br#"{"alg":"HS256","typ":"JWT"}"#),
+            URL_SAFE_NO_PAD.encode(payload)
+        );
+        Ok((format!("{}.{}", unsigned, self.sign(&unsigned)), expires))
+    }
+
+    pub fn verify(&self, token: &str) -> Result<Claims, String> {
+        let parts: Vec<_> = token.split('.').collect();
+        if parts.len() != 3 {
+            return Err("invalid token".into());
+        }
+        let unsigned = format!("{}.{}", parts[0], parts[1]);
+        // Go compares the encoded signature, so reject noncanonical encodings too.
+        let expected = self.sign(&unsigned);
+        let mut mac =
+            Hmac::<Sha256>::new_from_slice(&self.secret).expect("HMAC accepts any key size");
+        mac.update(parts[2].as_bytes());
+        let signature_tag = mac.finalize().into_bytes();
+        let mut expected_mac =
+            Hmac::<Sha256>::new_from_slice(&self.secret).expect("HMAC accepts any key size");
+        expected_mac.update(expected.as_bytes());
+        if expected_mac.verify_slice(&signature_tag).is_err() {
+            return Err("invalid token signature".into());
+        }
+        let payload =
+            decode_go_base64(parts[1], true).map_err(|_| "invalid token payload".to_string())?;
+        let claims: Claims = serde_json::from_slice(&crate::go_json::normalize(&payload))
+            .map_err(|_| "invalid token claims".to_string())?;
+        if claims.issuer != "balance-server"
+            || claims.subject.is_empty()
+            || Utc::now().timestamp() >= claims.expires
+        {
+            return Err("expired or invalid token".into());
+        }
+        Ok(claims)
+    }
+
+    fn sign(&self, value: &str) -> String {
+        let mut mac =
+            Hmac::<Sha256>::new_from_slice(&self.secret).expect("HMAC accepts any key size");
+        mac.update(value.as_bytes());
+        URL_SAFE_NO_PAD.encode(mac.finalize().into_bytes())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn jwt_claims_follow_go_scalar_and_duplicate_field_rules() {
+        let manager = TokenManager::new("secret", TimeDelta::minutes(1));
+        let claims = r#"{"SUB":"old","sub":"user","sub":null,"ISS":"balance-server","EXP":4102444800,"exp":null,"EMAIL":null,"iat":null,"jti":"id","JTI":null}"#;
+        // The Go verifier does not inspect or decode the JWT header.
+        let unsigned = format!("ignored-header.{}", URL_SAFE_NO_PAD.encode(claims));
+        let token = format!("{}.{}", unsigned, manager.sign(&unsigned));
+        let actual = manager.verify(&token).unwrap();
+        assert_eq!(actual.subject, "user");
+        assert_eq!(actual.email, "");
+        assert_eq!(actual.expires, 4_102_444_800);
+        assert_eq!(actual.id, "id");
+        for invalid in [r#"{"sub":1}"#, r#"{"iat":1.0}"#, r#"[]"#] {
+            assert!(serde_json::from_str::<Claims>(invalid).is_err());
+        }
+    }
+
+    #[test]
+    fn jwt_claims_replace_malformed_unicode_as_go_does() {
+        let manager = TokenManager::new("secret", TimeDelta::minutes(1));
+        for payload in [
+            br#"{"sub":"\ud800","iss":"balance-server","exp":4102444800}"#.as_slice(),
+            b"{\"sub\":\"\xff\",\"iss\":\"balance-server\",\"exp\":4102444800}".as_slice(),
+        ] {
+            let unsigned = format!("ignored.{}", URL_SAFE_NO_PAD.encode(payload));
+            let token = format!("{}.{}", unsigned, manager.sign(&unsigned));
+            assert_eq!(manager.verify(&token).unwrap().subject, "�");
+        }
+    }
+
+    #[test]
+    fn go_base64_ignores_line_breaks_and_unused_bits_only() {
+        assert_eq!(decode_go_base64("Zh\r\n", true).unwrap(), b"f");
+        assert_eq!(decode_go_base64("Zh\r\n", false).unwrap(), b"f");
+        for invalid in ["Zg=", "Z g", "Zg\t", "Z"] {
+            assert!(decode_go_base64(invalid, true).is_err());
+        }
+        assert_eq!(
+            password_parameters("m=65536,t=3,p=2suffix"),
+            Some((65536, 3, 2))
+        );
+        assert_eq!(password_parameters("m=65536,t= 3,p=2"), Some((65536, 3, 2)));
+        assert_eq!(password_parameters("m=65536,t=3,p=256"), None);
+        assert_eq!(password_parameters("m=+65536,t=3,p=2"), None);
+    }
+
+    #[test]
+    fn accepts_existing_password_hash_with_go_parameters() {
+        // Fixed salt and an independently computed libargon2 reference result.
+        // Go's argon2.IDKey uses this same Argon2id v19 representation.
+        let encoded = "$argon2id$v=19$m=65536,t=3,p=2$MDEyMzQ1Njc4OWFiY2RlZg$0R79n3aEb762TWx/8plp8QrcDJd8M65F5ccBf1ugp4Y";
+        assert!(verify_password(encoded, "Go-compatible-✓"));
+        assert!(!verify_password(encoded, "Go-compatible-x"));
+        assert_eq!(
+            hash_password("short").unwrap_err(),
+            "password must contain at least 8 characters"
+        );
+        assert_eq!(
+            hash_password(&"x".repeat(1025)).unwrap_err(),
+            "password is too long"
+        );
+        // The original Go implementation measures UTF-8 bytes, not characters.
+        assert!(hash_password("éééé").is_ok());
+        for malformed in [
+            "",
+            "$argon2i$v=19$m=65536,t=3,p=2$c2FsdA$aGFzaA",
+            "$argon2id$v=19$m=999999,t=3,p=2$c2FsdA$aGFzaA",
+            "$argon2id$v=19$m=65536,t=3,p=2$!$!",
+        ] {
+            assert!(!verify_password(malformed, "password"));
+        }
+    }
+
+    #[test]
+    fn accepts_go_jwt_wire_format_and_preserves_refresh_hashes() {
+        // Signed independently using Go's exact header, claims and HS256 encoding.
+        let token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJnby11c2VyIiwiZW1haWwiOiJnb0BleGFtcGxlLmNvbSIsImlzcyI6ImJhbGFuY2Utc2VydmVyIiwiaWF0IjoxNzAwMDAwMDAwLCJleHAiOjQxMDI0NDQ4MDAsImp0aSI6ImdvLXNlc3Npb24ifQ.BJrD7QmRGbRhYqS3EI1stCV5AEkocqOcFAJ3yC4MqeI";
+        let manager = TokenManager::new("go-fixture-secret", TimeDelta::minutes(15));
+        let claims = manager.verify(token).unwrap();
+        assert_eq!(claims.subject, "go-user");
+        assert_eq!(claims.email, "go@example.com");
+        assert_eq!(claims.expires, 4_102_444_800);
+        assert_eq!(claims.id, "go-session");
+        assert!(manager.verify(&format!("{token}=")).is_err());
+        assert!(
+            manager
+                .verify(&token.replace("eyJzdWI", "eyJzdWJ"))
+                .is_err()
+        );
+        assert_eq!(
+            token_hash("existing-go-refresh-token"),
+            "66e9debb6786892dfecaec6bff6761635629efae55baaeb78645820772911605"
+        );
+        assert_eq!(
+            URL_SAFE_NO_PAD
+                .decode(random_token(48).unwrap())
+                .unwrap()
+                .len(),
+            48
+        );
+        let id = identifier().unwrap();
+        assert_eq!(id.len(), 32);
+        assert!(
+            id.bytes()
+                .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
+        );
+    }
+
+    #[test]
+    fn passwords_and_tokens_round_trip() {
+        let encoded = hash_password("password123").unwrap();
+        assert!(encoded.starts_with("$argon2id$v=19$m=65536,t=3,p=2$"));
+        assert!(verify_password(&encoded, "password123"));
+        assert!(!verify_password(&encoded, "wrong"));
+        let manager = TokenManager::new("test secret", TimeDelta::minutes(15));
+        let (token, _) = manager.access_token("user", "user@example.com").unwrap();
+        assert_eq!(manager.verify(&token).unwrap().subject, "user");
+        assert!(
+            TokenManager::new("other", TimeDelta::minutes(15))
+                .verify(&token)
+                .is_err()
+        );
+        let (expired, _) = TokenManager::new("test secret", TimeDelta::seconds(-1))
+            .access_token("user", "email")
+            .unwrap();
+        assert!(manager.verify(&expired).is_err());
+    }
+}

src/config.rs

@@ -0,0 +1,160 @@
+use chrono::TimeDelta;
+use std::env;
+
+#[derive(Clone, Debug)]
+pub struct Config {
+    pub address: String,
+    pub database_path: String,
+    pub jwt_secret: String,
+    pub access_ttl: TimeDelta,
+    pub refresh_ttl: TimeDelta,
+    pub allow_registration: bool,
+    pub cors_origins: Vec<String>,
+}
+
+impl Config {
+    pub fn load() -> Result<Self, String> {
+        let access_ttl = duration("BALANCE_ACCESS_TTL", TimeDelta::minutes(15))?;
+        let refresh_ttl = duration("BALANCE_REFRESH_TTL", TimeDelta::days(30))?;
+        let allow_registration = match value("BALANCE_ALLOW_REGISTRATION").as_str() {
+            "" | "1" | "t" | "T" | "TRUE" | "true" | "True" => true,
+            "0" | "f" | "F" | "FALSE" | "false" | "False" => false,
+            _ => return Err("BALANCE_ALLOW_REGISTRATION must be true or false".into()),
+        };
+        let jwt_secret = value("BALANCE_JWT_SECRET");
+        if jwt_secret.len() < 32 {
+            return Err("BALANCE_JWT_SECRET must contain at least 32 characters".into());
+        }
+        Ok(Self {
+            address: fallback("BALANCE_ADDR", ":8080"),
+            database_path: fallback("BALANCE_DB_PATH", "./data/balance.db"),
+            jwt_secret,
+            access_ttl,
+            refresh_ttl,
+            allow_registration,
+            cors_origins: value("BALANCE_CORS_ORIGINS")
+                .split(',')
+                .map(str::trim)
+                .filter(|part| !part.is_empty())
+                .map(str::to_owned)
+                .collect(),
+        })
+    }
+}
+
+fn value(key: &str) -> String {
+    env::var(key).unwrap_or_default().trim().to_owned()
+}
+fn fallback(key: &str, default: &str) -> String {
+    let value = value(key);
+    if value.is_empty() {
+        default.to_owned()
+    } else {
+        value
+    }
+}
+fn duration(key: &str, default: TimeDelta) -> Result<TimeDelta, String> {
+    let value = value(key);
+    if value.is_empty() {
+        return Ok(default);
+    }
+    parse_go_duration(&value)
+        .ok_or_else(|| format!("{key} must be a Go duration such as 15m or 720h"))
+}
+
+/// Parse Go's signed, nanosecond-precision duration syntax and int64 range.
+pub fn parse_go_duration(input: &str) -> Option<TimeDelta> {
+    let (negative, mut remaining) = match input.as_bytes().first()? {
+        b'-' => (true, &input[1..]),
+        b'+' => (false, &input[1..]),
+        _ => (false, input),
+    };
+    if remaining == "0" {
+        return Some(TimeDelta::zero());
+    }
+    if remaining.is_empty() {
+        return None;
+    }
+    let mut total = 0i128;
+    while !remaining.is_empty() {
+        let integer_len = remaining.bytes().take_while(u8::is_ascii_digit).count();
+        let integer = if integer_len == 0 {
+            0
+        } else {
+            remaining[..integer_len].parse::<i128>().ok()?
+        };
+        remaining = &remaining[integer_len..];
+        let mut fraction = 0i128;
+        let mut scale = 1i128;
+        let mut fraction_len = 0;
+        if remaining.starts_with('.') {
+            remaining = &remaining[1..];
+            fraction_len = remaining.bytes().take_while(u8::is_ascii_digit).count();
+            // Beyond 18 decimal places no supported unit can affect nanoseconds.
+            for digit in remaining.bytes().take(fraction_len.min(18)) {
+                fraction = fraction * 10 + i128::from(digit - b'0');
+                scale *= 10;
+            }
+            remaining = &remaining[fraction_len..];
+        }
+        if integer_len == 0 && fraction_len == 0 {
+            return None;
+        }
+        let (unit_text, unit) = [
+            ("ns", 1i128),
+            ("us", 1_000),
+            ("µs", 1_000),
+            ("μs", 1_000),
+            ("ms", 1_000_000),
+            ("s", 1_000_000_000),
+            ("m", 60_000_000_000),
+            ("h", 3_600_000_000_000),
+        ]
+        .into_iter()
+        .find(|(unit, _)| remaining.starts_with(unit))?;
+        remaining = &remaining[unit_text.len()..];
+        total = total
+            .checked_add(integer.checked_mul(unit)?)?
+            .checked_add(fraction * unit / scale)?;
+        if total > i128::from(i64::MAX) + i128::from(negative) {
+            return None;
+        }
+    }
+    let nanos = i64::try_from(if negative { -total } else { total }).ok()?;
+    Some(TimeDelta::nanoseconds(nanos))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn duration_matches_go_syntax_and_range() {
+        assert_eq!(parse_go_duration("1h30m"), Some(TimeDelta::minutes(90)));
+        assert_eq!(
+            parse_go_duration("-.5s"),
+            Some(TimeDelta::milliseconds(-500))
+        );
+        assert_eq!(parse_go_duration("1µs"), Some(TimeDelta::microseconds(1)));
+        assert_eq!(parse_go_duration("1.s"), Some(TimeDelta::seconds(1)));
+        assert_eq!(parse_go_duration("0"), Some(TimeDelta::zero()));
+        assert_eq!(
+            parse_go_duration("9223372036854775807ns"),
+            Some(TimeDelta::nanoseconds(i64::MAX))
+        );
+        assert_eq!(
+            parse_go_duration("-9223372036854775808ns"),
+            Some(TimeDelta::nanoseconds(i64::MIN))
+        );
+        for invalid in [
+            "",
+            "1d",
+            "1",
+            "9223372036854775808ns",
+            "1h 2m",
+            ".s",
+            "--1s",
+        ] {
+            assert!(parse_go_duration(invalid).is_none(), "{invalid}");
+        }
+    }
+}

src/database.rs

@@ -0,0 +1,185 @@
+use rusqlite::Connection;
+use std::{error::Error, path::Path, time::Duration};
+
+/// Open the existing Go server's SQLite database without changing its schema.
+pub fn open(path: &str) -> Result<Connection, Box<dyn Error + Send + Sync>> {
+    if let Some(parent) = Path::new(path)
+        .parent()
+        .filter(|p| !p.as_os_str().is_empty())
+    {
+        let mut builder = std::fs::DirBuilder::new();
+        builder.recursive(true);
+        // Match Go's MkdirAll(..., 0750): the database directory must not be
+        // accessible to other local users under the usual 022 umask.
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::DirBuilderExt;
+            builder.mode(0o750);
+        }
+        builder.create(parent)?;
+    }
+    let db = Connection::open(path)?;
+    db.pragma_update(None, "foreign_keys", true)?;
+    db.pragma_update(None, "journal_mode", "WAL")?;
+    db.busy_timeout(Duration::from_secs(5))?;
+    migrate(&db)?;
+    Ok(db)
+}
+
+pub fn migrate(db: &Connection) -> rusqlite::Result<()> {
+    db.execute_batch(
+        "CREATE TABLE IF NOT EXISTS users (
+  id TEXT PRIMARY KEY,
+  email TEXT NOT NULL UNIQUE COLLATE NOCASE,
+  password_hash TEXT NOT NULL,
+  created_at TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS refresh_tokens (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  token_hash TEXT NOT NULL UNIQUE,
+  user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  expires_at TEXT NOT NULL,
+  created_at TEXT NOT NULL,
+  revoked_at TEXT
+);
+CREATE INDEX IF NOT EXISTS refresh_tokens_user_idx ON refresh_tokens(user_id);
+CREATE TABLE IF NOT EXISTS records (
+  user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  entity TEXT NOT NULL,
+  record_id TEXT NOT NULL,
+  payload TEXT,
+  updated_at TEXT NOT NULL,
+  deleted INTEGER NOT NULL DEFAULT 0,
+  version INTEGER NOT NULL,
+  PRIMARY KEY (user_id, entity, record_id)
+);
+CREATE TABLE IF NOT EXISTS changes (
+  sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+  user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  entity TEXT NOT NULL,
+  record_id TEXT NOT NULL,
+  payload TEXT,
+  updated_at TEXT NOT NULL,
+  deleted INTEGER NOT NULL,
+  version INTEGER NOT NULL
+);
+CREATE INDEX IF NOT EXISTS changes_user_sequence_idx ON changes(user_id, sequence);",
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn schema_is_repeatable_and_enforces_existing_constraints() {
+        let db = open(":memory:").unwrap();
+        migrate(&db).unwrap();
+        db.execute(
+            "INSERT INTO users VALUES ('u','Test@example.com','hash','now')",
+            [],
+        )
+        .unwrap();
+        assert!(
+            db.execute(
+                "INSERT INTO users VALUES ('v','test@example.com','hash','now')",
+                []
+            )
+            .is_err()
+        );
+        assert!(
+            db.execute(
+                "INSERT INTO records VALUES ('missing','category','c',NULL,'now',1,1)",
+                []
+            )
+            .is_err()
+        );
+        db.execute("INSERT INTO refresh_tokens(token_hash,user_id,expires_at,created_at) VALUES ('hash','u','then','now')", []).unwrap();
+        db.execute("DELETE FROM users WHERE id='u'", []).unwrap();
+        assert_eq!(
+            db.query_row("SELECT count(*) FROM refresh_tokens", [], |r| r
+                .get::<_, i64>(0))
+                .unwrap(),
+            0
+        );
+    }
+
+    #[test]
+    fn opens_populated_go_schema_without_changing_rows_or_sequence() {
+        let directory = std::env::temp_dir().join(format!(
+            "balance-go-db-{}",
+            crate::auth::identifier().unwrap()
+        ));
+        std::fs::create_dir(&directory).unwrap();
+        let path = directory.join("existing.sqlite");
+        {
+            let db = Connection::open(&path).unwrap();
+            // Use the original schema, so a divergence between implementations
+            // is caught even when Rust's migration is internally consistent.
+            let go_source = include_str!("../internal/database/database.go");
+            let schema = go_source
+                .split("const schema = `")
+                .nth(1)
+                .unwrap()
+                .split('`')
+                .next()
+                .unwrap();
+            db.execute_batch(schema).unwrap();
+            db.execute_batch(
+                "INSERT INTO users VALUES ('go-user','go@example.com','existing hash','2026-01-01T00:00:00.123456789Z');
+                 INSERT INTO refresh_tokens(id,token_hash,user_id,expires_at,created_at) VALUES (7,'existing-token-hash','go-user','2099-01-01T00:00:00Z','2026-01-01T00:00:00Z');
+                 INSERT INTO records VALUES ('go-user','transaction','record', '{\"amount\":12.5}', '2026-01-01T00:00:00.123456789Z',0,41);
+                 INSERT INTO changes VALUES (41,'go-user','transaction','record','{\"amount\":12.5}','2026-01-01T00:00:00.123456789Z',0,41);"
+            ).unwrap();
+        }
+        {
+            let db = open(path.to_str().unwrap()).unwrap();
+            let user: (String, String) = db
+                .query_row(
+                    "SELECT password_hash,created_at FROM users WHERE id='go-user'",
+                    [],
+                    |row| Ok((row.get(0)?, row.get(1)?)),
+                )
+                .unwrap();
+            assert_eq!(
+                user,
+                (
+                    "existing hash".into(),
+                    "2026-01-01T00:00:00.123456789Z".into()
+                )
+            );
+            assert_eq!(
+                db.query_row(
+                    "SELECT id FROM refresh_tokens WHERE revoked_at IS NULL",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )
+                .unwrap(),
+                7
+            );
+            assert_eq!(
+                db.query_row("SELECT payload FROM records", [], |r| r.get::<_, String>(0))
+                    .unwrap(),
+                "{\"amount\":12.5}"
+            );
+            db.execute("INSERT INTO changes(user_id,entity,record_id,payload,updated_at,deleted,version) VALUES ('go-user','transaction','next',NULL,'now',1,42)", []).unwrap();
+            assert_eq!(db.last_insert_rowid(), 42);
+            assert_eq!(
+                db.query_row("PRAGMA foreign_keys", [], |r| r.get::<_, i64>(0))
+                    .unwrap(),
+                1
+            );
+            assert_eq!(
+                db.query_row("PRAGMA journal_mode", [], |r| r.get::<_, String>(0))
+                    .unwrap(),
+                "wal"
+            );
+            assert_eq!(
+                db.query_row("PRAGMA busy_timeout", [], |r| r.get::<_, i64>(0))
+                    .unwrap(),
+                5000
+            );
+        }
+        std::fs::remove_dir_all(directory).unwrap();
+    }
+}

src/go_json.rs

@@ -0,0 +1,326 @@
+//! Compatibility with encoding/json's replacement of malformed Unicode in
+//! decoded JSON strings. Syntax errors remain errors for the JSON decoder.
+//!
+//! Apply this only after enforcing limits on the original request bytes.
+//! RawMessage/RawValue fields must retain their original input separately:
+//! Go does not replace malformed Unicode inside a RawMessage.
+
+/// Replace malformed UTF-8 and unpaired UTF-16 escapes inside JSON strings.
+/// Valid bytes, whitespace and escape spellings are preserved exactly.
+pub fn normalize(input: &[u8]) -> Vec<u8> {
+    let mut output = Vec::with_capacity(input.len());
+    let mut index = 0;
+    let mut in_string = false;
+    while index < input.len() {
+        let byte = input[index];
+        if !in_string {
+            output.push(byte);
+            in_string = byte == b'"';
+            index += 1;
+            continue;
+        }
+        match byte {
+            b'"' => {
+                output.push(byte);
+                in_string = false;
+                index += 1;
+            }
+            b'\\' => {
+                if let Some(code) = unicode_escape(&input[index..]) {
+                    if (0xD800..=0xDBFF).contains(&code) {
+                        if unicode_escape(&input[index + 6..])
+                            .is_some_and(|low| (0xDC00..=0xDFFF).contains(&low))
+                        {
+                            output.extend_from_slice(&input[index..index + 12]);
+                            index += 12;
+                            continue;
+                        }
+                        output.extend_from_slice(b"\\uFFFD");
+                    } else if (0xDC00..=0xDFFF).contains(&code) {
+                        output.extend_from_slice(b"\\uFFFD");
+                    } else {
+                        output.extend_from_slice(&input[index..index + 6]);
+                    }
+                    index += 6;
+                } else {
+                    // Skip an escaped quote/backslash as one unit; invalid
+                    // escape syntax is left intact for serde to reject.
+                    let end = (index + 2).min(input.len());
+                    output.extend_from_slice(&input[index..end]);
+                    index = end;
+                }
+            }
+            0..=0x7f => {
+                output.push(byte);
+                index += 1;
+            }
+            _ => {
+                let length = match byte {
+                    0xC2..=0xDF => 2,
+                    0xE0..=0xEF => 3,
+                    0xF0..=0xF4 => 4,
+                    _ => 0,
+                };
+                if length != 0
+                    && index + length <= input.len()
+                    && std::str::from_utf8(&input[index..index + length]).is_ok()
+                {
+                    output.extend_from_slice(&input[index..index + length]);
+                    index += length;
+                } else {
+                    // Go's utf8.DecodeRune consumes one byte for every error,
+                    // including each byte of an incomplete sequence.
+                    output.extend_from_slice("\u{FFFD}".as_bytes());
+                    index += 1;
+                }
+            }
+        }
+    }
+    output
+}
+
+fn unicode_escape(input: &[u8]) -> Option<u16> {
+    if input.len() < 6 || &input[..2] != b"\\u" {
+        return None;
+    }
+    input[2..6].iter().try_fold(0u16, |value, byte| {
+        let digit = match byte {
+            b'0'..=b'9' => byte - b'0',
+            b'a'..=b'f' => byte - b'a' + 10,
+            b'A'..=b'F' => byte - b'A' + 10,
+            _ => return None,
+        };
+        Some(value * 16 + u16::from(digit))
+    })
+}
+
+/// Replace named raw fields with numbered JSON strings before scalar decoding.
+/// The returned raw values retain the original byte spelling and whitespace.
+pub fn extract_raw_fields(input: &[u8], field: &str) -> Result<(Vec<u8>, Vec<Vec<u8>>), ()> {
+    let start = skip_space(input, 0);
+    let end = value_end(input, start)?;
+    let mut output = Vec::with_capacity(input.len());
+    let mut raw = Vec::new();
+    let mut copied = 0;
+    let mut index = start;
+    while index < end {
+        if input[index] != b'"' {
+            index += 1;
+            continue;
+        }
+        let string_end = quoted_end(input, index)?;
+        let colon = skip_space(input, string_end);
+        if input.get(colon) == Some(&b':') {
+            let key: String =
+                serde_json::from_slice(&normalize(&input[index..string_end])).map_err(|_| ())?;
+            if key.to_ascii_lowercase().replace('ſ', "s") == field {
+                let value_start = skip_space(input, colon + 1);
+                let value_end = value_end(input, value_start)?;
+                let bytes = &input[value_start..value_end];
+                // Validate raw syntax without converting numbers or strings.
+                serde_json::from_slice::<Box<serde_json::value::RawValue>>(&normalize(bytes))
+                    .map_err(|_| ())?;
+                output.extend_from_slice(&input[copied..value_start]);
+                output.extend_from_slice(format!("\"{}\"", raw.len()).as_bytes());
+                raw.push(bytes.to_vec());
+                copied = value_end;
+                index = value_end;
+                continue;
+            }
+        }
+        index = string_end;
+    }
+    output.extend_from_slice(&input[copied..]);
+    Ok((output, raw))
+}
+
+fn skip_space(input: &[u8], mut index: usize) -> usize {
+    while input
+        .get(index)
+        .is_some_and(|b| matches!(b, b' ' | b'\r' | b'\n' | b'\t'))
+    {
+        index += 1;
+    }
+    index
+}
+
+fn quoted_end(input: &[u8], mut index: usize) -> Result<usize, ()> {
+    index += 1;
+    while index < input.len() {
+        match input[index] {
+            b'"' => return Ok(index + 1),
+            b'\\' => index += 2,
+            _ => index += 1,
+        }
+    }
+    Err(())
+}
+
+fn value_end(input: &[u8], start: usize) -> Result<usize, ()> {
+    match input.get(start).ok_or(())? {
+        b'"' => quoted_end(input, start),
+        b'[' | b'{' => {
+            let mut depth = 1usize;
+            let mut index = start + 1;
+            while index < input.len() {
+                match input[index] {
+                    b'"' => {
+                        index = quoted_end(input, index)?;
+                        continue;
+                    }
+                    b'[' | b'{' => {
+                        depth += 1;
+                        if depth > 10_000 {
+                            return Err(());
+                        }
+                    }
+                    b']' | b'}' => {
+                        depth -= 1;
+                        if depth == 0 {
+                            return Ok(index + 1);
+                        }
+                    }
+                    _ => {}
+                }
+                index += 1;
+            }
+            Err(())
+        }
+        _ => {
+            let length = input[start..]
+                .iter()
+                .position(|b| matches!(b, b' ' | b'\r' | b'\n' | b'\t' | b',' | b']' | b'}'))
+                .unwrap_or(input.len() - start);
+            if length == 0 {
+                Err(())
+            } else {
+                Ok(start + length)
+            }
+        }
+    }
+}
+
+/// Go's encoding/json compacts RawMessage and escapes HTML-sensitive bytes,
+/// preserving duplicate keys, number spellings and malformed Unicode bytes.
+pub fn compact_raw(input: &[u8]) -> Vec<u8> {
+    let mut result = Vec::with_capacity(input.len());
+    let mut in_string = false;
+    let mut escaped = false;
+    let mut index = 0;
+    while index < input.len() {
+        let byte = input[index];
+        if !in_string && matches!(byte, b' ' | b'\n' | b'\r' | b'\t') {
+            index += 1;
+            continue;
+        }
+        if in_string && !escaped {
+            let replacement: Option<&[u8]> = match byte {
+                b'<' => Some(b"\\u003c"),
+                b'>' => Some(b"\\u003e"),
+                b'&' => Some(b"\\u0026"),
+                _ => None,
+            };
+            if let Some(value) = replacement {
+                result.extend_from_slice(value);
+                index += 1;
+                continue;
+            }
+            if input[index..].starts_with(&[0xE2, 0x80, 0xA8])
+                || input[index..].starts_with(&[0xE2, 0x80, 0xA9])
+            {
+                result.extend_from_slice(if input[index + 2] == 0xA8 {
+                    b"\\u2028"
+                } else {
+                    b"\\u2029"
+                });
+                index += 3;
+                continue;
+            }
+        }
+        result.push(byte);
+        if escaped {
+            escaped = false;
+        } else if in_string && byte == b'\\' {
+            escaped = true;
+        } else if byte == b'"' {
+            in_string = !in_string;
+        }
+        index += 1;
+    }
+    result
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn preserves_valid_json_and_literal_backslashes() {
+        let input =
+            br#" {"escaped":"\\ud800", "pair":"\ud83d\ude00", "quote":"\"", "slash":"\\"} "#;
+        assert_eq!(normalize(input), input);
+        serde_json::from_slice::<serde_json::Value>(&normalize(input)).unwrap();
+        let unicode = "{\"text\":\"Привет 🌍\"}".as_bytes();
+        assert_eq!(normalize(unicode), unicode);
+    }
+
+    #[test]
+    fn replaces_unpaired_surrogates_without_consuming_following_escapes() {
+        let input = br#"["\ud800", "\udc00", "\ud800\u0061", "\ud800\ud800\udc00"]"#;
+        let value: Vec<String> = serde_json::from_slice(&normalize(input)).unwrap();
+        assert_eq!(value, ["�", "�", "�a", "�𐀀"]);
+    }
+
+    #[test]
+    fn replaces_each_malformed_utf8_byte_as_go_does() {
+        for (input, expected) in [
+            (b"\"\xff\"".as_slice(), "�"),
+            (b"\"\xe1\x80\"".as_slice(), "��"),
+            (b"\"\xed\xa0\x80\"".as_slice(), "���"),
+            (b"\"\xf4\x90\x80\x80\"".as_slice(), "����"),
+        ] {
+            let decoded: String = serde_json::from_slice(&normalize(input)).unwrap();
+            assert_eq!(decoded, expected);
+        }
+    }
+
+    #[test]
+    fn leaves_invalid_json_syntax_rejected() {
+        for input in [
+            br#""\uZZZZ""#.as_slice(),
+            br#""\x""#.as_slice(),
+            b"\"raw\nnewline\"".as_slice(),
+            b"{\xff:1}".as_slice(),
+        ] {
+            assert!(serde_json::from_slice::<serde_json::Value>(&normalize(input)).is_err());
+        }
+    }
+
+    #[test]
+    fn raw_extraction_preserves_bytes_and_ignores_trailing_json() {
+        let input = b"{\"payload\": { \"x\":\"\\ud800\", \"x\":\"\xff\" },\"PAYLOAD\":null} {\"payload\":not-json}";
+        let (masked, values) = extract_raw_fields(input, "payload").unwrap();
+        assert_eq!(
+            values,
+            [
+                b"{ \"x\":\"\\ud800\", \"x\":\"\xff\" }".to_vec(),
+                b"null".to_vec()
+            ]
+        );
+        assert_eq!(
+            masked,
+            b"{\"payload\": \"0\",\"PAYLOAD\":\"1\"} {\"payload\":not-json}"
+        );
+        assert!(extract_raw_fields(br#"{"payload":{"bad":}}"#, "payload").is_err());
+        assert!(extract_raw_fields(br#"{"payload":[1,]}"#, "payload").is_err());
+    }
+
+    #[test]
+    fn raw_compaction_matches_go_html_escaping_without_decoding_strings() {
+        let input = "{ \"raw\":\"<&>\u{2028}\u{2029}\", \"escaped\":\"\\u003c\", \"dup\":1, \"dup\":2, \"n\":1.00e+999 }";
+        assert_eq!(
+            compact_raw(input.as_bytes()),
+            br#"{"raw":"\u003c\u0026\u003e\u2028\u2029","escaped":"\u003c","dup":1,"dup":2,"n":1.00e+999}"#
+        );
+    }
+}

src/main.rs

@@ -0,0 +1,49 @@
+mod auth;
+mod config;
+mod database;
+mod go_json;
+mod runtime;
+mod server;
+mod sync;
+
+use std::sync::{Arc, Mutex};
+fn main() {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .expect("create async runtime");
+    let result = runtime.block_on(run());
+    // Connection draining already has a ten-second budget. Blocking database
+    // or password jobs must not extend process shutdown beyond that budget.
+    runtime.shutdown_timeout(std::time::Duration::ZERO);
+    if let Err(e) = result {
+        eprintln!("Balance server startup failed: {e}");
+        std::process::exit(1);
+    }
+}
+async fn run() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let cfg = config::Config::load()?;
+    let db = database::open(&cfg.database_path)?;
+    let address = cfg.address.clone();
+    let listener = runtime::bind(&address).await?;
+    let state = Arc::new(server::State::new(cfg, Mutex::new(db)));
+    let app = axum::Router::new()
+        .fallback(server::handle)
+        .with_state(state);
+    eprintln!("Balance server listening on {address}");
+    runtime::serve(listener, app, runtime::Timeouts::default(), async {
+        #[cfg(unix)]
+        {
+            let mut term =
+                tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+                    .expect("signal handler");
+            tokio::select! { _ = tokio::signal::ctrl_c() => {}, _ = term.recv() => {} }
+        }
+        #[cfg(not(unix))]
+        {
+            let _ = tokio::signal::ctrl_c().await;
+        }
+    })
+    .await?;
+    Ok(())
+}

src/runtime.rs

@@ -0,0 +1,550 @@
+//! HTTP/1 transport deadlines matching net/http.Server's socket deadlines.
+use axum::{Router, body::Body, extract::ConnectInfo};
+use http_body::{Body as HttpBody, Frame, SizeHint};
+use hyper::{server::conn::http1, service::service_fn};
+use hyper_util::rt::TokioIo;
+use std::{
+    future::Future,
+    io,
+    net::SocketAddr,
+    pin::Pin,
+    sync::{Arc, Mutex},
+    task::{Context, Poll},
+    time::Duration,
+};
+use tokio::{
+    io::{AsyncRead, AsyncWrite, ReadBuf},
+    net::{TcpListener, TcpStream},
+    sync::watch,
+    task::JoinSet,
+    time::{Instant, Sleep},
+};
+use tower::ServiceExt;
+
+#[derive(Clone, Copy)]
+pub struct Timeouts {
+    pub header: Duration,
+    pub read: Duration,
+    pub write: Duration,
+    pub idle: Duration,
+    pub shutdown: Duration,
+}
+impl Default for Timeouts {
+    fn default() -> Self {
+        Self {
+            header: Duration::from_secs(5),
+            read: Duration::from_secs(15),
+            write: Duration::from_secs(30),
+            idle: Duration::from_secs(60),
+            shutdown: Duration::from_secs(10),
+        }
+    }
+}
+
+struct Deadlines {
+    started: Instant,
+    read: Option<Instant>,
+    write: Option<Instant>,
+    idle: bool,
+    response_done: bool,
+}
+struct DeadlineIo {
+    stream: TcpStream,
+    shared: Arc<Mutex<Deadlines>>,
+    timeouts: Timeouts,
+    read_timer: Pin<Box<Sleep>>,
+    write_timer: Pin<Box<Sleep>>,
+}
+fn expired(timer: &mut Pin<Box<Sleep>>, deadline: Option<Instant>, cx: &mut Context<'_>) -> bool {
+    if let Some(deadline) = deadline {
+        timer.as_mut().reset(deadline);
+        timer.as_mut().poll(cx).is_ready()
+    } else {
+        false
+    }
+}
+fn timed_out() -> io::Error {
+    io::Error::new(io::ErrorKind::TimedOut, "HTTP connection deadline exceeded")
+}
+impl AsyncRead for DeadlineIo {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        let deadline = self.shared.lock().unwrap().read;
+        if expired(&mut self.read_timer, deadline, cx) {
+            return Poll::Ready(Err(timed_out()));
+        }
+        let before = buf.filled().len();
+        let result = Pin::new(&mut self.stream).poll_read(cx, buf);
+        if matches!(result, Poll::Ready(Ok(()))) && buf.filled().len() > before {
+            let mut state = self.shared.lock().unwrap();
+            if state.idle {
+                state.idle = false;
+                state.started = Instant::now();
+                state.read = Some(state.started + self.timeouts.header);
+                cx.waker().wake_by_ref();
+            }
+        }
+        result
+    }
+}
+impl AsyncWrite for DeadlineIo {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        let deadline = self.shared.lock().unwrap().write;
+        if expired(&mut self.write_timer, deadline, cx) {
+            return Poll::Ready(Err(timed_out()));
+        }
+        Pin::new(&mut self.stream).poll_write(cx, buf)
+    }
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let deadline = self.shared.lock().unwrap().write;
+        if expired(&mut self.write_timer, deadline, cx) {
+            return Poll::Ready(Err(timed_out()));
+        }
+        let result = Pin::new(&mut self.stream).poll_flush(cx);
+        if matches!(result, Poll::Ready(Ok(()))) {
+            let mut state = self.shared.lock().unwrap();
+            if state.response_done {
+                state.write = None;
+                state.read = Some(Instant::now() + self.timeouts.idle);
+                state.idle = true;
+                state.response_done = false;
+                cx.waker().wake_by_ref();
+            }
+        }
+        result
+    }
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.stream).poll_shutdown(cx)
+    }
+}
+struct ResponseBody {
+    inner: Body,
+    shared: Arc<Mutex<Deadlines>>,
+}
+impl HttpBody for ResponseBody {
+    type Data = axum::body::Bytes;
+    type Error = axum::Error;
+    fn poll_frame(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        let frame = Pin::new(&mut self.inner).poll_frame(cx);
+        if frame.is_ready() && self.inner.is_end_stream() {
+            self.shared.lock().unwrap().response_done = true;
+        }
+        frame
+    }
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+    fn size_hint(&self) -> SizeHint {
+        self.inner.size_hint()
+    }
+}
+
+async fn connection(
+    stream: TcpStream,
+    peer: SocketAddr,
+    app: Router,
+    timeouts: Timeouts,
+    mut shutdown: watch::Receiver<bool>,
+) {
+    let now = Instant::now();
+    let state = Arc::new(Mutex::new(Deadlines {
+        started: now,
+        read: Some(now + timeouts.header),
+        write: None,
+        idle: false,
+        response_done: false,
+    }));
+    let io = DeadlineIo {
+        stream,
+        shared: Arc::clone(&state),
+        timeouts,
+        read_timer: Box::pin(tokio::time::sleep(timeouts.header)),
+        write_timer: Box::pin(tokio::time::sleep(timeouts.write)),
+    };
+    let service = service_fn(move |mut request: hyper::Request<hyper::body::Incoming>| {
+        request.extensions_mut().insert(ConnectInfo(peer));
+        let app = app.clone();
+        let state = Arc::clone(&state);
+        {
+            let mut deadlines = state.lock().unwrap();
+            if deadlines.idle {
+                deadlines.started = Instant::now();
+            }
+            deadlines.idle = false;
+            deadlines.response_done = false;
+            deadlines.read = Some(deadlines.started + timeouts.read);
+            deadlines.write = Some(Instant::now() + timeouts.write);
+        }
+        async move {
+            let response = app.oneshot(request.map(Body::new)).await?;
+            let (parts, body) = response.into_parts();
+            if body.is_end_stream() {
+                state.lock().unwrap().response_done = true;
+            }
+            Ok::<_, std::convert::Infallible>(hyper::Response::from_parts(
+                parts,
+                ResponseBody {
+                    inner: body,
+                    shared: state,
+                },
+            ))
+        }
+    });
+    let conn = http1::Builder::new().serve_connection(TokioIo::new(io), service);
+    tokio::pin!(conn);
+    tokio::select! {
+        _ = &mut conn => {},
+        _ = shutdown.changed() => { conn.as_mut().graceful_shutdown(); let _ = conn.await; }
+    }
+}
+
+/// Go's empty host listens on both IP families where IPv6 is available.
+pub async fn bind(address: &str) -> io::Result<TcpListener> {
+    if address.starts_with(':') {
+        match TcpListener::bind(format!("[::]{address}")).await {
+            Ok(listener) => Ok(listener),
+            Err(_) => TcpListener::bind(format!("0.0.0.0{address}")).await,
+        }
+    } else {
+        TcpListener::bind(address).await
+    }
+}
+
+pub async fn serve(
+    listener: TcpListener,
+    app: Router,
+    timeouts: Timeouts,
+    shutdown: impl Future<Output = ()>,
+) -> io::Result<()> {
+    let (stop, _) = watch::channel(false);
+    let mut connections = JoinSet::new();
+    tokio::pin!(shutdown);
+    loop {
+        tokio::select! {
+            biased;
+            _ = &mut shutdown => break,
+            finished = connections.join_next(), if !connections.is_empty() => { let _ = finished; },
+            accepted = listener.accept() => {
+                let (stream, peer) = accepted?;
+                stream.set_nodelay(true)?;
+                connections.spawn(connection(stream, peer, app.clone(), timeouts, stop.subscribe()));
+            }
+        }
+    }
+    drop(listener);
+    let _ = stop.send(true);
+    if tokio::time::timeout(timeouts.shutdown, async {
+        while connections.join_next().await.is_some() {}
+    })
+    .await
+    .is_err()
+    {
+        connections.abort_all();
+        while connections.join_next().await.is_some() {}
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::routing::{get, post};
+    use tokio::{
+        io::{AsyncReadExt, AsyncWriteExt},
+        sync::oneshot,
+    };
+
+    fn limits() -> Timeouts {
+        Timeouts {
+            header: Duration::from_millis(90),
+            read: Duration::from_millis(220),
+            write: Duration::from_millis(300),
+            idle: Duration::from_millis(500),
+            shutdown: Duration::from_millis(120),
+        }
+    }
+    async fn start(
+        app: Router,
+    ) -> (
+        SocketAddr,
+        oneshot::Sender<()>,
+        tokio::task::JoinHandle<io::Result<()>>,
+    ) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let (stop, stopped) = oneshot::channel();
+        let task = tokio::spawn(serve(listener, app, limits(), async {
+            let _ = stopped.await;
+        }));
+        (addr, stop, task)
+    }
+    async fn closed(stream: &mut TcpStream) {
+        let mut data = [0; 4096];
+        tokio::time::timeout(Duration::from_secs(2), async {
+            match stream.read(&mut data).await {
+                Ok(0) | Err(_) => {}
+                Ok(count) => panic!(
+                    "unexpected response before connection closed: {:?}",
+                    &data[..count]
+                ),
+            }
+        })
+        .await
+        .expect("connection deadline was not enforced");
+    }
+    async fn response(stream: &mut TcpStream) -> String {
+        let mut bytes = Vec::new();
+        while !bytes.ends_with(b"\r\n\r\n") {
+            bytes.push(
+                tokio::time::timeout(Duration::from_secs(2), stream.read_u8())
+                    .await
+                    .unwrap()
+                    .unwrap(),
+            );
+        }
+        let header = String::from_utf8(bytes).unwrap();
+        let len: usize = header
+            .lines()
+            .find_map(|l| {
+                l.to_ascii_lowercase()
+                    .strip_prefix("content-length:")
+                    .map(|s| s.trim().parse().unwrap())
+            })
+            .unwrap_or(0);
+        let mut body = vec![0; len];
+        stream.read_exact(&mut body).await.unwrap();
+        assert_eq!(body, b"ok");
+        header
+    }
+
+    #[tokio::test]
+    async fn incomplete_headers_expire_from_connection_start() {
+        let (addr, stop, task) = start(Router::new().route("/", get(|| async { "ok" }))).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream.write_all(b"GET / HTTP/1.1\r\nHost: ").await.unwrap();
+        let started = Instant::now();
+        closed(&mut stream).await;
+        assert!(started.elapsed() < Duration::from_millis(500));
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn body_deadline_includes_header_time_and_is_not_reset_by_progress() {
+        let app = Router::new().route(
+            "/",
+            post(|body: axum::body::Bytes| async move {
+                assert!(body.len() < 10);
+                "ok"
+            }),
+        );
+        let (addr, stop, task) = start(app).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        let started = Instant::now();
+        stream
+            .write_all(b"POST / HTTP/1.1\r\nHost: localhost\r\n")
+            .await
+            .unwrap();
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        stream
+            .write_all(b"Content-Length: 10\r\n\r\na")
+            .await
+            .unwrap();
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        stream.write_all(b"b").await.unwrap();
+        // Axum reports a body-read error, which may generate an HTTP error
+        // response; drain it and wait for the read deadline to close the socket.
+        let mut bytes = Vec::new();
+        tokio::time::timeout(Duration::from_secs(1), stream.read_to_end(&mut bytes))
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(started.elapsed() >= Duration::from_millis(200));
+        assert!(started.elapsed() < Duration::from_millis(400));
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn idle_deadline_is_longer_than_header_and_next_headers_get_own_deadline() {
+        let (addr, stop, task) = start(Router::new().route("/", get(|| async { "ok" }))).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        assert!(response(&mut stream).await.starts_with("HTTP/1.1 200"));
+        tokio::time::sleep(Duration::from_millis(180)).await;
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        assert!(response(&mut stream).await.starts_with("HTTP/1.1 200"));
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        stream.write_all(b"GET / HTTP/1.1\r\nHost: ").await.unwrap();
+        let started = Instant::now();
+        closed(&mut stream).await;
+        assert!(started.elapsed() < Duration::from_millis(250));
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn idle_keepalive_eventually_closes_without_another_request() {
+        let (addr, stop, task) = start(Router::new().route("/", get(|| async { "ok" }))).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        response(&mut stream).await;
+        let started = Instant::now();
+        closed(&mut stream).await;
+        assert!(started.elapsed() >= Duration::from_millis(450));
+        assert!(started.elapsed() < Duration::from_millis(800));
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn responses_cannot_be_written_after_write_deadline() {
+        let app = Router::new().route(
+            "/",
+            get(|| async {
+                tokio::time::sleep(Duration::from_millis(350)).await;
+                "ok"
+            }),
+        );
+        let (addr, stop, task) = start(app).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        closed(&mut stream).await;
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn shutdown_drains_active_requests_and_bounds_stuck_handlers() {
+        let (entered, notified) = oneshot::channel();
+        let entered = Arc::new(Mutex::new(Some(entered)));
+        let app = Router::new().route(
+            "/",
+            get(move || {
+                let entered = Arc::clone(&entered);
+                async move {
+                    if let Some(entered) = entered.lock().unwrap().take() {
+                        let _ = entered.send(());
+                    }
+                    std::future::pending::<()>().await;
+                    "ok"
+                }
+            }),
+        );
+        let (addr, stop, task) = start(app).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        notified.await.unwrap();
+        let started = Instant::now();
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+        assert!(started.elapsed() >= Duration::from_millis(100));
+        assert!(started.elapsed() < Duration::from_millis(400));
+        assert!(TcpStream::connect(addr).await.is_err());
+        closed(&mut stream).await;
+    }
+
+    #[tokio::test]
+    async fn shutdown_allows_in_flight_response_to_complete() {
+        let (entered, notified) = oneshot::channel();
+        let entered = Arc::new(Mutex::new(Some(entered)));
+        let app = Router::new().route(
+            "/",
+            get(move || {
+                let entered = Arc::clone(&entered);
+                async move {
+                    if let Some(entered) = entered.lock().unwrap().take() {
+                        let _ = entered.send(());
+                    }
+                    tokio::time::sleep(Duration::from_millis(30)).await;
+                    "ok"
+                }
+            }),
+        );
+        let (addr, stop, task) = start(app).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        notified.await.unwrap();
+        stop.send(()).unwrap();
+        assert!(response(&mut stream).await.starts_with("HTTP/1.1 200"));
+        task.await.unwrap().unwrap();
+        closed(&mut stream).await;
+    }
+    #[tokio::test]
+    async fn address_binding_supports_hostnames_ip_literals_and_empty_host() {
+        for address in ["127.0.0.1:0", "localhost:0", ":0"] {
+            let listener = bind(address).await.unwrap();
+            let local = listener.local_addr().unwrap();
+            assert_ne!(local.port(), 0);
+            if address == ":0" {
+                let client = TcpStream::connect(("127.0.0.1", local.port()))
+                    .await
+                    .unwrap();
+                let _ = listener.accept().await.unwrap();
+                drop(client);
+            }
+        }
+        // The IPv6 syntax remains accepted on hosts with an IPv6 loopback.
+        match bind("[::1]:0").await {
+            Ok(listener) => assert!(listener.local_addr().unwrap().is_ipv6()),
+            Err(error)
+                if matches!(
+                    error.kind(),
+                    io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
+                ) => {}
+            Err(error) => panic!("IPv6 listener: {error}"),
+        }
+    }
+    #[tokio::test]
+    async fn blocked_socket_writes_expire_without_waiting_for_client_reads() {
+        const LENGTH: usize = 16 * 1024 * 1024;
+        let app = Router::new().route("/", get(|| async { "x".repeat(LENGTH) }));
+        let (addr, stop, task) = start(app).await;
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        stream
+            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
+            .await
+            .unwrap();
+        // A response larger than the socket buffer must eventually block.
+        // The write deadline must wake that pending write and close the
+        // connection even though the client has not consumed any response.
+        tokio::time::sleep(Duration::from_millis(400)).await;
+        let mut received = Vec::new();
+        tokio::time::timeout(Duration::from_secs(2), stream.read_to_end(&mut received))
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(received.starts_with(b"HTTP/1.1 200"));
+        assert!(received.len() < LENGTH);
+        stop.send(()).unwrap();
+        task.await.unwrap().unwrap();
+    }
+}

src/sync.rs

@@ -0,0 +1,779 @@
+use chrono::{DateTime, Duration, NaiveDate, SecondsFormat, Utc};
+use rusqlite::{Connection, OptionalExtension, Transaction, params};
+use serde::{
+    Deserialize, Deserializer,
+    de::{self, DeserializeSeed, MapAccess, SeqAccess, Visitor},
+};
+#[cfg(test)]
+use serde_json::Value;
+use serde_json::{json, value::RawValue};
+use std::fmt;
+
+pub type SyncError = (u16, &'static str, String);
+
+#[derive(Default)]
+struct Request {
+    cursor: i64,
+    changes: Vec<Change>,
+    pull: Option<bool>,
+}
+
+#[derive(Default)]
+struct Change {
+    entity: String,
+    id: String,
+    deleted: bool,
+    updated_at: Option<DateTime<Utc>>,
+    payload: Option<Vec<u8>>,
+}
+
+// Go's decoder accepts null for ordinary scalar fields, matches keys ignoring
+// case, and lets the last duplicate field win. Keep those wire semantics.
+impl<'de> Deserialize<'de> for Request {
+    fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
+        struct RequestVisitor;
+        impl<'de> Visitor<'de> for RequestVisitor {
+            type Value = Request;
+            fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
+                f.write_str("a sync request")
+            }
+            fn visit_unit<E: de::Error>(self) -> Result<Request, E> {
+                Ok(Request::default())
+            }
+            fn visit_map<M: MapAccess<'de>>(self, mut m: M) -> Result<Request, M::Error> {
+                let mut r = Request::default();
+                while let Some(key) = m.next_key::<String>()? {
+                    match fold_key(&key).as_str() {
+                        "cursor" => {
+                            if let Some(v) = m.next_value::<Option<i64>>()? {
+                                r.cursor = v;
+                            }
+                        }
+                        "deviceid" => {
+                            let _: Option<String> = m.next_value()?;
+                        }
+                        "changes" => m.next_value_seed(ChangesSeed(&mut r.changes))?,
+                        "pull" => r.pull = m.next_value()?,
+                        _ => return Err(de::Error::custom("unknown field")),
+                    }
+                }
+                Ok(r)
+            }
+        }
+        d.deserialize_any(RequestVisitor)
+    }
+}
+
+// encoding/json reuses existing slice elements when a duplicate changes field
+// appears. Missing fields in those elements keep their preceding values.
+struct ChangesSeed<'a>(&'a mut Vec<Change>);
+impl<'de> DeserializeSeed<'de> for ChangesSeed<'_> {
+    type Value = ();
+    fn deserialize<D: Deserializer<'de>>(self, d: D) -> Result<(), D::Error> {
+        struct ChangesVisitor<'a>(&'a mut Vec<Change>);
+        impl<'de> Visitor<'de> for ChangesVisitor<'_> {
+            type Value = ();
+            fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
+                f.write_str("an array of changes")
+            }
+            fn visit_unit<E: de::Error>(self) -> Result<(), E> {
+                self.0.clear();
+                Ok(())
+            }
+            fn visit_seq<S: SeqAccess<'de>>(self, mut seq: S) -> Result<(), S::Error> {
+                let mut index = 0;
+                loop {
+                    if index == self.0.len() {
+                        self.0.push(Change::default());
+                    }
+                    if seq
+                        .next_element_seed(ChangeSeed(&mut self.0[index]))?
+                        .is_none()
+                    {
+                        break;
+                    }
+                    index += 1;
+                }
+                self.0.truncate(index);
+                Ok(())
+            }
+        }
+        d.deserialize_any(ChangesVisitor(self.0))
+    }
+}
+
+struct ChangeSeed<'a>(&'a mut Change);
+impl<'de> DeserializeSeed<'de> for ChangeSeed<'_> {
+    type Value = ();
+    fn deserialize<D: Deserializer<'de>>(self, d: D) -> Result<(), D::Error> {
+        struct ChangeVisitor<'a>(&'a mut Change);
+        impl<'de> Visitor<'de> for ChangeVisitor<'_> {
+            type Value = ();
+            fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result {
+                f.write_str("a sync change")
+            }
+            fn visit_unit<E: de::Error>(self) -> Result<(), E> {
+                Ok(())
+            }
+            fn visit_map<M: MapAccess<'de>>(self, mut m: M) -> Result<(), M::Error> {
+                let c = self.0;
+                while let Some(key) = m.next_key::<String>()? {
+                    match fold_key(&key).as_str() {
+                        "entity" => {
+                            if let Some(v) = m.next_value::<Option<String>>()? {
+                                c.entity = v;
+                            }
+                        }
+                        "id" => {
+                            if let Some(v) = m.next_value::<Option<String>>()? {
+                                c.id = v;
+                            }
+                        }
+                        "deleted" => {
+                            if let Some(v) = m.next_value::<Option<bool>>()? {
+                                c.deleted = v;
+                            }
+                        }
+                        "updatedat" => {
+                            // time.Time.UnmarshalJSON reads the quoted bytes directly;
+                            // JSON escapes inside a timestamp are not unescaped by Go.
+                            let raw = m.next_value::<Box<RawValue>>()?;
+                            if raw.get() != "null" {
+                                let value = raw
+                                    .get()
+                                    .strip_prefix('"')
+                                    .and_then(|v| v.strip_suffix('"'))
+                                    .ok_or_else(|| de::Error::custom("invalid time"))?;
+                                c.updated_at = Some(parse_time(value).map_err(de::Error::custom)?);
+                            }
+                        }
+                        "payload" => {
+                            c.payload =
+                                Some(m.next_value::<Box<RawValue>>()?.get().as_bytes().to_vec());
+                        }
+                        "version" | "sequence" => {
+                            let _: Option<i64> = m.next_value()?;
+                        }
+                        _ => return Err(de::Error::custom("unknown field")),
+                    }
+                }
+                Ok(())
+            }
+        }
+        d.deserialize_any(ChangeVisitor(self.0))
+    }
+}
+
+fn fold_key(s: &str) -> String {
+    // Unicode simple folding in encoding/json also equates long s with S.
+    s.chars()
+        .flat_map(|c| if c == 'ſ' { 's' } else { c }.to_lowercase())
+        .collect()
+}
+
+fn parse_time(s: &str) -> Result<DateTime<Utc>, &'static str> {
+    // Match Go's RFC3339 parser, including its currently enabled fallback:
+    // one-digit hours, comma fractions, offsets up to 24:60, nanosecond
+    // truncation, and rejection of leap seconds and lowercase t/z.
+    let invalid = "invalid RFC3339 timestamp";
+    if !s.is_ascii() {
+        return Err(invalid);
+    }
+    let (date, time) = s.split_once('T').ok_or(invalid)?;
+    let number = |v: &str, width: usize| -> Result<u32, &'static str> {
+        if v.len() != width || !v.bytes().all(|b| b.is_ascii_digit()) {
+            return Err(invalid);
+        }
+        v.parse().map_err(|_| invalid)
+    };
+    if date.len() != 10 {
+        return Err(invalid);
+    }
+    let mut date_parts = date.split('-');
+    let year = number(date_parts.next().ok_or(invalid)?, 4)?;
+    let month = number(date_parts.next().ok_or(invalid)?, 2)?;
+    let day = number(date_parts.next().ok_or(invalid)?, 2)?;
+    if date_parts.next().is_some() {
+        return Err(invalid);
+    }
+    let (clock, offset) = if let Some(clock) = time.strip_suffix('Z') {
+        (clock, 0i64)
+    } else {
+        if time.len() < 6 {
+            return Err(invalid);
+        }
+        let (clock, zone) = time.split_at(time.len() - 6);
+        if !matches!(zone.as_bytes()[0], b'+' | b'-') || zone.as_bytes()[3] != b':' {
+            return Err(invalid);
+        }
+        let hours = number(&zone[1..3], 2)?;
+        let minutes = number(&zone[4..6], 2)?;
+        if hours > 24 || minutes > 60 {
+            return Err(invalid);
+        }
+        let sign = if zone.starts_with('-') { -1 } else { 1 };
+        (clock, sign * i64::from(hours * 3600 + minutes * 60))
+    };
+    let (clock, nanos) = if let Some(index) = clock.find(['.', ',']) {
+        let fraction = &clock[index + 1..];
+        if fraction.is_empty() || !fraction.bytes().all(|b| b.is_ascii_digit()) {
+            return Err(invalid);
+        }
+        let precision = fraction.len().min(9);
+        let nanos = number(&fraction[..precision], precision)? * 10u32.pow((9 - precision) as u32);
+        (&clock[..index], nanos)
+    } else {
+        (clock, 0)
+    };
+    let mut parts = clock.split(':');
+    let hour = parts.next().ok_or(invalid)?;
+    if !(1..=2).contains(&hour.len()) {
+        return Err(invalid);
+    }
+    let hour = number(hour, hour.len())?;
+    let minute = number(parts.next().ok_or(invalid)?, 2)?;
+    let second = number(parts.next().ok_or(invalid)?, 2)?;
+    if parts.next().is_some() || second > 59 {
+        return Err(invalid);
+    }
+    let local = NaiveDate::from_ymd_opt(year as i32, month, day)
+        .and_then(|d| d.and_hms_nano_opt(hour, minute, second, nanos))
+        .ok_or(invalid)?;
+    local
+        .and_utc()
+        .checked_sub_signed(Duration::seconds(offset))
+        .ok_or(invalid)
+}
+
+fn timestamp(t: DateTime<Utc>) -> String {
+    // SecondsFormat::AutoSi pads to 3/6/9 digits, whereas Go strips every zero.
+    let s = t.to_rfc3339_opts(SecondsFormat::Nanos, true);
+    let s = s
+        .trim_end_matches('Z')
+        .trim_end_matches('0')
+        .trim_end_matches('.');
+    format!("{s}Z")
+}
+
+fn failure(status: u16, code: &'static str, message: &str) -> SyncError {
+    (status, code, message.to_owned())
+}
+
+pub fn synchronize_wire(
+    db: &mut Connection,
+    user_id: &str,
+    body: &[u8],
+) -> Result<Vec<u8>, SyncError> {
+    let invalid_json = || failure(400, "invalid_json", "Request body is not valid JSON");
+    let body = &body[..body.len().min(2 << 20)];
+    // The original decoder reads the first JSON value, without requiring EOF.
+    let (masked, raw_payloads) =
+        crate::go_json::extract_raw_fields(body, "payload").map_err(|_| invalid_json())?;
+    let normalized = crate::go_json::normalize(&masked);
+    let mut request = Request::deserialize(&mut serde_json::Deserializer::from_slice(&normalized))
+        .map_err(|_| invalid_json())?;
+    for change in &mut request.changes {
+        if let Some(placeholder) = &change.payload {
+            let index: String = serde_json::from_slice(placeholder).map_err(|_| invalid_json())?;
+            let index: usize = index.parse().map_err(|_| invalid_json())?;
+            change.payload = Some(raw_payloads.get(index).ok_or_else(invalid_json)?.clone());
+        }
+    }
+    if request.cursor < 0 || request.changes.len() > 5000 {
+        return Err(failure(
+            400,
+            "invalid_sync_request",
+            "Cursor or change count is invalid",
+        ));
+    }
+    for c in &request.changes {
+        let message = if !matches!(c.entity.as_str(), "transaction" | "category" | "budget") {
+            Some("entity must be transaction, category, or budget")
+        } else if c.id.is_empty() || c.id.len() > 128 {
+            Some("record id is invalid")
+        } else if c.updated_at.is_none() || c.updated_at == parse_time("0001-01-01T00:00:00Z").ok()
+        {
+            Some("updatedAt is required")
+        } else if c
+            .updated_at
+            .is_some_and(|time| time > Utc::now() + Duration::hours(24))
+        {
+            Some("updatedAt is too far in the future")
+        } else if !c.deleted && c.payload.is_none() {
+            Some("payload is required for active records")
+        } else if c.payload.as_ref().is_some_and(|p| p.len() > 256 * 1024) {
+            Some("payload is too large")
+        } else {
+            None
+        };
+        if let Some(message) = message {
+            return Err(failure(400, "invalid_change", message));
+        }
+    }
+    let tx = db
+        .transaction()
+        .map_err(|_| failure(500, "sync_failed", "Could not begin synchronization"))?;
+    for c in request.changes {
+        apply(&tx, user_id, c)
+            .map_err(|_| failure(500, "sync_failed", "Could not apply changes"))?;
+    }
+    let (changes, cursor, more) = if request.pull.unwrap_or(true) {
+        load(&tx, user_id, request.cursor)
+            .map_err(|_| failure(500, "sync_failed", "Could not load changes"))?
+    } else {
+        (Vec::new(), request.cursor, false)
+    };
+    tx.commit()
+        .map_err(|_| failure(500, "sync_failed", "Could not complete synchronization"))?;
+    let mut response = format!("{{\"cursor\":{cursor},\"changes\":[").into_bytes();
+    for (index, change) in changes.iter().enumerate() {
+        if index != 0 {
+            response.push(b',');
+        }
+        response.extend_from_slice(change);
+    }
+    response.extend_from_slice(
+        format!(
+            "],\"hasMore\":{more},\"serverTime\":{}}}\n",
+            serde_json::to_string(&timestamp(Utc::now())).expect("timestamp JSON")
+        )
+        .as_bytes(),
+    );
+    Ok(response)
+}
+
+#[cfg(test)]
+fn synchronize(db: &mut Connection, user_id: &str, body: &[u8]) -> Result<Value, SyncError> {
+    let response = synchronize_wire(db, user_id, body)?;
+    serde_json::from_slice(&crate::go_json::normalize(&response))
+        .map_err(|_| failure(500, "sync_failed", "Could not serialize synchronization"))
+}
+
+type InternalError = Box<dyn std::error::Error + Send + Sync>;
+
+fn apply(tx: &Transaction<'_>, user: &str, c: Change) -> Result<(), InternalError> {
+    let current: Option<(String, i64)> = tx
+        .query_row(
+            "SELECT updated_at,version FROM records WHERE user_id=? AND entity=? AND record_id=?",
+            params![user, c.entity, c.id],
+            |r| Ok((r.get(0)?, r.get(1)?)),
+        )
+        .optional()?;
+    let updated = c.updated_at.expect("validated timestamp");
+    let mut version = 1;
+    if let Some((time, old_version)) = current {
+        if updated <= parse_time(&time)? {
+            return Ok(());
+        }
+        version = old_version.wrapping_add(1);
+    }
+    let payload = if c.deleted {
+        None
+    } else {
+        c.payload.as_deref()
+    };
+    let time = timestamp(updated);
+    tx.execute("INSERT INTO records(user_id,entity,record_id,payload,updated_at,deleted,version) VALUES(?,?,?,CAST(? AS TEXT),?,?,?) ON CONFLICT(user_id,entity,record_id) DO UPDATE SET payload=excluded.payload,updated_at=excluded.updated_at,deleted=excluded.deleted,version=excluded.version", params![user,c.entity,c.id,payload,time,c.deleted,version])?;
+    tx.execute("INSERT INTO changes(user_id,entity,record_id,payload,updated_at,deleted,version) VALUES(?,?,?,CAST(? AS TEXT),?,?,?)", params![user,c.entity,c.id,payload,time,c.deleted,version])?;
+    Ok(())
+}
+
+fn load(
+    tx: &Transaction<'_>,
+    user: &str,
+    after: i64,
+) -> Result<(Vec<Vec<u8>>, i64, bool), InternalError> {
+    let mut stmt = tx.prepare("SELECT sequence,entity,record_id,payload,updated_at,deleted,version FROM changes WHERE user_id=? AND sequence>? ORDER BY sequence ASC LIMIT ?")?;
+    let mut rows = stmt.query(params![user, after, 501])?;
+    let mut result = Vec::new();
+    let mut cursor = after;
+    let mut more = false;
+    while let Some(r) = rows.next()? {
+        if result.len() == 500 {
+            more = true;
+            break;
+        }
+        let sequence: i64 = r.get(0)?;
+        let payload: Option<Vec<u8>> = match r.get_ref(3)? {
+            rusqlite::types::ValueRef::Null => None,
+            rusqlite::types::ValueRef::Text(bytes) | rusqlite::types::ValueRef::Blob(bytes) => {
+                Some(bytes.to_vec())
+            }
+            _ => return Err("invalid database payload".into()),
+        };
+        let time: String = r.get(4)?;
+        let version: i64 = r.get(6)?;
+        let mut item = json!({"entity": r.get::<_, String>(1)?, "id": r.get::<_, String>(2)?, "updatedAt": timestamp(parse_time(&time)?), "deleted": r.get::<_, bool>(5)?});
+        if sequence != 0 {
+            item["sequence"] = json!(sequence);
+        }
+        if version != 0 {
+            item["version"] = json!(version);
+        }
+        let mut encoded = serde_json::to_vec(&item)?;
+        if let Some(p) = payload {
+            serde_json::from_slice::<Box<RawValue>>(&crate::go_json::normalize(&p))?;
+            encoded.pop();
+            encoded.extend_from_slice(b",\"payload\":");
+            encoded.extend_from_slice(&crate::go_json::compact_raw(&p));
+            encoded.push(b'}');
+        }
+        result.push(encoded);
+        cursor = sequence;
+    }
+    Ok((result, cursor, more))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    fn db() -> Connection {
+        let db = crate::database::open(":memory:").unwrap();
+        db.execute(
+            "INSERT INTO users VALUES('u','u@example.com','hash','now')",
+            [],
+        )
+        .unwrap();
+        db.execute(
+            "INSERT INTO users VALUES('v','v@example.com','hash','now')",
+            [],
+        )
+        .unwrap();
+        db
+    }
+    fn change(id: &str, time: &str) -> Value {
+        json!({"entity":"category","id":id,"updatedAt":time,"payload":{"name":"Food"}})
+    }
+    fn sync(db: &mut Connection, body: Value) -> Result<Value, SyncError> {
+        synchronize(db, "u", &serde_json::to_vec(&body).unwrap())
+    }
+
+    #[test]
+    fn last_write_wins_and_replay_does_not_advance_sequence() {
+        let mut db = db();
+        let original = change("c", "2025-01-01T01:00:00+01:00");
+        let first = sync(&mut db, json!({"changes":[original.clone()]})).unwrap();
+        assert_eq!(first["cursor"], 1);
+        assert_eq!(first["changes"][0]["updatedAt"], "2025-01-01T00:00:00Z");
+        for c in [original, change("c", "2024-01-01T00:00:00Z")] {
+            let r = sync(&mut db, json!({"cursor":1,"changes":[c]})).unwrap();
+            assert_eq!(r["cursor"], 1);
+            assert_eq!(r["changes"], json!([]));
+        }
+        let deleted = json!({"entity":"category","id":"c","updatedAt":"2025-02-01T00:00:00.123400000Z","deleted":true,"payload":{"ignored":true}});
+        let r = sync(&mut db, json!({"cursor":1,"changes":[deleted]})).unwrap();
+        assert_eq!(r["changes"][0]["version"], 2);
+        assert_eq!(r["changes"][0]["deleted"], true);
+        assert!(r["changes"][0].get("payload").is_none());
+        assert_eq!(r["changes"][0]["updatedAt"], "2025-02-01T00:00:00.1234Z");
+    }
+
+    #[test]
+    fn pagination_and_user_isolation_preserve_global_sequence() {
+        let mut db = db();
+        let changes: Vec<Value> = (0..501)
+            .map(|i| change(&i.to_string(), "2025-01-01T00:00:00Z"))
+            .collect();
+        let r = sync(&mut db, json!({"changes":changes,"pull":false,"cursor":7})).unwrap();
+        assert_eq!(r["cursor"], 7);
+        assert_eq!(r["changes"], json!([]));
+        let r = sync(&mut db, json!({})).unwrap();
+        assert_eq!(r["changes"].as_array().unwrap().len(), 500);
+        assert_eq!(r["cursor"], 500);
+        assert_eq!(r["hasMore"], true);
+        let r = sync(&mut db, json!({"cursor":500})).unwrap();
+        assert_eq!(r["changes"].as_array().unwrap().len(), 1);
+        assert_eq!(r["cursor"], 501);
+        assert_eq!(r["hasMore"], false);
+        let r = synchronize(&mut db, "v", b"{}").unwrap();
+        assert_eq!(r["changes"], json!([]));
+        assert_eq!(r["cursor"], 0);
+    }
+
+    #[test]
+    fn all_pushes_roll_back_on_apply_or_pull_failure() {
+        let mut db = db();
+        db.execute_batch("CREATE TRIGGER reject_bad BEFORE INSERT ON records WHEN NEW.record_id='bad' BEGIN SELECT RAISE(ABORT, 'reject'); END;").unwrap();
+        let r = sync(
+            &mut db,
+            json!({"changes":[change("good","2025-01-01T00:00:00Z"),change("bad","2025-01-01T00:00:00Z")]}),
+        );
+        assert_eq!(r.unwrap_err().2, "Could not apply changes");
+        assert_eq!(
+            db.query_row("SELECT count(*) FROM records", [], |r| r.get::<_, i64>(0))
+                .unwrap(),
+            0
+        );
+        db.execute("INSERT INTO changes(user_id,entity,record_id,updated_at,deleted,version) VALUES('u','category','broken','invalid-time',1,1)", []).unwrap();
+        assert_eq!(
+            sync(
+                &mut db,
+                json!({"changes":[change("good","2025-01-01T00:00:00Z")]})
+            )
+            .unwrap_err()
+            .2,
+            "Could not load changes"
+        );
+        assert_eq!(
+            db.query_row("SELECT count(*) FROM records", [], |r| r.get::<_, i64>(0))
+                .unwrap(),
+            0
+        );
+    }
+
+    #[test]
+    fn strict_fields_and_go_null_defaults_and_validation() {
+        let mut db = db();
+        for raw in [
+            b"null".as_slice(),
+            b"{\"cursor\":null,\"deviceId\":null,\"changes\":null,\"pull\":null}",
+            b"{} {}",
+        ] {
+            assert!(synchronize(&mut db, "u", raw).is_ok());
+        }
+        for raw in [
+            b"{\"extra\":1}".as_slice(),
+            b"{\"changes\":[{\"extra\":1}]}",
+            b"{\"cursor\":1.0}",
+            b"{\"changes\":{}}",
+        ] {
+            assert_eq!(
+                synchronize(&mut db, "u", raw).unwrap_err().1,
+                "invalid_json"
+            );
+        }
+        let raw = br#"{"CHANGES":[{"ENTITY":"category","ID":"c","updatedAt":"2025-01-01T00:00:00Z","payload":null}]}"#;
+        let r = synchronize(&mut db, "u", raw).unwrap();
+        assert!(r["changes"][0].get("payload").unwrap().is_null());
+        assert_eq!(
+            sync(&mut db, json!({"cursor":-1})).unwrap_err().1,
+            "invalid_sync_request"
+        );
+        assert_eq!(
+            sync(&mut db, json!({"changes":[null]})).unwrap_err().2,
+            "entity must be transaction, category, or budget"
+        );
+        let mut c = change("c", "0001-01-01T00:00:00Z");
+        assert_eq!(
+            sync(&mut db, json!({"changes":[c.clone()]})).unwrap_err().2,
+            "updatedAt is required"
+        );
+        c["updatedAt"] = json!(timestamp(Utc::now() + Duration::hours(25)));
+        assert_eq!(
+            sync(&mut db, json!({"changes":[c]})).unwrap_err().2,
+            "updatedAt is too far in the future"
+        );
+        let mut c = change("c", "2025-01-01T00:00:00Z");
+        c["payload"] = json!("x".repeat(256 * 1024));
+        assert_eq!(
+            sync(&mut db, json!({"changes":[c]})).unwrap_err().2,
+            "payload is too large"
+        );
+    }
+    #[test]
+    fn go_timestamp_parser_accepts_fallbacks_and_rejects_chrono_extensions() {
+        for (input, expected) in [
+            ("2025-01-02T3:04:05Z", "2025-01-02T03:04:05Z"),
+            (
+                "2025-01-02T03:04:05,123456789123Z",
+                "2025-01-02T03:04:05.123456789Z",
+            ),
+            ("2025-01-02T03:04:05+24:60", "2025-01-01T02:04:05Z"),
+            ("0000-01-01T00:00:00Z", "0000-01-01T00:00:00Z"),
+        ] {
+            assert_eq!(timestamp(parse_time(input).unwrap()), expected, "{input}");
+        }
+        for input in [
+            "2025-01-02t03:04:05Z",
+            "2025-01-02T03:04:05z",
+            "2025-01-02T03:04:60Z",
+            "2025-01-02 03:04:05Z",
+            "2025-01-02T03:04:05+25:00",
+            "2025-01-02T03:04:05+01:61",
+            "2025-01-02T03:04:05.Z",
+            "2025-02-29T00:00:00Z",
+        ] {
+            assert!(parse_time(input).is_err(), "{input}");
+        }
+        let mut db = db();
+        let raw = br#"{"changes":[{"entity":"category","id":"c","updatedAt":"\u0032025-01-01T00:00:00Z","payload":null}]}"#;
+        assert_eq!(
+            synchronize(&mut db, "u", raw).unwrap_err().1,
+            "invalid_json"
+        );
+    }
+
+    #[test]
+    fn duplicate_fields_reuse_go_slice_elements_and_null_keeps_scalars() {
+        let mut db = db();
+        let raw = br#"{"changes":[{"entity":"category","id":"first","updatedAt":"2025-01-01T00:00:00Z","payload":null}],"changes":[{"id":"second","entity":null,"updatedAt":null}]}"#;
+        let result = synchronize(&mut db, "u", raw).unwrap();
+        assert_eq!(result["changes"][0]["id"], "second");
+        assert_eq!(result["changes"][0]["entity"], "category");
+        let raw = r#"{"curſor":1,"CHANGES":[{"entity":"category","id":"third","updatedAt":"2025-01-02T00:00:00Z","payload":true}]}"#;
+        let result = synchronize(&mut db, "u", raw.as_bytes()).unwrap();
+        assert_eq!(result["changes"][0]["id"], "third");
+        assert_eq!(result["changes"][0]["payload"], true);
+    }
+
+    #[test]
+    fn exact_payload_numbers_and_database_raw_text_are_preserved() {
+        let mut db = db();
+        let payload = r#"{ "large": 123456789012345678901234567890, "decimal": 0.123456789012345678901234567890, "exponent": 1e999 }"#;
+        let body = format!(
+            r#"{{"changes":[{{"entity":"transaction","id":"t","updatedAt":"2025-01-01T00:00:00Z","payload":{payload}}}]}}"#
+        );
+        let result = synchronize(&mut db, "u", body.as_bytes()).unwrap();
+        let stored: String = db
+            .query_row("SELECT payload FROM records", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(stored, payload);
+        let serialized = serde_json::to_string(&result["changes"][0]["payload"]).unwrap();
+        assert!(serialized.contains("123456789012345678901234567890"));
+        assert!(serialized.contains("0.123456789012345678901234567890"));
+        assert!(serialized.to_lowercase().contains("1e+999") || serialized.contains("1e999"));
+    }
+
+    #[test]
+    fn validation_limits_use_bytes_and_validate_whole_batch_before_writing() {
+        let mut db = db();
+        let mut invalid = change(&"я".repeat(65), "2025-01-01T00:00:00Z");
+        assert_eq!(
+            sync(
+                &mut db,
+                json!({"changes":[change("good","2025-01-01T00:00:00Z"),invalid.clone()]})
+            )
+            .unwrap_err()
+            .2,
+            "record id is invalid"
+        );
+        assert_eq!(
+            db.query_row("SELECT count(*) FROM records", [], |r| r.get::<_, i64>(0))
+                .unwrap(),
+            0
+        );
+        invalid["id"] = json!("я".repeat(64));
+        assert!(sync(&mut db, json!({"changes":[invalid]})).is_ok());
+        let changes: Vec<_> = (0..5001)
+            .map(|_| change("c", "2025-01-01T00:00:00Z"))
+            .collect();
+        assert_eq!(
+            sync(&mut db, json!({"changes":changes})).unwrap_err().1,
+            "invalid_sync_request"
+        );
+        let raw = format!(
+            r#"{{"changes":[{{"entity":"category","id":"large","updatedAt":"2025-01-01T00:00:00Z","payload":{{ {} }} }}]}}"#,
+            " ".repeat(256 * 1024)
+        );
+        assert_eq!(
+            synchronize(&mut db, "u", raw.as_bytes()).unwrap_err().2,
+            "payload is too large"
+        );
+        let mut trailing = b"{} ".to_vec();
+        trailing.resize((2 << 20) + 100, b' ');
+        assert!(synchronize(&mut db, "u", &trailing).is_ok());
+    }
+
+    #[test]
+    fn wire_preserves_raw_payload_duplicates_unicode_escapes_and_invalid_utf8() {
+        let mut db = db();
+        let payload = b"{ \"duplicate\":1, \"duplicate\":2, \"surrogate\":\"\\ud800\", \"invalid\":\"\xff\", \"html\":\"<&>\" }";
+        let mut body = br#"{"changes":[{"entity":"category","id":"\ud800","updatedAt":"2025-01-01T00:00:00Z","payload":"#.to_vec();
+        body.extend_from_slice(payload);
+        body.extend_from_slice(b"}]}");
+        let response = synchronize_wire(&mut db, "u", &body).unwrap();
+        let id = "\"id\":\"�\"".as_bytes();
+        assert!(response.windows(id.len()).any(|w| w == id));
+        let expected = b"\"payload\":{\"duplicate\":1,\"duplicate\":2,\"surrogate\":\"\\ud800\",\"invalid\":\"\xff\",\"html\":\"\\u003c\\u0026\\u003e\"}";
+        assert!(response.windows(expected.len()).any(|w| w == expected));
+        let stored: Vec<u8> = db
+            .query_row("SELECT CAST(payload AS BLOB) FROM records", [], |r| {
+                r.get(0)
+            })
+            .unwrap();
+        assert_eq!(stored, payload);
+        let storage_type: String = db
+            .query_row("SELECT typeof(payload) FROM records", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(storage_type, "text");
+        let pulled = synchronize_wire(&mut db, "u", b"{}").unwrap();
+        assert!(pulled.windows(expected.len()).any(|w| w == expected));
+    }
+
+    #[test]
+    fn raw_payload_limits_count_original_bytes_before_unicode_replacement() {
+        let mut db = db();
+        let mut body = br#"{"changes":[{"entity":"category","id":"c","updatedAt":"2025-01-01T00:00:00Z","payload":""#.to_vec();
+        body.extend(std::iter::repeat_n(0xff, 256 * 1024 - 2));
+        body.extend_from_slice(b"\"}]}");
+        assert!(synchronize_wire(&mut db, "u", &body).is_ok());
+        let body = b"{\"changes\":[{\"entity\":\"category\",\"id\":\"\xff\",\"updatedAt\":\"2025-01-02T00:00:00Z\",\"payload\":null}]}";
+        assert!(synchronize_wire(&mut db, "u", body).is_ok());
+    }
+
+    #[test]
+    fn tombstone_resurrection_and_sequential_batch_conflicts_keep_history() {
+        let mut db = db();
+        let initial = change("c", "2025-01-01T00:00:00Z");
+        let tombstone =
+            json!({"entity":"category","id":"c","updatedAt":"2025-01-02T00:00:00Z","deleted":true});
+        let newer = change("c", "2025-01-03T00:00:00Z");
+        let result = sync(
+            &mut db,
+            json!({"changes":[initial.clone(),tombstone.clone(),initial,newer.clone(),newer]}),
+        )
+        .unwrap();
+        let changes = result["changes"].as_array().unwrap();
+        assert_eq!(changes.len(), 3);
+        assert_eq!(changes[0]["version"], 1);
+        assert_eq!(changes[1]["version"], 2);
+        assert_eq!(changes[2]["version"], 3);
+        assert_eq!(changes[2]["deleted"], false);
+        assert!(changes[1].get("payload").is_none());
+        let isolated = synchronize(
+            &mut db,
+            "v",
+            &serde_json::to_vec(&json!({"changes":[tombstone]})).unwrap(),
+        )
+        .unwrap();
+        assert_eq!(isolated["cursor"], 4);
+        let user = sync(&mut db, json!({"cursor":3})).unwrap();
+        assert_eq!(user["cursor"], 3);
+        assert_eq!(user["changes"], json!([]));
+    }
+
+    #[test]
+    fn concurrent_device_retries_have_one_winner_and_unique_sequences() {
+        use std::sync::{Arc, Barrier, Mutex};
+        let db = Arc::new(Mutex::new(db()));
+        let barrier = Arc::new(Barrier::new(8));
+        let mut handles = Vec::new();
+        for index in 0..8 {
+            let db = Arc::clone(&db);
+            let barrier = Arc::clone(&barrier);
+            handles.push(std::thread::spawn(move || {
+                barrier.wait();
+                let c = change("shared", &format!("2025-01-01T00:00:0{index}Z"));
+                let mut db = db.lock().unwrap();
+                sync(&mut db, json!({"changes":[c.clone(), c],"pull":false})).unwrap();
+            }));
+        }
+        for handle in handles {
+            handle.join().unwrap();
+        }
+        let mut db = db.lock().unwrap();
+        let (time, version): (String, i64) = db
+            .query_row("SELECT updated_at,version FROM records", [], |r| {
+                Ok((r.get(0)?, r.get(1)?))
+            })
+            .unwrap();
+        assert_eq!(time, "2025-01-01T00:00:07Z");
+        let result = sync(&mut db, json!({})).unwrap();
+        let changes = result["changes"].as_array().unwrap();
+        assert_eq!(changes.len() as i64, version);
+        for (index, change) in changes.iter().enumerate() {
+            assert_eq!(change["sequence"], (index + 1) as i64);
+            assert_eq!(change["version"], (index + 1) as i64);
+        }
+    }
+}