Cleanup/miscellaneous

README.md

@@ -1,4 +1,4 @@
-# message-queue
+# foxmq
 Message queue MeshApp utilizing Tashi Consensus Engine
 
 ### Checking Out

src/cli/run.rs

@@ -1,6 +1,3 @@
-use std::net::SocketAddr;
-use std::path::{Path, PathBuf};
-use std::sync::Arc;
 use crate::cli::LogFormat;
 use crate::collections::HashMap;
 use crate::config;
@@ -12,6 +9,9 @@ use crate::mqtt::{KeepAlive, TceState};
 use crate::transaction::AddNodeTransaction;
 use color_eyre::eyre;
 use color_eyre::eyre::Context;
+use std::net::SocketAddr;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
 use tashi_consensus_engine::quic::QuicSocket;
 use tashi_consensus_engine::{
     Certificate, Platform, RootCertificates, SecretKey, UnknownConnectionAction,

src/cli/user.rs

@@ -3,12 +3,12 @@ use std::fs::{File, OpenOptions};
 use std::io::{BufRead, Write};
 use std::path::{Path, PathBuf};
 
-use color_eyre::eyre::WrapErr;
-use rand::distributions::Uniform;
-use rand::Rng;
 use crate::cli::LogFormat;
 use crate::collections::HashMap;
 use crate::config::users::{AuthConfig, User, UsersConfig};
+use color_eyre::eyre::WrapErr;
+use rand::distributions::Uniform;
+use rand::Rng;
 
 const DEFAULT_PASSWORD_LEN: usize = 12;
 

src/config/permissions.rs

@@ -55,13 +55,13 @@ impl PermissionsConfig {
         transaction_type: TransactionType,
     ) -> bool {
         // Allows everything if no topics config was found.
-        topics_config.map_or(true, |perms| {
+        topics_config.is_none_or(|perms| {
             perms
                 .topic
                 .iter()
                 .find(|k| k.filter.matches_topic(topic_name))
-                .map_or(true, |k| {
-                    k.allowed.iter().any(|k| *k == transaction_type)
+                .is_none_or(|k| {
+                    k.allowed.contains(&transaction_type)
                         || !k.denied.iter().all(|k| *k == transaction_type)
                 })
         })

src/mqtt/broker/connection.rs

@@ -1,9 +1,9 @@
+use std::cmp;
 use std::fmt::{Debug, Display};
 use std::pin::Pin;
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
-use std::{cmp, iter};
 
 use bytes::BytesMut;
 use color_eyre::eyre;
@@ -888,9 +888,11 @@ impl<S: MqttSocket> Connection<S> {
             self.send(Packet::UnsubAck(
                 UnsubAck {
                     pkid: unsub.pkid,
-                    reasons: iter::repeat(UnsubAckReason::PacketIdentifierInUse)
-                        .take(unsub.filters.len())
-                        .collect(),
+                    reasons: std::iter::repeat_n(
+                        UnsubAckReason::PacketIdentifierInUse,
+                        unsub.filters.len(),
+                    )
+                    .collect(),
                 },
                 None,
             ))

src/mqtt/broker/websocket.rs

@@ -250,6 +250,5 @@ fn find_client_ip(connected_addr: IpAddr, x_forwarded_for: &HeaderValue) -> Opti
                 })
                 .ok()
         })
-        .filter(|ip| ip != &connected_addr)
-        .next_back()
+        .rfind(|ip| ip != &connected_addr)
 }

src/mqtt/client_id.rs

@@ -192,8 +192,6 @@ impl PartialEq<ClientId> for str {
     }
 }
 
-
-
 impl From<ClientId> for String {
     fn from(value: ClientId) -> Self {
         value.as_str().into()

src/mqtt/publish.rs

@@ -363,7 +363,7 @@ pub fn txn_to_packet(
             Bytes::copy_from_slice(txn.topic.as_bytes()),
             txn.payload.0.clone(),
         ),
-        (!sub_ids.is_empty() || txn.properties.is_some()).then(|| {
+        (!sub_ids.is_empty() || txn.properties.is_some() || include_broker_timestamps).then(|| {
             macro_rules! clone_prop {
                 ($prop:ident) => {
                     txn.properties

src/mqtt/trie/node.rs

@@ -19,8 +19,7 @@ pub(super) struct NodeLeaf<T> {
 
 impl<T> NodeLeaf<T> {
     pub(super) fn all(&self, f: &mut impl FnMut(&T) -> bool) -> bool {
-        self.exact_val.as_ref().map_or(true, &mut *f)
-            && self.descendant_val.as_ref().map_or(true, f)
+        self.exact_val.as_ref().is_none_or(&mut *f) && self.descendant_val.as_ref().is_none_or(f)
     }
 
     fn is_empty(&self) -> bool {

tests/.gitignore

@@ -1 +1,2 @@
 /node_modules
+/foxmq-single.d/

tests/Dockerfile

@@ -1,6 +1,6 @@
 FROM rust:1-bookworm
 
-WORKDIR /home/message-queue
+WORKDIR /home/foxmq
 
 # The minimum number of files required to allow `cargo fetch` to cache all dependencies.
 # We do this separately from the `COPY ./ .` so most code changes don't invalidate the `cargo fetch` layer.
@@ -35,6 +35,6 @@ RUN cargo build --release -p tokio -p tracing -p tashi-consensus-engine
 
 COPY ./ .
 
-WORKDIR /home/message-queue
+WORKDIR /home/foxmq
 
 RUN cargo build --release

tests/docker-compose.yaml

@@ -9,9 +9,9 @@ services:
       dockerfile: tests/Dockerfile
       ssh:
         - default
-    image: message-queue-test
+    image: foxmq-test
     volumes:
-      - ./foxmq.d:/home/message-queue/foxmq.d
+      - ./foxmq.d:/home/foxmq/foxmq.d
     ports:
       - "1883:1883"
       - "8080:8080"
@@ -30,15 +30,15 @@ services:
       --secret-key-file foxmq.d/key_0.pem
       --tls-cert-file=foxmq.d/key_0.crt
       --server-name broker1.example.com
-      /home/message-queue/foxmq.d
+      /home/foxmq/foxmq.d
 
   broker2:
     container_name: broker2
     depends_on:
       - broker1
-    image: message-queue-test
+    image: foxmq-test
     volumes:
-      - ./foxmq.d:/home/message-queue/foxmq.d
+      - ./foxmq.d:/home/foxmq/foxmq.d
     ports:
       - "1884:1883"
       - "8081:8080"
@@ -56,7 +56,7 @@ services:
       --secret-key-file foxmq.d/key_1.pem
       --tls-cert-file=foxmq.d/key_1.crt
       --server-name broker2.example.com
-      /home/message-queue/foxmq.d
+      /home/foxmq/foxmq.d
 
 networks:
   mesh-app-bridge:

tests/foxmq-root-ca.d/docker-compose.yaml

@@ -9,9 +9,9 @@ services:
       dockerfile: tests/Dockerfile
       ssh:
         - default
-    image: message-queue-test
+    image: foxmq-test
     volumes:
-      - .:/home/message-queue/foxmq.d
+      - .:/home/foxmq/foxmq.d
     ports:
       - "1883:1883"
     networks:
@@ -28,16 +28,16 @@ services:
       --cluster-root-cert=foxmq.d/ca.crt
       --cluster-accept-peer-with-cert
       --server-name broker1.example.com
-      /home/message-queue/foxmq.d
+      /home/foxmq/foxmq.d
 
   broker2:
     container_name: broker2
     depends_on:
       # Ensure we start after `broker1` for proper IP address assignment
       - broker1
-    image: message-queue-test
+    image: foxmq-test
     volumes:
-      - .:/home/message-queue/foxmq.d
+      - .:/home/foxmq/foxmq.d
     ports:
       - "1884:1883"
     networks:
@@ -53,16 +53,16 @@ services:
       --cluster-root-cert=foxmq.d/ca.crt
       --cluster-accept-peer-with-cert
       --server-name broker2.example.com
-      /home/message-queue/foxmq.d
+      /home/foxmq/foxmq.d
 
   broker3:
     container_name: broker3
     depends_on:
       # Ensure we start after `broker2` for proper IP address assignment
       - broker2
-    image: message-queue-test
+    image: foxmq-test
     volumes:
-      - .:/home/message-queue/foxmq.d
+      - .:/home/foxmq/foxmq.d
     ports:
       - "1885:1883"
     networks:
@@ -78,16 +78,16 @@ services:
       --cluster-root-cert=foxmq.d/ca.crt
       --cluster-accept-peer-with-cert
       --server-name broker2.example.com
-      /home/message-queue/foxmq.d
+      /home/foxmq/foxmq.d
 
   broker4:
     container_name: broker4
     depends_on:
       # Ensure we start after `broker3` for proper IP address assignment\
       - broker3
-    image: message-queue-test
+    image: foxmq-test
     volumes:
-      - .:/home/message-queue/foxmq.d
+      - .:/home/foxmq/foxmq.d
     ports:
       - "1886:1883"
     networks:
@@ -104,7 +104,7 @@ services:
       --cluster-root-cert=foxmq.d/ca.crt
       --cluster-accept-peer-with-cert
       --server-name broker4.example.com
-      /home/message-queue/foxmq.d
+      /home/foxmq/foxmq.d
 
 networks:
   mesh-app-bridge:

tests/mqtt/consensus-timestamps.test.js

@@ -0,0 +1,78 @@
+const mqtt = require("mqtt");
+const events = require("node:events");
+
+describe("Consensus Timestamps", () => {
+    test("receive timestamp_received user property when requested", async () => {
+        // Client 1: Subscriber
+        console.log("Connecting Subscriber...");
+        const subClient = await mqtt.connectAsync("mqtt://127.0.0.1:1883", { protocolVersion: 5 });
+
+        // Client 2: Publisher
+        console.log("Connecting Publisher...");
+        const pubClient = await mqtt.connectAsync("mqtt://127.0.0.1:1883", { protocolVersion: 5 });
+
+        // Setup message collector on Subscriber BEFORE verifying subscription or publishing
+        const messagesReceived = [];
+        const completionPromise = new Promise((resolve, reject) => {
+            const timeout = setTimeout(() => {
+                reject(new Error("Timeout waiting for messages"));
+            }, 10000);
+
+            subClient.on('message', (topic, message, packet) => {
+                console.log(`Received: ${topic} ${message.toString()}`);
+                messagesReceived.push({
+                    msg: message.toString(),
+                    props: packet.properties
+                });
+
+                if (messagesReceived.length >= 3) {
+                    clearTimeout(timeout);
+                    resolve();
+                }
+            });
+        });
+
+        // Request timestamps via subscription properties
+        console.log("Subscribing...");
+        await subClient.subscribeAsync("timestamp/test", {
+            qos: 1,
+            properties: {
+                userProperties: {
+                    include_broker_timestamps: "true"
+                }
+            }
+        });
+        console.log("Subscribed!");
+
+        // Publish messages from Client 2
+        console.log("Publishing...");
+        await pubClient.publishAsync("timestamp/test", "message 1", { qos: 1 });
+        await pubClient.publishAsync("timestamp/test", "message 2", { qos: 1 });
+        await pubClient.publishAsync("timestamp/test", "message 3", { qos: 1 });
+        console.log("Published!");
+
+        // Wait for subscriber to get them
+        await completionPromise;
+
+        // Verify content and timestamps
+        expect(messagesReceived.length).toBe(3);
+
+        let lastTimestamp = "";
+        for (const m of messagesReceived) {
+            expect(m.props).toBeDefined();
+            expect(m.props.userProperties).toBeDefined();
+            expect(m.props.userProperties.timestamp_received).toBeDefined();
+
+            const currentTs = m.props.userProperties.timestamp_received;
+            if (lastTimestamp !== "") {
+                expect(currentTs >= lastTimestamp).toBe(true);
+            }
+            lastTimestamp = currentTs;
+        }
+
+        console.log("Timestamps verified:", messagesReceived.map(m => m.props.userProperties.timestamp_received));
+
+        await subClient.endAsync();
+        await pubClient.endAsync();
+    }, 30000);
+});