feat: protobuf definition for publish records

Cargo.lock

@@ -224,6 +224,13 @@ version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
 
+[[package]]
+name = "chat-proto"
+version = "0.1.0"
+dependencies = [
+ "prost",
+]
+
 [[package]]
 name = "chat-store"
 version = "0.1.0"
@@ -231,10 +238,12 @@ dependencies = [
  "anyhow",
  "axum",
  "base64",
+ "chat-proto",
  "clap",
  "ed25519-dalek",
  "hex",
  "logos-delivery",
+ "prost",
  "reqwest",
  "serde",
  "serde_json",
@@ -910,6 +919,15 @@ version = "1.70.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695"
 
+[[package]]
+name = "itertools"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285"
+dependencies = [
+ "either",
+]
+
 [[package]]
 name = "itoa"
 version = "1.0.18"
@@ -1229,6 +1247,29 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "prost"
+version = "0.14.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1"
+dependencies = [
+ "bytes",
+ "prost-derive",
+]
+
+[[package]]
+name = "prost-derive"
+version = "0.14.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
+dependencies = [
+ "anyhow",
+ "itertools",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "quote"
 version = "1.0.45"

Cargo.toml

@@ -15,6 +15,11 @@ path = "src/main.rs"
 anyhow = "1.0"
 axum = "0.7"
 base64 = "0.22"
+# Wire schema for the delivery submissions, shared with libchat.
+# TODO: switch to `{ git = "https://github.com/logos-messaging/chat_proto", rev = "..." }`
+# once the store submission schemas land on chat_proto's main. Path dep only so
+# this builds ahead of that merge.
+chat-proto = { path = "../chat_proto/gen/rust" }
 clap = { version = "4", features = ["derive"] }
 ed25519-dalek = "2.2.0"
 hex = "0.4"
@@ -22,6 +27,7 @@ hex = "0.4"
 # the sibling libchat checkout; links the native liblogosdelivery (enter the
 # libchat dev shell or set LOGOS_DELIVERY_LIB_DIR to build a runnable binary).
 logos-delivery = { path = "../libchat/extensions/logos-delivery-rust" }
+prost = "0.14"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 # SQLite via sqlx with a bundled libsqlite3 — no sqlcipher/OpenSSL, no system libs.

examples/delivery_smoke_test.rs

@@ -1,10 +1,10 @@
 //! End-to-end smoke test for the logos-delivery write path.
 //!
 //! Publishes a signed keypackage bundle and an account device-list bundle over
-//! the delivery network — the same JSON submissions the HTTP POST endpoints
+//! the delivery network — the protobuf submissions the store subscribes for,
-//! take, on the store's subscription topics — then polls the HTTP query API
+//! carrying the same fields the HTTP POST endpoints take as JSON — then polls
-//! until both appear. Requires a chat-store running with delivery ingestion
+//! the HTTP query API until both appear. Requires a chat-store running with
-//! enabled (the default) on the same network preset.
+//! delivery ingestion enabled (the default) on the same network preset.
 //!
 //! ```text
 //! cargo run -- --bind 127.0.0.1:8080 --db tmp/chat-store.db   # terminal 1
@@ -15,12 +15,12 @@
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use anyhow::{Context, Result, bail};
-use base64::Engine;
+use chat_proto::logoschat::store::{AccountSubmissionV1, KeyPackageSubmissionV1};
-use base64::engine::general_purpose::STANDARD as BASE64;
 use ed25519_dalek::{Signer, SigningKey};
 use logos_delivery::{P2pConfig, ThreadedDeliveryWrapper};
+use prost::Message;
+use prost::bytes::Bytes;
 use reqwest::blocking::Client;
-use serde_json::json;
 
 /// Must match `ingest::KEYPACKAGE_SUBMIT_TOPIC` / `ingest::ACCOUNT_SUBMIT_TOPIC`
 /// (examples cannot import from the binary crate).
@@ -59,14 +59,14 @@ fn main() -> Result<()> {
     // `timestamp_ms_le[8] || key_package`.
     let mut kp_payload = seed_ms.to_le_bytes().to_vec();
     kp_payload.extend_from_slice(b"delivery-smoke-keypackage");
-    publish_json(
+    publish(
         &node,
         KEYPACKAGE_SUBMIT_TOPIC,
-        &json!({
+        &KeyPackageSubmissionV1 {
-            "device_id": device_id,
+            device_id: Bytes::copy_from_slice(device_key.verifying_key().as_bytes()),
-            "payload": BASE64.encode(&kp_payload),
+            payload: Bytes::copy_from_slice(&kp_payload),
-            "signature": BASE64.encode(device_key.sign(&kp_payload).to_bytes()),
+            signature: Bytes::copy_from_slice(&device_key.sign(&kp_payload).to_bytes()),
-        }),
+        },
     )?;
     println!("published keypackage submission on {KEYPACKAGE_SUBMIT_TOPIC}");
 
@@ -75,14 +75,14 @@ fn main() -> Result<()> {
     acct_payload.push(1u8);
     acct_payload.extend_from_slice(&1u64.to_le_bytes());
     acct_payload.extend_from_slice(b"delivery-smoke-devices");
-    publish_json(
+    publish(
         &node,
         ACCOUNT_SUBMIT_TOPIC,
-        &json!({
+        &AccountSubmissionV1 {
-            "account_pub": account_pub,
+            account_pub: Bytes::copy_from_slice(account_key.verifying_key().as_bytes()),
-            "payload": BASE64.encode(&acct_payload),
+            payload: Bytes::copy_from_slice(&acct_payload),
-            "signature": BASE64.encode(account_key.sign(&acct_payload).to_bytes()),
+            signature: Bytes::copy_from_slice(&account_key.sign(&acct_payload).to_bytes()),
-        }),
+        },
     )?;
     println!("published account submission on {ACCOUNT_SUBMIT_TOPIC}");
 
@@ -103,12 +103,12 @@ fn derived_key(seed_ms: u64, salt: u8) -> SigningKey {
     SigningKey::from_bytes(&bytes)
 }
 
-fn publish_json(
+fn publish<M: Message>(
     node: &ThreadedDeliveryWrapper<Vec<u8>>,
     topic: &str,
-    body: &serde_json::Value,
+    submission: &M,
 ) -> Result<()> {
-    node.publish(topic, body.to_string().as_bytes())
+    node.publish(topic, &submission.encode_to_vec())
         .map_err(|e| anyhow::anyhow!("publish on {topic}: {e}"))
 }
 

src/handlers.rs

@@ -33,13 +33,14 @@ pub fn router(store: Arc<Store>) -> Router {
         .with_state(store)
 }
 
-/// `POST /v0/keypackage` — same submission the logos-delivery subscriber
+/// `POST /v0/keypackage` — the same submission the logos-delivery subscriber
-/// accepts; verification and storage live in [`submit::apply_keypackage`].
+/// accepts, in JSON rather than protobuf; verification and storage live in
+/// [`submit::apply_keypackage`].
 async fn submit(
     State(store): State<Arc<Store>>,
     Json(req): Json<SubmitRequest>,
 ) -> Result<StatusCode, ApiError> {
-    submit::apply_keypackage(&store, &req).await?;
+    submit::apply_keypackage(&store, &req.decode()?).await?;
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -74,14 +75,14 @@ pub struct FetchAccountResponse {
 ///
 /// The server verifies the Ed25519 signature and then stores exactly one blob
 /// per `account_pub`, replacing any previous value. Clients should re-publish
-/// whenever they add or rotate LocalIdentities. Same submission the
+/// whenever they add or rotate LocalIdentities. The same submission the
-/// logos-delivery subscriber accepts; the shared rules live in
+/// logos-delivery subscriber accepts, in JSON rather than protobuf; the shared
-/// [`submit::apply_account`].
+/// rules live in [`submit::apply_account`].
 async fn submit_account(
     State(store): State<Arc<Store>>,
     Json(req): Json<SubmitAccountRequest>,
 ) -> Result<StatusCode, ApiError> {
-    submit::apply_account(&store, &req).await?;
+    submit::apply_account(&store, &req.decode()?).await?;
     Ok(StatusCode::NO_CONTENT)
 }
 

src/ingest.rs

@@ -3,18 +3,23 @@
 //! Runs an embedded logos-delivery node, subscribes to the store submission
 //! content topics, and feeds every received submission through the same
 //! verification + storage pipeline as the HTTP POST endpoints (`submit`).
-//! Publishing is fire-and-forget on the client side, so rejected submissions
+//! Submissions arrive as protobuf on this wire — matching the `/proto` content
-//! are only logged here — consumers verify every bundle on retrieval anyway.
+//! topics they are published on — and carry the same fields the JSON POST
+//! bodies do. Publishing is fire-and-forget on the client side, so rejected
+//! submissions are only logged here; consumers verify every bundle on
+//! retrieval anyway.
 
 use std::sync::Arc;
 
 use anyhow::{Context, Result};
-use logos_delivery::ThreadedDeliveryWrapper;
+use chat_proto::logoschat::store::{AccountSubmissionV1, KeyPackageSubmissionV1};
 pub use logos_delivery::P2pConfig;
+use logos_delivery::ThreadedDeliveryWrapper;
+use prost::Message;
 use tracing::{debug, warn};
 
 use crate::store::Store;
-use crate::submit::{self, SubmitAccountRequest, SubmitRequest};
+use crate::submit::{self, Submission};
 
 /// Content topic carrying keypackage submissions. Must match what libchat's
 /// `DeliveryRegistry` publishes: delivery address `store-keypackage-v0` mapped
@@ -24,10 +29,10 @@ pub const KEYPACKAGE_SUBMIT_TOPIC: &str = "/logos-chat/1/store-keypackage-v0/pro
 /// Content topic carrying account device-list bundle submissions.
 pub const ACCOUNT_SUBMIT_TOPIC: &str = "/logos-chat/1/store-account-v0/proto";
 
-/// A raw submission taken off the wire; the JSON body is parsed (and its
+/// A raw submission taken off the wire; the protobuf body is decoded (and its
 /// signature verified) on the ingest thread, not the node callback.
 #[derive(Clone)]
-enum Submission {
+enum Received {
     KeyPackage(Vec<u8>),
     Account(Vec<u8>),
 }
@@ -43,8 +48,8 @@ pub fn start(store: Arc<Store>, cfg: P2pConfig, runtime: tokio::runtime::Handle)
     let mut node = ThreadedDeliveryWrapper::start(cfg, |event| {
         let msg = event.into_received()?;
         let wrap = match msg.content_topic() {
-            KEYPACKAGE_SUBMIT_TOPIC => Submission::KeyPackage as fn(Vec<u8>) -> Submission,
+            KEYPACKAGE_SUBMIT_TOPIC => Received::KeyPackage as fn(Vec<u8>) -> Received,
-            ACCOUNT_SUBMIT_TOPIC => Submission::Account,
+            ACCOUNT_SUBMIT_TOPIC => Received::Account,
             _ => return None,
         };
         msg.into_payload().map(wrap)
@@ -62,10 +67,10 @@ pub fn start(store: Arc<Store>, cfg: P2pConfig, runtime: tokio::runtime::Handle)
         .spawn(move || {
             // Keep the node alive: dropping the last wrapper clone stops it.
             let _node = node;
-            while let Ok(submission) = inbound.recv() {
+            while let Ok(received) = inbound.recv() {
-                match submission {
+                match received {
-                    Submission::KeyPackage(bytes) => ingest_keypackage(&store, &runtime, &bytes),
+                    Received::KeyPackage(bytes) => ingest_keypackage(&store, &runtime, &bytes),
-                    Submission::Account(bytes) => ingest_account(&store, &runtime, &bytes),
+                    Received::Account(bytes) => ingest_account(&store, &runtime, &bytes),
                 }
             }
         })
@@ -74,29 +79,45 @@ pub fn start(store: Arc<Store>, cfg: P2pConfig, runtime: tokio::runtime::Handle)
 }
 
 fn ingest_keypackage(store: &Store, runtime: &tokio::runtime::Handle, bytes: &[u8]) {
-    let req: SubmitRequest = match serde_json::from_slice(bytes) {
+    let wire = match KeyPackageSubmissionV1::decode(bytes) {
-        Ok(req) => req,
+        Ok(wire) => wire,
         Err(e) => {
-            warn!("keypackage submission: invalid JSON: {e}");
+            warn!("keypackage submission: invalid protobuf: {e}");
             return;
         }
     };
-    match runtime.block_on(submit::apply_keypackage(store, &req)) {
+    let submission = match Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature) {
-        Ok(()) => debug!(device_id = %req.device_id, "stored keypackage from delivery"),
+        Ok(submission) => submission,
-        Err(e) => warn!(device_id = %req.device_id, "keypackage submission rejected: {e}"),
+        Err(e) => {
+            warn!("keypackage submission rejected: {e}");
+            return;
+        }
+    };
+    let device_id = submission.key_hex();
+    match runtime.block_on(submit::apply_keypackage(store, &submission)) {
+        Ok(()) => debug!(%device_id, "stored keypackage from delivery"),
+        Err(e) => warn!(%device_id, "keypackage submission rejected: {e}"),
     }
 }
 
 fn ingest_account(store: &Store, runtime: &tokio::runtime::Handle, bytes: &[u8]) {
-    let req: SubmitAccountRequest = match serde_json::from_slice(bytes) {
+    let wire = match AccountSubmissionV1::decode(bytes) {
-        Ok(req) => req,
+        Ok(wire) => wire,
         Err(e) => {
-            warn!("account submission: invalid JSON: {e}");
+            warn!("account submission: invalid protobuf: {e}");
             return;
         }
     };
-    match runtime.block_on(submit::apply_account(store, &req)) {
+    let submission = match Submission::account(&wire.account_pub, &wire.payload, &wire.signature) {
-        Ok(()) => debug!(account_pub = %req.account_pub, "stored account bundle from delivery"),
+        Ok(submission) => submission,
-        Err(e) => warn!(account_pub = %req.account_pub, "account submission rejected: {e}"),
+        Err(e) => {
+            warn!("account submission rejected: {e}");
+            return;
+        }
+    };
+    let account_pub = submission.key_hex();
+    match runtime.block_on(submit::apply_account(store, &submission)) {
+        Ok(()) => debug!(%account_pub, "stored account bundle from delivery"),
+        Err(e) => warn!(%account_pub, "account submission rejected: {e}"),
     }
 }

src/submit.rs

@@ -1,9 +1,11 @@
 //! Shared submission pipeline.
 //!
-//! Both write paths — the HTTP POST endpoints and the logos-delivery
+//! The two write paths carry the same fields in the encoding native to each
-//! subscriber (`ingest`) — carry the same JSON submissions and feed them
+//! wire: the HTTP POST endpoints take a JSON body with hex and base64 strings,
-//! through these functions, so signature verification and storage rules are
+//! while the logos-delivery subscriber (`ingest`) takes a protobuf message
-//! identical no matter which wire delivered the request.
+//! with raw bytes. Both decode into a [`Submission`] and feed it through these
+//! functions, so signature verification and storage rules are identical no
+//! matter which wire delivered the request.
 
 use base64::Engine;
 use base64::engine::general_purpose::STANDARD as BASE64;
@@ -26,6 +28,13 @@ pub struct SubmitRequest {
     pub signature: String,
 }
 
+impl SubmitRequest {
+    /// Decode the JSON body's hex + base64 fields into a [`Submission`].
+    pub fn decode(&self) -> Result<Submission, SubmitError> {
+        Submission::from_encoded(&self.device_id, &self.payload, &self.signature, DEVICE_ID)
+    }
+}
+
 /// A signed account device-list bundle submission.
 ///
 /// The `payload` is intentionally opaque to the server. Clients are expected
@@ -45,6 +54,18 @@ pub struct SubmitAccountRequest {
     pub signature: String,
 }
 
+impl SubmitAccountRequest {
+    /// Decode the JSON body's hex + base64 fields into a [`Submission`].
+    pub fn decode(&self) -> Result<Submission, SubmitError> {
+        Submission::from_encoded(
+            &self.account_pub,
+            &self.payload,
+            &self.signature,
+            ACCOUNT_PUB,
+        )
+    }
+}
+
 #[derive(Debug)]
 pub enum SubmitError {
     /// Malformed submission or failed signature verification.
@@ -67,70 +88,134 @@ impl std::fmt::Display for SubmitError {
     }
 }
 
-/// Decoded and signature-verified fields common to both submission kinds.
+/// A submission decoded off either wire, before verification: the raw key,
-struct Verified {
+/// payload and signature bytes that both encodings ultimately carry.
+#[derive(Debug)]
+pub struct Submission {
+    key: [u8; 32],
     payload: Vec<u8>,
     signature: [u8; 64],
 }
 
+impl Submission {
+    /// A keypackage submission from raw bytes, as the protobuf wire carries them.
+    pub fn keypackage(
+        device_id: &[u8],
+        payload: &[u8],
+        signature: &[u8],
+    ) -> Result<Self, SubmitError> {
+        Self::from_bytes(device_id, payload, signature, DEVICE_ID)
+    }
+
+    /// An account device-list submission from raw bytes.
+    pub fn account(
+        account_pub: &[u8],
+        payload: &[u8],
+        signature: &[u8],
+    ) -> Result<Self, SubmitError> {
+        Self::from_bytes(account_pub, payload, signature, ACCOUNT_PUB)
+    }
+
+    fn from_bytes(
+        key: &[u8],
+        payload: &[u8],
+        signature: &[u8],
+        key_field: KeyField,
+    ) -> Result<Self, SubmitError> {
+        Ok(Self {
+            key: key
+                .try_into()
+                .map_err(|_| SubmitError::Invalid(key_field.not_32_bytes))?,
+            payload: payload.to_vec(),
+            signature: signature
+                .try_into()
+                .map_err(|_| SubmitError::Invalid("signature: must be 64 bytes"))?,
+        })
+    }
+
+    /// From the HTTP body's hex key and base64 payload/signature.
+    fn from_encoded(
+        key_hex: &str,
+        payload_b64: &str,
+        signature_b64: &str,
+        key_field: KeyField,
+    ) -> Result<Self, SubmitError> {
+        let key: [u8; 32] = hex::decode(key_hex)
+            .ok()
+            .and_then(|b| b.try_into().ok())
+            .ok_or(SubmitError::Invalid(key_field.not_hex))?;
+        let payload = BASE64
+            .decode(payload_b64)
+            .map_err(|_| SubmitError::Invalid("payload: not valid base64"))?;
+        let signature: [u8; 64] = BASE64
+            .decode(signature_b64)
+            .ok()
+            .and_then(|b| b.try_into().ok())
+            .ok_or(SubmitError::Invalid(
+                "signature: must be base64 of 64 bytes",
+            ))?;
+        Ok(Self {
+            key,
+            payload,
+            signature,
+        })
+    }
+
+    /// Hex of the submission's key — the store's lookup key. Canonical
+    /// lowercase, so the same key always lands in the same row whichever wire
+    /// and whichever casing it arrived in.
+    pub fn key_hex(&self) -> String {
+        hex::encode(self.key)
+    }
+}
+
 /// Verify proof-of-possession before persisting. `payload` is opaque — the
 /// server only checks that `signature` over the received payload bytes is
-/// valid under `key_hex`'s key. A valid signature means the submitter holds
+/// valid under the submission's key. A valid signature means the submitter
-/// that key. This rejects junk early (DoS mitigation); consumers still verify
+/// holds that key. This rejects junk early (DoS mitigation); consumers still
-/// on retrieve, the server is not a trusted authority.
+/// verify on retrieve, the server is not a trusted authority.
-fn verify(
+fn verify(submission: &Submission, key_field: KeyField) -> Result<(), SubmitError> {
-    key_hex: &str,
+    let verifying_key = VerifyingKey::from_bytes(&submission.key)
-    payload_b64: &str,
+        .map_err(|_| SubmitError::Invalid(key_field.not_key))?;
-    signature_b64: &str,
-    key_field: KeyField,
-) -> Result<Verified, SubmitError> {
-    let pubkey: [u8; 32] = hex::decode(key_hex)
-        .ok()
-        .and_then(|b| b.try_into().ok())
-        .ok_or(SubmitError::Invalid(key_field.not_hex))?;
-    let payload = BASE64
-        .decode(payload_b64)
-        .map_err(|_| SubmitError::Invalid("payload: not valid base64"))?;
-    let signature: [u8; 64] = BASE64
-        .decode(signature_b64)
-        .ok()
-        .and_then(|b| b.try_into().ok())
-        .ok_or(SubmitError::Invalid("signature: must be base64 of 64 bytes"))?;
-
-    let verifying_key =
-        VerifyingKey::from_bytes(&pubkey).map_err(|_| SubmitError::Invalid(key_field.not_key))?;
     verifying_key
-        .verify_strict(&payload, &Signature::from_bytes(&signature))
+        .verify_strict(
+            &submission.payload,
+            &Signature::from_bytes(&submission.signature),
+        )
         .map_err(|_| SubmitError::Invalid("signature: verification failed"))?;
-    Ok(Verified { payload, signature })
+    Ok(())
 }
 
 /// Error messages named after the submission's key field, so HTTP responses
 /// and ingest logs point at the right field.
+#[derive(Clone, Copy)]
 struct KeyField {
     not_hex: &'static str,
+    not_32_bytes: &'static str,
     not_key: &'static str,
 }
 
 const DEVICE_ID: KeyField = KeyField {
     not_hex: "device_id: must be hex of a 32-byte key",
+    not_32_bytes: "device_id: must be 32 bytes",
     not_key: "device_id: not a valid ed25519 key",
 };
 
 const ACCOUNT_PUB: KeyField = KeyField {
     not_hex: "account_pub: must be hex of a 32-byte key",
+    not_32_bytes: "account_pub: must be 32 bytes",
     not_key: "account_pub: not a valid ed25519 key",
 };
 
 /// Verify and store a keypackage bundle submission.
-pub async fn apply_keypackage(store: &Store, req: &SubmitRequest) -> Result<(), SubmitError> {
+pub async fn apply_keypackage(store: &Store, submission: &Submission) -> Result<(), SubmitError> {
-    let verified = verify(&req.device_id, &req.payload, &req.signature, DEVICE_ID)?;
+    verify(submission, DEVICE_ID)?;
     store
         .insert(
-            &req.device_id,
+            &submission.key_hex(),
             &StoredKeyPackageBundle {
-                payload: verified.payload,
+                payload: submission.payload.clone(),
-                signature: verified.signature.to_vec(),
+                signature: submission.signature.to_vec(),
             },
         )
         .await
@@ -138,23 +223,23 @@ pub async fn apply_keypackage(store: &Store, req: &SubmitRequest) -> Result<(),
 }
 
 /// Verify and upsert an account device-list bundle submission.
-pub async fn apply_account(store: &Store, req: &SubmitAccountRequest) -> Result<(), SubmitError> {
+pub async fn apply_account(store: &Store, submission: &Submission) -> Result<(), SubmitError> {
-    let verified = verify(&req.account_pub, &req.payload, &req.signature, ACCOUNT_PUB)?;
+    verify(submission, ACCOUNT_PUB)?;
 
     // Read the bundle's lamport so the store can reject replays. Safe to trust:
     // the signature over `payload` was just verified, so the lamport can't be
     // forged without the account key.
-    let lamport = crate::store::payload_lamport(&verified.payload).ok_or(SubmitError::Invalid(
+    let lamport = crate::store::payload_lamport(&submission.payload).ok_or(SubmitError::Invalid(
         "payload: too short to contain a lamport header",
     ))?;
 
     let applied = store
         .upsert_account(
-            &req.account_pub,
+            &submission.key_hex(),
             lamport,
             &StoredAccountBundle {
-                payload: verified.payload,
+                payload: submission.payload.clone(),
-                signature: verified.signature.to_vec(),
+                signature: submission.signature.to_vec(),
                 updated_at: 0, // filled in by store
             },
         )
@@ -170,7 +255,10 @@ pub async fn apply_account(store: &Store, req: &SubmitAccountRequest) -> Result<
 mod tests {
     use std::path::Path;
 
+    use chat_proto::logoschat::store::{AccountSubmissionV1, KeyPackageSubmissionV1};
     use ed25519_dalek::{Signer, SigningKey};
+    use prost::Message;
+    use prost::bytes::Bytes;
     use serde_json::json;
 
     use super::*;
@@ -182,8 +270,18 @@ mod tests {
         SigningKey::from_bytes(&[seed; 32])
     }
 
-    /// A submission exactly as it arrives off the delivery wire: the JSON body
+    /// A keypackage submission exactly as it arrives off the delivery wire:
-    /// libchat's `DeliveryRegistry` publishes (same shape as the POST body).
+    /// the protobuf message libchat's `DeliveryRegistry` publishes.
+    fn keypackage_submission_proto(key: &SigningKey, payload: &[u8]) -> Vec<u8> {
+        KeyPackageSubmissionV1 {
+            device_id: Bytes::copy_from_slice(key.verifying_key().as_bytes()),
+            payload: Bytes::copy_from_slice(payload),
+            signature: Bytes::copy_from_slice(&key.sign(payload).to_bytes()),
+        }
+        .encode_to_vec()
+    }
+
+    /// The same submission as the HTTP POST body.
     fn keypackage_submission_json(key: &SigningKey, payload: &[u8]) -> Vec<u8> {
         json!({
             "device_id": hex::encode(key.verifying_key().as_bytes()),
@@ -201,25 +299,49 @@ mod tests {
         p
     }
 
-    fn account_submission(key: &SigningKey, payload: &[u8]) -> SubmitAccountRequest {
+    fn account_submission(key: &SigningKey, payload: &[u8]) -> Submission {
-        SubmitAccountRequest {
+        Submission::account(
-            account_pub: hex::encode(key.verifying_key().as_bytes()),
+            key.verifying_key().as_bytes(),
-            payload: BASE64.encode(payload),
+            payload,
-            signature: BASE64.encode(key.sign(payload).to_bytes()),
+            &key.sign(payload).to_bytes(),
-        }
+        )
+        .unwrap()
+    }
+
+    /// The protobuf wire and the JSON body must decode to the same submission,
+    /// so the store applies identical rules whichever path delivered it.
+    #[tokio::test]
+    async fn proto_and_json_keypackage_wires_agree() {
+        let key = signing_key(1);
+        let payload = b"ts-and-keypackage-bytes".to_vec();
+
+        let wire = KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, &payload)[..])
+            .unwrap();
+        let from_proto =
+            Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature).unwrap();
+
+        let body: SubmitRequest =
+            serde_json::from_slice(&keypackage_submission_json(&key, &payload)).unwrap();
+        let from_json = body.decode().unwrap();
+
+        assert_eq!(from_proto.key_hex(), from_json.key_hex());
+        assert_eq!(from_proto.payload, from_json.payload);
+        assert_eq!(from_proto.signature, from_json.signature);
     }
 
     #[tokio::test]
-    async fn wire_json_keypackage_is_parsed_verified_and_stored() {
+    async fn wire_proto_keypackage_is_parsed_verified_and_stored() {
         let store = Store::open(Path::new(":memory:")).await.unwrap();
         let key = signing_key(1);
         let payload = b"ts-and-keypackage-bytes".to_vec();
 
-        let wire = keypackage_submission_json(&key, &payload);
+        let wire = KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, &payload)[..])
-        let req: SubmitRequest = serde_json::from_slice(&wire).unwrap();
+            .unwrap();
-        apply_keypackage(&store, &req).await.unwrap();
+        let submission =
+            Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature).unwrap();
+        apply_keypackage(&store, &submission).await.unwrap();
 
-        let stored = store.latest(&req.device_id).await.unwrap().unwrap();
+        let stored = store.latest(&submission.key_hex()).await.unwrap().unwrap();
         assert_eq!(stored.payload, payload);
     }
 
@@ -228,13 +350,37 @@ mod tests {
         let store = Store::open(Path::new(":memory:")).await.unwrap();
         let key = signing_key(2);
 
-        let mut req: SubmitRequest =
+        let mut wire =
-            serde_json::from_slice(&keypackage_submission_json(&key, b"original")).unwrap();
+            KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, b"original")[..])
-        req.payload = BASE64.encode(b"tampered");
+                .unwrap();
+        wire.payload = Bytes::from_static(b"tampered");
+        let submission =
+            Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature).unwrap();
 
-        let err = apply_keypackage(&store, &req).await.unwrap_err();
+        let err = apply_keypackage(&store, &submission).await.unwrap_err();
-        assert!(matches!(err, SubmitError::Invalid("signature: verification failed")));
+        assert!(matches!(
-        assert!(store.latest(&req.device_id).await.unwrap().is_none());
+            err,
+            SubmitError::Invalid("signature: verification failed")
+        ));
+        assert!(store.latest(&submission.key_hex()).await.unwrap().is_none());
+    }
+
+    /// A truncated or foreign payload on the submission topics must be dropped,
+    /// not panic the ingest thread.
+    #[test]
+    fn non_protobuf_bytes_are_rejected() {
+        assert!(KeyPackageSubmissionV1::decode(&b"{\"device_id\":\"xx\"}"[..]).is_err());
+        assert!(AccountSubmissionV1::decode(&[0xff, 0xff, 0xff][..]).is_err());
+    }
+
+    /// A key that is not 32 bytes is rejected at decode, before verification.
+    #[test]
+    fn short_key_is_rejected() {
+        let err = Submission::keypackage(&[1u8; 31], b"payload", &[0u8; 64]).unwrap_err();
+        assert!(matches!(
+            err,
+            SubmitError::Invalid("device_id: must be 32 bytes")
+        ));
     }
 
     #[tokio::test]