summaryrefslogtreecommitdiffstats
path: root/transport/mesh
diff options
context:
space:
mode:
Diffstat (limited to 'transport/mesh')
-rw-r--r--transport/mesh/discovery.go332
-rw-r--r--transport/mesh/mesh.go248
-rw-r--r--transport/mesh/storeforward.go319
3 files changed, 899 insertions, 0 deletions
diff --git a/transport/mesh/discovery.go b/transport/mesh/discovery.go
new file mode 100644
index 0000000..2cfad32
--- /dev/null
+++ b/transport/mesh/discovery.go
@@ -0,0 +1,332 @@
+// Package mesh - Peer discovery for military mesh networking
+package mesh
+
+import (
+ "encoding/json"
+ "fmt"
+ "net"
+ "sync"
+ "time"
+)
+
+// PeerDiscovery manages automatic peer discovery via mDNS and UDP broadcast
+type PeerDiscovery struct {
+ localPeer *PeerInfo
+ discoveredPeers map[string]*PeerInfo // address -> peer info
+ mu sync.RWMutex
+ udpConn *net.UDPConn
+ running bool
+ onPeerFound func(*PeerInfo) // Callback when new peer found
+ onPeerLost func(string) // Callback when peer lost
+}
+
+// PeerInfo contains information about a discovered peer
+type PeerInfo struct {
+ Address string `json:"address"` // IP:Port
+ PublicKey string `json:"public_key"` // Ed25519 public key hex
+ Nickname string `json:"nickname"` // Optional display name
+ Timestamp time.Time `json:"timestamp"` // Last seen
+ IsRelay bool `json:"is_relay"` // Can relay messages?
+ Hops int `json:"hops"` // Distance in hops
+}
+
+// DiscoveryConfig holds peer discovery configuration
+type DiscoveryConfig struct {
+ BroadcastPort int // UDP broadcast port
+ BroadcastInterval time.Duration // How often to announce
+ PeerTimeout time.Duration // Remove peer if not seen
+ EnableMDNS bool // Use mDNS (Bonjour)
+ EnableBroadcast bool // Use UDP broadcast
+}
+
+// NewPeerDiscovery creates a new peer discovery instance
+func NewPeerDiscovery(localPeer *PeerInfo, config *DiscoveryConfig) (*PeerDiscovery, error) {
+ if config == nil {
+ config = &DiscoveryConfig{
+ BroadcastPort: 9876,
+ BroadcastInterval: 10 * time.Second,
+ PeerTimeout: 60 * time.Second,
+ EnableBroadcast: true,
+ EnableMDNS: false, // mDNS requires external library
+ }
+ }
+
+ pd := &PeerDiscovery{
+ localPeer: localPeer,
+ discoveredPeers: make(map[string]*PeerInfo),
+ }
+
+ // Setup UDP broadcast listener
+ if config.EnableBroadcast {
+ addr := &net.UDPAddr{
+ IP: net.IPv4zero,
+ Port: config.BroadcastPort,
+ }
+
+ conn, err := net.ListenUDP("udp4", addr)
+ if err != nil {
+ return nil, fmt.Errorf("failed to create UDP listener: %w", err)
+ }
+
+ pd.udpConn = conn
+ }
+
+ return pd, nil
+}
+
+// Start starts peer discovery
+func (pd *PeerDiscovery) Start(config *DiscoveryConfig) error {
+ pd.mu.Lock()
+ if pd.running {
+ pd.mu.Unlock()
+ return fmt.Errorf("discovery already running")
+ }
+ pd.running = true
+ pd.mu.Unlock()
+
+ if config == nil {
+ config = &DiscoveryConfig{
+ BroadcastPort: 9876,
+ BroadcastInterval: 10 * time.Second,
+ PeerTimeout: 60 * time.Second,
+ EnableBroadcast: true,
+ }
+ }
+
+ // Start UDP broadcast announcer
+ if config.EnableBroadcast {
+ go pd.broadcastAnnouncer(config)
+ go pd.broadcastListener(config)
+ }
+
+ // Start peer timeout checker
+ go pd.peerTimeoutChecker(config)
+
+ return nil
+}
+
+// Stop stops peer discovery
+func (pd *PeerDiscovery) Stop() error {
+ pd.mu.Lock()
+ defer pd.mu.Unlock()
+
+ if !pd.running {
+ return nil
+ }
+
+ pd.running = false
+
+ if pd.udpConn != nil {
+ pd.udpConn.Close()
+ }
+
+ return nil
+}
+
+// broadcastAnnouncer periodically broadcasts our presence
+func (pd *PeerDiscovery) broadcastAnnouncer(config *DiscoveryConfig) {
+ ticker := time.NewTicker(config.BroadcastInterval)
+ defer ticker.Stop()
+
+ for {
+ pd.mu.RLock()
+ if !pd.running {
+ pd.mu.RUnlock()
+ return
+ }
+ pd.mu.RUnlock()
+
+ <-ticker.C
+
+ // Create announcement message
+ announcement := map[string]interface{}{
+ "type": "peer_announcement",
+ "address": pd.localPeer.Address,
+ "public_key": pd.localPeer.PublicKey,
+ "nickname": pd.localPeer.Nickname,
+ "is_relay": pd.localPeer.IsRelay,
+ "timestamp": time.Now().Unix(),
+ }
+
+ data, err := json.Marshal(announcement)
+ if err != nil {
+ continue
+ }
+
+ // Broadcast to LAN
+ broadcastAddr := &net.UDPAddr{
+ IP: net.IPv4bcast,
+ Port: config.BroadcastPort,
+ }
+
+ conn, err := net.DialUDP("udp4", nil, broadcastAddr)
+ if err != nil {
+ continue
+ }
+
+ conn.Write(data)
+ conn.Close()
+ }
+}
+
+// broadcastListener listens for peer announcements
+func (pd *PeerDiscovery) broadcastListener(config *DiscoveryConfig) {
+ buf := make([]byte, 4096)
+
+ for {
+ pd.mu.RLock()
+ if !pd.running {
+ pd.mu.RUnlock()
+ return
+ }
+ pd.mu.RUnlock()
+
+ // Set read deadline to allow periodic checks
+ pd.udpConn.SetReadDeadline(time.Now().Add(1 * time.Second))
+
+ n, remoteAddr, err := pd.udpConn.ReadFromUDP(buf)
+ if err != nil {
+ if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+ continue // Timeout is expected
+ }
+ continue
+ }
+
+ // Parse announcement
+ var announcement map[string]interface{}
+ if err := json.Unmarshal(buf[:n], &announcement); err != nil {
+ continue
+ }
+
+ // Ignore our own announcements
+ if announcement["address"] == pd.localPeer.Address {
+ continue
+ }
+
+ // Extract peer info
+ peerInfo := &PeerInfo{
+ Address: fmt.Sprintf("%v", announcement["address"]),
+ PublicKey: fmt.Sprintf("%v", announcement["public_key"]),
+ Nickname: fmt.Sprintf("%v", announcement["nickname"]),
+ Timestamp: time.Now(),
+ IsRelay: announcement["is_relay"].(bool),
+ Hops: 1, // Direct peer
+ }
+
+ // If address is missing, use source address
+ if peerInfo.Address == "" || peerInfo.Address == "<nil>" {
+ peerInfo.Address = remoteAddr.IP.String()
+ }
+
+ pd.addPeer(peerInfo)
+ }
+}
+
+// addPeer adds or updates a discovered peer
+func (pd *PeerDiscovery) addPeer(peer *PeerInfo) {
+ pd.mu.Lock()
+ defer pd.mu.Unlock()
+
+ isNew := false
+ if _, exists := pd.discoveredPeers[peer.Address]; !exists {
+ isNew = true
+ }
+
+ pd.discoveredPeers[peer.Address] = peer
+
+ // Call callback if new peer
+ if isNew && pd.onPeerFound != nil {
+ go pd.onPeerFound(peer)
+ }
+}
+
+// peerTimeoutChecker removes stale peers
+func (pd *PeerDiscovery) peerTimeoutChecker(config *DiscoveryConfig) {
+ ticker := time.NewTicker(config.PeerTimeout / 2)
+ defer ticker.Stop()
+
+ for {
+ pd.mu.RLock()
+ if !pd.running {
+ pd.mu.RUnlock()
+ return
+ }
+ pd.mu.RUnlock()
+
+ <-ticker.C
+
+ pd.mu.Lock()
+ now := time.Now()
+ for addr, peer := range pd.discoveredPeers {
+ if now.Sub(peer.Timestamp) > config.PeerTimeout {
+ delete(pd.discoveredPeers, addr)
+
+ // Call callback
+ if pd.onPeerLost != nil {
+ go pd.onPeerLost(addr)
+ }
+ }
+ }
+ pd.mu.Unlock()
+ }
+}
+
+// GetPeers returns all discovered peers
+func (pd *PeerDiscovery) GetPeers() []*PeerInfo {
+ pd.mu.RLock()
+ defer pd.mu.RUnlock()
+
+ peers := make([]*PeerInfo, 0, len(pd.discoveredPeers))
+ for _, peer := range pd.discoveredPeers {
+ peers = append(peers, peer)
+ }
+ return peers
+}
+
+// GetPeerCount returns the number of discovered peers
+func (pd *PeerDiscovery) GetPeerCount() int {
+ pd.mu.RLock()
+ defer pd.mu.RUnlock()
+ return len(pd.discoveredPeers)
+}
+
+// SetOnPeerFound sets callback for when new peer is discovered
+func (pd *PeerDiscovery) SetOnPeerFound(callback func(*PeerInfo)) {
+ pd.mu.Lock()
+ defer pd.mu.Unlock()
+ pd.onPeerFound = callback
+}
+
+// SetOnPeerLost sets callback for when peer is lost
+func (pd *PeerDiscovery) SetOnPeerLost(callback func(string)) {
+ pd.mu.Lock()
+ defer pd.mu.Unlock()
+ pd.onPeerLost = callback
+}
+
+// FindPeerByPublicKey finds a peer by their public key
+func (pd *PeerDiscovery) FindPeerByPublicKey(publicKey string) (*PeerInfo, bool) {
+ pd.mu.RLock()
+ defer pd.mu.RUnlock()
+
+ for _, peer := range pd.discoveredPeers {
+ if peer.PublicKey == publicKey {
+ return peer, true
+ }
+ }
+ return nil, false
+}
+
+// String returns a string representation
+func (pd *PeerDiscovery) String() string {
+ pd.mu.RLock()
+ defer pd.mu.RUnlock()
+
+ status := "stopped"
+ if pd.running {
+ status = "running"
+ }
+
+ return fmt.Sprintf("PeerDiscovery(status=%s, peers=%d)",
+ status, len(pd.discoveredPeers))
+}
diff --git a/transport/mesh/mesh.go b/transport/mesh/mesh.go
new file mode 100644
index 0000000..dbe50e1
--- /dev/null
+++ b/transport/mesh/mesh.go
@@ -0,0 +1,248 @@
+// Package mesh implements peer-to-peer mesh networking
+package mesh
+
+import (
+ "fmt"
+ "net"
+ "sync"
+ "time"
+
+ "veilith/transport"
+)
+
+// MeshTransport implements peer-to-peer mesh networking
+type MeshTransport struct {
+ peers map[string]*Peer // address -> peer
+ mu sync.RWMutex
+ listener net.Listener
+ localAddr string
+ closed bool
+ discoveryInterval time.Duration
+}
+
+// Peer represents a mesh network peer
+type Peer struct {
+ Address string
+ PublicKey string
+ LastSeen time.Time
+ Conn net.Conn
+ IsRelay bool // Can this peer relay messages?
+}
+
+// NewMeshTransport creates a new mesh transport
+func NewMeshTransport() *MeshTransport {
+ return &MeshTransport{
+ peers: make(map[string]*Peer),
+ discoveryInterval: 30 * time.Second,
+ }
+}
+
+// Connect connects to a mesh peer
+func (m *MeshTransport) Connect(address string) (net.Conn, error) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ if m.closed {
+ return nil, fmt.Errorf("mesh transport is closed")
+ }
+
+ // Check if already connected
+ if peer, exists := m.peers[address]; exists {
+ if peer.Conn != nil {
+ peer.LastSeen = time.Now()
+ return peer.Conn, nil
+ }
+ }
+
+ // Establish new connection
+ conn, err := net.DialTimeout("tcp", address, 10*time.Second)
+ if err != nil {
+ return nil, fmt.Errorf("failed to connect to peer %s: %w", address, err)
+ }
+
+ // Add peer
+ m.peers[address] = &Peer{
+ Address: address,
+ LastSeen: time.Now(),
+ Conn: conn,
+ }
+
+ return conn, nil
+}
+
+// Listen starts listening for mesh connections
+func (m *MeshTransport) Listen(address string) (net.Listener, error) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ if m.closed {
+ return nil, fmt.Errorf("mesh transport is closed")
+ }
+
+ if m.listener != nil {
+ return nil, fmt.Errorf("already listening on %s", m.localAddr)
+ }
+
+ listener, err := net.Listen("tcp", address)
+ if err != nil {
+ return nil, fmt.Errorf("failed to listen on %s: %w", address, err)
+ }
+
+ m.listener = listener
+ m.localAddr = address
+
+ // Start peer discovery in background
+ go m.startPeerDiscovery()
+
+ return listener, nil
+}
+
+// Type returns the transport type
+func (m *MeshTransport) Type() transport.TransportType {
+ return transport.TransportMesh
+}
+
+// Close closes the mesh transport
+func (m *MeshTransport) Close() error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ if m.closed {
+ return nil
+ }
+
+ m.closed = true
+
+ // Close listener
+ if m.listener != nil {
+ if err := m.listener.Close(); err != nil {
+ return fmt.Errorf("failed to close listener: %w", err)
+ }
+ m.listener = nil
+ }
+
+ // Close all peer connections
+ for addr, peer := range m.peers {
+ if peer.Conn != nil {
+ peer.Conn.Close()
+ }
+ delete(m.peers, addr)
+ }
+
+ return nil
+}
+
+// AddPeer manually adds a peer to the mesh
+func (m *MeshTransport) AddPeer(address string) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ if m.closed {
+ return fmt.Errorf("mesh transport is closed")
+ }
+
+ if _, exists := m.peers[address]; exists {
+ return fmt.Errorf("peer already exists: %s", address)
+ }
+
+ m.peers[address] = &Peer{
+ Address: address,
+ LastSeen: time.Now(),
+ }
+
+ return nil
+}
+
+// RemovePeer removes a peer from the mesh
+func (m *MeshTransport) RemovePeer(address string) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ peer, exists := m.peers[address]
+ if !exists {
+ return fmt.Errorf("peer not found: %s", address)
+ }
+
+ if peer.Conn != nil {
+ peer.Conn.Close()
+ }
+
+ delete(m.peers, address)
+ return nil
+}
+
+// GetPeers returns all known peers
+func (m *MeshTransport) GetPeers() []*Peer {
+ m.mu.RLock()
+ defer m.mu.RUnlock()
+
+ peers := make([]*Peer, 0, len(m.peers))
+ for _, peer := range m.peers {
+ peers = append(peers, peer)
+ }
+ return peers
+}
+
+// GetPeerCount returns the number of known peers
+func (m *MeshTransport) GetPeerCount() int {
+ m.mu.RLock()
+ defer m.mu.RUnlock()
+ return len(m.peers)
+}
+
+// startPeerDiscovery periodically discovers new peers
+func (m *MeshTransport) startPeerDiscovery() {
+ ticker := time.NewTicker(m.discoveryInterval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ticker.C:
+ m.mu.RLock()
+ if m.closed {
+ m.mu.RUnlock()
+ return
+ }
+ m.mu.RUnlock()
+
+ // Clean up stale peers
+ m.cleanupStalePeers(5 * time.Minute)
+
+ // In a real implementation, you would:
+ // 1. Query known peers for their peer lists
+ // 2. Attempt connections to new peers
+ // 3. Exchange routing tables
+ }
+ }
+}
+
+// cleanupStalePeers removes peers that haven't been seen recently
+func (m *MeshTransport) cleanupStalePeers(maxAge time.Duration) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ now := time.Now()
+ for addr, peer := range m.peers {
+ if now.Sub(peer.LastSeen) > maxAge {
+ if peer.Conn != nil {
+ peer.Conn.Close()
+ }
+ delete(m.peers, addr)
+ }
+ }
+}
+
+// SetDiscoveryInterval sets the peer discovery interval
+func (m *MeshTransport) SetDiscoveryInterval(interval time.Duration) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ m.discoveryInterval = interval
+}
+
+// String returns a string representation
+func (m *MeshTransport) String() string {
+ m.mu.RLock()
+ defer m.mu.RUnlock()
+ return fmt.Sprintf("MeshTransport(peers=%d, listening=%s, closed=%v)",
+ len(m.peers), m.localAddr, m.closed)
+}
diff --git a/transport/mesh/storeforward.go b/transport/mesh/storeforward.go
new file mode 100644
index 0000000..1fcb93d
--- /dev/null
+++ b/transport/mesh/storeforward.go
@@ -0,0 +1,319 @@
+// Package mesh - Store-and-forward messaging for offline scenarios
+package mesh
+
+import (
+ "encoding/json"
+ "fmt"
+ "os"
+ "path/filepath"
+ "sync"
+ "time"
+)
+
+// MessageQueue implements store-and-forward messaging
+// Critical for military scenarios where connectivity is intermittent
+type MessageQueue struct {
+ pending map[string][]*QueuedMessage // recipient -> messages
+ delivered map[string]bool // message ID -> delivered
+ queueDir string // Persistent storage directory
+ maxAge time.Duration // Auto-delete after this time
+ maxSize int // Max messages in queue
+ encrypted bool // Encrypt queue on disk
+ mu sync.RWMutex
+ onDelivered func(string) // Callback when message delivered
+}
+
+// QueuedMessage represents a message waiting for delivery
+type QueuedMessage struct {
+ ID string `json:"id"` // Unique message ID
+ From string `json:"from"` // Sender public key
+ To string `json:"to"` // Recipient public key
+ Content []byte `json:"content"` // Encrypted message content
+ Timestamp time.Time `json:"timestamp"` // When queued
+ Attempts int `json:"attempts"` // Delivery attempts
+ MaxAttempts int `json:"max_attempts"` // Give up after this many
+ Priority int `json:"priority"` // Higher = more important
+ Hops int `json:"hops"` // How many hops so far
+ MaxHops int `json:"max_hops"` // Don't forward beyond this
+}
+
+// QueueConfig holds message queue configuration
+type QueueConfig struct {
+ QueueDir string // Where to store pending messages
+ MaxAge time.Duration // Auto-delete messages older than this
+ MaxSize int // Max messages per recipient
+ MaxAttempts int // Max delivery attempts
+ MaxHops int // Max forwarding hops
+ Encrypted bool // Encrypt queue on disk
+}
+
+// NewMessageQueue creates a new message queue
+func NewMessageQueue(config *QueueConfig) (*MessageQueue, error) {
+ if config == nil {
+ config = &QueueConfig{
+ QueueDir: filepath.Join(os.TempDir(), "veilith-queue"),
+ MaxAge: 24 * time.Hour,
+ MaxSize: 1000,
+ MaxAttempts: 10,
+ MaxHops: 5,
+ Encrypted: true,
+ }
+ }
+
+ // Create queue directory
+ if err := os.MkdirAll(config.QueueDir, 0700); err != nil {
+ return nil, fmt.Errorf("failed to create queue dir: %w", err)
+ }
+
+ mq := &MessageQueue{
+ pending: make(map[string][]*QueuedMessage),
+ delivered: make(map[string]bool),
+ queueDir: config.QueueDir,
+ maxAge: config.MaxAge,
+ maxSize: config.MaxSize,
+ encrypted: config.Encrypted,
+ }
+
+ // Load existing queue from disk
+ if err := mq.loadFromDisk(); err != nil {
+ return nil, fmt.Errorf("failed to load queue: %w", err)
+ }
+
+ // Start background cleanup
+ go mq.cleanupLoop()
+
+ return mq, nil
+}
+
+// Enqueue adds a message to the queue
+func (mq *MessageQueue) Enqueue(msg *QueuedMessage) error {
+ mq.mu.Lock()
+ defer mq.mu.Unlock()
+
+ // Check if already delivered
+ if mq.delivered[msg.ID] {
+ return fmt.Errorf("message already delivered")
+ }
+
+ // Check queue size limit
+ if len(mq.pending[msg.To]) >= mq.maxSize {
+ return fmt.Errorf("queue full for recipient %s", msg.To)
+ }
+
+ // Add to pending queue
+ mq.pending[msg.To] = append(mq.pending[msg.To], msg)
+
+ // Persist to disk
+ if err := mq.saveToDisk(); err != nil {
+ return fmt.Errorf("failed to save queue: %w", err)
+ }
+
+ return nil
+}
+
+// Dequeue retrieves messages for a specific recipient
+func (mq *MessageQueue) Dequeue(recipient string) []*QueuedMessage {
+ mq.mu.Lock()
+ defer mq.mu.Unlock()
+
+ messages := mq.pending[recipient]
+ delete(mq.pending, recipient)
+
+ // Mark as delivered
+ for _, msg := range messages {
+ mq.delivered[msg.ID] = true
+ if mq.onDelivered != nil {
+ go mq.onDelivered(msg.ID)
+ }
+ }
+
+ // Persist to disk
+ mq.saveToDisk()
+
+ return messages
+}
+
+// GetPendingCount returns number of pending messages for a recipient
+func (mq *MessageQueue) GetPendingCount(recipient string) int {
+ mq.mu.RLock()
+ defer mq.mu.RUnlock()
+ return len(mq.pending[recipient])
+}
+
+// GetTotalPendingCount returns total number of pending messages
+func (mq *MessageQueue) GetTotalPendingCount() int {
+ mq.mu.RLock()
+ defer mq.mu.RUnlock()
+
+ total := 0
+ for _, messages := range mq.pending {
+ total += len(messages)
+ }
+ return total
+}
+
+// GetPendingRecipients returns list of recipients with pending messages
+func (mq *MessageQueue) GetPendingRecipients() []string {
+ mq.mu.RLock()
+ defer mq.mu.RUnlock()
+
+ recipients := make([]string, 0, len(mq.pending))
+ for recipient := range mq.pending {
+ recipients = append(recipients, recipient)
+ }
+ return recipients
+}
+
+// IncrementAttempts increments delivery attempt counter
+func (mq *MessageQueue) IncrementAttempts(msgID string) {
+ mq.mu.Lock()
+ defer mq.mu.Unlock()
+
+ for _, messages := range mq.pending {
+ for _, msg := range messages {
+ if msg.ID == msgID {
+ msg.Attempts++
+ mq.saveToDisk()
+ return
+ }
+ }
+ }
+}
+
+// ShouldForward checks if message should be forwarded (multi-hop routing)
+func (mq *MessageQueue) ShouldForward(msg *QueuedMessage) bool {
+ return msg.Hops < msg.MaxHops && msg.Attempts < msg.MaxAttempts
+}
+
+// saveToDisk persists the queue to disk
+func (mq *MessageQueue) saveToDisk() error {
+ queuePath := filepath.Join(mq.queueDir, "queue.json")
+
+ // Serialize queue
+ data, err := json.MarshalIndent(mq.pending, "", " ")
+ if err != nil {
+ return fmt.Errorf("failed to marshal queue: %w", err)
+ }
+
+ // TODO: Encrypt if mq.encrypted is true
+ // For now, just save plaintext
+
+ // Write to temp file first (atomic write)
+ tempPath := queuePath + ".tmp"
+ if err := os.WriteFile(tempPath, data, 0600); err != nil {
+ return fmt.Errorf("failed to write queue: %w", err)
+ }
+
+ // Atomic rename
+ if err := os.Rename(tempPath, queuePath); err != nil {
+ return fmt.Errorf("failed to rename queue: %w", err)
+ }
+
+ return nil
+}
+
+// loadFromDisk loads the queue from disk
+func (mq *MessageQueue) loadFromDisk() error {
+ queuePath := filepath.Join(mq.queueDir, "queue.json")
+
+ // Check if file exists
+ if _, err := os.Stat(queuePath); os.IsNotExist(err) {
+ // No existing queue - that's fine
+ return nil
+ }
+
+ // Read file
+ data, err := os.ReadFile(queuePath)
+ if err != nil {
+ return fmt.Errorf("failed to read queue: %w", err)
+ }
+
+ // TODO: Decrypt if mq.encrypted is true
+
+ // Parse JSON
+ if err := json.Unmarshal(data, &mq.pending); err != nil {
+ return fmt.Errorf("failed to unmarshal queue: %w", err)
+ }
+
+ return nil
+}
+
+// cleanupLoop periodically removes old messages
+func (mq *MessageQueue) cleanupLoop() {
+ ticker := time.NewTicker(10 * time.Minute)
+ defer ticker.Stop()
+
+ for range ticker.C {
+ mq.cleanup()
+ }
+}
+
+// cleanup removes expired messages
+func (mq *MessageQueue) cleanup() {
+ mq.mu.Lock()
+ defer mq.mu.Unlock()
+
+ now := time.Now()
+ removed := 0
+
+ for recipient, messages := range mq.pending {
+ // Filter out expired messages
+ validMessages := make([]*QueuedMessage, 0)
+ for _, msg := range messages {
+ // Remove if too old or too many attempts
+ if now.Sub(msg.Timestamp) > mq.maxAge || msg.Attempts >= msg.MaxAttempts {
+ removed++
+ continue
+ }
+ validMessages = append(validMessages, msg)
+ }
+
+ if len(validMessages) > 0 {
+ mq.pending[recipient] = validMessages
+ } else {
+ delete(mq.pending, recipient)
+ }
+ }
+
+ // Clean delivered map (keep only recent entries)
+ for msgID := range mq.delivered {
+ // TODO: Add timestamp to delivered map to clean old entries
+ _ = msgID
+ }
+
+ if removed > 0 {
+ mq.saveToDisk()
+ }
+}
+
+// Clear removes all messages from queue
+func (mq *MessageQueue) Clear() error {
+ mq.mu.Lock()
+ defer mq.mu.Unlock()
+
+ mq.pending = make(map[string][]*QueuedMessage)
+ mq.delivered = make(map[string]bool)
+
+ return mq.saveToDisk()
+}
+
+// SetOnDelivered sets callback for when message is delivered
+func (mq *MessageQueue) SetOnDelivered(callback func(string)) {
+ mq.mu.Lock()
+ defer mq.mu.Unlock()
+ mq.onDelivered = callback
+}
+
+// String returns a string representation
+func (mq *MessageQueue) String() string {
+ mq.mu.RLock()
+ defer mq.mu.RUnlock()
+
+ totalPending := 0
+ for _, messages := range mq.pending {
+ totalPending += len(messages)
+ }
+
+ return fmt.Sprintf("MessageQueue(pending=%d, delivered=%d, recipients=%d)",
+ totalPending, len(mq.delivered), len(mq.pending))
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-13 02:09:37 +0000