leptoskit_rpc 0.3.0

//! The live protocol: the frames a client and a server exchange over a
//! WebSocket, and the patches that keep a 2,000-message channel's update at
//! about one message's size.
//!
//! The frames are JSON, negotiated as `leptoskit.live.v1.json`. A client
//! opens `/live`, sends [`ClientFrame::Hello`], subscribes to queries with
//! [`ClientFrame::Sub`], and receives a [`ServerFrame::Transition`] whenever
//! a result changes. Every update comes with the version it reflects, so a
//! client can tell which writes it has seen.
//!
//! ```text
//! client → server                                   server → client
//! hello   { resume?: { version, queries? } }        welcome     { connection, version, heartbeat_seconds }
//! auth    { token }                                 transition  { from, to, updates: [ { id, ok | patch | err } ] }
//! sub     { id, query, args }                       done        { request, ok | err, version }
//! unsub   { id }                                    auth_status { refused | expiring }
//! mutate  { request, name, args, key? }             reconnect   { after_ms }
//! act     { request, name, args }                   pong
//! stream  { stream, name, args, credits }           item        { stream, value, cursor? }
//! credit  { stream, n }                             end         { stream, reason, err? }
//! cancel  { stream }
//! ping
//! ```
//!
//! # Versions
//!
//! Every commit to a live table takes the next version of the database's
//! change log. A transition's `to` is the version its results reflect, and
//! a `done` for a mutation or an action names the version its writes
//! committed at: once the client has applied a transition with `to` at or
//! past it, every result it holds includes them. The server sends an empty
//! transition when a commit changes none of a connection's results but the
//! connection is waiting for it, so a mutation never waits for a change
//! that won't come.
//!
//! # Resuming
//!
//! A client that reconnects sends the last `to` it applied, and the queries
//! it holds, in [`ClientFrame::Hello`]. When the change log still reaches
//! back that far, the server subscribes them again and sends one transition
//! with only the results that changed since; otherwise the transition is
//! [`full`](Transition::full).
//!
//! # Server-sent events
//!
//! Where a WebSocket can't get through, the same frames travel over HTTP:
//! `GET /live/events` streams server frames as server-sent events, each
//! transition with its `to` as the event's `id`, and the client posts its
//! own frames to `POST /live/frames?connection=<id>`, a JSON array at a
//! time, with the `connection` from `welcome`. A reconnecting event source
//! sends `Last-Event-ID`, which stands in for the hello's resume version.
//!
//! # Streams
//!
//! A stream is server output over time that isn't a query over rows: a
//! job's progress, a model's tokens, a log tail. [`ClientFrame::Stream`]
//! opens one with a number of credits, and the server sends at most that
//! many [`ServerFrame::Item`]s until [`ClientFrame::Credit`] grants more: it
//! never reads ahead of what the client has room for. A stream ends with
//! [`ServerFrame::End`], whose `reason` is one of [`EndReason`]'s names,
//! or when the client sends [`ClientFrame::Cancel`].
//!
//! [`patch`] turns two results into an [`Update::Patch`] when both are lists
//! of rows with an `id`, and into [`Update::Ok`] with the whole value
//! otherwise. Diffs keep steady-state frames small, and never lose data:
//! applying the same updates in order gives the same result as replacing
//! the value.

use serde::Deserialize;
use serde::Serialize;
use serde_json::Value;

/// The WebSocket subprotocol every live connection speaks.
pub const SUBPROTOCOL: &str = "leptoskit.live.v1.json";

/// How often the server sends a WebSocket ping, in seconds.
pub const HEARTBEAT_SECONDS: u64 = 30;

/// How many queries one connection may subscribe to.
pub const MAX_SUBSCRIPTIONS: usize = 256;

/// The largest result one subscription may hold, in bytes of JSON.
pub const MAX_RESULT_BYTES: usize = 512 * 1024;

/// The largest frame a connection may send, in bytes.
pub const MAX_FRAME_BYTES: usize = 1024 * 1024;

/// How many streams one connection may hold open at once.
pub const MAX_STREAMS: usize = 64;

/// How many actions one connection may have running at once.
pub const MAX_ACTIONS: usize = 16;

/// How many channels one connection may be in at once.
pub const MAX_CHANNELS: usize = 64;

/// The longest channel instance key, its join arguments serialized, in
/// bytes. Each instance is a room the server keeps while anyone is in it.
pub const MAX_CHANNEL_KEY_BYTES: usize = 4 * 1024;

/// The largest presence state one member may show, in bytes of JSON. Every
/// join and leave sends every member's state to every member.
pub const MAX_PRESENCE_BYTES: usize = 16 * 1024;

/// Where a server-sent events connection reads server frames, next to the
/// WebSocket's `/live`.
pub const EVENTS_PATH: &str = "/live/events";

/// Where a server-sent events connection posts its client frames, with
/// `?connection=<id>` from [`ServerFrame::Welcome`].
pub const FRAMES_PATH: &str = "/live/frames";

/// What a client sends.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ClientFrame {
	/// The first frame: what the client already has, so the server can send
	/// only what changed.
	Hello {
		/// The last version the client applied, when it is resuming.
		#[serde(default, skip_serializing_if = "Option::is_none")]
		resume: Option<Resume>,
	},
	/// Who the connection is, for a native app that can't send cookies on
	/// the upgrade request.
	Auth {
		/// The app's bearer token.
		token: String,
	},
	/// Subscribe to a query.
	Sub {
		/// The client's name for the subscription, echoed in every update.
		id: u32,
		/// The operation's name.
		query: String,
		/// Its arguments, as one JSON object.
		#[serde(default)]
		args: Value,
	},
	/// Stop a subscription.
	Unsub {
		/// The id the subscription was made with.
		id: u32,
	},
	/// Join a channel. `channel` is the instance key, the join arguments
	/// serialized; `name` is the channel operation.
	Join {
		/// The instance key: the join arguments, serialized.
		channel: String,
		/// The channel operation's name.
		name: String,
		/// The join arguments.
		#[serde(default)]
		args: Value,
		/// What this connection shows the others, or `null` for the default.
		#[serde(default)]
		state: Value,
	},
	/// Leave a channel.
	Leave {
		/// The instance key the connection joined with.
		channel: String,
	},
	/// Send a signal to everyone else in a channel.
	Signal {
		/// The instance key the connection joined with.
		channel: String,
		/// The message.
		message: Value,
	},
	/// Run a mutation. Its `request` echoes in [`ServerFrame::Done`].
	Mutate {
		/// The client's name for the call.
		request: u32,
		/// The operation's name.
		name: String,
		/// Its arguments, as one JSON object.
		#[serde(default)]
		args: Value,
		/// The idempotency key that makes a retry run once.
		#[serde(default, skip_serializing_if = "Option::is_none")]
		key: Option<String>,
	},
	/// Run an action. Its `request` echoes in [`ServerFrame::Done`]. An
	/// action may call the outside world, so a client never sends one
	/// twice.
	Act {
		/// The client's name for the call.
		request: u32,
		/// The operation's name.
		name: String,
		/// Its arguments, as one JSON object.
		#[serde(default)]
		args: Value,
	},
	/// Open a stream with room for `credits` items.
	Stream {
		/// The client's name for the stream, echoed in its items and its
		/// end.
		stream: u32,
		/// The stream operation's name.
		name: String,
		/// Its arguments, as one JSON object.
		#[serde(default)]
		args: Value,
		/// How many items the server may send before it hears of more room.
		#[serde(default)]
		credits: u32,
	},
	/// Room for `n` more items on a stream.
	Credit {
		/// The stream's id.
		stream: u32,
		/// How many more items the client has room for.
		n: u32,
	},
	/// Stop a stream. The server sends nothing more for it.
	Cancel {
		/// The stream's id.
		stream: u32,
	},
	/// Check the connection is alive.
	Ping,
	/// A frame this version doesn't know: ignored, so a newer client can
	/// talk to an older server.
	#[serde(other)]
	Unknown,
}

/// What a resuming client already applied.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Resume {
	/// The last version the client applied: the `to` of the last
	/// transition.
	pub version: u64,
	/// The queries the client holds results for. The server subscribes
	/// them again without a `done` each, and sends one transition with the
	/// results that changed since `version`. Empty for a client that sends
	/// a `sub` for each instead.
	#[serde(default, skip_serializing_if = "Vec::is_empty")]
	pub queries: Vec<ResumeQuery>,
}

/// A query a resuming client already holds the result of.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ResumeQuery {
	/// The client's name for the subscription, as in [`ClientFrame::Sub`].
	pub id: u32,
	/// The operation's name.
	pub query: String,
	/// Its arguments, as one JSON object.
	#[serde(default)]
	pub args: Value,
}

/// What a server sends.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ServerFrame {
	/// The connection is open.
	Welcome {
		/// An opaque name for the connection, for logs.
		connection: String,
		/// The version the server starts from.
		version: u64,
		/// How often the server pings, in seconds.
		heartbeat_seconds: u64,
		/// Whether the app was built against a spec the server can no longer
		/// serve.
		#[serde(default)]
		outdated_client: bool,
	},
	/// Results changed.
	Transition(Transition),
	/// A mutation or a one-shot call finished.
	Done {
		/// The client's name for the call.
		request: u32,
		/// The result, when it succeeded.
		#[serde(default, skip_serializing_if = "Option::is_none")]
		ok: Option<Value>,
		/// Why it failed, when it did.
		#[serde(default, skip_serializing_if = "Option::is_none")]
		err: Option<WireError>,
		/// The version the call's writes committed at, which a transition
		/// reaching it includes; for a call that wrote no live table, the
		/// version the connection had already reached. For a subscription's
		/// first value, the version the connection had reached.
		version: u64,
	},
	/// What happened to the connection's identity.
	AuthStatus {
		/// Whether the server refused the credentials, ending the
		/// connection's subscriptions.
		#[serde(default)]
		refused: bool,
		/// Whether the credentials expire soon, so the client renews them.
		#[serde(default)]
		expiring: bool,
	},
	/// The server is going away: reconnect after `after_ms`.
	Reconnect {
		/// How long to wait, with jitter already applied.
		after_ms: u64,
	},
	/// A signal on a channel.
	Signal {
		/// The channel's name.
		channel: String,
		/// Who sent it, when the server can say.
		#[serde(default, skip_serializing_if = "Option::is_none")]
		from: Option<String>,
		/// The signal's message.
		message: Value,
	},
	/// Who is in a channel, after a join or a leave.
	Presence {
		/// The instance key.
		channel: String,
		/// Everyone now in the channel.
		members: Vec<PresenceMember>,
	},
	/// A channel operation was refused, such as when the caller may not
	/// join.
	ChannelError {
		/// The instance key.
		channel: String,
		/// Why it failed.
		err: WireError,
	},
	/// An item from a stream.
	Item {
		/// The stream's subscription id.
		stream: u32,
		/// The item.
		value: Value,
		/// Where a resumable stream reached with this item: the client
		/// re-opens it from here after a reconnect.
		#[serde(default, skip_serializing_if = "Option::is_none")]
		cursor: Option<Value>,
	},
	/// A stream ended.
	End {
		/// The stream's subscription id.
		stream: u32,
		/// Why it ended: one of [`EndReason`]'s names.
		reason: String,
		/// Why it failed, when `reason` is `failed`.
		#[serde(default, skip_serializing_if = "Option::is_none")]
		err: Option<WireError>,
	},
	/// The answer to [`ClientFrame::Ping`].
	Pong,
	/// A frame this version doesn't know.
	#[serde(other)]
	Unknown,
}

/// Why a stream ended, as [`ServerFrame::End`] names it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum EndReason {
	/// It ran out of items.
	Done,
	/// It failed, or never opened.
	Failed(WireError),
	/// The connection or the server went away first, or the connection
	/// changed who it is. A resumable stream opens again from its last
	/// cursor.
	Interrupted,
}

impl EndReason {
	/// The `reason` a frame carries: `done`, `failed`, or `interrupted`.
	#[must_use]
	pub fn as_str(&self) -> &'static str {
		match self {
			Self::Done => "done",
			Self::Failed(_) => "failed",
			Self::Interrupted => "interrupted",
		}
	}

	/// The frame that ends `stream` for this reason.
	#[must_use]
	pub fn frame(self, stream: u32) -> ServerFrame {
		let reason = self.as_str().to_owned();
		let err = match self {
			Self::Failed(error) => Some(error),
			Self::Done | Self::Interrupted => None,
		};
		ServerFrame::End {
			stream,
			reason,
			err,
		}
	}

	/// The reason an [`End`](ServerFrame::End) frame gives. A reason this
	/// version doesn't know is a failure.
	#[must_use]
	pub fn from_frame(reason: &str, err: Option<WireError>) -> Self {
		match (reason, err) {
			("done", None) => Self::Done,
			("interrupted", None) => Self::Interrupted,
			(_, Some(error)) => Self::Failed(error),
			(other, None) => {
				Self::Failed(WireError {
					status: 500,
					message: format!("the stream ended: {other}"),
				})
			}
		}
	}
}

/// Every result that changed in one database state.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Transition {
	/// The version the client last had.
	pub from: u64,
	/// The version these updates reach. Applying them in order leaves the
	/// client at this version.
	pub to: u64,
	/// The results that changed.
	pub updates: Vec<SubscriptionUpdate>,
	/// Whether every update repeats its whole value, because the client
	/// missed intermediate versions.
	#[serde(default, skip_serializing_if = "is_false")]
	pub full: bool,
}

/// One subscription's new result.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct SubscriptionUpdate {
	/// The subscription's client-chosen id.
	pub id: u32,
	/// What the result is now.
	#[serde(flatten)]
	pub update: Update,
}

/// How a result changed.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Update {
	/// The result, whole.
	Ok(Value),
	/// The changes to a list of rows with ids.
	Patch(Patch),
	/// The query failed; the subscription stays, and retries on the next
	/// change.
	Err(WireError),
}

/// The changes to a keyed list: rows to drop, and rows to put in order.
///
/// Apply the removals first, then each upsert at its index, in order: it
/// replaces the row with its id there, or goes in there as a new row. An
/// upsert at an index past the end appends.
///
/// A [`Page`](crate::Page)'s result is an object whose `items` are the
/// keyed list: its patch names that `field`, and carries the object's other
/// fields that changed in `set`.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct Patch {
	/// The ids of the rows to remove.
	#[serde(default, skip_serializing_if = "Vec::is_empty")]
	pub remove: Vec<String>,
	/// The rows to insert or replace, in the order they belong.
	#[serde(default, skip_serializing_if = "Vec::is_empty")]
	pub upsert: Vec<Upsert>,
	/// The field of an object result that holds the keyed list, when the
	/// result is a page rather than a list.
	#[serde(default, skip_serializing_if = "Option::is_none")]
	pub field: Option<String>,
	/// The object's other fields that changed, whole.
	#[serde(default, skip_serializing_if = "serde_json::Map::is_empty")]
	pub set: serde_json::Map<String, Value>,
}

/// One row of a [`Patch`].
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Upsert {
	/// Where the row goes, after the removals.
	pub at: usize,
	/// The row, as a JSON object with an `id`.
	pub value: Value,
}

/// Someone in a channel.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PresenceMember {
	/// Who: the caller id the server signs them with.
	pub id: String,
	/// What the channel's presence says about them.
	#[serde(default)]
	pub state: Value,
}

/// A failure, as a frame reports it.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WireError {
	/// The HTTP status the same failure would carry.
	pub status: u16,
	/// The message the caller reads.
	pub message: String,
}

// The constructors compile with the server's live module, which the `live`
// feature builds (`live` implies `server`); in builds without it, such as
// an app server that depends on the auth crate alone, they are dead code.
#[cfg(feature = "live")]
impl WireError {
	/// The failure `error` carries.
	pub(crate) fn new(error: &crate::Error) -> Self {
		Self {
			status: error.status(),
			message: error.message().to_owned(),
		}
	}

	/// A failure that isn't an operation's, such as a frame that doesn't
	/// parse.
	pub(crate) fn invalid(message: impl Into<String>) -> Self {
		Self {
			status: 400,
			message: message.into(),
		}
	}
}

impl std::fmt::Display for WireError {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		f.write_str(&self.message)
	}
}

impl std::error::Error for WireError {}

#[allow(
	clippy::trivially_copy_pass_by_ref,
	reason = "serde calls `skip_serializing_if` with a reference"
)]
const fn is_false(value: &bool) -> bool {
	!*value
}

/// The update that turns `old` into `new`: a [`Patch`] when both are lists
/// of rows with string `id`s whose existing rows keep their order, or both
/// are pages of such rows, and the whole value otherwise.
#[must_use]
pub fn patch(old: &Value, new: &Value) -> Update {
	if old == new {
		return Update::Ok(new.clone());
	}

	if let (Some(before), Some(after)) = (page_fields(old), page_fields(new)) {
		return match diff_rows(&before["items"], &after["items"]) {
			Some(mut patch) => {
				patch.field = Some("items".to_owned());
				patch.set = after
					.iter()
					.filter(|(name, value)| *name != "items" && before.get(*name) != Some(value))
					.map(|(name, value)| (name.clone(), value.clone()))
					.collect();
				Update::Patch(patch)
			}
			None => Update::Ok(new.clone()),
		};
	}

	match diff_rows(old, new) {
		Some(patch) => Update::Patch(patch),
		None => Update::Ok(new.clone()),
	}
}

/// A page's fields, when `value` is shaped exactly like one: `items`,
/// `end` and `has_more`.
fn page_fields(value: &Value) -> Option<&serde_json::Map<String, Value>> {
	let fields = value.as_object()?;
	let shaped = fields.len() == 3
		&& fields.get("items").is_some_and(Value::is_array)
		&& fields.contains_key("end")
		&& fields.get("has_more").is_some_and(Value::is_boolean);
	shaped.then_some(fields)
}

/// The patch between two lists of rows with ids, when one can say it.
fn diff_rows(old: &Value, new: &Value) -> Option<Patch> {
	match (rows(old), rows(new)) {
		(Some(before), Some(after)) => diff(&before, &after),
		_ => None,
	}
}

/// The `(id, row)` pairs of a JSON array of objects with string ids.
fn rows(value: &Value) -> Option<Vec<(String, Value)>> {
	let items = value.as_array()?;
	items
		.iter()
		.map(|item| {
			let id = item.get("id")?.as_str()?;
			Some((id.to_owned(), item.clone()))
		})
		.collect()
}

/// The patch from `before` to `after`, or `None` when a patch can't express
/// the change: a row moved relative to the rows that stayed, or two rows
/// share an id.
fn diff(before: &[(String, Value)], after: &[(String, Value)]) -> Option<Patch> {
	let before_ids: std::collections::HashSet<&str> =
		before.iter().map(|(id, _)| id.as_str()).collect();

	if before_ids.len() != before.len() {
		return None;
	}

	let after_ids: std::collections::HashSet<&str> =
		after.iter().map(|(id, _)| id.as_str()).collect();

	if after_ids.len() != after.len() {
		return None;
	}

	let remove: Vec<String> = before
		.iter()
		.filter(|(id, _)| !after_ids.contains(id.as_str()))
		.map(|(id, _)| id.clone())
		.collect();

	// Rows that stayed must keep their relative order: a patch can move a
	// row only by re-sending it, and a swap would need both.
	let stayed: Vec<&str> = before
		.iter()
		.map(|(id, _)| id.as_str())
		.filter(|id| after_ids.contains(id))
		.collect();
	let after_stayed: Vec<&str> = after
		.iter()
		.map(|(id, _)| id.as_str())
		.filter(|id| before_ids.contains(id))
		.collect();

	if stayed != after_stayed {
		return None;
	}

	let previous: std::collections::HashMap<&str, &Value> =
		before.iter().map(|(id, row)| (id.as_str(), row)).collect();
	let upsert: Vec<Upsert> = after
		.iter()
		.enumerate()
		.filter(|(_, (id, row))| previous.get(id.as_str()) != Some(&row))
		.map(|(at, (_, row))| {
			Upsert {
				at,
				value: row.clone(),
			}
		})
		.collect();

	Some(Patch {
		remove,
		upsert,
		..Patch::default()
	})
}

/// Applies an [`Update`] to a value. A patch on something that isn't a
/// keyed list leaves `value` alone: the server only sends patches for what
/// it described.
pub fn apply(value: &mut Value, update: &Update) {
	match update {
		Update::Ok(next) => *value = next.clone(),
		Update::Err(_) => {}
		Update::Patch(patch) => {
			let list = match &patch.field {
				Some(field) => {
					let Some(fields) = value.as_object_mut() else {
						return;
					};

					for (name, changed) in &patch.set {
						fields.insert(name.clone(), changed.clone());
					}

					fields.get_mut(field)
				}
				None => Some(value),
			};
			let Some(items) = list.and_then(Value::as_array_mut) else {
				return;
			};
			items.retain(|item| {
				item.get("id")
					.and_then(Value::as_str)
					.is_none_or(|id| !patch.remove.iter().any(|removed| removed == id))
			});
			// Upserts come in order of where they land: the row already at
			// that place is the one being replaced, or else the upsert is a
			// new row that goes in there.
			for upsert in &patch.upsert {
				let id = upsert.value.get("id").and_then(Value::as_str);
				let at = upsert.at.min(items.len());

				match items.get_mut(at) {
					Some(slot) if id.is_some() && slot.get("id").and_then(Value::as_str) == id => {
						*slot = upsert.value.clone();
					}
					_ => items.insert(at, upsert.value.clone()),
				}
			}
		}
	}
}

#[cfg(test)]
mod tests {
	use serde_json::json;

	use super::*;

	#[test]
	fn frames_round_trip_through_json() {
		let frames = [
			ClientFrame::Hello {
				resume: Some(Resume {
					version: 7,
					queries: vec![ResumeQuery {
						id: 1,
						query: "messages".into(),
						args: json!({ "channel": "general" }),
					}],
				}),
			},
			ClientFrame::Act {
				request: 3,
				name: "summarize".into(),
				args: json!({}),
			},
			ClientFrame::Auth {
				token: "abc".into(),
			},
			ClientFrame::Sub {
				id: 1,
				query: "messages".into(),
				args: json!({ "channel": "general" }),
			},
			ClientFrame::Unsub { id: 1 },
			ClientFrame::Mutate {
				request: 2,
				name: "send_message".into(),
				args: json!({}),
				key: Some("k".into()),
			},
			ClientFrame::Stream {
				stream: 5,
				name: "tokens".into(),
				args: json!({ "prompt": "hi" }),
				credits: 8,
			},
			ClientFrame::Credit { stream: 5, n: 4 },
			ClientFrame::Cancel { stream: 5 },
			ClientFrame::Ping,
		];

		for frame in frames {
			let json = serde_json::to_string(&frame).unwrap();
			let back: ClientFrame = serde_json::from_str(&json).unwrap();
			assert_eq!(back, frame);
		}

		let frames = [
			ServerFrame::Welcome {
				connection: "c1".into(),
				version: 0,
				heartbeat_seconds: 30,
				outdated_client: false,
			},
			ServerFrame::Transition(Transition {
				from: 0,
				to: 1,
				updates: vec![SubscriptionUpdate {
					id: 1,
					update: Update::Ok(json!([{ "id": "a" }])),
				}],
				full: false,
			}),
			ServerFrame::Done {
				request: 2,
				ok: Some(json!("ok")),
				err: None,
				version: 1,
			},
			ServerFrame::AuthStatus {
				refused: false,
				expiring: true,
			},
			ServerFrame::Reconnect { after_ms: 250 },
			ServerFrame::Item {
				stream: 5,
				value: json!("tok"),
				cursor: Some(json!(3)),
			},
			EndReason::Failed(WireError {
				status: 500,
				message: "gone".into(),
			})
			.frame(5),
			EndReason::Done.frame(5),
			ServerFrame::Pong,
		];

		for frame in frames {
			let json = serde_json::to_string(&frame).unwrap();
			let back: ServerFrame = serde_json::from_str(&json).unwrap();
			assert_eq!(back, frame);
		}
	}

	#[test]
	fn an_older_stream_frame_still_parses() {
		assert_eq!(
			serde_json::from_str::<ServerFrame>(r#"{"type":"end","stream":2,"reason":"done"}"#)
				.unwrap(),
			EndReason::Done.frame(2)
		);
		assert_eq!(
			serde_json::from_str::<ServerFrame>(r#"{"type":"item","stream":2,"value":1}"#).unwrap(),
			ServerFrame::Item {
				stream: 2,
				value: json!(1),
				cursor: None,
			}
		);
		assert_eq!(EndReason::from_frame("done", None), EndReason::Done);
		assert_eq!(
			EndReason::from_frame("interrupted", None),
			EndReason::Interrupted
		);
		assert!(matches!(
			EndReason::from_frame("later", None),
			EndReason::Failed(_)
		));
	}

	#[test]
	fn an_older_hello_still_parses() {
		assert_eq!(
			serde_json::from_str::<ClientFrame>(r#"{"type":"hello","resume":{"version":4}}"#)
				.unwrap(),
			ClientFrame::Hello {
				resume: Some(Resume {
					version: 4,
					queries: Vec::new(),
				}),
			}
		);
	}

	#[test]
	fn unknown_frames_are_ignored() {
		assert_eq!(
			serde_json::from_str::<ClientFrame>(r#"{"type":"future"}"#).unwrap(),
			ClientFrame::Unknown
		);
		assert_eq!(
			serde_json::from_str::<ServerFrame>(r#"{"type":"future"}"#).unwrap(),
			ServerFrame::Unknown
		);
	}

	#[test]
	fn lists_of_rows_change_by_patch() {
		let before = json!([
			{ "id": "a", "body": "one" },
			{ "id": "b", "body": "two" },
			{ "id": "c", "body": "three" }
		]);
		let after = json!([
			{ "id": "a", "body": "one" },
			{ "id": "b", "body": "TWO" },
			{ "id": "d", "body": "four" }
		]);

		let Update::Patch(patch) = patch(&before, &after) else {
			panic!("a keyed list patches");
		};
		assert_eq!(patch.remove, ["c"]);
		assert_eq!(
			patch.upsert,
			[
				Upsert {
					at: 1,
					value: json!({ "id": "b", "body": "TWO" })
				},
				Upsert {
					at: 2,
					value: json!({ "id": "d", "body": "four" })
				}
			]
		);

		let mut applied = before.clone();
		apply(&mut applied, &Update::Patch(patch));
		assert_eq!(applied, after);
	}

	#[test]
	fn a_row_inserted_before_others_goes_in_rather_than_over() {
		let before = json!([{ "id": "b" }, { "id": "d" }]);
		let after = json!([{ "id": "a" }, { "id": "b" }, { "id": "c" }, { "id": "d", "n": 1 }]);
		let update = patch(&before, &after);
		assert!(matches!(update, Update::Patch(_)));
		let mut applied = before.clone();
		apply(&mut applied, &update);
		assert_eq!(applied, after);
	}

	#[test]
	fn a_page_changes_by_a_patch_of_its_items() {
		let before = json!({
			"items": [{ "id": "a", "n": 1 }, { "id": "b", "n": 2 }],
			"end": "\"b\"",
			"has_more": false
		});
		let after = json!({
			"items": [{ "id": "a", "n": 1 }, { "id": "b", "n": 3 }, { "id": "c", "n": 4 }],
			"end": "\"c\"",
			"has_more": true
		});

		let Update::Patch(patch) = patch(&before, &after) else {
			panic!("a page patches");
		};
		assert_eq!(patch.field.as_deref(), Some("items"));
		assert_eq!(patch.upsert.len(), 2);
		assert_eq!(patch.set.len(), 2, "end and has_more changed");

		let mut applied = before.clone();
		apply(&mut applied, &Update::Patch(patch));
		assert_eq!(applied, after);

		// Something only page-like is sent whole.
		let other = json!({ "items": [{ "id": "a" }], "total": 1 });
		let grown = json!({ "items": [{ "id": "a" }, { "id": "b" }], "total": 2 });
		assert_eq!(patch_of(&other, &grown), Update::Ok(grown.clone()));
	}

	fn patch_of(old: &Value, new: &Value) -> Update {
		patch(old, new)
	}

	#[test]
	fn a_reorder_or_a_shape_without_ids_sends_the_whole_value() {
		let before = json!([{ "id": "a" }, { "id": "b" }]);
		let ast = json!([{ "id": "b" }, { "id": "a" }]);
		assert_eq!(patch(&before, &ast), Update::Ok(ast.clone()));

		let object = json!({ "id": "a" });
		let other = json!({ "id": "b" });
		assert_eq!(patch(&object, &other), Update::Ok(other.clone()));

		let missing = json!([{ "body": "no id" }]);
		assert_eq!(patch(&before, &missing), Update::Ok(missing.clone()));
	}

	#[test]
	fn applying_an_ok_replaces_the_value_and_an_error_keeps_it() {
		let mut value = json!([{ "id": "a" }]);
		apply(&mut value, &Update::Ok(json!([])));
		assert_eq!(value, json!([]));

		let kept = value.clone();
		apply(
			&mut value,
			&Update::Err(WireError {
				status: 500,
				message: "no connection".to_owned(),
			}),
		);
		assert_eq!(value, kept);
	}
}