DebTik/internal/bwtest/manager.go

package bwtest

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"sync"
)

const DefaultConfigPath = "/etc/debtik/bwtest.json"

// Manager coordinates the BwTest server and client subsystems
type Manager struct {
	mu         sync.RWMutex
	configPath string
	server     *BwTestServer
	client     *BwTestClient
	history    []*HistoricalTestRecord
}

// NewManager initializes the Bandwidth Test manager and starts the background server
func NewManager(cfgPath string) *Manager {
	if cfgPath == "" {
		cfgPath = DefaultConfigPath
	}

	cfg := ServerConfig{
		Enabled:        true,
		Port:           2000,
		ListenAddr:     "0.0.0.0",
		AllowAnonymous: true,
	}

	m := &Manager{
		configPath: cfgPath,
		history:    make([]*HistoricalTestRecord, 0, 50),
	}

	// Try loading persisted configuration
	if loaded, err := m.loadConfig(); err == nil {
		cfg = *loaded
	}

	m.server = NewServer(cfg)
	m.client = NewClient()

	if cfg.Enabled {
		if err := m.server.Start(); err != nil {
			fmt.Printf("[BwTestManager] Warning: failed to start server on port %d: %v\n", cfg.Port, err)
		}
	}

	return m
}

func (m *Manager) loadConfig() (*ServerConfig, error) {
	data, err := os.ReadFile(m.configPath)
	if err != nil {
		return nil, err
	}
	var cfg ServerConfig
	if err := json.Unmarshal(data, &cfg); err != nil {
		return nil, err
	}
	return &cfg, nil
}

func (m *Manager) saveConfigLocked(cfg ServerConfig) error {
	_ = os.MkdirAll("/etc/debtik", 0755)
	data, err := json.MarshalIndent(cfg, "", "  ")
	if err != nil {
		return err
	}
	return os.WriteFile(m.configPath, data, 0644)
}

// GetServerStatus returns metrics and status for the local BwTest server
func (m *Manager) GetServerStatus() ServerStatus {
	return m.server.Status()
}

// SetServerConfig updates server configuration (enabled state, port)
func (m *Manager) SetServerConfig(cfg ServerConfig) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if cfg.Port <= 0 {
		cfg.Port = 2000
	}
	if cfg.ListenAddr == "" {
		cfg.ListenAddr = "0.0.0.0"
	}

	m.server.Stop()
	m.server.config = cfg

	if cfg.Enabled {
		if err := m.server.Start(); err != nil {
			return err
		}
	}

	return m.saveConfigLocked(cfg)
}

// StartTest initiates an asynchronous bandwidth test against the specified target
func (m *Manager) StartTest(ctx context.Context, req SessionRequest) error {
	return m.client.Run(ctx, req)
}

// StopTest terminates any ongoing bandwidth test
func (m *Manager) StopTest() {
	m.client.Stop()
}

// GetTestStatus retrieves current live test status and telemetry samples
func (m *Manager) GetTestStatus() TestStatus {
	st := m.client.GetStatus()

	// If test completed with a summary, persist into history
	if !st.Running && st.Summary != nil && st.Error == "" {
		m.mu.Lock()
		alreadyExists := false
		for _, h := range m.history {
			if h.ID == st.SessionID {
				alreadyExists = true
				break
			}
		}
		if !alreadyExists && st.SessionID != "" {
			record := &HistoricalTestRecord{
				ID:        st.SessionID,
				Target:    fmt.Sprintf("%s:%d", st.Target, st.Port),
				Protocol:  st.Protocol,
				Direction: st.Direction,
				Timestamp: st.StartTime,
				Summary:   st.Summary,
			}
			m.history = append([]*HistoricalTestRecord{record}, m.history...)
			if len(m.history) > 50 {
				m.history = m.history[:50]
			}
		}
		m.mu.Unlock()
	}

	return st
}

// SubscribeClient returns a stream of real-time telemetry events
func (m *Manager) SubscribeClient() (<-chan LiveSample, func()) {
	return m.client.Subscribe()
}

// GetHistory returns past completed bandwidth test results
func (m *Manager) GetHistory() []*HistoricalTestRecord {
	m.mu.RLock()
	defer m.mu.RUnlock()
	res := make([]*HistoricalTestRecord, len(m.history))
	copy(res, m.history)
	return res
}

// Stop shuts down the manager and server listeners cleanly
func (m *Manager) Stop() {
	m.client.Stop()
	m.server.Stop()
}