feat(sync): add typed flat patch boundary
req: sync/002 req: sync/003
This commit is contained in:
slhx agent
@@ -0,0 +1,11 @@
|
||||
[package]
|
||||
name = "hemx-sync"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
|
||||
[dependencies]
|
||||
hemx-core = { path = "../hemx-core" }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
|
||||
[dev-dependencies]
|
||||
serde_json = "1"
|
||||
@@ -0,0 +1,140 @@
|
||||
const DATABASE = "hemx-sync-v1";
|
||||
const STORE = "patches";
|
||||
const SCHEMA_VERSION = 1;
|
||||
const EVENT = "hemx:sync-patch";
|
||||
const root = document.querySelector("[data-hemx-root]");
|
||||
let database;
|
||||
let pumping = false;
|
||||
|
||||
function requestResult(request) {
|
||||
return new Promise((resolve, reject) => {
|
||||
request.addEventListener("success", () => resolve(request.result), { once: true });
|
||||
request.addEventListener("error", () => reject(request.error), { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
function transactionDone(transaction) {
|
||||
return new Promise((resolve, reject) => {
|
||||
transaction.addEventListener("complete", resolve, { once: true });
|
||||
transaction.addEventListener("abort", () => reject(transaction.error), { once: true });
|
||||
transaction.addEventListener("error", () => reject(transaction.error), { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
async function openDatabase() {
|
||||
const request = indexedDB.open(DATABASE, 1);
|
||||
request.addEventListener("upgradeneeded", () => {
|
||||
if (!request.result.objectStoreNames.contains(STORE)) {
|
||||
request.result.createObjectStore(STORE, { keyPath: "idempotencyKey" });
|
||||
}
|
||||
});
|
||||
return requestResult(request);
|
||||
}
|
||||
|
||||
function validIdentifier(value) {
|
||||
return typeof value === "string" && value.length > 0 && value.length <= 128 && /^[A-Za-z0-9:_.-]+$/.test(value);
|
||||
}
|
||||
|
||||
function validKey(value) {
|
||||
return typeof value === "string"
|
||||
&& value.length > 0
|
||||
&& value.length <= 64
|
||||
&& /^[A-Za-z][A-Za-z0-9_-]*$/.test(value)
|
||||
&& !["schemaVersion", "idempotencyKey", "operationId", "key", "value"].includes(value);
|
||||
}
|
||||
|
||||
function validatePatch(patch) {
|
||||
if (!patch || Object.getPrototypeOf(patch) !== Object.prototype) throw new Error("patch must be an object");
|
||||
const keys = Object.keys(patch).sort();
|
||||
const expected = ["idempotencyKey", "key", "operationId", "schemaVersion", "value"];
|
||||
if (keys.length !== expected.length || keys.some((key, index) => key !== expected[index])) {
|
||||
throw new Error("patch fields do not match schema");
|
||||
}
|
||||
if (patch.schemaVersion !== SCHEMA_VERSION) throw new Error(`unsupported patch schema version ${patch.schemaVersion}`);
|
||||
if (!validIdentifier(patch.idempotencyKey)) throw new Error("invalid idempotencyKey");
|
||||
if (!validIdentifier(patch.operationId)) throw new Error("invalid operationId");
|
||||
if (!validKey(patch.key)) throw new Error("invalid patch key");
|
||||
if (!["string", "number", "boolean"].includes(typeof patch.value)
|
||||
|| (typeof patch.value === "number" && !Number.isSafeInteger(patch.value))
|
||||
|| (typeof patch.value === "string" && patch.value.length > 4096)) {
|
||||
throw new Error("invalid patch value");
|
||||
}
|
||||
return patch;
|
||||
}
|
||||
|
||||
async function allPatches() {
|
||||
const transaction = database.transaction(STORE, "readonly");
|
||||
const done = transactionDone(transaction);
|
||||
const patches = await requestResult(transaction.objectStore(STORE).getAll());
|
||||
await done;
|
||||
return patches.sort((left, right) => left.queuedAt - right.queuedAt || left.idempotencyKey.localeCompare(right.idempotencyKey));
|
||||
}
|
||||
|
||||
async function persist(patch) {
|
||||
const transaction = database.transaction(STORE, "readwrite");
|
||||
const done = transactionDone(transaction);
|
||||
transaction.objectStore(STORE).add({ ...patch, queuedAt: Date.now() });
|
||||
await done;
|
||||
root?.setAttribute("data-hemx-sync-pending", String((await allPatches()).length));
|
||||
}
|
||||
|
||||
async function remove(idempotencyKey) {
|
||||
const transaction = database.transaction(STORE, "readwrite");
|
||||
const done = transactionDone(transaction);
|
||||
transaction.objectStore(STORE).delete(idempotencyKey);
|
||||
await done;
|
||||
}
|
||||
|
||||
async function pump() {
|
||||
if (pumping || !navigator.onLine) return;
|
||||
pumping = true;
|
||||
try {
|
||||
for (const stored of await allPatches()) {
|
||||
const { queuedAt: _queuedAt, ...patch } = stored;
|
||||
const endpoint = root?.getAttribute("data-sync-endpoint") || "/sync/patches";
|
||||
const response = await fetch(endpoint, {
|
||||
method: "POST",
|
||||
credentials: "same-origin",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(patch),
|
||||
});
|
||||
if (!response.ok) {
|
||||
root?.setAttribute("data-hemx-sync-error", `upload-${response.status}`);
|
||||
return;
|
||||
}
|
||||
const acknowledgement = await response.json();
|
||||
if (acknowledgement.idempotencyKey !== patch.idempotencyKey
|
||||
|| acknowledgement.operationId !== patch.operationId) {
|
||||
root?.setAttribute("data-hemx-sync-error", "acknowledgement-mismatch");
|
||||
return;
|
||||
}
|
||||
await remove(patch.idempotencyKey);
|
||||
root?.setAttribute("data-hemx-sync-ack", acknowledgement.idempotencyKey);
|
||||
}
|
||||
root?.setAttribute("data-hemx-sync-pending", String((await allPatches()).length));
|
||||
} catch {
|
||||
root?.setAttribute("data-hemx-sync-error", "offline");
|
||||
} finally {
|
||||
pumping = false;
|
||||
}
|
||||
}
|
||||
|
||||
async function start() {
|
||||
if (!root) return;
|
||||
database = await openDatabase();
|
||||
root.setAttribute("data-hemx-sync-ready", "");
|
||||
root.setAttribute("data-hemx-sync-pending", String((await allPatches()).length));
|
||||
document.addEventListener(EVENT, async (event) => {
|
||||
try {
|
||||
const patch = validatePatch(JSON.parse(event.detail));
|
||||
await persist(patch);
|
||||
await pump();
|
||||
} catch (error) {
|
||||
root.setAttribute("data-hemx-sync-error", error instanceof Error ? error.message : String(error));
|
||||
}
|
||||
});
|
||||
window.addEventListener("online", () => pump());
|
||||
await pump();
|
||||
}
|
||||
|
||||
start().catch((error) => root?.setAttribute("data-hemx-sync-error", error instanceof Error ? error.message : String(error)));
|
||||
@@ -0,0 +1,277 @@
|
||||
use hemx_core::{Effect, IntoEffect};
|
||||
use serde::{de, Deserialize, Deserializer, Serialize};
|
||||
use std::{error::Error, fmt};
|
||||
|
||||
pub const PATCH_SCHEMA_VERSION: u16 = 1;
|
||||
pub const PATCH_EVENT: &str = "hemx:sync-patch";
|
||||
const INTERACTION_ID: &str = "$hemx-interaction";
|
||||
pub const BROWSER_RUNTIME: &str = include_str!("../runtime/hemx-sync.js");
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
|
||||
#[serde(untagged)]
|
||||
pub enum PatchValue {
|
||||
Boolean(bool),
|
||||
Integer(i64),
|
||||
String(String),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct FlatPatch {
|
||||
// req: sync/003
|
||||
schema_version: u16,
|
||||
idempotency_key: String,
|
||||
operation_id: String,
|
||||
key: String,
|
||||
value: PatchValue,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
struct FlatPatchWire {
|
||||
schema_version: u16,
|
||||
idempotency_key: String,
|
||||
operation_id: String,
|
||||
key: String,
|
||||
value: PatchValue,
|
||||
}
|
||||
|
||||
impl<'de> Deserialize<'de> for FlatPatch {
|
||||
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
let wire = FlatPatchWire::deserialize(deserializer)?;
|
||||
let patch = Self {
|
||||
schema_version: wire.schema_version,
|
||||
idempotency_key: wire.idempotency_key,
|
||||
operation_id: wire.operation_id,
|
||||
key: wire.key,
|
||||
value: wire.value,
|
||||
};
|
||||
patch.validate().map_err(de::Error::custom)?;
|
||||
Ok(patch)
|
||||
}
|
||||
}
|
||||
|
||||
impl FlatPatch {
|
||||
pub fn for_interaction(key: impl Into<String>, value: PatchValue) -> Result<Self, PatchError> {
|
||||
let key = key.into();
|
||||
validate_key(&key)?;
|
||||
validate_value(&value)?;
|
||||
Ok(Self {
|
||||
schema_version: PATCH_SCHEMA_VERSION,
|
||||
idempotency_key: INTERACTION_ID.to_owned(),
|
||||
operation_id: INTERACTION_ID.to_owned(),
|
||||
key,
|
||||
value,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn new(
|
||||
idempotency_key: impl Into<String>,
|
||||
operation_id: impl Into<String>,
|
||||
key: impl Into<String>,
|
||||
value: PatchValue,
|
||||
) -> Result<Self, PatchError> {
|
||||
let patch = Self {
|
||||
schema_version: PATCH_SCHEMA_VERSION,
|
||||
idempotency_key: idempotency_key.into(),
|
||||
operation_id: operation_id.into(),
|
||||
key: key.into(),
|
||||
value,
|
||||
};
|
||||
patch.validate()?;
|
||||
Ok(patch)
|
||||
}
|
||||
|
||||
pub fn payload(&self) -> String {
|
||||
let value = match &self.value {
|
||||
PatchValue::Boolean(value) => value.to_string(),
|
||||
PatchValue::Integer(value) => value.to_string(),
|
||||
PatchValue::String(value) => json_string(value),
|
||||
};
|
||||
format!(
|
||||
r#"{{"schemaVersion":{},"idempotencyKey":{},"operationId":{},"key":{},"value":{value}}}"#,
|
||||
self.schema_version,
|
||||
json_string(&self.idempotency_key),
|
||||
json_string(&self.operation_id),
|
||||
json_string(&self.key),
|
||||
)
|
||||
}
|
||||
|
||||
pub fn validate(&self) -> Result<(), PatchError> {
|
||||
if self.schema_version != PATCH_SCHEMA_VERSION {
|
||||
return Err(PatchError::SchemaVersion(self.schema_version));
|
||||
}
|
||||
if self.idempotency_key != INTERACTION_ID || self.operation_id != INTERACTION_ID {
|
||||
validate_identifier("idempotency_key", &self.idempotency_key, 128)?;
|
||||
validate_identifier("operation_id", &self.operation_id, 128)?;
|
||||
}
|
||||
validate_key(&self.key)?;
|
||||
validate_value(&self.value)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||
pub enum PatchError {
|
||||
Empty(&'static str),
|
||||
InvalidCharacter(&'static str),
|
||||
TooLong(&'static str),
|
||||
ReservedKey,
|
||||
SchemaVersion(u16),
|
||||
ValueTooLong,
|
||||
IntegerOutOfRange,
|
||||
}
|
||||
|
||||
impl fmt::Display for PatchError {
|
||||
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
Self::Empty(field) => write!(formatter, "{field} must not be empty"),
|
||||
Self::InvalidCharacter(field) => {
|
||||
write!(formatter, "{field} contains an invalid character")
|
||||
}
|
||||
Self::TooLong(field) => write!(formatter, "{field} is too long"),
|
||||
Self::ReservedKey => formatter.write_str("patch key is reserved"),
|
||||
Self::SchemaVersion(version) => {
|
||||
write!(formatter, "unsupported patch schema version {version}")
|
||||
}
|
||||
Self::ValueTooLong => formatter.write_str("patch string value is too long"),
|
||||
Self::IntegerOutOfRange => {
|
||||
formatter.write_str("patch integer value exceeds JavaScript's safe range")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Error for PatchError {}
|
||||
|
||||
fn json_string(value: &str) -> String {
|
||||
const HEX: &[u8; 16] = b"0123456789abcdef";
|
||||
let mut encoded = String::with_capacity(value.len() + 2);
|
||||
encoded.push('"');
|
||||
for character in value.chars() {
|
||||
match character {
|
||||
'"' => encoded.push_str("\\\""),
|
||||
'\\' => encoded.push_str("\\\\"),
|
||||
'\n' => encoded.push_str("\\n"),
|
||||
'\r' => encoded.push_str("\\r"),
|
||||
'\t' => encoded.push_str("\\t"),
|
||||
character if character <= '\u{1f}' => {
|
||||
let byte = character as u8;
|
||||
encoded.push_str("\\u00");
|
||||
encoded.push(HEX[(byte >> 4) as usize] as char);
|
||||
encoded.push(HEX[(byte & 0x0f) as usize] as char);
|
||||
}
|
||||
character => encoded.push(character),
|
||||
}
|
||||
}
|
||||
encoded.push('"');
|
||||
encoded
|
||||
}
|
||||
|
||||
fn validate_value(value: &PatchValue) -> Result<(), PatchError> {
|
||||
match value {
|
||||
PatchValue::String(value) if value.len() > 4096 => Err(PatchError::ValueTooLong),
|
||||
PatchValue::Integer(value) if value.unsigned_abs() > 9_007_199_254_740_991 => {
|
||||
Err(PatchError::IntegerOutOfRange)
|
||||
}
|
||||
_ => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_identifier(field: &'static str, value: &str, maximum: usize) -> Result<(), PatchError> {
|
||||
if value.is_empty() {
|
||||
return Err(PatchError::Empty(field));
|
||||
}
|
||||
if value.len() > maximum {
|
||||
return Err(PatchError::TooLong(field));
|
||||
}
|
||||
if !value
|
||||
.bytes()
|
||||
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b':' | b'_' | b'-' | b'.'))
|
||||
{
|
||||
return Err(PatchError::InvalidCharacter(field));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_key(key: &str) -> Result<(), PatchError> {
|
||||
if key.is_empty() {
|
||||
return Err(PatchError::Empty("key"));
|
||||
}
|
||||
if key.len() > 64 {
|
||||
return Err(PatchError::TooLong("key"));
|
||||
}
|
||||
if matches!(
|
||||
key,
|
||||
"schemaVersion" | "idempotencyKey" | "operationId" | "key" | "value"
|
||||
) {
|
||||
return Err(PatchError::ReservedKey);
|
||||
}
|
||||
let mut bytes = key.bytes();
|
||||
if !bytes.next().is_some_and(|byte| byte.is_ascii_alphabetic())
|
||||
|| !bytes.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
|
||||
{
|
||||
return Err(PatchError::InvalidCharacter("key"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||
pub struct SyncEffect(Effect); // req: sync/002
|
||||
|
||||
impl SyncEffect {
|
||||
pub fn send_patch(patch: FlatPatch) -> Self {
|
||||
patch.validate().expect("FlatPatch must remain valid");
|
||||
Self(Effect::Emit {
|
||||
name: PATCH_EVENT.to_owned(),
|
||||
payload: patch.payload(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl IntoEffect for SyncEffect {
|
||||
fn append_to(self, ops: &mut Vec<Effect>) {
|
||||
self.0.append_to(ops);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn schema_is_flat_and_rejects_reserved_keys() {
|
||||
let patch = FlatPatch::new(
|
||||
"actor:1",
|
||||
"move-card-to-done",
|
||||
"cardColumn",
|
||||
PatchValue::String("done".to_owned()),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
patch.payload(),
|
||||
r#"{"schemaVersion":1,"idempotencyKey":"actor:1","operationId":"move-card-to-done","key":"cardColumn","value":"done"}"#
|
||||
);
|
||||
assert_eq!(
|
||||
FlatPatch::new("actor:1", "move", "value", PatchValue::Integer(1)),
|
||||
Err(PatchError::ReservedKey)
|
||||
);
|
||||
assert_eq!(
|
||||
FlatPatch::new(
|
||||
"actor:1",
|
||||
"move",
|
||||
"rank",
|
||||
PatchValue::Integer(9_007_199_254_740_992),
|
||||
),
|
||||
Err(PatchError::IntegerOutOfRange)
|
||||
);
|
||||
assert!(serde_json::from_str::<FlatPatch>(
|
||||
r#"{"schemaVersion":1,"idempotencyKey":"actor:1","operationId":"move","key":"rank","value":9007199254740992}"#,
|
||||
)
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("safe range"));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user