path: root/transport/mesh/storeforward.go

// 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))
}