yovision/Sense/cmd/sense-api/main.go

package main

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"yovision/sense/internal/auth"
	"yovision/sense/internal/config"
	"yovision/sense/internal/controlapi"
	"yovision/sense/internal/mtx"
	"yovision/sense/internal/onvif"
	"yovision/sense/internal/probe"
	"yovision/sense/internal/reconcile"
	"yovision/sense/internal/store"
)

var version = "dev"

func main() {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
	if err := run(logger); err != nil {
		logger.Error("Sense stopped", "error", err)
		os.Exit(1)
	}
}

func run(logger *slog.Logger) error {
	cfg, err := config.Load()
	if err != nil {
		return fmt.Errorf("load configuration: %w", err)
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	repository, err := store.OpenRepository(ctx, cfg.DatabaseDriver, cfg.DatabaseDSN)
	if err != nil {
		return err
	}
	defer repository.Close()
	var controlHandler http.Handler
	if cfg.ControlAPIEnabled {
		controlStore, ok := repository.(store.ControlRepository)
		if !ok {
			return errors.New("selected repository does not support Sense Control API")
		}
		authenticator, err := auth.LoadStaticSHA256(cfg.ControlAuthFile)
		if err != nil {
			return err
		}
		cursors, err := controlapi.LoadCursorCodec(cfg.ControlCursorKeyFile)
		if err != nil {
			return err
		}
		controlHandler = controlapi.NewHTTPHandler(controlStore, authenticator, cursors)
	}
	mediaClient, err := mtx.NewClient(cfg.MediaMTXURL, nil)
	if err != nil {
		return err
	}

	credentials := onvif.EnvCredentials{}
	var cameraAdapter onvif.Adapter = onvif.UnavailableAdapter{}
	if cfg.ONVIFMode == "standard" {
		cameraAdapter = onvif.NewHTTPAdapter(credentials, nil, onvif.HTTPOptions{
			RTSPRewriteHost: cfg.RTSPRewriteHost,
			RTSPRewritePort: cfg.RTSPRewritePort,
			StripRTSPQuery:  cfg.RTSPStripQuery,
		})
	}
	discovery := onvif.NewRouter(cameraAdapter, credentials)
	reconciler := reconcile.New(repository, discovery, mediaClient)
	checker := probe.New(repository, mediaClient)
	report := func(err error) {
		// Domain and MediaMTX errors intentionally omit stream URIs and credentials.
		logger.Warn("background convergence error", "error", err)
	}
	var background sync.WaitGroup
	background.Add(2)
	go func() {
		defer background.Done()
		reconciler.Run(ctx, cfg.ReconcileInterval, report)
	}()
	go func() {
		defer background.Done()
		checker.Run(ctx, cfg.ProbeInterval, report)
	}()

	mux := http.NewServeMux()
	mux.HandleFunc("GET /healthz", func(writer http.ResponseWriter, _ *http.Request) {
		writer.Header().Set("Content-Type", "application/json")
		writer.WriteHeader(http.StatusOK)
		_, _ = writer.Write([]byte(`{"status":"ok"}`))
	})
	mux.HandleFunc("GET /readyz", func(writer http.ResponseWriter, _ *http.Request) {
		writer.Header().Set("Content-Type", "application/json")
		writer.WriteHeader(http.StatusOK)
		_, _ = writer.Write([]byte(`{"status":"ready"}`))
	})
	if cfg.ControlAPIEnabled {
		mux.Handle("/api/v1/", controlHandler)
	}

	server := &http.Server{
		Addr: cfg.HTTPAddress, Handler: mux,
		ReadHeaderTimeout: 5 * time.Second,
		ReadTimeout:       15 * time.Second,
		WriteTimeout:      15 * time.Second,
		IdleTimeout:       60 * time.Second,
	}
	serverErrors := make(chan error, 1)
	go func() {
		logger.Info("Sense listening", "address", cfg.HTTPAddress, "version", version,
			"control_api_enabled", cfg.ControlAPIEnabled)
		serverErrors <- server.ListenAndServe()
	}()

	select {
	case <-ctx.Done():
	case serverErr := <-serverErrors:
		if !errors.Is(serverErr, http.ErrServerClosed) {
			stop()
			background.Wait()
			return fmt.Errorf("serve HTTP: %w", serverErr)
		}
	}
	stop()
	shutdownContext, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	shutdownErr := server.Shutdown(shutdownContext)
	background.Wait()
	if shutdownErr != nil {
		return fmt.Errorf("shutdown HTTP server: %w", shutdownErr)
	}
	return nil
}