internet_use_control/cmd/control/main.go

// Command control is the Internet Use Control Control Service.
//
// It is the single source of truth for the family's device rules
// (SYSTEM_MAP.md). Phase 3: publishes each device's desired state to the broker,
// retained, and consumes device presence and telemetry reports from it.
package main

import (
	"context"
	"crypto/rand"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"math/big"
	"net/http"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"sync/atomic"
	"syscall"
	"time"

	"golang.org/x/crypto/bcrypt"

	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/alert"
	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/auth"
	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/config"
	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/hub"
	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/mqtt"
	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store"
	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/web"
)

// version is injected with -ldflags at build time.
var version = "dev"

// republishInterval is how often the retained config is refreshed from SQLite.
// The broker is transport, not storage: if its retained store is ever lost, the
// service repairs it from the database without a restart.
const republishInterval = 15 * time.Minute

// republishPoll is how often the loop re-evaluates the next schedule boundary.
// It bounds how late a transition can be after the schedule is edited, and it
// keeps the service from sleeping past a boundary that did not exist yet when
// the loop last looked.
const republishPoll = 30 * time.Second

// Maintenance (Phase 7): the daily usage rollup is rebuilt periodically so it
// is repaired from the detail table, and the append-only audit log is capped so
// it cannot grow without bound (AGENTS.md §7 retention).
const (
	maintenanceInterval = 15 * time.Minute
	auditMaxRows        = 20000
)

func main() {
	if len(os.Args) > 1 {
		switch os.Args[1] {
		case "hash-password":
			hashPassword(os.Args[2:])
			return
		case "mqtt-clients":
			mqttClients(os.Args[2:])
			return
		case "keycode":
			keycode(os.Args[2:])
			return
		case "rollup":
			rollup(os.Args[2:])
			return
		case "alerts":
			alertsCommand(os.Args[2:])
			return
		case "usage":
			usageCommand(os.Args[2:])
			return
		case "version":
			fmt.Println(version)
			return
		}
	}
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
}

func run() error {
	started := time.Now()

	cfg, err := config.Load()
	if err != nil {
		return err
	}
	cfg.Version = version

	log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: cfg.LogLevel}))
	slog.SetDefault(log)

	st, err := store.Open(cfg.DBPath)
	if err != nil {
		return err
	}
	defer st.Close()

	if err := st.Migrate(context.Background()); err != nil {
		return err
	}

	// Backfill the daily rollup from any telemetry written before Phase 7, then
	// cap the audit log. Both are idempotent.
	if n, err := st.RebuildRollups(context.Background()); err != nil {
		return err
	} else if n > 0 {
		log.Info("usage rollups rebuilt", "rows", n)
	}
	if n, err := st.CapAudit(context.Background(), auditMaxRows); err != nil {
		log.Error("audit cap", "err", err)
	} else if n > 0 {
		log.Info("audit log capped", "removed", n, "max", auditMaxRows)
	}

	// Phase 8: seed the shared device keycode from environment.d. The stored
	// value is the runtime copy that `control keycode` can change; the
	// environment is the deploy-time source. A malformed hash stops start-up
	// rather than silently disabling the keycode path.
	if seeded, err := st.SeedKeycodeHash(context.Background(), cfg.DeviceKeycodeHash); err != nil {
		return err
	} else if seeded {
		log.Info("device keycode seeded from CONTROL_DEVICE_KEYCODE_HASH")
	}

	am, err := auth.New(cfg.SessionSecret, cfg.SessionTTL, cfg.CookieSecure,
		cfg.AdminUsername, cfg.AdminPasswordHash, cfg.AdminPassword)
	if err != nil {
		return err
	}
	if !am.Hashed() {
		log.Warn("admin password is plaintext; set CONTROL_ADMIN_PASSWORD_HASH to use bcrypt")
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	go maintenanceLoop(ctx, log, st)

	// The broker link (Phase 3). It is optional on purpose: the dashboard must
	// keep working while the broker is down, and the device keeps its last state
	// (AGENTS.md §10 rule 2).
	var (
		link *mqtt.Link
		pub  web.Publisher
	)
	// alerts holds the Phase 9 engine once it exists. The broker link is built
	// before the web server (and therefore before the engine), so a status
	// payload that carries a tamper report (FR-17) reaches the engine through
	// this pointer. Absent a configured engine, the tamper is still logged by
	// the link and nothing is delivered.
	var alerts atomic.Pointer[alert.Engine]
	if cfg.MQTTEnabled {
		link, err = mqtt.New(mqtt.Options{
			Broker:      cfg.MQTTBroker,
			ClientID:    cfg.MQTTClientID,
			Username:    cfg.MQTTUsername,
			Password:    cfg.MQTTPassword,
			CAFile:      cfg.MQTTCAFile,
			Actor:       "device",
			Location:    cfg.ScheduleLocation,
			Log:         log,
			Devices:     st,
			Source:      st,
			Reports:     st,
			OnConnected: func() { republish(ctx, log, link, st) },
			OnTamper: func(tctx context.Context, dev store.Device, t mqtt.Tamper) {
				engine := alerts.Load()
				if engine == nil {
					log.Warn("tamper received but alerts are disabled", "device", dev.Name, "kind", t.Kind)
					return
				}
				engine.Raise(tctx, dev, store.AlertTamper, alert.TamperMessage(dev, t.Kind, t.Detail))
			},
		})
		if err != nil {
			return err
		}
		pub = link
		link.Start(ctx)
		defer link.Stop()
		go republishLoop(ctx, log, link, st, cfg.ScheduleLocation)
		if !cfg.MQTTTLS() {
			log.Warn("mqtt is running without TLS; the shared broker has no TLS listener yet (AGENTS.md §10 rule 6)",
				"broker", cfg.MQTTBroker)
		}
	} else {
		log.Warn("mqtt is disabled; device config stays in SQLite and is not published")
	}

	srv, err := web.New(cfg, st, am, log, pub)
	if err != nil {
		return err
	}

	// Phase 9: the alert engine turns a limit hit or a too-long offline spell
	// into one Apprise message to Sam (FR-13). It is optional; /ask and
	// /notify keep working without it.
	if cfg.AlertsEnabled {
		engine := alert.New(st, srv.Apprise(), log, cfg.AlertOfflineAfter, cfg.ScheduleLocation)
		engine.Interval = cfg.AlertInterval
		alerts.Store(engine)
		go engine.Run(ctx)
		if !cfg.AppriseConfigured() {
			log.Warn("alerts are enabled but Apprise is not configured; alerts will be recorded as undelivered")
		} else {
			log.Info("alert engine started", "offline_after", cfg.AlertOfflineAfter.String())
		}
	} else {
		log.Info("alert engine disabled")
	}

	httpServer := &http.Server{
		Addr:              cfg.Addr,
		Handler:           srv.Handler(),
		ReadHeaderTimeout: 10 * time.Second,
		ReadTimeout:       30 * time.Second,
		WriteTimeout:      30 * time.Second,
		IdleTimeout:       60 * time.Second,
	}

	errCh := make(chan error, 1)
	go func() {
		log.Info("control service listening",
			"addr", cfg.Addr, "db", cfg.DBPath, "env", cfg.AppEnv, "version", version)
		if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			errCh <- err
		}
	}()

	stop := make(chan os.Signal, 1)
	signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)

	select {
	case err := <-errCh:
		return err
	case sig := <-stop:
		log.Info("shutting down", "signal", sig.String())
	}
	cancel()

	ctxShutdown, cancelShutdown := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancelShutdown()
	if err := httpServer.Shutdown(ctxShutdown); err != nil {
		return fmt.Errorf("shutdown: %w", err)
	}
	log.Info("stopped", "uptime", time.Since(started).Round(time.Second).String())
	return nil
}

// republish pushes every device's retained config. It runs once per connect and
// whenever the desired state may have changed, so the broker's view is rebuilt
// from SQLite, the source of truth. It first clears any manual lock that has
// expired, so the published state is never stale.
func republish(ctx context.Context, log *slog.Logger, link *mqtt.Link, st *store.Store) {
	if link == nil {
		return
	}
	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	if n, err := st.ExpireLocks(ctx); err != nil {
		log.Error("expire locks", "err", err)
	} else if n > 0 {
		log.Info("expired timed locks", "count", n)
	}
	n, err := link.PublishAll(ctx)
	if err != nil {
		log.Error("mqtt publish all", "published", n, "err", err)
		return
	}
	log.Info("mqtt retained config refreshed", "devices", n)
}

// republishLoop refreshes the retained config when a schedule boundary is
// crossed and otherwise on a long interval. It re-evaluates often instead of
// sleeping until one computed boundary, so a schedule edited after the loop
// started still turns on and off on time (FR-2) without a poll from the device.
func republishLoop(ctx context.Context, log *slog.Logger, link *mqtt.Link, st *store.Store, loc *time.Location) {
	if loc == nil {
		loc = time.Local
	}
	lastEval := time.Now()
	lastPublish := lastEval
	for {
		select {
		case <-ctx.Done():
			return
		case <-time.After(republishPoll):
		}
		if !link.Connected() {
			lastEval = time.Now()
			continue
		}
		now := time.Now()
		shouldPublish := !now.Before(lastPublish.Add(republishInterval))
		// A boundary between the last evaluation and now has been crossed.
		if boundary, err := st.NextScheduleBoundary(ctx, lastEval.In(loc)); err != nil {
			log.Error("next schedule boundary", "err", err)
		} else if boundary != nil && !boundary.After(now) {
			shouldPublish = true
		}
		if shouldPublish {
			republish(ctx, log, link, st)
			lastPublish = time.Now()
		}
		lastEval = now
	}
}

// maintenanceLoop keeps derived data healthy without blocking the service: it
// rebuilds the daily rollups and caps the audit log on a fixed interval.
func maintenanceLoop(ctx context.Context, log *slog.Logger, st *store.Store) {
	ticker := time.NewTicker(maintenanceInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			runMaintenance(ctx, log, st)
		}
	}
}

func runMaintenance(ctx context.Context, log *slog.Logger, st *store.Store) {
	mctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	if n, err := st.RebuildRollups(mctx); err != nil {
		log.Error("usage rollups", "err", err)
	} else if n > 0 {
		log.Debug("usage rollups rebuilt", "rows", n)
	}
	if n, err := st.CapAudit(mctx, auditMaxRows); err != nil {
		log.Error("audit cap", "err", err)
	} else if n > 0 {
		log.Info("audit log capped", "removed", n, "max", auditMaxRows)
	}
}

// alertsCommand prints the most recent alerts, newest first, one per line as
// tab-separated fields:
//
//	<id>\t<kind>\t<delivered|pending>\t<created>\t<device>\t<message>
//
// It is a read-only operational view. The Phase 9 verification script uses it
// to check that an alert was raised and delivered.
func alertsCommand(args []string) {
	dbPath := env("CONTROL_DB_PATH", "/data/control.db")
	limit := 20
	for i := 0; i < len(args); i++ {
		switch args[i] {
		case "--db", "-db":
			if i+1 < len(args) {
				dbPath = args[i+1]
				i++
			}
		case "--limit", "-limit":
			if i+1 < len(args) {
				if n, err := strconv.Atoi(args[i+1]); err == nil {
					limit = n
				}
				i++
			}
		}
	}

	st, err := store.Open(dbPath)
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	defer st.Close()
	if err := st.Migrate(context.Background()); err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	list, err := st.ListAlerts(context.Background(), limit)
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	for _, a := range list {
		delivered := "pending"
		if a.Delivered {
			delivered = "delivered"
		}
		device := a.DeviceName
		if device == "" {
			device = "-"
		}
		fmt.Printf("%d\t%s\t%s\t%s\t%s\t%s\n",
			a.ID, a.Kind, delivered, a.CreatedAt.Format(time.RFC3339), device, a.Message)
	}
}

// usageCommand prints one day's per-device screen time. --detail reads the raw
// telemetry detail table instead of the derived rollup, so an operator can
// check the daily answer independently.
func usageCommand(args []string) {
	dbPath := env("CONTROL_DB_PATH", "/data/control.db")
	day := time.Now().Format("2006-01-02")
	detail := false
	for i := 0; i < len(args); i++ {
		switch args[i] {
		case "--db", "-db":
			if i+1 < len(args) {
				dbPath = args[i+1]
				i++
			}
		case "--day", "-day":
			if i+1 < len(args) {
				day = args[i+1]
				i++
			}
		case "--detail", "-detail":
			detail = true
		}
	}

	st, err := store.Open(dbPath)
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	defer st.Close()
	ctx := context.Background()
	if err := st.Migrate(ctx); err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}

	source := "rollup"
	var rows []store.DeviceDayUsage
	if detail {
		source = "detail"
		rows, err = st.DetailUsageByDeviceForDay(ctx, day)
	} else {
		rows, err = st.UsageByDeviceForDay(ctx, day)
	}
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}

	fmt.Printf("day=%s source=%s\n", day, source)
	total := 0
	for _, u := range rows {
		screen := hub.ScreenSeconds(u.AppSeconds, u.DomainSeconds)
		total += screen
		fmt.Printf("%s\t%s\tapp=%d\tdomain=%d\tscreen=%d\n",
			u.Name, u.Child, u.AppSeconds, u.DomainSeconds, screen)
	}
	fmt.Printf("total_screen=%d\n", total)
}

// rollup rebuilds the daily usage rollup from the telemetry detail table. It is
// idempotent and safe to run while the service is up; it exits non-zero on a
// failure so a script can tell.
func rollup(args []string) {
	dbPath := env("CONTROL_DB_PATH", "/data/control.db")
	for i := 0; i < len(args); i++ {
		if (args[i] == "--db" || args[i] == "-db") && i+1 < len(args) {
			dbPath = args[i+1]
			i++
		}
	}

	st, err := store.Open(dbPath)
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	defer st.Close()
	if err := st.Migrate(context.Background()); err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	n, err := st.RebuildRollups(context.Background())
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	fmt.Printf("rebuilt %d daily rollup rows from %s\n", n, dbPath)
}

// mqttClients prints the MQTT identity of every registered device, one device
// per line, as tab-separated fields:
//
//	<token>\t<config topic>\t<platform>\t<child>\t<name>
//
// provision-clients.sh uses it to create one broker account per device. Real
// tabs, not padded columns: names contain spaces, so a space-separated field
// would not be safe to parse. The token is the MQTT username and the middle
// level of the device's topics (deploy/mosquitto/README.md). It is not a secret
// and this never prints one.
func mqttClients(args []string) {
	dbPath := env("CONTROL_DB_PATH", "/data/control.db")
	for i := 0; i < len(args); i++ {
		if (args[i] == "--db" || args[i] == "-db") && i+1 < len(args) {
			dbPath = args[i+1]
			i++
		}
	}

	st, err := store.Open(dbPath)
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	defer st.Close()

	devices, err := st.ListDevices(context.Background())
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}

	for _, dev := range devices {
		token := strings.TrimSuffix(strings.TrimPrefix(dev.Topic, mqtt.TopicPrefix+"/"), mqtt.SuffixConfig)
		fmt.Printf("%s\t%s\t%s\t%s\t%s\n", token, dev.Topic, dev.Platform, dev.Child, dev.Name)
	}
}

func env(key, fallback string) string {
	if v := strings.TrimSpace(os.Getenv(key)); v != "" {
		return v
	}
	return fallback
}

// keycode sets the shared device keycode (AGENTS.md §10 rule 3). It reads the
// code from a positional argument or stdin, or generates a random six-digit
// code with --generate, and stores only its bcrypt hash in SQLite. The code
// itself is printed once, and only when it was generated here, so it can be set
// on each device; it is never logged by the running service.
func keycode(args []string) {
	dbPath := env("CONTROL_DB_PATH", "/data/control.db")
	generate := false
	code := ""
	for i := 0; i < len(args); i++ {
		switch args[i] {
		case "--db", "-db":
			if i+1 < len(args) {
				dbPath = args[i+1]
				i++
			}
		case "--generate", "-generate":
			generate = true
		default:
			if code == "" {
				code = args[i]
			}
		}
	}

	if generate {
		n, err := rand.Int(rand.Reader, big.NewInt(1000000))
		if err != nil {
			fmt.Fprintln(os.Stderr, "control:", err)
			os.Exit(1)
		}
		code = fmt.Sprintf("%06d", n.Int64())
	} else if code == "" {
		b, err := io.ReadAll(os.Stdin)
		if err != nil {
			fmt.Fprintln(os.Stderr, "control:", err)
			os.Exit(1)
		}
		code = strings.TrimSpace(string(b))
	}
	if len(code) < 4 || len(code) > 64 {
		fmt.Fprintln(os.Stderr, "control: keycode must be 4 to 64 characters")
		os.Exit(2)
	}

	st, err := store.Open(dbPath)
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	defer st.Close()
	if err := st.Migrate(context.Background()); err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	hash, err := bcrypt.GenerateFromPassword([]byte(code), bcrypt.DefaultCost)
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	if err := st.SetKeycodeHash(context.Background(), string(hash), "admin"); err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	if generate {
		fmt.Printf("device keycode set: %s\n", code)
		return
	}
	fmt.Println("device keycode updated")
}

// hashPassword prints a bcrypt hash for the password given as an argument or
// read from stdin. It never prints the password itself.
func hashPassword(args []string) {
	var password string
	if len(args) > 0 {
		password = args[0]
	} else {
		b, err := io.ReadAll(os.Stdin)
		if err != nil {
			fmt.Fprintln(os.Stderr, "control:", err)
			os.Exit(1)
		}
		password = strings.TrimRight(string(b), "\r\n")
	}
	if password == "" {
		fmt.Fprintln(os.Stderr, "control: empty password")
		os.Exit(2)
	}
	hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
	if err != nil {
		fmt.Fprintln(os.Stderr, "control:", err)
		os.Exit(1)
	}
	fmt.Println(string(hash))
}