Add optional MoQ live-stream publisher

Adds a dedicated MoQ build path that publishes H.264 live streams to a configurable relay, with retry handling, stream selection, Annex B framing, and Docker packaging. Standard builds retain a no-op implementation.

Dockerfile.moq

@@ -0,0 +1,74 @@
+ARG GO_IMAGE=golang:1.24-trixie
+ARG RUNTIME_IMAGE=debian:trixie-slim
+ARG VERSION=0.0.0
+
+FROM ${GO_IMAGE} AS build-machinery
+
+ARG VERSION
+ENV CGO_ENABLED=1
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    ca-certificates \
+    git \
+    libavcodec-dev \
+    libavutil-dev \
+    libswscale-dev \
+    pkg-config \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /src
+COPY machinery ./machinery
+COPY .git ./.git
+
+RUN cd machinery && \
+    go mod download && \
+    if [ -z "${VERSION}" ] || [ "${VERSION}" = "0.0.0" ]; then \
+        VERSION=$(git -C /src describe --tags --always 2>/dev/null || echo "0.0.0"); \
+    fi && \
+    go build \
+        -tags moq,timetzdata,netgo,osusergo \
+        -ldflags "-s -w -X github.com/kerberos-io/agent/machinery/src/utils.VERSION=${VERSION}" \
+        -o /out/main \
+        ./main.go && \
+    cp -r data /out/data && \
+    printf '%s' "${VERSION}" > /out/version && \
+    mkdir -p /out/data/cloud /out/data/snapshots /out/data/log /out/data/recordings /out/data/capture-test /out/data/config
+
+FROM node:22-alpine AS build-ui
+
+WORKDIR /src/ui
+COPY ui ./
+RUN mkdir -p /src/machinery && \
+    yarn config set network-timeout 300000 && \
+    yarn install --frozen-lockfile && \
+    yarn build
+
+FROM ${RUNTIME_IMAGE}
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    ca-certificates \
+    curl \
+    ffmpeg \
+    libcap2-bin \
+    libstdc++6 \
+    && rm -rf /var/lib/apt/lists/* && \
+    groupadd --system kerberosio && \
+    useradd --system --gid kerberosio --groups video --create-home agent
+
+COPY --from=build-machinery --chown=root:root /out/main /home/agent/main
+COPY --from=build-machinery --chown=agent:kerberosio /out/data /home/agent/data
+COPY --from=build-machinery --chown=root:root /out/version /home/agent/version
+COPY --from=build-ui --chown=agent:kerberosio /src/machinery/www /home/agent/www
+
+RUN cp /home/agent/data/config/config.json /home/agent/data/config.template.json && \
+    chown agent:kerberosio /home/agent/data/config.template.json && \
+    setcap 'cap_net_bind_service=+ep' /home/agent/main && \
+    /home/agent/main -action version
+
+USER agent
+WORKDIR /home/agent
+
+EXPOSE 80
+HEALTHCHECK CMD curl --fail http://localhost:80 || exit 1
+CMD ["./main", "-action", "run", "-port", "80"]

machinery/go.mod

@@ -26,6 +26,7 @@ require (
 	github.com/kerberos-io/joy4 v1.0.64
 	github.com/kerberos-io/onvif v1.2.2
 	github.com/minio/minio-go/v6 v6.0.57
+	github.com/moq-dev/moq-go v0.5.7
 	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
 	github.com/pion/interceptor v0.1.40
@@ -95,6 +96,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/montanaflynn/stats v0.7.1 // indirect
+	github.com/moq-dev/moq-go-ffi v0.3.7 // indirect
 	github.com/nxadm/tail v1.4.11 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.3 // indirect
 	github.com/pion/datachannel v1.5.10 // indirect

machinery/go.sum

@@ -845,6 +845,10 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE=
 github.com/montanaflynn/stats v0.7.1/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
+github.com/moq-dev/moq-go v0.5.7 h1:LfFpgAU8FRMcnU85L5Lb03HzxhYbQ2+BhT/pJQdjy5U=
+github.com/moq-dev/moq-go v0.5.7/go.mod h1:5K8zjKKjWe5lzfCtlfxoAfdCr3KX6EBuTOZsz0WBnzw=
+github.com/moq-dev/moq-go-ffi v0.3.7 h1:+xwPOzTJHvB0tuTnW6znbvvOgT4yLnN7sMqXXZYh02M=
+github.com/moq-dev/moq-go-ffi v0.3.7/go.mod h1:zxpOlUetvaoxWBnbXTdILtLLUytaoVuLvO36lftwWO0=
 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6Oo2LfFZAehjjQMERAvZLEDnQ=
 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646/go.mod h1:jpp1/29i3P1S/RLdc7JQKbRpFeM1dOBd8T9ki5s+AY8=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=

machinery/src/cloud/livemoq/annexb.go

@@ -0,0 +1,29 @@
+package livemoq
+
+import "strings"
+
+var annexBStartCode = []byte{0x00, 0x00, 0x00, 0x01}
+
+// EnsureAnnexB restores the start code stripped by the Agent capture queue.
+func EnsureAnnexB(payload []byte) []byte {
+	if hasAnnexBStartCode(payload) {
+		return payload
+	}
+
+	framed := make([]byte, 0, len(annexBStartCode)+len(payload))
+	framed = append(framed, annexBStartCode...)
+	return append(framed, payload...)
+}
+
+func BroadcastPath(prefix string, deviceKey string) string {
+	prefix = strings.Trim(prefix, "/")
+	if prefix == "" {
+		prefix = "devices"
+	}
+	return prefix + "/" + strings.Trim(deviceKey, "/") + "/live.hang"
+}
+
+func hasAnnexBStartCode(payload []byte) bool {
+	return len(payload) >= 4 && payload[0] == 0 && payload[1] == 0 &&
+		((payload[2] == 0 && payload[3] == 1) || payload[2] == 1)
+}

machinery/src/cloud/livemoq/annexb_test.go

@@ -0,0 +1,47 @@
+package livemoq
+
+import (
+	"bytes"
+	"testing"
+)
+
+func TestEnsureAnnexB(t *testing.T) {
+	tests := []struct {
+		name    string
+		payload []byte
+		want    []byte
+	}{
+		{
+			name:    "missing start code",
+			payload: []byte{0x41, 0x01},
+			want:    []byte{0x00, 0x00, 0x00, 0x01, 0x41, 0x01},
+		},
+		{
+			name:    "four byte start code",
+			payload: []byte{0x00, 0x00, 0x00, 0x01, 0x65},
+			want:    []byte{0x00, 0x00, 0x00, 0x01, 0x65},
+		},
+		{
+			name:    "three byte start code",
+			payload: []byte{0x00, 0x00, 0x01, 0x41},
+			want:    []byte{0x00, 0x00, 0x01, 0x41},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if got := EnsureAnnexB(test.payload); !bytes.Equal(got, test.want) {
+				t.Fatalf("EnsureAnnexB() = %x, want %x", got, test.want)
+			}
+		})
+	}
+}
+
+func TestBroadcastPath(t *testing.T) {
+	if got := BroadcastPath("/devices/", "/camera-1/"); got != "devices/camera-1/live.hang" {
+		t.Fatalf("BroadcastPath() = %q", got)
+	}
+	if got := BroadcastPath("", "camera-1"); got != "devices/camera-1/live.hang" {
+		t.Fatalf("BroadcastPath() default = %q", got)
+	}
+}

machinery/src/cloud/livemoq_disabled.go

@@ -0,0 +1,8 @@
+//go:build !moq
+
+package cloud
+
+import "github.com/kerberos-io/agent/machinery/src/models"
+
+// StartLiveStreamMoQ is disabled in the standard Agent build.
+func StartLiveStreamMoQ(_ *models.Configuration, _ *models.Communication, _ bool) {}

machinery/src/cloud/livemoq_enabled.go

@@ -0,0 +1,159 @@
+//go:build moq
+
+package cloud
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/cloud/livemoq"
+	"github.com/kerberos-io/agent/machinery/src/log"
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/agent/machinery/src/packets"
+	"github.com/moq-dev/moq-go/moq"
+)
+
+const (
+	defaultMoQRelayURL = "https://relay.uug.ai/anon"
+	minMoQRetryDelay   = time.Second
+	maxMoQRetryDelay   = 30 * time.Second
+)
+
+type liveMoQConfig struct {
+	relayURL    string
+	broadcast   string
+	quality     string
+	sourceLabel string
+	queue       *packets.Queue
+}
+
+// StartLiveStreamMoQ starts the publisher only in the dedicated MoQ build and
+// only when explicitly enabled by the deployment.
+func StartLiveStreamMoQ(configuration *models.Configuration, communication *models.Communication, subStreamEnabled bool) {
+	if os.Getenv("AGENT_LIVE_MOQ_ENABLED") != "true" {
+		return
+	}
+
+	config := configuration.Config
+	if config.Offline == "true" || config.Capture.Liveview == "false" {
+		log.Log.Info("cloud.StartLiveStreamMoQ(): disabled by Agent live-view configuration")
+		return
+	}
+	if config.Key == "" {
+		log.Log.Warning("cloud.StartLiveStreamMoQ(): AGENT_KEY is required")
+		return
+	}
+
+	quality := os.Getenv("AGENT_LIVE_MOQ_QUALITY")
+	if quality == "" {
+		quality = models.StreamQualityAuto
+	}
+	useSub := models.SelectSubStreamForQuality(config, quality, subStreamEnabled)
+	queue := communication.Queue
+	sourceLabel := "main"
+	if useSub && communication.SubQueue != nil {
+		queue = communication.SubQueue
+		sourceLabel = "sub"
+	}
+	if queue == nil {
+		log.Log.Warning("cloud.StartLiveStreamMoQ(): selected packet queue is unavailable")
+		return
+	}
+
+	relayURL := os.Getenv("AGENT_LIVE_MOQ_URL")
+	if relayURL == "" {
+		relayURL = defaultMoQRelayURL
+	}
+	publisherConfig := liveMoQConfig{
+		relayURL:    relayURL,
+		broadcast:   livemoq.BroadcastPath(os.Getenv("AGENT_LIVE_MOQ_BROADCAST_PREFIX"), config.Key),
+		quality:     quality,
+		sourceLabel: sourceLabel,
+		queue:       queue,
+	}
+
+	ctx := context.Background()
+	if communication.Context != nil {
+		ctx = *communication.Context
+	}
+	go runLiveStreamMoQ(ctx, publisherConfig)
+}
+
+func runLiveStreamMoQ(ctx context.Context, config liveMoQConfig) {
+	log.Log.Info(fmt.Sprintf(
+		"cloud.runLiveStreamMoQ(): publishing %s stream (quality=%s) to %s/%s",
+		config.sourceLabel, config.quality, strings.TrimRight(config.relayURL, "/"), config.broadcast,
+	))
+
+	retryDelay := minMoQRetryDelay
+	for ctx.Err() == nil {
+		connectedAt := time.Now()
+		err := publishLiveStreamMoQ(ctx, config)
+		if ctx.Err() != nil {
+			return
+		}
+		log.Log.Warning("cloud.runLiveStreamMoQ(): publisher stopped: " + err.Error())
+		if time.Since(connectedAt) >= time.Minute {
+			retryDelay = minMoQRetryDelay
+		}
+
+		timer := time.NewTimer(retryDelay)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-timer.C:
+		}
+		if retryDelay < maxMoQRetryDelay {
+			retryDelay *= 2
+			if retryDelay > maxMoQRetryDelay {
+				retryDelay = maxMoQRetryDelay
+			}
+		}
+	}
+}
+
+func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error {
+	client, err := moq.Dial(ctx, config.relayURL)
+	if err != nil {
+		return fmt.Errorf("connect to relay: %w", err)
+	}
+	defer client.Close()
+
+	broadcast, err := client.CreateBroadcast(config.broadcast)
+	if err != nil {
+		return fmt.Errorf("create broadcast: %w", err)
+	}
+	defer broadcast.Finish()
+
+	stream, err := broadcast.PublishMediaStream("avc3")
+	if err != nil {
+		return fmt.Errorf("create H.264 media stream: %w", err)
+	}
+	defer stream.Finish()
+
+	cursor := config.queue.Latest()
+	writing := false
+	for {
+		packet, err := cursor.ReadPacket()
+		if err != nil {
+			return fmt.Errorf("read packet: %w", err)
+		}
+		if !packet.IsVideo || len(packet.Data) == 0 || !strings.EqualFold(packet.Codec, "H264") {
+			continue
+		}
+		if !writing {
+			if !packet.IsKeyFrame {
+				continue
+			}
+			writing = true
+			log.Log.Info("cloud.publishLiveStreamMoQ(): first H.264 keyframe received; broadcast is live")
+		}
+		if err := stream.Write(livemoq.EnsureAnnexB(packet.Data)); err != nil {
+			return fmt.Errorf("write H.264 access unit: %w", err)
+		}
+	}
+}

machinery/src/components/kerberos.go

@@ -305,6 +305,10 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	// watching.
 	go cloud.HandleLiveStreamHLS(configuration, communication, mqttClient, subStreamEnabled)
 
+	// MoQ is available only in the dedicated CGO/glibc build. The standard
+	// static Alpine build resolves this hook to a no-op.
+	cloud.StartLiveStreamMoQ(configuration, communication, subStreamEnabled)
+
 	// Handle livestream HD (high resolution over WEBRTC). Both the main and sub
 	// stream are exposed as separate broadcasters so a viewer can request the
 	// high (main) or low (sub) resolution per peer connection; "auto" prefers the