// 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(), "khimera-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)) }