Merge pull request #306 from kerberos-io/fix/heartbeat-offline-tus-retry

fix(cloud): prevent cameras going offline from unbounded TUS retries

machinery/go.mod

@@ -32,6 +32,7 @@ require (
 	github.com/pion/rtp v1.8.19
 	github.com/pion/webrtc/v4 v4.1.2
 	github.com/sirupsen/logrus v1.9.3
+	github.com/stretchr/testify v1.10.0
 	github.com/swaggo/files v1.0.1
 	github.com/swaggo/gin-swagger v1.6.0
 	github.com/swaggo/swag v1.16.4
@@ -58,6 +59,7 @@ require (
 	github.com/clbanning/mxj v1.8.4 // indirect
 	github.com/clbanning/mxj/v2 v2.7.0 // indirect
 	github.com/cloudwego/base64x v0.1.5 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/elastic/go-windows v1.0.2 // indirect
 	github.com/elgs/gostrgen v0.0.0-20161222160715-9d61ae07eeae // indirect
 	github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5 // indirect
@@ -108,6 +110,7 @@ require (
 	github.com/pion/stun/v3 v3.0.0 // indirect
 	github.com/pion/transport/v3 v3.0.7 // indirect
 	github.com/pion/turn/v4 v4.0.0 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.12 // indirect

machinery/src/cloud/cloud.go

@@ -231,14 +231,20 @@ func rawJSONOrEmptyArray(b []byte) json.RawMessage {
 func HandleHeartBeat(configuration *models.Configuration, communication *models.Communication, uptimeStart time.Time) {
 	log.Log.Debug("cloud.HandleHeartBeat(): started")
 
+	// Bound every heartbeat POST so a stalled connection (e.g. a saturated uplink
+	// or an unresponsive Hub/Vault) fails fast on this cycle instead of blocking
+	// the whole heartbeat loop indefinitely. A hung POST would otherwise stop all
+	// further heartbeats, and Hub marks a camera offline once its last heartbeat is
+	// older than 180s even while capture is healthy.
+	const heartbeatHTTPTimeout = 30 * time.Second
 	var client *http.Client
 	if os.Getenv("AGENT_TLS_INSECURE") == "true" {
 		tr := &http.Transport{
 			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 		}
-		client = &http.Client{Transport: tr}
+		client = &http.Client{Transport: tr, Timeout: heartbeatHTTPTimeout}
 	} else {
-		client = &http.Client{}
+		client = &http.Client{Timeout: heartbeatHTTPTimeout}
 	}
 
 	kerberosAgentVersion := utils.VERSION

machinery/src/cloud/tus_client.go

@@ -163,6 +163,15 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 	// never consume the retry budget (matching the legacy single-POST behaviour).
 	lastStatus := 0
 
+	// highWaterOffset is the furthest server-acknowledged offset observed across
+	// all attempts (via HEAD or PATCH). It lets the retry budget be refreshed only
+	// on GENUINE net forward progress. Without it, a server that keeps resetting the
+	// offset — e.g. a persistent 409 ERR_MISMATCHED_OFFSET where HEAD reports 0 again
+	// while the first chunk still "succeeds" — would refresh the budget every attempt
+	// and loop forever, wedging the upload worker on one recording and saturating the
+	// uplink.
+	highWaterOffset := int64(0)
+
 	for attempt := 0; attempt < maxAttempts; attempt++ {
 		// (1) Ensure we have an active upload URL, creating one if needed.
 		if uploadURL == "" {
@@ -196,6 +205,15 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 			continue
 		}
 
+		// The furthest offset any previous attempt reached. If this attempt pushes
+		// past it (via HEAD showing server-side progress or a successful PATCH) we made
+		// genuine net progress and may refresh the retry budget; if not, a repeated
+		// failure at the same spot must count against maxAttempts.
+		startHighWater := highWaterOffset
+		if offset > highWaterOffset {
+			highWaterOffset = offset
+		}
+
 		// (3) All bytes are present but the upload was not finalized (e.g. the
 		// completion hook failed). A completed tus upload cannot be re-finalized
 		// with another PATCH, so delete it and re-upload to force a clean finalize.
@@ -216,7 +234,6 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 		// checkpointing the offset after each one so an interruption resumes from the
 		// last completed chunk instead of re-uploading everything.
 		chunkSize := tusChunkSize()
-		progressed := false
 		patchFailed := false
 		var lastBody string
 		loggedProgressBucket := tusProgressBucket(offset, size)
@@ -244,10 +261,10 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 				patchFailed = true
 				break
 			}
-			if newOffset > offset {
-				progressed = true
-			}
 			offset = newOffset
+			if offset > highWaterOffset {
+				highWaterOffset = offset
+			}
 			lastBody = respBody
 			logTusUploadProgress(label, offset, size, &loggedProgressBucket)
 			if offset < size {
@@ -256,10 +273,13 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 			}
 		}
 		if patchFailed {
-			if progressed {
-				// Forward progress refreshes the retry budget: maxAttempts bounds the
-				// number of consecutive failures, not the number of chunks needed for
-				// a large recording.
+			if highWaterOffset > startHighWater {
+				// Genuine net progress (we advanced past the furthest point any previous
+				// attempt reached) refreshes the retry budget: maxAttempts bounds the
+				// number of consecutive *non-progressing* failures, not the number of
+				// chunks needed for a large recording. A server that keeps rejecting the
+				// same offset (no net progress, e.g. a persistent ERR_MISMATCHED_OFFSET)
+				// therefore gives up after maxAttempts instead of retrying forever.
 				attempt = -1
 			}
 			continue

machinery/src/cloud/tus_client_test.go

@@ -49,6 +49,14 @@ type fakeTus struct {
 	// after storing the bytes, simulating a failed completion hook.
 	failFinalize int
 
+	// loseProgress simulates a vault that never durably retains the in-progress
+	// upload: every PATCH is acknowledged (the response advertises the advanced
+	// offset) but the stored offset is immediately reset to 0. HEAD therefore
+	// keeps reporting 0 and the next chunk — sent at the advanced offset — is
+	// rejected with 409, reproducing the cross-replica ERR_MISMATCHED_OFFSET
+	// loop that previously wedged the agent's upload worker forever.
+	loseProgress bool
+
 	// requests records the headers of every received request (in order) so
 	// tests can assert which auth/routing headers the client sent per method.
 	requests []recordedRequest
@@ -155,6 +163,31 @@ func (s *fakeTus) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusNotFound)
 			return
 		}
+		if s.loseProgress {
+			reqOffset, _ := strconv.ParseInt(r.Header.Get("Upload-Offset"), 10, 64)
+			s.mu.Lock()
+			cur := u.offset
+			if reqOffset != cur {
+				// The offset the client resumes from no longer matches what this
+				// "replica" retained, so reject like a vault returning
+				// ERR_MISMATCHED_OFFSET.
+				s.mu.Unlock()
+				w.Header().Set("Upload-Offset", strconv.FormatInt(cur, 10))
+				w.WriteHeader(http.StatusConflict)
+				return
+			}
+			n, _ := io.Copy(io.Discard, r.Body)
+			s.lastPatchBytes = n
+			s.patchSizes = append(s.patchSizes, n)
+			// Advertise progress to the client, then immediately forget it so the
+			// next chunk (sent at the advanced offset) mismatches again.
+			reported := cur + n
+			u.offset = 0
+			s.mu.Unlock()
+			w.Header().Set("Upload-Offset", strconv.FormatInt(reported, 10))
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
 		n, _ := io.Copy(io.Discard, r.Body)
 		s.mu.Lock()
 		u.offset += n
@@ -393,6 +426,61 @@ func TestUploadVaultResumable_NetworkErrorKeepsRetryBudget(t *testing.T) {
 	}
 }
 
+func TestUploadVaultResumable_MismatchedOffsetGivesUp(t *testing.T) {
+	// A vault that never durably retains the in-progress upload (offset resets to
+	// 0 between chunks) makes every resume "progress" by one chunk and then fail
+	// the next chunk with 409. Before the high-water gating fix this refreshed the
+	// retry budget every attempt and looped forever, wedging the upload worker and
+	// saturating the uplink (which starved heartbeats and reported the camera
+	// offline). The loop must now be bounded: give up after a fixed number of
+	// non-progressing attempts and report responded=true so the caller re-queues.
+	srv := newFakeTus()
+	srv.loseProgress = true
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	// Keep the between-attempt back-off tiny so the test stays fast.
+	oldDelay := tusBackoffBaseDelay
+	tusBackoffBaseDelay = time.Millisecond
+	defer func() { tusBackoffBaseDelay = oldDelay }()
+
+	// Force multiple chunks so there is always a second chunk to be rejected.
+	t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "4096")
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	withRecording(t, fileName, bytes.Repeat([]byte("m"), 12288))
+
+	done := make(chan struct{})
+	var uploaded, responded bool
+	var upErr error
+	go func() {
+		uploaded, responded, _, _, upErr = uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("resumable upload did not terminate: the retry loop is unbounded on a persistent mismatched offset")
+	}
+
+	if uploaded {
+		t.Fatal("expected uploaded=false when the vault never retains the offset")
+	}
+	if !responded {
+		t.Fatal("expected responded=true (the vault answered) so the caller re-queues the recording")
+	}
+	if upErr == nil {
+		t.Fatal("expected an error when the upload cannot complete")
+	}
+
+	// The bounded retry budget must cap the number of PATCH requests. Two PATCHes
+	// per attempt across a handful of attempts stays comfortably below this.
+	if count, _ := srv.patchCounts(); count > 50 {
+		t.Fatalf("expected a bounded number of PATCH requests, got %d (retry loop not bounded)", count)
+	}
+}
+
 func TestUploadVaultResumable_FinalizeRetry(t *testing.T) {
 	srv := newFakeTus()
 	srv.failFinalize = 1

machinery/src/components/kerberos.go

@@ -341,6 +341,11 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	communication.HandleONVIF = make(chan models.OnvifAction, 10)
 	go onvif.HandleONVIFActions(configuration, communication)
 
+	// Handle ONVIF event stream — opt-in via Capture.ONVIFMotion="true".
+	// Stops when the agent's shared context is cancelled. The function
+	// is a no-op if ONVIFMotion is not enabled.
+	go onvif.HandleONVIFEventStream(*communication.Context, configuration, communication)
+
 	communication.HandleAudio = make(chan models.AudioDataPartial, 10)
 	if rtspBackChannelClient.HasBackChannel {
 		communication.HasBackChannel = true

machinery/src/computervision/main.go

@@ -47,7 +47,7 @@ func ProcessMotion(motionCursor *packets.QueueCursor, configuration *models.Conf
 
 	if motionDisabled {
 
-		log.Log.Info("computervision.main.ProcessMotion(): motion detection disabled (pixelChangeThreshold set to 0), skipping.")
+		log.Log.Warning("computervision.main.ProcessMotion(): motion detection is DISABLED because pixelChangeThreshold is set to 0 or less (nil/unset would default to 150). If motion detection is expected to be running, set capture.pixelChangeThreshold to a positive value (150 recommended) or AGENT_CAPTURE_PIXEL_CHANGE, then restart/update the agent.")
 
 	} else if continuousMode && !hasMotionRegion {
 
@@ -231,13 +231,14 @@ func ProcessMotion(motionCursor *packets.QueueCursor, configuration *models.Conf
 													"mainWidth":  configuration.Config.Capture.IPCamera.Width,
 													"mainHeight": configuration.Config.Capture.IPCamera.Height,
 													"regions":    motionRectangles,
-													"polygon":    regionPolygons,												// Motion sensitivity = the pixel-change threshold that must
-												// be exceeded before motion triggers. The live view renders
-												// a reference square of sqrt(threshold) px (in this MOTION
-												// frame's pixel space) so the user can visually gauge how
-												// large a moving object must be before it is detected.
-												"pixelChangeThreshold": pixelThreshold,												},
+													"polygon":    regionPolygons, // Motion sensitivity = the pixel-change threshold that must
+													// be exceeded before motion triggers. The live view renders
+													// a reference square of sqrt(threshold) px (in this MOTION
+													// frame's pixel space) so the user can visually gauge how
+													// large a moving object must be before it is detected.
+												"pixelChangeThreshold": pixelThreshold,
 											},
+										},
 										}
 										payload, err := models.PackageMQTTMessage(configuration, message)
 										if err == nil {

machinery/src/config/main.go

@@ -377,6 +377,9 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 			case "AGENT_CAPTURE_MOTION":
 				configuration.Config.Capture.Motion = value
 				break
+			case "AGENT_CAPTURE_ONVIF_MOTION":
+				configuration.Config.Capture.ONVIFMotion = value
+				break
 			case "AGENT_CAPTURE_SNAPSHOTS":
 				configuration.Config.Capture.Snapshots = value
 				break
@@ -648,6 +651,17 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply
 		}
 	}
 
+	// Motion sensitivity: nil/unset must still resolve to the default (150), not
+	// be left nil. An explicit 0 (temporary "disable motion detection" switch
+	// from the UI) is a real, non-nil value and must NOT be touched here. Only
+	// applied for the effective configuration (applyDefaults), not for the
+	// separate global/custom views, so a missing value in one layer can still be
+	// inherited from the other instead of being masked by this default.
+	if applyDefaults && configuration.Config.Capture.PixelChangeThreshold == nil {
+		defaultPixelChangeThreshold := 150
+		configuration.Config.Capture.PixelChangeThreshold = &defaultPixelChangeThreshold
+	}
+
 	// Signing is a new feature, so if empty we set default values. Only applied
 	// for the effective configuration (applyDefaults), not for the separate
 	// global/custom views.

machinery/src/models/config.go

@@ -75,6 +75,14 @@ type Capture struct {
 	Fragmented            string      `json:"fragmented,omitempty" bson:"fragmented,omitempty"`
 	FragmentedDuration    int64       `json:"fragmentedduration,omitempty" bson:"fragmentedduration,omitempty"`
 	PixelChangeThreshold  *int        `json:"pixelChangeThreshold,omitempty"`
+	// ONVIFMotion routes the camera's ONVIF motion events into the
+	// agent's motion-triggered recording pipeline. When "true" the
+	// agent opens an event/stream against the configured ONVIF
+	// endpoint and forwards Motion+Active events to HandleMotion.
+	// Requires Capture.IPCamera.ONVIFXAddr / ONVIFUsername /
+	// ONVIFPassword to be set. Default empty (disabled) keeps the
+	// existing pixel-diff motion detection as the only source.
+	ONVIFMotion string `json:"onvif_motion,omitempty" bson:"onvif_motion,omitempty"`
 }
 
 // IPCamera configuration, such as the RTSP url of the IPCamera and the FPS.

machinery/src/onvif/events.go

@@ -0,0 +1,225 @@
+package onvif
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/log"
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/onvif/event/stream"
+)
+
+// The library handles in-stream reconnect; these guards cover the
+// initial-connect path the library cannot see.
+const (
+	initialBackoff = time.Second
+	maxBackoff     = 5 * time.Minute
+)
+
+// HandleONVIFEventStream opens an event/stream against the configured
+// ONVIF camera and routes Motion events into communication.HandleMotion.
+//
+// Behind the Capture.ONVIFMotion flag; the goroutine returns
+// immediately when not enabled. The flag is read once at start, so
+// toggling at runtime requires an agent restart. On transient
+// construction failure (camera not yet ready at boot, brief network
+// blip, credential reload) the goroutine retries with exponential
+// backoff. Exits when ctx is cancelled.
+func HandleONVIFEventStream(ctx context.Context, configuration *models.Configuration, communication *models.Communication) {
+	log.Log.Debug("onvif.HandleONVIFEventStream(): started")
+	defer log.Log.Debug("onvif.HandleONVIFEventStream(): finished")
+
+	if !isONVIFMotionEnabled(configuration.Config.Capture.ONVIFMotion) {
+		return
+	}
+	if configuration.Config.Capture.IPCamera.ONVIFXAddr == "" {
+		log.Log.Warning("onvif.HandleONVIFEventStream(): ONVIFMotion enabled but ONVIFXAddr is empty; nothing to do")
+		return
+	}
+
+	backoff := initialBackoff
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		recoverable := runStreamOnce(ctx, configuration, communication)
+		if !recoverable {
+			return
+		}
+		if !sleepCtx(ctx, backoff) {
+			return
+		}
+		backoff *= 2
+		if backoff > maxBackoff {
+			backoff = maxBackoff
+		}
+	}
+}
+
+// runStreamOnce returns true when the caller should retry construction
+// (transient failure), false on clean ctx-driven shutdown.
+func runStreamOnce(ctx context.Context, configuration *models.Configuration, communication *models.Communication) (retry bool) {
+	camera := configuration.Config.Capture.IPCamera
+
+	device, _, err := ConnectToOnvifDevice(&camera)
+	if err != nil {
+		log.Log.Error("onvif.HandleONVIFEventStream(): connect: " + err.Error())
+		return true
+	}
+
+	deviceID := resolveDeviceID(configuration.Name, camera.ONVIFXAddr)
+	s, err := stream.NewStream(ctx, device, stream.Options{DeviceID: deviceID})
+	if err != nil {
+		log.Log.Error("onvif.HandleONVIFEventStream(): open stream: " + err.Error())
+		return true
+	}
+	defer func() {
+		if err := s.Close(); err != nil {
+			log.Log.Debug("onvif.HandleONVIFEventStream(): close: " + err.Error())
+		}
+	}()
+
+	log.Log.Info("onvif.HandleONVIFEventStream(): consuming events for " + deviceID)
+
+	// recovering = the first successful event after an error streak
+	// logs a recovery line so on-call operators see the clear-of-
+	// condition for the ERROR they were paged on.
+	var recovering bool
+	for {
+		select {
+		case <-ctx.Done():
+			return false
+		case ev, ok := <-s.Events():
+			if !ok {
+				return false
+			}
+			if recovering {
+				log.Log.Info("onvif.HandleONVIFEventStream(): event stream recovered for " + deviceID)
+				recovering = false
+			}
+			dispatchEvent(ctx, ev, configuration, communication)
+		case e, ok := <-s.Errors():
+			if !ok {
+				return false
+			}
+			recovering = true
+			logStreamError(e)
+		}
+	}
+}
+
+// dispatchEvent routes motion-active events to HandleMotion.
+//
+// The ctx pre-check + ctx-in-select guards a shutdown race: the agent
+// closes HandleMotion shortly after cancelling ctx, and a stale event
+// reaching the send would otherwise panic on a closed channel.
+func dispatchEvent(ctx context.Context, ev stream.Event, configuration *models.Configuration, communication *models.Communication) {
+	topic := sanitiseTopic(ev.Topic)
+	if ev.Kind != stream.KindMotion {
+		log.Log.Debug("onvif.dispatchEvent(): non-motion event " + ev.Kind.String() + " topic=" + topic)
+		return
+	}
+	if ev.State != stream.StateActive {
+		return
+	}
+	if !isTransition(ev.Operation) {
+		log.Log.Debug("onvif.dispatchEvent(): " + ev.Operation.String() + " is not a transition, not a trigger: topic=" + topic)
+		return
+	}
+	if configuration.Config.Capture.Recording == "false" {
+		return
+	}
+	if ctx.Err() != nil {
+		return
+	}
+
+	dataToPass := models.MotionDataPartial{
+		Timestamp:       time.Now().Unix(),
+		NumberOfChanges: 0, // ONVIF does not quantify motion area.
+	}
+	select {
+	case <-ctx.Done():
+	case communication.HandleMotion <- dataToPass:
+		// Logged on the send, not before it: this line records that a
+		// recording started, so a dropped event must not leave one.
+		log.Log.Debug("onvif.dispatchEvent(): recording trigger " + ev.Kind.String() + " topic=" + topic)
+	default:
+		log.Log.Debug("onvif.dispatchEvent(): HandleMotion full, dropping ONVIF motion event")
+	}
+}
+
+// isTransition reports whether an operation represents a state change.
+// A camera replays every property's current state as Initialized on
+// each new subscription and announces removals as Deleted; neither is
+// motion starting. Absent (Unknown) counts — PropertyOperation is
+// optional per WS-Notification and many non-property events omit it.
+func isTransition(op stream.PropertyOperation) bool {
+	return op == stream.PropertyChanged || op == stream.PropertyUnknown
+}
+
+// maxLoggedTopic bounds a topic in the log; the wire imposes no limit,
+// and the reject path logs every event received.
+const maxLoggedTopic = 256
+
+// sanitiseTopic makes a camera-controlled topic safe to concatenate
+// into a log line. logrus's coloured text formatter writes the message
+// unquoted, so a raw newline would let a camera forge entries in the
+// log being used to diagnose it.
+func sanitiseTopic(topic string) string {
+	if len(topic) > maxLoggedTopic {
+		topic = topic[:maxLoggedTopic] + "…(truncated)"
+	}
+	quoted := strconv.Quote(topic)
+	return quoted[1 : len(quoted)-1]
+}
+
+// logStreamError logs at a level matching severity: recreate is loud
+// because it usually means the camera is offline; pull and renew are
+// debug because the library recovers from them automatically.
+func logStreamError(e error) {
+	var recreate stream.ErrRecreateFailed
+	var pull stream.ErrPullFailed
+	var renew stream.ErrRenewFailed
+	switch {
+	case errors.As(e, &recreate):
+		log.Log.Error("onvif.HandleONVIFEventStream(): subscription recreate failed (camera may be offline): " + recreate.Err.Error())
+	case errors.As(e, &renew):
+		log.Log.Debug("onvif.HandleONVIFEventStream(): renew failed (will recover via pull/recreate): " + renew.Err.Error())
+	case errors.As(e, &pull):
+		log.Log.Debug("onvif.HandleONVIFEventStream(): pull failed (will retry): " + pull.Err.Error())
+	default:
+		log.Log.Info("onvif.HandleONVIFEventStream(): stream error: " + e.Error())
+	}
+}
+
+func isONVIFMotionEnabled(v string) bool {
+	return strings.EqualFold(strings.TrimSpace(v), "true")
+}
+
+// resolveDeviceID falls back from operator-supplied name to ONVIF
+// endpoint to a constant placeholder so log lines always have
+// something to grep.
+func resolveDeviceID(configName, xaddr string) string {
+	if n := strings.TrimSpace(configName); n != "" {
+		return n
+	}
+	if x := strings.TrimSpace(xaddr); x != "" {
+		return x
+	}
+	return "unknown"
+}
+
+// sleepCtx returns false if ctx was cancelled, true if d elapsed.
+func sleepCtx(ctx context.Context, d time.Duration) bool {
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-t.C:
+		return true
+	}
+}

machinery/src/onvif/events_test.go

@@ -0,0 +1,375 @@
+package onvif
+
+import (
+	"bytes"
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/onvif/event/stream"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func makeConfig(recording, onvifMotion, name string) *models.Configuration {
+	return &models.Configuration{
+		Name: name,
+		Config: models.Config{
+			Capture: models.Capture{
+				Recording:   recording,
+				ONVIFMotion: onvifMotion,
+			},
+		},
+	}
+}
+
+func makeCommunication(buffer int) *models.Communication {
+	return &models.Communication{
+		HandleMotion: make(chan models.MotionDataPartial, buffer),
+	}
+}
+
+// --- dispatchEvent ---------------------------------------------------
+
+func TestDispatchEvent_MotionActive_SendsToHandleMotion(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case m := <-comm.HandleMotion:
+		assert.NotZero(t, m.Timestamp)
+	case <-time.After(time.Second):
+		t.Fatal("expected motion data on HandleMotion")
+	}
+}
+
+func TestDispatchEvent_MotionInactive_DoesNotSend(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateInactive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case <-comm.HandleMotion:
+		t.Fatal("inactive motion must not reach HandleMotion (motion-stop is a follow-up)")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestDispatchEvent_NonMotionKindIgnored(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindDigitalInput, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case <-comm.HandleMotion:
+		t.Fatal("non-motion kinds must not reach HandleMotion")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+// captureDebugLog redirects logrus to a buffer at debug level for the
+// duration of a test and returns what was written. It mutates package
+// globals, so callers must not run in parallel.
+func captureDebugLog(t *testing.T) *bytes.Buffer {
+	t.Helper()
+	var buf bytes.Buffer
+	prevOut, prevLevel := logrus.StandardLogger().Out, logrus.GetLevel()
+	logrus.SetOutput(&buf)
+	logrus.SetLevel(logrus.DebugLevel)
+	t.Cleanup(func() {
+		logrus.SetOutput(prevOut)
+		logrus.SetLevel(prevLevel)
+	})
+	return &buf
+}
+
+// TestDispatchEvent_LogsTheTriggeringTopic — a dispatched event is what
+// actually starts a recording, so its topic is the one an operator needs
+// when a camera records for the wrong reason (or the right reason and
+// nobody can prove which). Rejected events were already logged; without
+// this the triggering topic is only knowable by elimination.
+func TestDispatchEvent_LogsTheTriggeringTopic(t *testing.T) {
+	buf := captureDebugLog(t)
+
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{
+		Kind:  stream.KindMotion,
+		State: stream.StateActive,
+		Topic: "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1",
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	assert.Contains(t, buf.String(), "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1",
+		"the dispatched event's topic must appear in the log")
+	assert.Contains(t, buf.String(), "Motion",
+		"the dispatched event's Kind must appear in the log")
+}
+
+// TestDispatchEvent_PropertyOperation — a camera replays the current
+// state of every property topic as Initialized whenever a pull-point
+// subscription is created. If that counts as a trigger, every
+// reconnect restarts a recording for any motion property that happens
+// to be active, and a flapping subscription manufactures motion out of
+// nothing. Only reject Initialized specifically: PropertyOperation is
+// optional per WS-Notification and absent on many non-property events,
+// which decode reports as PropertyUnknown.
+func TestDispatchEvent_PropertyOperation(t *testing.T) {
+	tests := []struct {
+		name     string
+		op       stream.PropertyOperation
+		wantSend bool
+	}{
+		{"changed is a real transition", stream.PropertyChanged, true},
+		{"absent attribute still counts", stream.PropertyUnknown, true},
+		{"initialized is a subscription state replay", stream.PropertyInitialized, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := makeConfig("true", "true", "cam-1")
+			comm := makeCommunication(1)
+			ev := stream.Event{
+				Kind:      stream.KindMotion,
+				State:     stream.StateActive,
+				Operation: tt.op,
+				Topic:     "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1",
+			}
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			dispatchEvent(ctx, ev, cfg, comm)
+
+			select {
+			case <-comm.HandleMotion:
+				if !tt.wantSend {
+					t.Fatalf("%v must not trigger a recording", tt.op)
+				}
+			case <-time.After(100 * time.Millisecond):
+				if tt.wantSend {
+					t.Fatalf("%v must trigger a recording", tt.op)
+				}
+			}
+		})
+	}
+}
+
+func TestDispatchEvent_RecordingDisabled_DoesNotSend(t *testing.T) {
+	cfg := makeConfig("false", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case <-comm.HandleMotion:
+		t.Fatal("Recording=false must gate the send (matches computervision behaviour)")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestDispatchEvent_HandleMotionFull_DropsRatherThanBlocks(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	// Pre-fill the buffer so the next send would block.
+	comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)}
+	comm.HandleMotion <- models.MotionDataPartial{}
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	done := make(chan struct{})
+	go func() {
+		dispatchEvent(ctx, ev, cfg, comm)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("dispatchEvent must drop when HandleMotion is full, not block")
+	}
+}
+
+func TestDispatchEvent_CtxCancelledAndHandleMotionClosed_DoesNotPanic(t *testing.T) {
+	// Regression for the shutdown race: between cancel() and
+	// close(HandleMotion) the agent leaves a 3s window. If dispatchEvent
+	// runs in that window AFTER the channel is closed, a non-protected
+	// send would panic. The ctx pre-check must short-circuit before the
+	// send is attempted.
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)}
+	close(comm.HandleMotion)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // already cancelled, matching the shutdown sequence
+
+	assert.NotPanics(t, func() {
+		dispatchEvent(ctx, ev, cfg, comm)
+	})
+}
+
+// --- isONVIFMotionEnabled --------------------------------------------
+
+func TestIsONVIFMotionEnabled_CaseAndWhitespace(t *testing.T) {
+	tests := []struct {
+		in   string
+		want bool
+	}{
+		{"true", true},
+		{"True", true},
+		{"TRUE", true},
+		{" true", true},
+		{"true ", true},
+		{"  true  ", true},
+		{"false", false},
+		{"False", false},
+		{"", false},
+		{"yes", false},
+		{"1", false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.in, func(t *testing.T) {
+			assert.Equal(t, tc.want, isONVIFMotionEnabled(tc.in))
+		})
+	}
+}
+
+// --- resolveDeviceID -------------------------------------------------
+
+func TestResolveDeviceID_FallbackChain(t *testing.T) {
+	tests := []struct {
+		name    string
+		cfgName string
+		xaddr   string
+		want    string
+	}{
+		{"name_set", "front-door", "192.168.1.10", "front-door"},
+		{"name_empty_xaddr_set", "", "192.168.1.10", "192.168.1.10"},
+		{"name_whitespace_only_xaddr_set", "  ", "192.168.1.10", "192.168.1.10"},
+		{"both_empty", "", "", "unknown"},
+		{"name_with_trailing_whitespace", "cam-2  ", "192.168.1.10", "cam-2"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, resolveDeviceID(tc.cfgName, tc.xaddr))
+		})
+	}
+}
+
+// TestDispatchEvent_OnlyRealTransitionsTrigger — a camera replays every
+// property topic's state on each new subscription (Initialized) and
+// announces removals (Deleted). Neither is a motion transition, and a
+// flapping pull-point would otherwise manufacture recordings out of
+// replayed state. PropertyOperation is optional per WS-Notification, so
+// absent (Unknown) still counts — many non-property events omit it.
+func TestDispatchEvent_OnlyRealTransitionsTrigger(t *testing.T) {
+	tests := []struct {
+		op       stream.PropertyOperation
+		wantSend bool
+	}{
+		{stream.PropertyChanged, true},
+		{stream.PropertyUnknown, true},
+		{stream.PropertyInitialized, false},
+		{stream.PropertyDeleted, false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.op.String(), func(t *testing.T) {
+			cfg := makeConfig("true", "true", "cam-1")
+			comm := makeCommunication(1)
+			ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive, Operation: tt.op}
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			dispatchEvent(ctx, ev, cfg, comm)
+
+			if tt.wantSend {
+				require.Len(t, comm.HandleMotion, 1, "%v must trigger a recording", tt.op)
+				return
+			}
+			require.Empty(t, comm.HandleMotion, "%v must not trigger a recording", tt.op)
+		})
+	}
+}
+
+// TestSanitiseTopic — ev.Topic is camera-controlled and reaches the log
+// unmodified. logrus's coloured text formatter (the default) writes the
+// message without quoting, so an embedded newline forges whole log
+// lines: a compromised camera can fabricate ERROR entries or spoof
+// another device's id, in the logs an operator is reading to diagnose
+// that very camera. Length is also unbounded on the wire, and the
+// reject path logs every event, so an oversized topic is a cheap way to
+// evict a container's whole retained history.
+func TestSanitiseTopic(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want string
+	}{
+		{"ordinary topic passes through", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1"},
+		{"newline cannot forge a line", "a\nERRO[fake] boom", `a\nERRO[fake] boom`},
+		{"carriage return", "a\rb", `a\rb`},
+		{"tab", "a\tb", `a\tb`},
+		{"NUL", "a\x00b", `a\x00b`},
+		{"empty", "", ""},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := sanitiseTopic(tt.in)
+			assert.Equal(t, tt.want, got)
+			assert.NotContains(t, got, "\n", "no raw newline may survive")
+			assert.NotContains(t, got, "\r", "no raw carriage return may survive")
+		})
+	}
+}
+
+func TestSanitiseTopic_Truncates(t *testing.T) {
+	got := sanitiseTopic(strings.Repeat("x", maxLoggedTopic*2))
+	assert.LessOrEqual(t, len(got), maxLoggedTopic+len("…(truncated)"))
+	assert.Contains(t, got, "truncated")
+}
+
+// TestDispatchEvent_LogsTriggerOnlyWhenSent — the trigger line is the
+// record that a recording started. Logging it before the send means a
+// dropped event (full channel, or shutdown) leaves a line claiming a
+// recording that never began.
+func TestDispatchEvent_LogsTriggerOnlyWhenSent(t *testing.T) {
+	buf := captureDebugLog(t)
+
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)}
+	comm.HandleMotion <- models.MotionDataPartial{} // full
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive, Topic: "tns1:VideoSource/MotionAlarm"}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	assert.NotContains(t, buf.String(), "recording trigger",
+		"a dropped event must not be logged as a trigger")
+	assert.Contains(t, buf.String(), "dropping", "the drop itself must still be logged")
+}