Split 5 of 7: capture adapters, camera producer hub and agent DataService

go/cmd/wendy-agent/main.go

@@ -16,6 +16,7 @@ import (
 	"path/filepath"
 	"runtime"
 	"strconv"
+	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -30,6 +31,7 @@ import (
 	"github.com/wendylabsinc/wendy/go/internal/agent/configpartition"
 	"github.com/wendylabsinc/wendy/go/internal/agent/container"
 	agentcontainerd "github.com/wendylabsinc/wendy/go/internal/agent/containerd"
+	agentdata "github.com/wendylabsinc/wendy/go/internal/agent/data"
 	"github.com/wendylabsinc/wendy/go/internal/agent/dbusproxy"
 	"github.com/wendylabsinc/wendy/go/internal/agent/hardware"
 	"github.com/wendylabsinc/wendy/go/internal/agent/hostexec"
@@ -283,11 +285,31 @@ func main() {
 	provisioningSvcV2 := services.NewProvisioningServiceV2(provisioningSvc)
 	audioSvcV2 := services.NewAudioServiceV2(audioSvc)
 	telemetrySvcV2 := services.NewTelemetryServiceV2(logger, broadcaster, telemetryBuf)
+	dataRoot := os.Getenv("WENDY_DATA_DIR")
+	dataManager, err := agentdata.NewManager(dataRoot)
+	if err != nil {
+		logger.Fatal("Failed to initialize episode data manager", zap.Error(err))
+	}
+	dataManager.SetConsensusProvider(func(ctx context.Context) (timesync.Consensus, error) {
+		return timesync.QueryConsensus(ctx, timesync.Servers)
+	})
+	dataManager.SetWarnLogger(func(msg string) { logger.Warn(msg) })
+	// Episode store bounds. The enforced quota is the smaller of a fifth of the
+	// data filesystem and this cap, and eviction preserves the reserve as free
+	// space. A device whose data partition wants different bounds sets these
+	// rather than being stuck with the built-in numbers.
+	dataManager.SetQuota(
+		envBytes(logger, "WENDY_DATA_MAX_BYTES", agentdata.DefaultMaxQuotaBytes),
+		envBytes(logger, "WENDY_DATA_RESERVE_BYTES", agentdata.DefaultReserveBytes),
+	)
+	dataSvc := services.NewDataService(dataManager)
+	dataSvc.SetAudioService(audioSvc)
 	// ROS 2 inspection requires the containerd-backed sidecar runtime; the
 	// service is only registered when containerd connected (WDY-1332).
 	var ros2Svc *services.ROS2Service
 	if ctrdClient != nil {
 		ros2Svc = services.NewROS2Service(logger, ctrdClient, agentcontainerd.ROS2BagDir)
+		dataSvc.SetROS2Service(ros2Svc)
 	}
 
 	// OTEL receivers.
@@ -298,10 +320,22 @@ func main() {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	// The video service is constructed before the app socket managers because it
+	// owns the camera producer the sensor sockets subscribe apps to, and every
+	// per-app sensor socket must be built with that provider already registered.
+	videoSvc := services.NewVideoService(ctx, logger)
+	dataSvc.SetVideoService(videoSvc)
+	defer videoSvc.Shutdown()
+
 	notificationSender := services.NewCloudNotificationSender(logger, provisioningSvc)
 	systemAPISocketManager := services.NewAppSystemAPISocketManager(ctx, logger, notificationSender)
+	appDataSocketManager := services.NewAppDataSocketManager(ctx, logger, dataManager)
+	appSensorSocketManager := services.NewAppSensorSocketManager(ctx, logger, dataManager)
+	appSensorSocketManager.AddProvider(videoSvc)
 	if ctrdClient != nil {
 		ctrdClient.SetAppSystemAPISocketProvider(systemAPISocketManager)
+		ctrdClient.SetAppDataSocketProvider(appDataSocketManager)
+		ctrdClient.SetAppSensorSocketProvider(appSensorSocketManager)
 		ctrdClient.RestoreAppSystemAPISockets(ctx)
 	}
 
@@ -310,8 +344,6 @@ func main() {
 
 	startROS2BatteryMonitor(ctx, logger, configPath)
 
-	videoSvc := services.NewVideoService(ctx, logger)
-	defer videoSvc.Shutdown()
 	// Network cameras have to be found before they can be listed, so probe
 	// periodically rather than only when a client asks.
 	videoSvc.StartDiscovery()
@@ -572,6 +604,7 @@ func main() {
 		agentpbv2.RegisterWendyProvisioningServiceServer(srv, provisioningSvcV2)
 		agentpbv2.RegisterWendyAudioServiceServer(srv, audioSvcV2)
 		agentpbv2.RegisterWendyTelemetryServiceServer(srv, telemetrySvcV2)
+		agentpbv2.RegisterDataServiceServer(srv, dataSvc)
 		agentpbv2.RegisterWendyMeshServiceServer(srv, meshSvc)
 		agentpbv2.RegisterWendyBuildServiceServer(srv, buildSvc)
 		if ros2Svc != nil {
@@ -1195,3 +1228,21 @@ func handleUtilityCommand(args []string) (bool, int) {
 	fmt.Printf("Opening %s in default browser...\n", rawURL)
 	return true, 0
 }
+
+// envBytes reads a byte count from the environment, falling back to fallback
+// when the variable is unset. A value that is present but unusable is reported
+// rather than silently ignored: a device configured with a bad quota should
+// learn that its configuration did not take.
+func envBytes(logger *zap.Logger, name string, fallback int64) int64 {
+	raw, ok := os.LookupEnv(name)
+	if !ok || strings.TrimSpace(raw) == "" {
+		return fallback
+	}
+	v, err := strconv.ParseInt(strings.TrimSpace(raw), 10, 64)
+	if err != nil || v < 0 {
+		logger.Warn("ignoring unusable byte count in environment; using the default",
+			zap.String("variable", name), zap.String("value", raw), zap.Int64("default", fallback))
+		return fallback
+	}
+	return v
+}