Merge pull request #299 from kerberos-io/feature/remote-recording

feature/remote-recording

machinery/data/config/config.json

@@ -7,7 +7,7 @@
 	"offline": "false",
 	"auto_clean": "true",
 	"remove_after_upload": "true",
-	"max_directory_size": 100,
+	"max_directory_size": 0,
 	"timezone": "Africa/Ceuta",
 	"capture": {
 		"name": "",

machinery/go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/kellydunn/golang-geo v0.7.0
 	github.com/kerberos-io/joy4 v1.0.64
-	github.com/kerberos-io/onvif v1.0.0
+	github.com/kerberos-io/onvif v1.2.1
 	github.com/minio/minio-go/v6 v6.0.57
 	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
@@ -76,7 +76,6 @@ require (
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 // indirect
-	github.com/icholy/digest v0.1.23 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/juju/errors v1.0.0 // indirect

machinery/go.sum

@@ -755,8 +755,6 @@ github.com/iancoleman/strcase v0.3.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20220319035150-800ac71e25c2/go.mod h1:aYm2/VgdVmcIU8iMfdMvDMsRAQjcfZSKFby6HOFvi/w=
-github.com/icholy/digest v0.1.23 h1:4hX2pIloP0aDx7RJW0JewhPPy3R8kU+vWKdxPsCCGtY=
-github.com/icholy/digest v0.1.23/go.mod h1:QNrsSGQ5v7v9cReDI0+eyjsXGUoRSUZQHeQ5C4XLa0Y=
 github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
@@ -776,8 +774,8 @@ github.com/kellydunn/golang-geo v0.7.0 h1:A5j0/BvNgGwY6Yb6inXQxzYwlPHc6WVZR+Mrar
 github.com/kellydunn/golang-geo v0.7.0/go.mod h1:YYlQPJ+DPEzrHx8kT3oPHC/NjyvCCXE+IuKGKdrjrcU=
 github.com/kerberos-io/joy4 v1.0.64 h1:gTUSotHSOhp9mNqEecgq88tQHvpj7TjmrvPUsPm0idg=
 github.com/kerberos-io/joy4 v1.0.64/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU=
-github.com/kerberos-io/onvif v1.0.0 h1:pLJrK6skPkK+5Bj4XfqHUkQ2I+p5pwELnp+kQTJWXiQ=
-github.com/kerberos-io/onvif v1.0.0/go.mod h1:P1kUcCfeotJSlL1jwGseH6NSnCwWiuJLl3gAzafnLbA=
+github.com/kerberos-io/onvif v1.2.1 h1:+vxyHPylt0ufK8gv7FL+KzhJUeComMGrTmP5KxT2YEc=
+github.com/kerberos-io/onvif v1.2.1/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j0HLHbNSE=
@@ -2109,7 +2107,6 @@ gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C
 gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gotest.tools/v3 v3.5.1 h1:EENdUnS3pdur5nybKYIh2Vfgc8IUNBjxDPSjtiJcOzU=
 gotest.tools/v3 v3.5.1/go.mod h1:isy3WKz7GK6uNw/sbHzfKBLvlvXwUyV06n6brMxxopU=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

machinery/src/capture/main.go

@@ -9,6 +9,7 @@ import (
 	"strconv"
 	"time"
 
+	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/gin-gonic/gin"
 	"github.com/kerberos-io/agent/machinery/src/conditions"
 	"github.com/kerberos-io/agent/machinery/src/encryption"
@@ -20,6 +21,35 @@ import (
 	"go.opentelemetry.io/otel/trace"
 )
 
+// publishRecordingState notifies the hub (and ultimately the live-view UI) that
+// this camera started ("recording": true) or stopped ("recording": false)
+// recording, so the frontend can show a "recording" indicator while the agent
+// is recording (e.g. a motion clip triggered manually from the live view or by
+// motion detection). It is a best-effort broadcast: when no hub/MQTT is
+// configured (or the agent is offline) it is a no-op, and a missed message is
+// self-healed by the frontend's safety timeout.
+func publishRecordingState(mqttClient mqtt.Client, hubKey string, configuration *models.Configuration, recording bool) {
+	if mqttClient == nil || hubKey == "" || configuration.Config.Offline == "true" {
+		return
+	}
+	message := models.Message{
+		Payload: models.Payload{
+			Action:   "recording",
+			DeviceId: configuration.Config.Key,
+			Value: map[string]interface{}{
+				"timestamp": time.Now().Unix(),
+				"recording": recording,
+			},
+		},
+	}
+	payload, err := models.PackageMQTTMessage(configuration, message)
+	if err == nil {
+		mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
+	} else {
+		log.Log.Error("capture.main.publishRecordingState(): failed to package MQTT message: " + err.Error())
+	}
+}
+
 func CleanupRecordingDirectory(configDirectory string, configuration *models.Configuration) {
 	autoClean := configuration.Config.AutoClean
 	if autoClean != "true" {
@@ -188,11 +218,16 @@ func pickRecordingToCleanup(recordingsDirectory, cloudDirectory string) (string,
 	return "", false, os.ErrNotExist
 }
 
-func HandleRecordStream(queue *packets.Queue, configDirectory string, configuration *models.Configuration, communication *models.Communication, rtspClient RTSPClient) {
+func HandleRecordStream(queue *packets.Queue, configDirectory string, configuration *models.Configuration, communication *models.Communication, rtspClient RTSPClient, mqttClient mqtt.Client) {
 
 	config := configuration.Config
+	hubKey := config.HubKey
 	loc, _ := time.LoadLocation(config.Timezone)
 
+	// Start each capture session with manual recording off, so a leftover
+	// request from before a restart/reconnect doesn't silently persist.
+	communication.IsRecordingManual.UnSet()
+
 	if config.Capture.Recording == "false" {
 		log.Log.Info("capture.main.HandleRecordStream(): disabled, we will not record anything.")
 	} else {
@@ -357,6 +392,9 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 
 					recordingStatus = "idle"
 
+					// Notify the hub / live-view UI that this camera stopped recording.
+					publishRecordingState(mqttClient, hubKey, configuration, false)
+
 					// Clean up the recording directory if necessary.
 					CleanupRecordingDirectory(configDirectory, configuration)
 				}
@@ -433,6 +471,9 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
 					recordingStatus = "started"
 
+					// Notify the hub / live-view UI that this camera started recording.
+					publishRecordingState(mqttClient, hubKey, configuration, true)
+
 				} else if start {
 
 					writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
@@ -510,6 +551,9 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 
 					recordingStatus = "idle"
 
+					// Notify the hub / live-view UI that this camera stopped recording.
+					publishRecordingState(mqttClient, hubKey, configuration, false)
+
 					// Clean up the recording directory if necessary.
 					CleanupRecordingDirectory(configDirectory, configuration)
 				}
@@ -626,6 +670,14 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					default:
 					}
 
+					// While a manual recording is active, keep it alive: refresh the
+					// motion timestamp every iteration so the post-recording timeout
+					// never fires. The clip still rolls over at maxRecordingPeriod and
+					// is restarted below, until the viewer stops the manual recording.
+					if communication.IsRecordingManual.IsSet() {
+						motionTimestamp = now
+					}
+
 					if start && (motionTimestamp+postRecording-now < 0 || now-startRecording > maxRecordingPeriod-500) && nextPkt.IsKeyFrame {
 						log.Log.Info("capture.main.HandleRecordStream(motiondetection): timestamp+postRecording-now < 0  - " + strconv.FormatInt(motionTimestamp+postRecording-now, 10) + " < 0")
 						log.Log.Info("capture.main.HandleRecordStream(motiondetection): now-startRecording > maxRecordingPeriod-500 - " + strconv.FormatInt(now-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-500, 10))
@@ -657,6 +709,9 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 							log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
 						}
 						start = true
+
+						// Notify the hub / live-view UI that this camera started recording.
+						publishRecordingState(mqttClient, hubKey, configuration, true)
 					}
 					if start {
 						writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
@@ -691,6 +746,19 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 				mp4Video.Close(&config)
 				log.Log.Info("capture.main.HandleRecordStream(motiondetection): file save: " + name)
 
+				// Notify the hub / live-view UI that this camera stopped recording.
+				publishRecordingState(mqttClient, hubKey, configuration, false)
+
+				// If the viewer still has a manual recording running, this clip just
+				// rolled over at the max length — immediately kick off the next
+				// segment so recording stays continuous until they stop it.
+				if communication.IsRecordingManual.IsSet() {
+					select {
+					case communication.HandleMotion <- models.MotionDataPartial{Timestamp: time.Now().Unix(), NumberOfChanges: 100000000}:
+					default:
+					}
+				}
+
 				// Update the name of the recording with the duration.
 				// We will update the name of the recording with the duration in milliseconds.
 				if mp4Video.VideoTotalDuration > 0 {

machinery/src/components/kerberos.go

@@ -75,6 +75,7 @@ func Bootstrap(ctx context.Context, configDirectory string, configuration *model
 	communication.HandleLiveHDPeers = make(chan string, 1)
 	communication.HandleLiveHLS = make(chan string, 1)
 	communication.IsConfiguring = abool.New()
+	communication.IsRecordingManual = abool.New()
 
 	cameraSettings := &models.Camera{}
 
@@ -309,7 +310,7 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	go cloud.HandleLiveStreamHD(configuration, communication, mqttClient, rtspClient, rtspSubClient, subStreamEnabled)
 
 	// Handle recording, will write an mp4 to disk.
-	go capture.HandleRecordStream(queue, configDirectory, configuration, communication, rtspClient)
+	go capture.HandleRecordStream(queue, configDirectory, configuration, communication, rtspClient, mqttClient)
 
 	// Handle processing of motion
 	communication.HandleMotion = make(chan models.MotionDataPartial, 10)

machinery/src/models/communication.go

@@ -44,9 +44,15 @@ type Communication struct {
 	// HandleLiveHLS is the live HLS viewer keepalive. It carries the requested
 	// quality tier ("auto"|"high"|"low"; empty => auto) so the producer can switch
 	// the live session between the main and sub stream on demand.
-	HandleLiveHLS       chan string
-	HandleONVIF         chan OnvifAction
-	IsConfiguring       *abool.AtomicBool
+	HandleLiveHLS chan string
+	HandleONVIF   chan OnvifAction
+	IsConfiguring *abool.AtomicBool
+	// IsRecordingManual is set while a viewer has requested a manual recording
+	// from the live view (the record button). While set, the motion-based
+	// recorder keeps recording (it does not auto-close on the post-recording
+	// timeout) until the viewer stops it again. It is independent of motion
+	// detection so it also works when nothing is moving.
+	IsRecordingManual   *abool.AtomicBool
 	Queue               *packets.Queue
 	SubQueue            *packets.Queue
 	Image               string

machinery/src/models/mqtt.go

@@ -150,6 +150,10 @@ type AudioPayload struct {
 // We received a recording request, we'll send it to the motion handler.
 type RecordPayload struct {
 	Timestamp int64 `json:"timestamp"` // timestamp of the recording request.
+	// Recording toggles a manual recording from the live view: true starts a
+	// recording (and keeps it running), false stops it. Older clients that only
+	// send a timestamp default to false; the live view always sets it explicitly.
+	Recording bool `json:"recording"`
 }
 
 // We received a preset position request, we'll request it through onvif and send it back.

machinery/src/routers/mqtt/main.go

@@ -375,11 +375,29 @@ func HandleRecording(mqttClient mqtt.Client, hubKey string, payload models.Paylo
 	var recordPayload models.RecordPayload
 	json.Unmarshal(jsonData, &recordPayload)
 
-	if recordPayload.Timestamp != 0 {
-		motionDataPartial := models.MotionDataPartial{
-			Timestamp: recordPayload.Timestamp,
+	timestamp := recordPayload.Timestamp
+	if timestamp == 0 {
+		timestamp = time.Now().Unix()
+	}
+
+	if recordPayload.Recording {
+		// Start a manual recording from the live view (record button). Keep it
+		// running until the viewer stops it again — the motion recorder honours
+		// communication.IsRecordingManual and won't auto-close on the
+		// post-recording timeout while it's set. We also inject a motion event
+		// so the recording starts immediately, even when nothing is moving.
+		log.Log.Info("routers.mqtt.main.HandleRecording(): manual recording started.")
+		communication.IsRecordingManual.Set()
+		select {
+		case communication.HandleMotion <- models.MotionDataPartial{Timestamp: timestamp, NumberOfChanges: 100000000}:
+		default:
+			log.Log.Warning("routers.mqtt.main.HandleRecording(): motion channel full, manual recording start not queued.")
 		}
-		communication.HandleMotion <- motionDataPartial
+	} else {
+		// Stop the manual recording; the motion recorder closes the clip once the
+		// post-recording window elapses.
+		log.Log.Info("routers.mqtt.main.HandleRecording(): manual recording stopped.")
+		communication.IsRecordingManual.UnSet()
 	}
 }