jacobh2
15056dc7df
- src/sync/: pull with UNION ALL paging and cursor_ahead; push processing of SPEC §14.4.4 steps 1 to 12 in one BEGIN IMMEDIATE with a savepoint per mutation, LWW with delete-wins, tombstones, clock_skew, replay from sync_mutations, sync_batch for more than 3 notifiable changes; 90-day pruning. - src/audit/: SqlAuditWriter and the §14.4.7 diff (now the default in AppState::new). - src/push/: bounded PushHub (100, drop oldest), FCM HTTP v1 client with service-account JWT (RS256, jsonwebtoken on aws-lc-rs) and token cache, sender with 1/5/25 s retries and token clearing; disabled without credentials. - src/tasks/: start installs push and spawns the 06:00 generation (and at start) and 04:00 pruning loops on the injectable clock. - cli: rotate-epoch; serve uses into_make_service_with_connect_info; stale doc comments updated; /i/ empty code gets the HTML 404. - tests: sync_api, push_fcm (local stub of Google's endpoints), scheduler_daily; skeleton routes test now asserts no route is 501. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TQd3pEP9UeN2uifbHknQ7B
89 lines
3.4 KiB
Rust
89 lines
3.4 KiB
Rust
//! Sync: E-31 pull and E-32 push (SPEC §14.4). The protocol itself is in `crate::sync`.
|
|
|
|
use std::collections::HashMap;
|
|
|
|
use axum::extract::{Query, State};
|
|
use serde::Deserialize;
|
|
use serde_json::Value;
|
|
|
|
use crate::AppState;
|
|
use crate::api::{ApiJson, JsonBody};
|
|
use crate::auth::AuthDevice;
|
|
use crate::error::AppError;
|
|
use crate::openapi::responses::{BadRequest, Conflict, PayloadTooLarge, Unprocessable};
|
|
use crate::sync;
|
|
use crate::wire::{SyncLimitSchema, SyncPullResponse, SyncPushRequest, SyncPushResponse};
|
|
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/sync",
|
|
operation_id = "syncPull",
|
|
summary = "E-31 Pull changes since a cursor",
|
|
description = "SPEC §14.4.2.",
|
|
params(
|
|
("since" = i64, Query, description = "The highest `server_version` the phone has applied (0 initially)."),
|
|
("limit" = Option<SyncLimitSchema>, Query, description = "Rows per page, 1 to 1000 (`invalid_request` otherwise)."),
|
|
),
|
|
responses(
|
|
(status = 200, description = "One page. Every array is present (possibly empty) and sorted by `server_version`.", body = SyncPullResponse, content_type = "application/json; charset=utf-8"),
|
|
(status = 400, response = BadRequest),
|
|
(status = 409, response = Conflict),
|
|
)
|
|
)]
|
|
pub async fn sync_pull(
|
|
_device: AuthDevice,
|
|
State(state): State<AppState>,
|
|
Query(query): Query<HashMap<String, String>>,
|
|
) -> Result<JsonBody<SyncPullResponse>, AppError> {
|
|
let int = |key: &str| -> Result<Option<i64>, AppError> {
|
|
query
|
|
.get(key)
|
|
.map(|v| {
|
|
v.parse::<i64>().map_err(|_| {
|
|
AppError::invalid_request(format!("{key} must be an integer")).with_field(key)
|
|
})
|
|
})
|
|
.transpose()
|
|
};
|
|
let since = int("since")?
|
|
.ok_or_else(|| AppError::invalid_request("since is required").with_field("since"))?;
|
|
if since < 0 {
|
|
return Err(AppError::invalid_request("since must be 0 or more").with_field("since"));
|
|
}
|
|
let limit = sync::pull::page_size(int("limit")?)?;
|
|
Ok(JsonBody(sync::pull::pull(&state, since, limit).await?))
|
|
}
|
|
|
|
/// The E-32 body as the handler reads it: the mutations stay raw JSON so that a malformed mutation is rejected on
|
|
/// its own (`validation_failed`, SPEC §14.4.4 step 3) instead of failing the request. The schema in the OpenAPI
|
|
/// document is [`SyncPushRequest`].
|
|
#[derive(Debug, Deserialize)]
|
|
pub struct RawPushRequest {
|
|
pub mutations: Vec<Value>,
|
|
}
|
|
|
|
#[utoipa::path(
|
|
post,
|
|
path = "/sync",
|
|
operation_id = "syncPush",
|
|
summary = "E-32 Push local expense and settlement changes",
|
|
description = "SPEC §14.4.3 and §14.4.4. 1 to 100 mutations, body at most 1 MiB. One result per mutation, in request order.",
|
|
request_body(content = SyncPushRequest, content_type = "application/json; charset=utf-8"),
|
|
responses(
|
|
(status = 200, description = "Per-mutation results.", body = SyncPushResponse, content_type = "application/json; charset=utf-8"),
|
|
(status = 400, response = BadRequest),
|
|
(status = 409, response = Conflict),
|
|
(status = 413, response = PayloadTooLarge),
|
|
(status = 422, response = Unprocessable),
|
|
)
|
|
)]
|
|
pub async fn sync_push(
|
|
device: AuthDevice,
|
|
State(state): State<AppState>,
|
|
ApiJson(body): ApiJson<RawPushRequest>,
|
|
) -> Result<JsonBody<sync::push::PushResponse>, AppError> {
|
|
Ok(JsonBody(
|
|
sync::push::push(&state, &device, body.mutations).await?,
|
|
))
|
|
}
|