feat: refactor modules splition

Cargo.toml

@@ -15,15 +15,10 @@ path = "src/main.rs"
 anyhow = "1.0"
 axum = "0.7"
 base64 = "0.22"
-# Wire schema for the delivery submissions, shared with libchat. Pinned to a rev
-# rather than a branch so the wire format can only change deliberately.
 chat-proto = { git = "https://github.com/logos-messaging/chat_proto", rev = "e05c21229a18bab655c9aec06409189447d24a6b" }
 clap = { version = "4", features = ["derive"] }
 ed25519-dalek = "2.2.0"
 hex = "0.4"
-# Embedded logos-delivery node for subscription-based ingestion. Links the
-# native liblogosdelivery, which this repo's flake builds: `nix develop` exports
-# LOGOS_DELIVERY_LIB_DIR, without which the binary fails at link.
 logos-delivery = { git = "https://github.com/logos-messaging/libchat", rev = "5c55c2ee76bebd1dbd5eb4bfa95d9e71acd0fa14" }
 prost = "0.14"
 serde = { version = "1.0", features = ["derive"] }

examples/delivery_smoke_test.rs

@@ -22,7 +22,7 @@ use prost::Message;
 use prost::bytes::Bytes;
 use reqwest::blocking::Client;
 
-/// Must match `ingest::KEYPACKAGE_SUBMIT_TOPIC` / `ingest::ACCOUNT_SUBMIT_TOPIC`
+/// Must match `delivery::KEYPACKAGE_SUBMIT_TOPIC` / `delivery::ACCOUNT_SUBMIT_TOPIC`
 /// (examples cannot import from the binary crate).
 const KEYPACKAGE_SUBMIT_TOPIC: &str = "/logos-chat/1/store-keypackage-v0/proto";
 const ACCOUNT_SUBMIT_TOPIC: &str = "/logos-chat/1/store-account-v0/proto";
@@ -91,7 +91,11 @@ fn main() -> Result<()> {
         &format!("{base}/v0/keypackage/{device_id}"),
         "keypackage",
     )?;
-    poll_until_stored(&client, &format!("{base}/v0/account/{account_pub}"), "account")?;
+    poll_until_stored(
+        &client,
+        &format!("{base}/v0/account/{account_pub}"),
+        "account",
+    )?;
 
     println!("delivery smoke test passed");
     Ok(())
@@ -123,7 +127,10 @@ fn poll_until_stored(client: &Client, url: &str, what: &str) -> Result<()> {
             .with_context(|| format!("GET {url}"))?
             .status();
         if status.is_success() {
-            println!("GET {url} -> {status} ({what} stored, {:?})", started.elapsed());
+            println!(
+                "GET {url} -> {status} ({what} stored, {:?})",
+                started.elapsed()
+            );
             return Ok(());
         }
         if started.elapsed() > POLL_BUDGET {

examples/smoke_test.rs

@@ -56,7 +56,10 @@ fn test_keypackage(client: &Client, base: &str) -> Result<()> {
             "signature": BASE64.encode(key.sign(&payload).to_bytes()),
         }))
         .send()?;
-    println!("POST /v0/keypackage        -> {} (expect 204)", resp.status());
+    println!(
+        "POST /v0/keypackage        -> {} (expect 204)",
+        resp.status()
+    );
 
     let resp = client
         .get(format!("{base}/v0/keypackage/{device_id}"))
@@ -91,9 +94,14 @@ fn test_account(client: &Client, base: &str) -> Result<()> {
         .post(format!("{base}/v0/account"))
         .json(&body)
         .send()?;
-    println!("POST /v0/account           -> {} (expect 204)", resp.status());
+    println!(
+        "POST /v0/account           -> {} (expect 204)",
+        resp.status()
+    );
 
-    let resp = client.get(format!("{base}/v0/account/{account_pub}")).send()?;
+    let resp = client
+        .get(format!("{base}/v0/account/{account_pub}"))
+        .send()?;
     println!(
         "GET  /v0/account/<id>      -> {} (expect 200) {}",
         resp.status(),
@@ -105,7 +113,10 @@ fn test_account(client: &Client, base: &str) -> Result<()> {
         .post(format!("{base}/v0/account"))
         .json(&body)
         .send()?;
-    println!("POST /v0/account (replay)  -> {} (expect 409)", resp.status());
+    println!(
+        "POST /v0/account (replay)  -> {} (expect 409)",
+        resp.status()
+    );
     Ok(())
 }
 

src/bundle.rs

@@ -0,0 +1,277 @@
+//! Bundles: an opaque payload plus its signature, published under the key that
+//! signed it. This is what the store holds and what both write paths carry.
+//!
+//! Each wire encodes a bundle in its own way — the HTTP POST endpoints take a
+//! JSON body with hex and base64 strings, the logos-delivery subscriber
+//! (`delivery`) takes a protobuf message with raw bytes — and both decode into
+//! a [`Bundle`]. Verification and storage rules below are therefore identical
+//! no matter which wire carried it.
+
+use ed25519_dalek::{Signature, VerifyingKey};
+
+use crate::store::{Store, StoredAccountBundle, StoredKeyPackageBundle};
+
+#[derive(Debug)]
+pub enum BundleError {
+    /// Malformed bundle or failed signature verification.
+    Invalid(&'static str),
+    /// Valid account bundle whose lamport is not newer than the stored one.
+    Stale,
+    /// Storage failure.
+    Internal(anyhow::Error),
+}
+
+impl std::fmt::Display for BundleError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            BundleError::Invalid(msg) => write!(f, "{msg}"),
+            BundleError::Stale => {
+                write!(f, "stale bundle: lamport is not newer than the stored one")
+            }
+            BundleError::Internal(err) => write!(f, "internal: {err}"),
+        }
+    }
+}
+
+/// A bundle decoded off either wire, before verification: the raw key,
+/// payload and signature bytes that both encodings ultimately carry.
+#[derive(Debug)]
+pub struct Bundle {
+    key: [u8; 32],
+    payload: Vec<u8>,
+    signature: [u8; 64],
+}
+
+impl Bundle {
+    /// From raw bytes, as the protobuf wire carries them. Keypackage and
+    /// account bundles are the same shape here — which kind it is comes from
+    /// the topic it arrived on and the `apply_*` it is passed to.
+    pub fn from_bytes(key: &[u8], payload: &[u8], signature: &[u8]) -> Result<Self, BundleError> {
+        Ok(Self {
+            key: key
+                .try_into()
+                .map_err(|_| BundleError::Invalid("key: must be 32 bytes"))?,
+            payload: payload.to_vec(),
+            signature: signature
+                .try_into()
+                .map_err(|_| BundleError::Invalid("signature: must be 64 bytes"))?,
+        })
+    }
+
+    /// Hex of the bundle'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 the bundle's key. A valid signature means the submitter
+/// holds that key. This rejects junk early (DoS mitigation); consumers still
+/// verify on retrieve, the server is not a trusted authority.
+fn verify(bundle: &Bundle) -> Result<(), BundleError> {
+    let verifying_key = VerifyingKey::from_bytes(&bundle.key)
+        .map_err(|_| BundleError::Invalid("key: not a valid ed25519 key"))?;
+    verifying_key
+        .verify_strict(&bundle.payload, &Signature::from_bytes(&bundle.signature))
+        .map_err(|_| BundleError::Invalid("signature: verification failed"))?;
+    Ok(())
+}
+
+/// Verify and store a keypackage bundle.
+pub async fn apply_keypackage(store: &Store, bundle: &Bundle) -> Result<(), BundleError> {
+    verify(bundle)?;
+    store
+        .insert(
+            &bundle.key_hex(),
+            &StoredKeyPackageBundle {
+                payload: bundle.payload.clone(),
+                signature: bundle.signature.to_vec(),
+            },
+        )
+        .await
+        .map_err(BundleError::Internal)
+}
+
+/// Verify and upsert an account device-list bundle.
+pub async fn apply_account(store: &Store, bundle: &Bundle) -> Result<(), BundleError> {
+    verify(bundle)?;
+
+    // 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(&bundle.payload).ok_or(BundleError::Invalid(
+        "payload: too short to contain a lamport header",
+    ))?;
+
+    let applied = store
+        .upsert_account(
+            &bundle.key_hex(),
+            lamport,
+            &StoredAccountBundle {
+                payload: bundle.payload.clone(),
+                signature: bundle.signature.to_vec(),
+                updated_at: 0, // filled in by store
+            },
+        )
+        .await
+        .map_err(BundleError::Internal)?;
+    if !applied {
+        return Err(BundleError::Stale);
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use std::path::Path;
+
+    use base64::Engine;
+    use base64::engine::general_purpose::STANDARD as BASE64;
+    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::*;
+    // The JSON body lives with the HTTP wire that owns it; this test asserts
+    // that wire and the protobuf one decode to the same bundle.
+    use crate::handlers::SubmitKeyPackageRequest;
+
+    /// Must match `BUNDLE_DOMAIN` in `store.rs` (kept private there).
+    const ACCOUNT_BUNDLE_DOMAIN: &[u8] = b"libchat:account-device-bundle\0";
+
+    fn signing_key(seed: u8) -> SigningKey {
+        SigningKey::from_bytes(&[seed; 32])
+    }
+
+    /// A keypackage bundle exactly as it arrives off the delivery wire:
+    /// 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 bundle 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()),
+            "payload": BASE64.encode(payload),
+            "signature": BASE64.encode(key.sign(payload).to_bytes()),
+        })
+        .to_string()
+        .into_bytes()
+    }
+
+    fn account_payload(lamport: u64) -> Vec<u8> {
+        let mut p = ACCOUNT_BUNDLE_DOMAIN.to_vec();
+        p.push(1u8); // version
+        p.extend_from_slice(&lamport.to_le_bytes());
+        p
+    }
+
+    fn account_submission(key: &SigningKey, payload: &[u8]) -> Bundle {
+        Bundle::from_bytes(
+            key.verifying_key().as_bytes(),
+            payload,
+            &key.sign(payload).to_bytes(),
+        )
+        .unwrap()
+    }
+
+    /// The protobuf wire and the JSON body must decode to the same bundle,
+    /// 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 =
+            Bundle::from_bytes(&wire.device_id, &wire.payload, &wire.signature).unwrap();
+
+        let body: SubmitKeyPackageRequest =
+            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_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 = KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, &payload)[..])
+            .unwrap();
+        let bundle = Bundle::from_bytes(&wire.device_id, &wire.payload, &wire.signature).unwrap();
+        apply_keypackage(&store, &bundle).await.unwrap();
+
+        let stored = store.latest(&bundle.key_hex()).await.unwrap().unwrap();
+        assert_eq!(stored.payload, payload);
+    }
+
+    #[tokio::test]
+    async fn tampered_keypackage_bundle_is_rejected() {
+        let store = Store::open(Path::new(":memory:")).await.unwrap();
+        let key = signing_key(2);
+
+        let mut wire =
+            KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, b"original")[..])
+                .unwrap();
+        wire.payload = Bytes::from_static(b"tampered");
+        let bundle = Bundle::from_bytes(&wire.device_id, &wire.payload, &wire.signature).unwrap();
+
+        let err = apply_keypackage(&store, &bundle).await.unwrap_err();
+        assert!(matches!(
+            err,
+            BundleError::Invalid("signature: verification failed")
+        ));
+        assert!(store.latest(&bundle.key_hex()).await.unwrap().is_none());
+    }
+
+    /// A truncated or foreign payload on the bundle 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 = Bundle::from_bytes(&[1u8; 31], b"payload", &[0u8; 64]).unwrap_err();
+        assert!(matches!(err, BundleError::Invalid("key: must be 32 bytes")));
+    }
+
+    #[tokio::test]
+    async fn account_bundle_upserts_and_rejects_stale_replay() {
+        let store = Store::open(Path::new(":memory:")).await.unwrap();
+        let key = signing_key(3);
+
+        apply_account(&store, &account_submission(&key, &account_payload(1)))
+            .await
+            .unwrap();
+        apply_account(&store, &account_submission(&key, &account_payload(2)))
+            .await
+            .unwrap();
+
+        // Replaying the lamport-2 bundle (as a delivery duplicate would) is stale.
+        let err = apply_account(&store, &account_submission(&key, &account_payload(2)))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, BundleError::Stale));
+    }
+}

src/delivery.rs

@@ -1,13 +1,13 @@
-//! logos-delivery ingestion.
+//! The logos-delivery write path.
 //!
-//! Runs an embedded logos-delivery node, subscribes to the store submission
+//! Runs an embedded logos-delivery node, subscribes to the store's content
-//! content topics, and feeds every received submission through the same
+//! topics, and feeds every bundle it receives through the same verification +
-//! verification + storage pipeline as the HTTP POST endpoints (`submit`).
+//! storage pipeline as the HTTP POST endpoints (`bundle`). Bundles arrive as
-//! Submissions arrive as protobuf on this wire — matching the `/proto` content
+//! protobuf on this wire — matching the `/proto` content topics they are
-//! topics they are published on — and carry the same fields the JSON POST
+//! published on — carrying the same fields the JSON POST bodies do.
-//! bodies do. Publishing is fire-and-forget on the client side, so rejected
+//!
-//! submissions are only logged here; consumers verify every bundle on
+//! Publishing is fire-and-forget on the client side, so rejected bundles are
-//! retrieval anyway.
+//! only logged here; consumers verify every bundle on retrieval anyway.
 
 use std::sync::Arc;
 
@@ -18,18 +18,18 @@ use logos_delivery::ThreadedDeliveryWrapper;
 use prost::Message;
 use tracing::{debug, warn};
 
+use crate::bundle::{Bundle, apply_account, apply_keypackage};
 use crate::store::Store;
-use crate::submit::{self, Submission};
 
-/// Content topic carrying keypackage submissions. Must match what libchat's
+/// Content topic carrying keypackage bundles. Must match what libchat's
-/// `DeliveryRegistry` publishes: delivery address `store-keypackage-v0` mapped
+/// `ContactRegistry` publishes: delivery address `store-keypackage-v0` mapped
 /// through the `/logos-chat/1/{address}/proto` content-topic scheme.
 pub const KEYPACKAGE_SUBMIT_TOPIC: &str = "/logos-chat/1/store-keypackage-v0/proto";
 
-/// Content topic carrying account device-list bundle submissions.
+/// Content topic carrying account device-list bundles.
 pub const ACCOUNT_SUBMIT_TOPIC: &str = "/logos-chat/1/store-account-v0/proto";
 
-/// A raw submission taken off the wire; the protobuf body is decoded (and its
+/// A raw bundle taken off the wire; the protobuf body is decoded (and its
 /// signature verified) on the ingest thread, not the node callback.
 #[derive(Clone)]
 enum Received {
@@ -37,12 +37,12 @@ enum Received {
     Account(Vec<u8>),
 }
 
-/// Start the embedded node, subscribe to the submission topics, and spawn the
+/// Start the embedded node, subscribe to the bundle topics, and spawn the
 /// ingest thread. The node lives as long as the returned thread does — i.e.
 /// the whole process; there is no shutdown handshake for this testnet service.
 ///
 /// `runtime` is the server's tokio handle: the store is async, but the
-/// delivery wrapper hands messages to a plain thread, so each submission is
+/// delivery wrapper hands messages to a plain thread, so each bundle is
 /// bridged back with `block_on`.
 pub fn start(store: Arc<Store>, cfg: P2pConfig, runtime: tokio::runtime::Handle) -> Result<()> {
     let mut node = ThreadedDeliveryWrapper::start(cfg, |event| {
@@ -57,9 +57,9 @@ pub fn start(store: Arc<Store>, cfg: P2pConfig, runtime: tokio::runtime::Handle)
     .context("start embedded logos-delivery node")?;
 
     node.subscribe(KEYPACKAGE_SUBMIT_TOPIC)
-        .context("subscribe keypackage submissions")?;
+        .context("subscribe keypackage bundles")?;
     node.subscribe(ACCOUNT_SUBMIT_TOPIC)
-        .context("subscribe account submissions")?;
+        .context("subscribe account bundles")?;
 
     let inbound = node.inbound_queue();
     std::thread::Builder::new()
@@ -82,21 +82,21 @@ fn ingest_keypackage(store: &Store, runtime: &tokio::runtime::Handle, bytes: &[u
     let wire = match KeyPackageSubmissionV1::decode(bytes) {
         Ok(wire) => wire,
         Err(e) => {
-            warn!("keypackage submission: invalid protobuf: {e}");
+            warn!("keypackage bundle: invalid protobuf: {e}");
             return;
         }
     };
-    let submission = match Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature) {
+    let bundle = match Bundle::from_bytes(&wire.device_id, &wire.payload, &wire.signature) {
-        Ok(submission) => submission,
+        Ok(bundle) => bundle,
         Err(e) => {
-            warn!("keypackage submission rejected: {e}");
+            warn!("keypackage bundle rejected: {e}");
             return;
         }
     };
-    let device_id = submission.key_hex();
+    let device_id = bundle.key_hex();
-    match runtime.block_on(submit::apply_keypackage(store, &submission)) {
+    match runtime.block_on(apply_keypackage(store, &bundle)) {
         Ok(()) => debug!(%device_id, "stored keypackage from delivery"),
-        Err(e) => warn!(%device_id, "keypackage submission rejected: {e}"),
+        Err(e) => warn!(%device_id, "keypackage bundle rejected: {e}"),
     }
 }
 
@@ -104,20 +104,20 @@ fn ingest_account(store: &Store, runtime: &tokio::runtime::Handle, bytes: &[u8])
     let wire = match AccountSubmissionV1::decode(bytes) {
         Ok(wire) => wire,
         Err(e) => {
-            warn!("account submission: invalid protobuf: {e}");
+            warn!("account bundle: invalid protobuf: {e}");
             return;
         }
     };
-    let submission = match Submission::account(&wire.account_pub, &wire.payload, &wire.signature) {
+    let bundle = match Bundle::from_bytes(&wire.account_pub, &wire.payload, &wire.signature) {
-        Ok(submission) => submission,
+        Ok(bundle) => bundle,
         Err(e) => {
-            warn!("account submission rejected: {e}");
+            warn!("account bundle rejected: {e}");
             return;
         }
     };
-    let account_pub = submission.key_hex();
+    let account_pub = bundle.key_hex();
-    match runtime.block_on(submit::apply_account(store, &submission)) {
+    match runtime.block_on(apply_account(store, &bundle)) {
         Ok(()) => debug!(%account_pub, "stored account bundle from delivery"),
-        Err(e) => warn!(%account_pub, "account submission rejected: {e}"),
+        Err(e) => warn!(%account_pub, "account bundle rejected: {e}"),
     }
 }

src/handlers.rs

@@ -7,13 +7,78 @@ use axum::routing::{get, post};
 use axum::{Json, Router};
 use base64::Engine;
 use base64::engine::general_purpose::STANDARD as BASE64;
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 
+use crate::bundle::{self, Bundle, BundleError};
 use crate::store::Store;
-use crate::submit::{self, SubmitAccountRequest, SubmitError, SubmitRequest};
+
+/// A signed keypackage bundle.
+#[derive(Debug, Deserialize)]
+pub struct SubmitKeyPackageRequest {
+    /// Hex of the 32-byte Ed25519 device verifying key. Used to verify the
+    /// signature and as the storage/lookup key. `payload` stays opaque.
+    pub device_id: String,
+    /// base64 of the signed payload. Opaque to the server — it never decodes it.
+    pub payload: String,
+    /// base64 of the 64-byte Ed25519 signature over `payload`. Verifying it
+    /// under `device_id`'s key is proof-of-possession: only the holder of that
+    /// key can publish under this `device_id`.
+    pub signature: String,
+}
+
+impl SubmitKeyPackageRequest {
+    /// Decode the JSON body's hex + base64 fields into a [`Bundle`].
+    pub fn decode(&self) -> Result<Bundle, BundleError> {
+        decode_bundle(&self.device_id, &self.payload, &self.signature)
+    }
+}
+
+/// A signed account device-list bundle.
+///
+/// The `payload` is intentionally opaque to the server. Clients are expected
+/// to encode a lamport-timestamped list of device (LocalIdentity) Ed25519
+/// public keys inside it so that consumers can detect stale bundles. The server
+/// only verifies that `signature` is a valid Ed25519 signature over `payload`
+/// made by the key identified by `account_pub`.
+#[derive(Debug, Deserialize)]
+pub struct SubmitAccountRequest {
+    /// Hex of the 32-byte Ed25519 account (AccountAddress) verifying key.
+    /// Acts as both the storage key and the verification key.
+    pub account_pub: String,
+    /// base64 of the opaque signed payload (lamport-ts + device pubkeys, etc.).
+    pub payload: String,
+    /// base64 of the 64-byte Ed25519 signature over `payload` made by the
+    /// account key. Proof-of-possession: only the account holder can publish.
+    pub signature: String,
+}
+
+impl SubmitAccountRequest {
+    /// Decode the JSON body's hex + base64 fields into a [`Bundle`].
+    pub fn decode(&self) -> Result<Bundle, BundleError> {
+        decode_bundle(&self.account_pub, &self.payload, &self.signature)
+    }
+}
+
+/// This wire spells a bundle out in text: the key as hex, payload and signature
+/// as base64. Undo that here — the delivery wire hands over raw bytes already —
+/// and let [`Bundle::from_bytes`] enforce the lengths.
+fn decode_bundle(
+    key_hex: &str,
+    payload_b64: &str,
+    signature_b64: &str,
+) -> Result<Bundle, BundleError> {
+    let key = hex::decode(key_hex).map_err(|_| BundleError::Invalid("key: must be hex"))?;
+    let payload = BASE64
+        .decode(payload_b64)
+        .map_err(|_| BundleError::Invalid("payload: not valid base64"))?;
+    let signature = BASE64
+        .decode(signature_b64)
+        .map_err(|_| BundleError::Invalid("signature: not valid base64"))?;
+    Bundle::from_bytes(&key, &payload, &signature)
+}
 
 #[derive(Debug, Serialize)]
-pub struct FetchResponse {
+pub struct FetchKeyPackageResponse {
     /// base64 of the stored payload; consumers verify `signature` over it.
     pub payload: String,
     pub signature: String,
@@ -26,36 +91,32 @@ struct ErrorBody {
 
 pub fn router(store: Arc<Store>) -> Router {
     Router::new()
-        .route("/v0/keypackage", post(submit))
+        .route("/v0/keypackage", post(submit_key_package))
-        .route("/v0/keypackage/:device_id", get(fetch))
+        .route("/v0/keypackage/:device_id", get(fetch_key_package))
         .route("/v0/account", post(submit_account))
         .route("/v0/account/:account_pub", get(fetch_account))
         .with_state(store)
 }
 
-/// `POST /v0/keypackage` — the same submission the logos-delivery subscriber
+/// `POST /v0/keypackage` — the same bundle the logos-delivery subscriber
 /// accepts, in JSON rather than protobuf; verification and storage live in
-/// [`submit::apply_keypackage`].
+/// [`bundle::apply_keypackage`].
-async fn submit(
+async fn submit_key_package(
     State(store): State<Arc<Store>>,
-    Json(req): Json<SubmitRequest>,
+    Json(req): Json<SubmitKeyPackageRequest>,
 ) -> Result<StatusCode, ApiError> {
-    submit::apply_keypackage(&store, &req.decode()?).await?;
+    bundle::apply_keypackage(&store, &req.decode()?).await?;
     Ok(StatusCode::NO_CONTENT)
 }
 
-async fn fetch(
+async fn fetch_key_package(
     State(store): State<Arc<Store>>,
     Path(device_id): Path<String>,
-) -> Result<Json<FetchResponse>, ApiError> {
+) -> Result<Json<FetchKeyPackageResponse>, ApiError> {
-    let Some(bundle) = store
+    let Some(bundle) = store.latest(&device_id).await.map_err(ApiError::internal)? else {
-        .latest(&device_id)
-        .await
-        .map_err(ApiError::internal)?
-    else {
         return Err(ApiError::not_found("no keypackage for device"));
     };
-    Ok(Json(FetchResponse {
+    Ok(Json(FetchKeyPackageResponse {
         payload: BASE64.encode(&bundle.payload),
         signature: BASE64.encode(&bundle.signature),
     }))
@@ -77,12 +138,12 @@ pub struct FetchAccountResponse {
 /// per `account_pub`, replacing any previous value. Clients should re-publish
 /// whenever they add or rotate LocalIdentities. The same submission the
 /// logos-delivery subscriber accepts, in JSON rather than protobuf; the shared
-/// rules live in [`submit::apply_account`].
+/// rules live in [`bundle::apply_account`].
 async fn submit_account(
     State(store): State<Arc<Store>>,
     Json(req): Json<SubmitAccountRequest>,
 ) -> Result<StatusCode, ApiError> {
-    submit::apply_account(&store, &req.decode()?).await?;
+    bundle::apply_account(&store, &req.decode()?).await?;
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -129,18 +190,18 @@ impl ApiError {
     }
 }
 
-impl From<SubmitError> for ApiError {
+impl From<BundleError> for ApiError {
-    fn from(err: SubmitError) -> Self {
+    fn from(err: BundleError) -> Self {
         match err {
-            SubmitError::Invalid(msg) => Self {
+            BundleError::Invalid(msg) => Self {
                 status: StatusCode::BAD_REQUEST,
                 message: msg.into(),
             },
-            SubmitError::Stale => Self {
+            BundleError::Stale => Self {
                 status: StatusCode::CONFLICT,
                 message: err.to_string(),
             },
-            SubmitError::Internal(inner) => Self::internal(inner),
+            BundleError::Internal(inner) => Self::internal(inner),
         }
     }
 }

src/main.rs

@@ -1,14 +1,14 @@
 //! Testnet KeyPackage Registry service.
 //!
 //! Throwaway service for issue #110 — replaced by λLEZ in v0.3. No libchat-core
-//! dependency; the embedded logos-delivery node comes from the sibling libchat
+//! dependency; the embedded logos-delivery node comes from libchat's transport
-//! checkout's transport crate.
+//! crate, pulled in as a pinned git dependency.
 //!
 //! Submissions arrive on either write path — both feed the same verification +
-//! storage pipeline (`submit`):
+//! storage pipeline (`bundle`):
 //!   - HTTP POST (below), synchronous and acknowledged;
-//!   - logos-delivery subscription (`ingest`, disable with `--no-delivery`):
+//!   - logos-delivery subscription (`delivery`, disable with `--no-delivery`):
-//!     clients publish the same JSON on the store content topics.
+//!     clients publish protobuf submissions on the store content topics.
 //!
 //! HTTP wire:
 //!   POST /v0/keypackage             — submit a signed keypackage bundle
@@ -16,10 +16,10 @@
 //!   POST /v0/account                — upsert a signed account device-list bundle
 //!   GET  /v0/account/{account_pub}  — fetch the account device-list bundle
 
+mod bundle;
+mod delivery;
 mod handlers;
-mod ingest;
 mod store;
-mod submit;
 
 use std::net::SocketAddr;
 use std::path::PathBuf;
@@ -33,7 +33,10 @@ use tracing_subscriber::EnvFilter;
 use store::Store;
 
 #[derive(Parser, Debug)]
-#[command(name = "chat-store", about = "Testnet Chat Store (KeyPackage + account directory)")]
+#[command(
+    name = "chat-store",
+    about = "Testnet Chat Store (KeyPackage + account directory)"
+)]
 struct Cli {
     /// Address to bind the HTTP server.
     #[arg(long, default_value = "0.0.0.0:8080")]
@@ -80,11 +83,7 @@ async fn main() -> Result<()> {
 
     let cli = Cli::parse();
 
-    let store = Arc::new(
+    let store = Arc::new(Store::open(&cli.db).await.context("failed to open store")?);
-        Store::open(&cli.db)
-            .await
-            .context("failed to open store")?,
-    );
 
     let prune_store = store.clone();
     let max_per_id = cli.max_per_identity;
@@ -107,9 +106,9 @@ async fn main() -> Result<()> {
         // Blocks for a few seconds while the node starts and finds peers;
         // deliberately before serving HTTP so a subscriber failure is a
         // startup error, not a silent half-running store.
-        ingest::start(
+        delivery::start(
             store.clone(),
-            ingest::P2pConfig {
+            delivery::P2pConfig {
                 preset: cli.preset.clone(),
                 port: cli.p2p_port,
                 log_level: "ERROR".into(),

src/submit.rs

@@ -1,404 +0,0 @@
-//! Shared submission pipeline.
-//!
-//! The two write paths carry the same fields in the encoding native to each
-//! wire: the HTTP POST endpoints take a JSON body with hex and base64 strings,
-//! while the logos-delivery subscriber (`ingest`) takes a protobuf message
-//! 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;
-use ed25519_dalek::{Signature, VerifyingKey};
-use serde::Deserialize;
-
-use crate::store::{Store, StoredAccountBundle, StoredKeyPackageBundle};
-
-/// A signed keypackage bundle submission.
-#[derive(Debug, Deserialize)]
-pub struct SubmitRequest {
-    /// Hex of the 32-byte Ed25519 device verifying key. Used to verify the
-    /// signature and as the storage/lookup key. `payload` stays opaque.
-    pub device_id: String,
-    /// base64 of the signed payload. Opaque to the server — it never decodes it.
-    pub payload: String,
-    /// base64 of the 64-byte Ed25519 signature over `payload`. Verifying it
-    /// under `device_id`'s key is proof-of-possession: only the holder of that
-    /// key can publish under this `device_id`.
-    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
-/// to encode a lamport-timestamped list of device (LocalIdentity) Ed25519
-/// public keys inside it so that consumers can detect stale bundles. The server
-/// only verifies that `signature` is a valid Ed25519 signature over `payload`
-/// made by the key identified by `account_pub`.
-#[derive(Debug, Deserialize)]
-pub struct SubmitAccountRequest {
-    /// Hex of the 32-byte Ed25519 account (AccountAddress) verifying key.
-    /// Acts as both the storage key and the verification key.
-    pub account_pub: String,
-    /// base64 of the opaque signed payload (lamport-ts + device pubkeys, etc.).
-    pub payload: String,
-    /// base64 of the 64-byte Ed25519 signature over `payload` made by the
-    /// account key. Proof-of-possession: only the account holder can publish.
-    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.
-    Invalid(&'static str),
-    /// Valid account bundle whose lamport is not newer than the stored one.
-    Stale,
-    /// Storage failure.
-    Internal(anyhow::Error),
-}
-
-impl std::fmt::Display for SubmitError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            SubmitError::Invalid(msg) => write!(f, "{msg}"),
-            SubmitError::Stale => {
-                write!(f, "stale bundle: lamport is not newer than the stored one")
-            }
-            SubmitError::Internal(err) => write!(f, "internal: {err}"),
-        }
-    }
-}
-
-/// A submission decoded off either wire, before verification: the raw key,
-/// 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 the submission's key. A valid signature means the submitter
-/// holds that key. This rejects junk early (DoS mitigation); consumers still
-/// verify on retrieve, the server is not a trusted authority.
-fn verify(submission: &Submission, key_field: KeyField) -> Result<(), SubmitError> {
-    let verifying_key = VerifyingKey::from_bytes(&submission.key)
-        .map_err(|_| SubmitError::Invalid(key_field.not_key))?;
-    verifying_key
-        .verify_strict(
-            &submission.payload,
-            &Signature::from_bytes(&submission.signature),
-        )
-        .map_err(|_| SubmitError::Invalid("signature: verification failed"))?;
-    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, submission: &Submission) -> Result<(), SubmitError> {
-    verify(submission, DEVICE_ID)?;
-    store
-        .insert(
-            &submission.key_hex(),
-            &StoredKeyPackageBundle {
-                payload: submission.payload.clone(),
-                signature: submission.signature.to_vec(),
-            },
-        )
-        .await
-        .map_err(SubmitError::Internal)
-}
-
-/// Verify and upsert an account device-list bundle submission.
-pub async fn apply_account(store: &Store, submission: &Submission) -> Result<(), SubmitError> {
-    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(&submission.payload).ok_or(SubmitError::Invalid(
-        "payload: too short to contain a lamport header",
-    ))?;
-
-    let applied = store
-        .upsert_account(
-            &submission.key_hex(),
-            lamport,
-            &StoredAccountBundle {
-                payload: submission.payload.clone(),
-                signature: submission.signature.to_vec(),
-                updated_at: 0, // filled in by store
-            },
-        )
-        .await
-        .map_err(SubmitError::Internal)?;
-    if !applied {
-        return Err(SubmitError::Stale);
-    }
-    Ok(())
-}
-
-#[cfg(test)]
-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::*;
-
-    /// Must match `BUNDLE_DOMAIN` in `store.rs` (kept private there).
-    const ACCOUNT_BUNDLE_DOMAIN: &[u8] = b"libchat:account-device-bundle\0";
-
-    fn signing_key(seed: u8) -> SigningKey {
-        SigningKey::from_bytes(&[seed; 32])
-    }
-
-    /// A keypackage submission exactly as it arrives off the delivery wire:
-    /// 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()),
-            "payload": BASE64.encode(payload),
-            "signature": BASE64.encode(key.sign(payload).to_bytes()),
-        })
-        .to_string()
-        .into_bytes()
-    }
-
-    fn account_payload(lamport: u64) -> Vec<u8> {
-        let mut p = ACCOUNT_BUNDLE_DOMAIN.to_vec();
-        p.push(1u8); // version
-        p.extend_from_slice(&lamport.to_le_bytes());
-        p
-    }
-
-    fn account_submission(key: &SigningKey, payload: &[u8]) -> Submission {
-        Submission::account(
-            key.verifying_key().as_bytes(),
-            payload,
-            &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_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 = KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, &payload)[..])
-            .unwrap();
-        let submission =
-            Submission::keypackage(&wire.device_id, &wire.payload, &wire.signature).unwrap();
-        apply_keypackage(&store, &submission).await.unwrap();
-
-        let stored = store.latest(&submission.key_hex()).await.unwrap().unwrap();
-        assert_eq!(stored.payload, payload);
-    }
-
-    #[tokio::test]
-    async fn tampered_keypackage_submission_is_rejected() {
-        let store = Store::open(Path::new(":memory:")).await.unwrap();
-        let key = signing_key(2);
-
-        let mut wire =
-            KeyPackageSubmissionV1::decode(&keypackage_submission_proto(&key, b"original")[..])
-                .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, &submission).await.unwrap_err();
-        assert!(matches!(
-            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]
-    async fn account_submission_upserts_and_rejects_stale_replay() {
-        let store = Store::open(Path::new(":memory:")).await.unwrap();
-        let key = signing_key(3);
-
-        apply_account(&store, &account_submission(&key, &account_payload(1)))
-            .await
-            .unwrap();
-        apply_account(&store, &account_submission(&key, &account_payload(2)))
-            .await
-            .unwrap();
-
-        // Replaying the lamport-2 bundle (as a delivery duplicate would) is stale.
-        let err = apply_account(&store, &account_submission(&key, &account_payload(2)))
-            .await
-            .unwrap_err();
-        assert!(matches!(err, SubmitError::Stale));
-    }
-}