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
+ );
+ }
+}