From 6ac667009fcaaaa7d1eef66d3776244458f84d39 Mon Sep 17 00:00:00 2001 From: dawn <90008@gaze.systems> Date: Sat, 21 Feb 2026 20:20:22 +0300 Subject: [PATCH] fjall: parse operation to store types more efficiently --- Cargo.lock | 1 + Cargo.toml | 1 + src/plc_fjall.rs | 440 ++++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 437 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1c8ce64..bcd1dd4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -35,6 +35,7 @@ dependencies = [ "governor", "http-body-util", "log", + "multibase", "native-tls", "opentelemetry", "opentelemetry-otlp", diff --git a/Cargo.toml b/Cargo.toml index d91f7d3..8f9580a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,4 +48,5 @@ cid = "0.11.1" rmp-serde = "1.3.1" bincode = "1.3.3" serde_bytes = "0.11.19" +multibase = "0.9.2" diff --git a/src/plc_fjall.rs b/src/plc_fjall.rs index 107dca2..6e1bc86 100644 --- a/src/plc_fjall.rs +++ b/src/plc_fjall.rs @@ -1,7 +1,9 @@ use crate::{Dt, ExportPage, Op as CommonOp, PageBoundaryState}; -use data_encoding::BASE32_NOPAD; +use data_encoding::{BASE32_NOPAD, BASE64URL}; use fjall::{Database, Keyspace, KeyspaceCreateOptions, OwnedWriteBatch, PersistMode}; use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; +use std::fmt; use std::path::Path; use std::sync::Arc; use std::time::Instant; @@ -78,6 +80,340 @@ fn decode_timestamp(key: &[u8]) -> anyhow::Result
{ .ok_or_else(|| anyhow::anyhow!("invalid timestamp {micros}")) } +/// base64url-encoded ECDSA signature → raw bytes +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Signature(#[serde(with = "serde_bytes")] Vec); + +impl Signature { + fn from_base64url(s: &str) -> anyhow::Result { + BASE64URL + .decode(s.as_bytes()) + .map(Self) + .map_err(|e| anyhow::anyhow!("invalid base64url sig: {e}")) + } +} + +impl fmt::Display for Signature { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&BASE64URL.encode(&self.0)) + } +} + +/// did:key:z... → raw multicodec public key bytes +#[derive(Debug, Clone, Serialize, Deserialize)] +struct DidKey(#[serde(with = "serde_bytes")] Vec); + +impl DidKey { + fn from_did_key(s: &str) -> anyhow::Result { + let multibase_str = s + .strip_prefix("did:key:") + .ok_or_else(|| anyhow::anyhow!("missing did:key: prefix in {s}"))?; + let (_base, bytes) = multibase::decode(multibase_str) + .map_err(|e| anyhow::anyhow!("invalid multibase in did:key {s}: {e}"))?; + Ok(Self(bytes)) + } +} + +impl fmt::Display for DidKey { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "did:key:{}", + multibase::encode(multibase::Base::Base58Btc, &self.0) + ) + } +} + +/// CID string → binary CID bytes +#[derive(Debug, Clone, Serialize, Deserialize)] +struct PlcCid(#[serde(with = "serde_bytes")] Vec); + +impl PlcCid { + fn from_cid_str(s: &str) -> anyhow::Result { + let cid = IpldCid::try_from(s)?; + let mut buf = Vec::new(); + cid.write_bytes(&mut buf) + .map_err(|e| anyhow::anyhow!("failed to encode cid {s}: {e}"))?; + Ok(Self(buf)) + } +} + +impl fmt::Display for PlcCid { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let cid = IpldCid::try_from(self.0.as_slice()).map_err(|_| fmt::Error)?; + write!(f, "{cid}") + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +enum Aka { + Atproto(String), + Other(String), +} + +impl Aka { + fn from_str(s: &str) -> Self { + if let Some(stripped) = s.strip_prefix("at://") { + Self::Atproto(stripped.to_string()) + } else { + Self::Other(s.to_string()) + } + } +} + +impl fmt::Display for Aka { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Atproto(h) => write!(f, "at://{h}"), + Self::Other(s) => f.write_str(s), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum OpType { + PlcOperation, + Create, + PlcTombstone, + Other(String), +} + +impl OpType { + fn from_str(s: &str) -> Self { + match s { + "plc_operation" => Self::PlcOperation, + "create" => Self::Create, + "plc_tombstone" => Self::PlcTombstone, + other => Self::Other(other.to_string()), + } + } + + fn as_str(&self) -> &str { + match self { + Self::PlcOperation => "plc_operation", + Self::Create => "create", + Self::PlcTombstone => "plc_tombstone", + Self::Other(s) => s, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct StoredService { + r#type: String, + endpoint: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct StoredOp { + op_type: OpType, + sig: Signature, + prev: Option, + + rotation_keys: Option>, + verification_methods: Option>, + also_known_as: Option>, + services: Option>, + + // legacy create fields + signing_key: Option, + recovery_key: Option, + handle: Option, + service: Option, + + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + unknown: BTreeMap, +} + +impl StoredOp { + fn from_json_value(v: &serde_json::Value) -> anyhow::Result { + let obj = v + .as_object() + .ok_or_else(|| anyhow::anyhow!("operation is not an object"))?; + + let mut known_keys = Vec::new(); + let mut get = |key: &'static str| { + known_keys.push(key); + obj.get(key) + }; + + let op_type = get("type") + .and_then(|t| t.as_str()) + .map(OpType::from_str) + .ok_or_else(|| anyhow::anyhow!("missing type field"))?; + + let sig = get("sig") + .and_then(|s| s.as_str()) + .ok_or_else(|| anyhow::anyhow!("missing sig field")) + .and_then(Signature::from_base64url)?; + + let prev = match get("prev").and_then(|p| p.as_str()) { + Some(s) => Some(PlcCid::from_cid_str(s)?), + None => None, + }; + + let rotation_keys = get("rotationKeys") + .and_then(|v| v.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|v| v.as_str()) + .map(DidKey::from_did_key) + .collect::>>() + }) + .transpose()?; + + let verification_methods = get("verificationMethods") + .and_then(|v| v.as_object()) + .map(|map| { + map.iter() + .map(|(k, v)| { + let key = DidKey::from_did_key(v.as_str().unwrap_or_default())?; + Ok((k.clone(), key)) + }) + .collect::>>() + }) + .transpose()?; + + let also_known_as = get("alsoKnownAs").and_then(|v| v.as_array()).map(|arr| { + arr.iter() + .filter_map(|v| v.as_str()) + .map(Aka::from_str) + .collect() + }); + + let services = get("services").and_then(|v| v.as_object()).map(|map| { + map.iter() + .filter_map(|(k, v)| { + let svc = StoredService { + r#type: v.get("type")?.as_str()?.to_string(), + endpoint: v.get("endpoint")?.as_str()?.to_string(), + }; + Some((k.clone(), svc)) + }) + .collect() + }); + + let signing_key = get("signingKey") + .and_then(|v| v.as_str()) + .map(DidKey::from_did_key) + .transpose()?; + + let recovery_key = get("recoveryKey") + .and_then(|v| v.as_str()) + .map(DidKey::from_did_key) + .transpose()?; + + let handle = get("handle") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + let service = get("service") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + let mut unknown = BTreeMap::new(); + for (k, v) in obj { + if !known_keys.contains(&k.as_str()) { + unknown.insert(k.clone(), v.clone()); + } + } + + Ok(Self { + op_type, + sig, + prev, + rotation_keys, + verification_methods, + also_known_as, + services, + signing_key, + recovery_key, + handle, + service, + unknown, + }) + } + + fn to_json_value(&self) -> serde_json::Value { + let mut map = serde_json::Map::new(); + + map.insert("type".into(), self.op_type.as_str().into()); + map.insert("sig".into(), self.sig.to_string().into()); + map.insert( + "prev".into(), + self.prev + .as_ref() + .map(|c| serde_json::Value::String(c.to_string())) + .unwrap_or(serde_json::Value::Null), + ); + + if let Some(keys) = &self.rotation_keys { + map.insert( + "rotationKeys".into(), + keys.iter() + .map(|k| serde_json::Value::String(k.to_string())) + .collect::>() + .into(), + ); + } + + if let Some(methods) = &self.verification_methods { + let obj: serde_json::Map = methods + .iter() + .map(|(k, v)| (k.clone(), serde_json::Value::String(v.to_string()))) + .collect(); + map.insert("verificationMethods".into(), obj.into()); + } + + if let Some(aka) = &self.also_known_as { + map.insert( + "alsoKnownAs".into(), + aka.iter() + .map(|h| serde_json::Value::String(h.to_string())) + .collect::>() + .into(), + ); + } + + if let Some(services) = &self.services { + let obj: serde_json::Map = services + .iter() + .map(|(k, svc)| { + ( + k.clone(), + serde_json::json!({ + "type": svc.r#type, + "endpoint": svc.endpoint, + }), + ) + }) + .collect(); + map.insert("services".into(), obj.into()); + } + + // legacy create fields + if let Some(key) = &self.signing_key { + map.insert("signingKey".into(), key.to_string().into()); + } + if let Some(key) = &self.recovery_key { + map.insert("recoveryKey".into(), key.to_string().into()); + } + if let Some(handle) = &self.handle { + map.insert("handle".into(), handle.clone().into()); + } + if let Some(service) = &self.service { + map.insert("service".into(), service.clone().into()); + } + + for (k, v) in &self.unknown { + map.insert(k.clone(), v.clone()); + } + + serde_json::Value::Object(map) + } +} + // we have our own Op struct for fjall since we dont want to have to convert Value back to RawValue #[derive(Debug, Serialize)] pub struct Op { @@ -96,7 +432,7 @@ struct DbOp { #[serde(with = "serde_bytes")] pub did: Vec, pub nullified: bool, - pub operation: serde_json::Value, + pub operation: StoredOp, } #[derive(Clone)] @@ -168,10 +504,12 @@ impl FjallDb { let mut encoded_did = Vec::with_capacity(15); encode_did(&mut encoded_did, &op.did)?; + let json_val: serde_json::Value = serde_json::from_str(op.operation.get())?; + let stored = StoredOp::from_json_value(&json_val)?; let db_op = DbOp { did: encoded_did, nullified: op.nullified, - operation: serde_json::to_value(&op.operation)?, + operation: stored, }; let value = rmp_serde::to_vec(&db_op)?; batch.insert(&self.inner.ops, &ts_key, &value); @@ -219,7 +557,7 @@ impl FjallDb { cid, created_at: ts, nullified: op.nullified, - operation: op.operation, + operation: op.operation.to_json_value(), }) })) } @@ -256,7 +594,7 @@ impl FjallDb { cid, created_at, nullified: db_op.nullified, - operation: db_op.operation, + operation: db_op.operation.to_json_value(), }) })) } @@ -350,3 +688,95 @@ pub async fn pages_to_fjall( ); Ok("pages_to_fjall") } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn signature_roundtrip() { + let original = "9NuYV7AqwHVTc0YuWzNV3CJafsSZWH7qCxHRUIP2xWlB-YexXC1OaYAnUayiCXLVzRQ8WBXIqF-SvZdNalwcjA"; + let sig = Signature::from_base64url(original).unwrap(); + assert_eq!(sig.0.len(), 64); + assert_eq!(sig.to_string(), original); + } + + #[test] + fn did_key_roundtrip() { + let original = "did:key:zQ3shhCGUqDKjStzuDxPkTxN6ujddP4RkEKJJouJGRRkaLGbg"; + let key = DidKey::from_did_key(original).unwrap(); + assert_eq!(key.to_string(), original); + } + + #[test] + fn plc_cid_roundtrip() { + let original = "bafyreigp6shzy6dlcxuowwoxz7u5nemdrkad2my5zwzpwilcnhih7bw6zm"; + let cid = PlcCid::from_cid_str(original).unwrap(); + assert_eq!(cid.to_string(), original); + } + + #[test] + fn handle_roundtrip() { + let h = Aka::from_str("at://alice.bsky.social"); + assert_eq!(h, Aka::Atproto("alice.bsky.social".to_string())); + assert_eq!(h.to_string(), "at://alice.bsky.social"); + } + + #[test] + fn handle_without_prefix() { + // According to DID spec, alsoKnownAs should be URIs. + // If an alternative URI scheme is used, it will preserve it. + let h = Aka::from_str("https://something.else"); + assert_eq!(h, Aka::Other("https://something.else".to_string())); + assert_eq!(h.to_string(), "https://something.else"); + } + + #[test] + fn op_type_roundtrip() { + assert_eq!(OpType::from_str("plc_operation").as_str(), "plc_operation"); + assert_eq!(OpType::from_str("create").as_str(), "create"); + assert_eq!(OpType::from_str("plc_tombstone").as_str(), "plc_tombstone"); + assert_eq!(OpType::from_str("weird_thing").as_str(), "weird_thing"); + } + + #[test] + fn stored_op_fixture_roundtrip() { + let fixtures = [ + "tests/fixtures/log_bskyapp.json", + "tests/fixtures/log_legacy_dholms.json", + "tests/fixtures/log_nullification.json", + "tests/fixtures/log_tombstone.json", + ]; + + let mut total_json_size = 0; + let mut total_packed_size = 0; + + for path in fixtures { + let data = std::fs::read_to_string(path).unwrap(); + let entries: Vec = serde_json::from_str(&data).unwrap(); + + for entry in &entries { + let op = &entry["operation"]; + let stored = StoredOp::from_json_value(op) + .unwrap_or_else(|e| panic!("failed to parse op in {path}: {e}\n{op}")); + + // msgpack verification + let packed = rmp_serde::to_vec(&stored).unwrap(); + let unpacked: StoredOp = rmp_serde::from_slice(&packed).unwrap(); + + let reconstructed = unpacked.to_json_value(); + assert_eq!(*op, reconstructed, "roundtrip mismatch in {path}"); + + total_json_size += serde_json::to_vec(op).unwrap().len(); + total_packed_size += packed.len(); + } + } + + println!( + "json size: {} bytes, msgpack size: {} bytes, saved: {} bytes", + total_json_size, + total_packed_size, + total_json_size as isize - total_packed_size as isize + ); + } +} -- 2.51.2