add traffic counter

This introduces a new wrapper that counts messages and payload bytes per
peer, split by kind (hello, helloReply, sync, ping, pong), each sent,
received and refused.

The traffic counter goes around the firewall, which means messages
dropped outbound are still counted as sent, and messages dropped inbound
don't count as received.

1 files changed, 16 insertions, 5 deletions

diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index 9075ae8..cc7f25f 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -559,10 +559,12 @@ impl Transport for MyceliumTransport {
     }
 
     fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
-        let outbound = format_message(peer, message)?;
+        let (outbound, payload_len) = format_message(peer, message)?;
         Ok(EncodedMessage {
             peer: peer.to_owned(),
             body: serde_json::to_vec(&outbound)?.into(),
+            kind: message.kind(),
+            payload_len,
         })
     }
 
@@ -685,7 +687,8 @@ impl Transport for MyceliumTransport {
                         //payload_bytes = raw_bytes.len(),
                         "Received message"
                     );
-                    if let Err(error) = handler(sender, parsed).await {
+                    let payload_len = base64_decoded_len(&inbound_message.payload);
+                    if let Err(error) = handler(sender, parsed, payload_len).await {
                         tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message");
                     }
                 }
@@ -720,6 +723,12 @@ fn encode_payload(payload: &Vec<u8>) -> String {
     BASE64.encode(payload)
 }
 
+/// Length of what padded base64 `encoded` decodes to.
+fn base64_decoded_len(encoded: &str) -> usize {
+    let padding = encoded.bytes().rev().take_while(|&b| b == b'=').count();
+    (encoded.len() / 4 * 3).saturating_sub(padding)
+}
+
 fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, TransportError> {
     let topic: &str = &decode_topic(topic_encoded)?;
     let bytes = BASE64.decode(payload)?;
@@ -754,10 +763,11 @@ fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), Tr
     Ok((sender, parsed))
 }
 
+/// The message for Mycelium's API, and the length of its payload.
 fn format_message(
     recipient: &str,
     message: &ParsedMessage,
-) -> Result<OutboundMessage, TransportError> {
+) -> Result<(OutboundMessage, usize), TransportError> {
     let (topic, payload_bytes) = match message {
         ParsedMessage::Hello(req) => (TOPIC_HELLO, hello_request_to_bytes(req)?),
         ParsedMessage::HelloReply(res) => (TOPIC_HELLO_REPLY, hello_reply_to_bytes(res)?),
@@ -765,13 +775,14 @@ fn format_message(
         ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?),
         ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?),
     };
-    Ok(OutboundMessage {
+    let outbound = OutboundMessage {
         dst: Destination {
             pk: recipient.to_owned(),
         },
         topic: encode_topic(topic),
         payload: encode_payload(&payload_bytes),
-    })
+    };
+    Ok((outbound, payload_bytes.len()))
 }
 
 // =====================================================================