teraWeb/backend/internal/logruntime/scheduler.go

// Package logruntime runs configured status-event cadences and feeds the
// existing API policy/alarm/syslog path. Durable persistence remains owned by
// tera-logd through the store's journal adapter.
package logruntime

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"tera-clone/services/tera-api/internal/domain"
	"tera-clone/services/tera-api/internal/serial"
)

type Backend interface {
	LogRetention(context.Context) (domain.LogRetentionSettings, error)
	RecordSystemLog(context.Context, domain.SystemLogEvent) (domain.SystemLogEvent, error)
	RecordLinkState(context.Context, domain.LinkStateObservation) (*domain.LinkLogEvent, error)
}

type Observer interface {
	Observe(context.Context, time.Time) ([]domain.LinkStateObservation, error)
}

type Options struct {
	Backend  Backend
	Observer Observer
	Clock    func() time.Time
	Tick     <-chan time.Time
	OnError  func(error)
}

type Scheduler struct {
	backend  Backend
	observer Observer
	clock    func() time.Time
	tick     <-chan time.Time
	onError  func(error)

	mu            sync.Mutex
	lastKeepalive time.Time
	lastPoll      time.Time
}

func NewScheduler(options Options) (*Scheduler, error) {
	if options.Backend == nil {
		return nil, errors.New("log scheduler backend is required")
	}
	clock := options.Clock
	if clock == nil {
		clock = time.Now
	}
	return &Scheduler{backend: options.Backend, observer: options.Observer, clock: clock, tick: options.Tick, onError: options.OnError}, nil
}

func (s *Scheduler) Run(ctx context.Context) {
	if ctx == nil {
		ctx = context.Background()
	}
	if s.tick != nil {
		for {
			select {
			case <-ctx.Done():
				return
			case now, ok := <-s.tick:
				if !ok {
					return
				}
				s.step(ctx, now.UTC())
			}
		}
	}
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case now := <-ticker.C:
			s.step(ctx, now.UTC())
		}
	}
}

func (s *Scheduler) step(ctx context.Context, now time.Time) {
	policy, err := s.backend.LogRetention(ctx)
	if err != nil {
		s.report(fmt.Errorf("read log cadence policy: %w", err))
		return
	}
	s.mu.Lock()
	keepaliveDue := policy.KeepaliveEnabled && due(s.lastKeepalive, now, time.Duration(policy.KeepaliveIntervalMinutes)*time.Minute)
	pollInterval := linkPollInterval(policy)
	pollDue := pollInterval > 0 && due(s.lastPoll, now, pollInterval)
	if keepaliveDue {
		s.lastKeepalive = now
	}
	if pollDue {
		s.lastPoll = now
	}
	s.mu.Unlock()

	if keepaliveDue {
		_, err := s.backend.RecordSystemLog(ctx, domain.SystemLogEvent{
			Timestamp: now, Category: "status", Severity: "info", Source: "tera-logd",
			Message: "log and link observer keepalive",
		})
		if err != nil {
			s.report(fmt.Errorf("record log keepalive: %w", err))
		}
	}
	if !pollDue || s.observer == nil {
		return
	}
	observations, err := s.observer.Observe(ctx, now)
	if err != nil {
		s.report(fmt.Errorf("observe physical links: %w", err))
		if len(observations) == 0 {
			return
		}
	}
	for _, observation := range observations {
		if _, err := s.backend.RecordLinkState(ctx, observation); err != nil {
			s.report(fmt.Errorf("record %s %s link: %w", observation.Kind, observation.Resource, err))
		}
	}
}

func due(last, now time.Time, interval time.Duration) bool {
	return interval > 0 && (last.IsZero() || !now.Before(last.Add(interval)))
}

func linkPollInterval(policy domain.LogRetentionSettings) time.Duration {
	if policy.SinglePortStatusEnabled {
		if policy.SinglePortStatusRealtime {
			return time.Second
		}
		return time.Duration(policy.SinglePortStatusIntervalMinutes) * time.Minute
	}
	if policy.AllPortStatusEnabled {
		return time.Duration(policy.AllPortStatusIntervalMinutes) * time.Minute
	}
	return 0
}

func (s *Scheduler) report(err error) {
	if err != nil && s.onError != nil {
		s.onError(err)
	}
}

type CompositeObserver []Observer

func (observers CompositeObserver) Observe(ctx context.Context, now time.Time) ([]domain.LinkStateObservation, error) {
	result := make([]domain.LinkStateObservation, 0)
	var combined error
	for _, observer := range observers {
		if observer == nil {
			continue
		}
		values, err := observer.Observe(ctx, now)
		if err != nil {
			combined = errors.Join(combined, err)
		}
		result = append(result, values...)
	}
	return result, combined
}

type EthernetObserver struct {
	SysfsRoot  string
	Interfaces map[string]string
}

func (o EthernetObserver) Observe(ctx context.Context, now time.Time) ([]domain.LinkStateObservation, error) {
	root := o.SysfsRoot
	if root == "" {
		root = "/sys/class/net"
	}
	names := make([]string, 0, len(o.Interfaces))
	for logical := range o.Interfaces {
		names = append(names, logical)
	}
	sort.Strings(names)
	result := make([]domain.LinkStateObservation, 0, len(names))
	for _, logical := range names {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		device := strings.TrimSpace(o.Interfaces[logical])
		state, signal := "unknown", "unknown"
		data, err := os.ReadFile(filepath.Join(root, device, "carrier"))
		if err == nil {
			switch strings.TrimSpace(string(data)) {
			case "0":
				state, signal = "down", "carrier"
			case "1":
				state, signal = "up", "carrier"
			}
		}
		result = append(result, domain.LinkStateObservation{Kind: "ethernet", Resource: logical, State: state, Signal: signal, ObservedAt: now})
	}
	return result, nil
}

func ParseEthernetInterfaces(value string) (map[string]string, error) {
	if strings.TrimSpace(value) == "" {
		return map[string]string{"Interface-1": "eth0", "Interface-2": "eth1"}, nil
	}
	result := make(map[string]string)
	for _, item := range strings.Split(value, ",") {
		parts := strings.SplitN(strings.TrimSpace(item), "=", 2)
		if len(parts) != 2 || strings.TrimSpace(parts[0]) == "" || strings.TrimSpace(parts[1]) == "" {
			return nil, errors.New("Ethernet log mapping must use logical=device entries")
		}
		logical, device := strings.TrimSpace(parts[0]), strings.TrimSpace(parts[1])
		if !strings.HasPrefix(logical, "Interface-") {
			return nil, errors.New("Ethernet log mapping logical names must use Interface-N")
		}
		if _, err := strconv.Atoi(strings.TrimPrefix(logical, "Interface-")); err != nil || strings.ContainsAny(device, `/\\`) {
			return nil, errors.New("Ethernet log mapping contains an invalid interface")
		}
		result[logical] = device
	}
	return result, nil
}

type SerialObserver struct {
	Controller serial.LinkStateController
	Probes     serial.CRProbeStateController
}

func (o SerialObserver) Observe(ctx context.Context, now time.Time) ([]domain.LinkStateObservation, error) {
	if o.Controller == nil && o.Probes == nil {
		return nil, errors.New("serial modem observer is unavailable")
	}
	result := make([]domain.LinkStateObservation, 0, serial.ChannelCount*2)
	var combined error
	if o.Controller != nil {
		states, err := o.Controller.LinkStates(ctx)
		if err != nil {
			combined = errors.Join(combined, err)
		} else {
			for _, state := range states {
				linkState, signal := state.State, state.Signal
				if !state.Available {
					if o.Probes != nil {
						continue
					}
					linkState, signal = "unknown", "unknown"
				}
				result = append(result, domain.LinkStateObservation{Kind: "serial", Resource: state.Channel, State: linkState, Signal: signal, ObservedAt: now})
			}
		}
	}
	if o.Probes != nil {
		probes, err := o.Probes.CRProbeStates(ctx)
		if err != nil {
			combined = errors.Join(combined, err)
		} else {
			for _, probe := range probes {
				if !probe.Enabled {
					continue
				}
				state := serial.CRConnectionStatus(probe, now)
				result = append(result, domain.LinkStateObservation{
					Kind: "serial", Resource: probe.Channel, State: state,
					Signal: "cr", ObservedAt: now,
				})
			}
		}
	}
	return result, combined
}