Adjust MinNormalGOPMs threshold to prevent false positives on loop seams

Add LastKeyframeGapMs to MP4 state and update fragment-flush logic to consider the previous keyframe gap before forcing a fragment boundary. Previously any unexpectedly short keyframe gap (< MinNormalGOPMs) would force a flush, which could cascade on streams that legitimately emit short GOPs. Now we only force a flush when the current gap is short and the prior gap was healthy (or unset), and we record the current gap for future checks. Also refactor the check to use a local gap variable and preserve the existing log message.

Dockerfile

@@ -95,7 +95,7 @@ RUN addgroup -S kerberosio && adduser -S agent -G kerberosio && addgroup agent v
 COPY --chown=0:0 --from=build-machinery /dist /
 COPY --chown=0:0 --from=build-ui /dist /
 
-RUN apk update && apk add ca-certificates curl libstdc++ libc6-compat --no-cache && rm -rf /var/cache/apk/*
+RUN apk update && apk add ca-certificates curl ffmpeg libstdc++ libc6-compat --no-cache && rm -rf /var/cache/apk/*
 
 ##################
 # Try running agent

Dockerfile.arm64

@@ -95,7 +95,7 @@ RUN addgroup -S kerberosio && adduser -S agent -G kerberosio && addgroup agent v
 COPY --chown=0:0 --from=build-machinery /dist /
 COPY --chown=0:0 --from=build-ui /dist /
 
-RUN apk update && apk add ca-certificates curl libstdc++ libc6-compat --no-cache && rm -rf /var/cache/apk/*
+RUN apk update && apk add ca-certificates curl ffmpeg libstdc++ libc6-compat --no-cache && rm -rf /var/cache/apk/*
 
 ##################
 # Try running agent

README.md

@@ -65,6 +65,7 @@ There are a myriad of cameras out there (USB, IP and other cameras), and it migh
 
 ### Contributing
 
+1. [Security vulnerability reporting](#security-vulnerability-reporting)
 1. [Contribute with Codespaces](#contribute-with-codespaces)
 2. [Develop and build](#develop-and-build)
 3. [Building from source](#building-from-source)
@@ -301,6 +302,10 @@ If we talk about video encoders and decoders (codecs) there are 2 major video co
 
 Conclusion: depending on the use case you might choose one over the other, and you can use both at the same time. For example you can use H264 (main stream) for livestreaming, and H265 (sub stream) for recording. If you wish to play recordings in a cross-platform and cross-browser environment, you might opt for H264 for better support.
 
+## Security vulnerability reporting
+
+If you found a potential security vulnerability, please use the private channels described in [SECURITY.md](SECURITY.md). Avoid opening public GitHub issues for sensitive findings.
+
 ## Contribute with Codespaces
 
 One of the major blockers for letting you contribute to an Open Source project is to set up your local development machine. Why? Because you might already have some tools and libraries installed that are used for other projects, and the libraries you would need for Kerberos Agent, for example FFmpeg, might require a different version. Welcome to dependency hell...

SECURITY.md

@@ -0,0 +1,40 @@
+# Security Policy
+
+## Supported Versions
+
+We only provide security fixes for the latest release series on the `master` branch.
+
+## Reporting a Vulnerability
+
+Please do **not** open a public GitHub issue for potential security vulnerabilities.
+
+Use one of the private channels below:
+
+1. Preferred: GitHub private vulnerability reporting
+   - https://github.com/kerberos-io/agent/security/advisories/new
+2. Fallback: Email
+   - support@kerberos.io
+   - Optional CC: support@uug.ai
+
+Please include:
+
+- A short summary and impact.
+- Reproduction steps or proof of concept.
+- Affected version(s), commit hash, or deployment details.
+- Any proposed mitigation/workaround.
+- Your preferred attribution name.
+
+For faster triage, use this subject format in email:
+
+`[Security][Kerberos Agent] <short title>`
+
+## Response Expectations
+
+- Acknowledgement target: within 3 business days.
+- Triage/update target: within 7 business days after acknowledgement.
+
+If you do not receive a response in time, please resend your report and include your original timestamp.
+
+## Disclosure and Credits
+
+We follow coordinated disclosure. After a fix is available, we will credit reporters unless they prefer to stay anonymous.

machinery/README.md

@@ -22,4 +22,8 @@ https://brianmacdonald.github.io/Ethonate/address#0xf4a759C9436E2280Ea9cdd23d314
 
 [**Docker Hub**](https://hub.docker.com/r/kerberos/agent) | [**Documentation**](https://doc.kerberos.io) | [**Website**](https://kerberos.io)
 
-Kerberos Open source (v3) is a cutting edge video surveillance management system made available as Open Source under the MIT License. This means that all the source code is available for you or your company, and you can use, transform and distribute the source code; as long you keep a reference of the original license. Kerberos Open Source (v3) can be used for commercial usage (which was not the case for v2). Read more [about the license here](LICENSE).
+Kerberos Open source (v3) is a cutting edge video surveillance management system made available as Open Source under the MIT License. This means that all the source code is available for you or your company, and you can use, transform and distribute the source code; as long you keep a reference of the original license. Kerberos Open Source (v3) can be used for commercial usage (which was not the case for v2). Read more [about the license here](LICENSE).
+
+## Security reporting
+
+For sensitive vulnerabilities, use private disclosure channels documented in [../SECURITY.md](../SECURITY.md).

machinery/src/components/Kerberos.go

@@ -21,6 +21,7 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/packets"
 	routers "github.com/kerberos-io/agent/machinery/src/routers/mqtt"
 	"github.com/kerberos-io/agent/machinery/src/utils"
+	"github.com/kerberos-io/agent/machinery/src/webrtc"
 	"github.com/tevino/abool"
 )
 
@@ -303,7 +304,7 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	}
 
 	// Handle livestream HD (high resolution over WEBRTC)
-	communication.HandleLiveHDHandshake = make(chan models.RequestHDStreamPayload, 10)
+	communication.HandleLiveHDHandshake = make(chan models.LiveHDHandshake, 100)
 	if subStreamEnabled {
 		livestreamHDCursor := subQueue.Latest()
 		go cloud.HandleLiveStreamHD(livestreamHDCursor, configuration, communication, mqttClient, rtspSubClient)
@@ -552,6 +553,11 @@ func GetDashboard(c *gin.Context, configDirectory string, configuration *models.
 	// The total number of recordings stored in the directory.
 	recordingDirectory := configDirectory + "/data/recordings"
 	numberOfRecordings := utils.NumberOfMP4sInDirectory(recordingDirectory)
+	activeWebRTCReaders := webrtc.GetActivePeerConnectionCount()
+	pendingWebRTCHandshakes := 0
+	if communication.HandleLiveHDHandshake != nil {
+		pendingWebRTCHandshakes = len(communication.HandleLiveHDHandshake)
+	}
 
 	// All days stored in this agent.
 	days := []string{}
@@ -574,6 +580,8 @@ func GetDashboard(c *gin.Context, configDirectory string, configuration *models.
 		"cameraOnline":       cameraIsOnline,
 		"cloudOnline":        cloudIsOnline,
 		"numberOfRecordings": numberOfRecordings,
+		"webrtcReaders":      activeWebRTCReaders,
+		"webrtcPending":      pendingWebRTCHandshakes,
 		"days":               days,
 		"latestEvents":       latestEvents,
 	})

machinery/src/models/Communication.go

@@ -8,6 +8,17 @@ import (
 	"github.com/tevino/abool"
 )
 
+type LiveHDSignalingCallbacks struct {
+	SendAnswer    func(sessionID string, sdp string) error
+	SendCandidate func(sessionID string, candidate string) error
+	SendError     func(sessionID string, message string) error
+}
+
+type LiveHDHandshake struct {
+	Payload   RequestHDStreamPayload
+	Signaling *LiveHDSignalingCallbacks
+}
+
 // The communication struct that is managing
 // all the communication between the different goroutines.
 type Communication struct {
@@ -27,7 +38,7 @@ type Communication struct {
 	HandleHeartBeat       chan string
 	HandleLiveSD          chan int64
 	HandleLiveHDKeepalive chan string
-	HandleLiveHDHandshake chan RequestHDStreamPayload
+	HandleLiveHDHandshake chan LiveHDHandshake
 	HandleLiveHDPeers     chan string
 	HandleONVIF           chan OnvifAction
 	IsConfiguring         *abool.AtomicBool

machinery/src/routers/mqtt/main.go

@@ -90,9 +90,10 @@ func ConfigureMQTT(configDirectory string, configuration *models.Configuration,
 
 		// Some extra options to make sure the connection behaves
 		// properly. More information here: github.com/eclipse/paho.mqtt.golang.
-		opts.SetCleanSession(true)
-		//opts.SetResumeSubs(true)
-		//opts.SetStore(mqtt.NewMemoryStore())
+		//opts.SetCleanSession(true)
+		opts.SetCleanSession(false)
+		opts.SetResumeSubs(true)
+		opts.SetStore(mqtt.NewMemoryStore())
 		opts.SetConnectRetry(true)
 		opts.SetAutoReconnect(true)
 		opts.SetConnectRetryInterval(5 * time.Second)
@@ -537,9 +538,13 @@ func HandleRequestHDStream(mqttClient mqtt.Client, hubKey string, payload models
 		if communication.CameraConnected {
 			// Set the Hub key, so we can send back the answer.
 			requestHDStreamPayload.HubKey = hubKey
-			select {
-			case communication.HandleLiveHDHandshake <- requestHDStreamPayload:
-			default:
+			if communication.HandleLiveHDHandshake == nil {
+				log.Log.Error("routers.mqtt.main.HandleRequestHDStream(): handshake channel is nil, dropping request")
+				return
+			}
+
+			communication.HandleLiveHDHandshake <- models.LiveHDHandshake{
+				Payload: requestHDStreamPayload,
 			}
 			log.Log.Info("routers.mqtt.main.HandleRequestHDStream(): received request to setup webrtc.")
 		} else {

machinery/src/routers/websocket/main.go

@@ -6,6 +6,7 @@ import (
 	"image"
 	"net/http"
 	"sync"
+	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/gorilla/websocket"
@@ -14,6 +15,7 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/models"
 	"github.com/kerberos-io/agent/machinery/src/packets"
 	"github.com/kerberos-io/agent/machinery/src/utils"
+	"github.com/kerberos-io/agent/machinery/src/webrtc"
 )
 
 type Message struct {
@@ -28,6 +30,23 @@ type Connection struct {
 	Cancels map[string]context.CancelFunc
 }
 
+func writeWebRTCError(connection *Connection, clientID string, sessionID string, errorMessage string) {
+	if connection == nil {
+		return
+	}
+
+	if err := connection.WriteJson(Message{
+		ClientID:    clientID,
+		MessageType: "webrtc-error",
+		Message: map[string]string{
+			"session_id": sessionID,
+			"message":    errorMessage,
+		},
+	}); err != nil {
+		log.Log.Error("routers.websocket.main.writeWebRTCError(): " + err.Error())
+	}
+}
+
 // Concurrency handling - sending messages
 func (c *Connection) WriteJson(message Message) error {
 	c.mu.Lock()
@@ -115,6 +134,82 @@ func WebsocketHandler(c *gin.Context, configuration *models.Configuration, commu
 						go ForwardSDStream(ctx, clientID, sockets[clientID], configuration, communication, captureDevice)
 					}
 				}
+
+			case "stream-hd":
+				sessionID := message.Message["session_id"]
+				sessionDescription := message.Message["sdp"]
+
+				if sessionID == "" || sessionDescription == "" {
+					writeWebRTCError(sockets[clientID], clientID, sessionID, "missing session_id or sdp")
+					break
+				}
+
+				if !communication.CameraConnected {
+					writeWebRTCError(sockets[clientID], clientID, sessionID, "camera is not connected")
+					break
+				}
+
+				if communication.HandleLiveHDHandshake == nil {
+					writeWebRTCError(sockets[clientID], clientID, sessionID, "webrtc liveview is not available")
+					break
+				}
+
+				handshake := models.LiveHDHandshake{
+					Payload: models.RequestHDStreamPayload{
+						Timestamp:          time.Now().Unix(),
+						SessionID:          sessionID,
+						SessionDescription: sessionDescription,
+					},
+					Signaling: &models.LiveHDSignalingCallbacks{
+						SendAnswer: func(callbackSessionID string, sdp string) error {
+							return sockets[clientID].WriteJson(Message{
+								ClientID:    clientID,
+								MessageType: "webrtc-answer",
+								Message: map[string]string{
+									"session_id": callbackSessionID,
+									"sdp":        sdp,
+								},
+							})
+						},
+						SendCandidate: func(callbackSessionID string, candidate string) error {
+							return sockets[clientID].WriteJson(Message{
+								ClientID:    clientID,
+								MessageType: "webrtc-candidate",
+								Message: map[string]string{
+									"session_id": callbackSessionID,
+									"candidate":  candidate,
+								},
+							})
+						},
+						SendError: func(callbackSessionID string, errorMessage string) error {
+							writeWebRTCError(sockets[clientID], clientID, callbackSessionID, errorMessage)
+							return nil
+						},
+					},
+				}
+
+				communication.HandleLiveHDHandshake <- handshake
+
+			case "webrtc-candidate":
+				sessionID := message.Message["session_id"]
+				candidate := message.Message["candidate"]
+
+				if sessionID == "" || candidate == "" {
+					writeWebRTCError(sockets[clientID], clientID, sessionID, "missing session_id or candidate")
+					break
+				}
+
+				if !communication.CameraConnected {
+					writeWebRTCError(sockets[clientID], clientID, sessionID, "camera is not connected")
+					break
+				}
+
+				key := configuration.Config.Key + "/" + sessionID
+				go webrtc.RegisterCandidates(key, models.ReceiveHDCandidatesPayload{
+					Timestamp: time.Now().Unix(),
+					SessionID: sessionID,
+					Candidate: candidate,
+				})
 			}
 
 			err = conn.ReadJSON(&message)

machinery/src/utils/main.go

@@ -27,7 +27,8 @@ import (
 
 // VERSION is the agent version. It defaults to "0.0.0" for local dev builds
 // and is overridden at build time via:
-//   go build -ldflags "-X github.com/kerberos-io/agent/machinery/src/utils.VERSION=v1.2.3"
+//
+//	go build -ldflags "-X github.com/kerberos-io/agent/machinery/src/utils.VERSION=v1.2.3"
 var VERSION = "0.0.0"
 
 const letterBytes = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
@@ -198,6 +199,13 @@ func GetMediaFormatted(files []os.FileInfo, recordingDirectory string, configura
 			timestampInt, err := strconv.ParseInt(timestamp, 10, 64)
 			if err == nil {
 
+				if eventFilter.TimestampOffsetStart > 0 {
+					// TimestampOffsetStart represents the newest lower bound to include.
+					if timestampInt < eventFilter.TimestampOffsetStart {
+						continue
+					}
+				}
+
 				// If we have an offset we will check if we should skip or not
 				if eventFilter.TimestampOffsetEnd > 0 {
 					// Medias are sorted from new to older. TimestampOffsetEnd holds the oldest

machinery/src/utils/main_test.go

@@ -0,0 +1,54 @@
+package utils
+
+import (
+	"os"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+type stubFileInfo struct {
+	name string
+}
+
+func (s stubFileInfo) Name() string       { return s.name }
+func (s stubFileInfo) Size() int64        { return 0 }
+func (s stubFileInfo) Mode() os.FileMode  { return 0 }
+func (s stubFileInfo) ModTime() time.Time { return time.Unix(0, 0) }
+func (s stubFileInfo) IsDir() bool        { return false }
+func (s stubFileInfo) Sys() interface{}   { return nil }
+
+func TestGetMediaFormattedHonorsTimestampRange(t *testing.T) {
+	configuration := &models.Configuration{}
+	configuration.Config.Timezone = "UTC"
+	configuration.Config.Name = "Front Door"
+	configuration.Config.Key = "camera-1"
+
+	files := []os.FileInfo{
+		stubFileInfo{name: "1700000200_6_7_8_9_10.mp4"},
+		stubFileInfo{name: "1700000100_6_7_8_9_10.mp4"},
+		stubFileInfo{name: "1700000000_6_7_8_9_10.mp4"},
+	}
+
+	media := GetMediaFormatted(files, "/tmp/recordings", configuration, models.EventFilter{
+		TimestampOffsetStart: 1700000050,
+		TimestampOffsetEnd:   1700000200,
+		NumberOfElements:     10,
+	})
+
+	if len(media) != 1 {
+		t.Fatalf("expected 1 media item in time range, got %d", len(media))
+	}
+
+	if media[0].Timestamp != "1700000100" {
+		t.Fatalf("expected timestamp 1700000100, got %s", media[0].Timestamp)
+	}
+
+	if media[0].CameraName != "Front Door" {
+		t.Fatalf("expected camera name to be preserved, got %s", media[0].CameraName)
+	}
+	if media[0].CameraKey != "camera-1" {
+		t.Fatalf("expected camera key to be preserved, got %s", media[0].CameraKey)
+	}
+}

machinery/src/video/mp4.go

@@ -32,6 +32,24 @@ const MacEpochOffset uint64 = 2082844800
 // resulting in ~3 second fragments (assuming a typical GOP interval).
 const FragmentDurationMs = 3000
 
+// MinNormalGOPMs is the maximum spacing between two consecutive IDRs that
+// we still consider an anomalous "loop/restart seam". When two keyframes
+// arrive closer than this, we treat the second one as an upstream
+// restart/loop-seam and force a fresh fragment so the seam IDR cannot end
+// up as a mid-fragment sync sample. The check only runs when the current
+// fragment has not yet reached FragmentDurationMs.
+//
+// This must be set well below the smallest plausible *legitimate* GOP
+// length. Typical IP cameras use GOP intervals of 1000-2000 ms, and the
+// arrival timing of consecutive IDRs can jitter by a few hundred ms due to
+// network/RTSP buffering. A threshold close to 1 s (e.g. 950) caused
+// false positives on cameras with ~1 s GOPs (warnings like
+// "gap=800 ms / 300 ms / 200 ms" while the stream itself was healthy).
+// 400 ms is comfortably below any realistic GOP yet still catches the
+// virtual-rtsp / ffmpeg loop-seam pattern (seam IDRs typically arrive
+// 100-200 ms after the prior IDR).
+const MinNormalGOPMs = 400
+
 type MP4 struct {
 	// FileName is the name of the file
 	FileName                string
@@ -56,6 +74,8 @@ type MP4 struct {
 	FreeBoxSize             int64
 	FragmentStartRawPTS     uint64            // Raw PTS for timing when to flush fragments
 	FragmentStartDTS        uint64            // Accumulated VideoTotalDuration at fragment start (matches tfdt)
+	LastKeyframeRawPTS      uint64            // Raw PTS of the most recently seen keyframe (in any fragment)
+	LastKeyframeGapMs       uint64            // Gap (ms) between the previous two consecutive keyframes
 	MoofBoxes               int64             // Number of moof boxes in the file
 	MoofBoxSizes            []int64           // Sizes of each moof box
 	SegmentDurations        []uint64          // Duration of each segment in timescale units
@@ -233,6 +253,24 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
 	var duration uint64
 	if nextPTS > 0 && nextPTS > mp4.VideoFullSample.DecodeTime {
 		duration = nextPTS - mp4.VideoFullSample.DecodeTime
+		// Guard against forward PTS jumps (e.g. when looping a source MP4
+		// through virtual-rtsp the upstream ffmpeg may insert a large offset
+		// at the loop boundary, or the RTSP stream may stall briefly).
+		// Without this clamp the sample gets a huge duration which appears
+		// as a discontinuity in the trun/sidx/mvhd and causes browsers
+		// (Video.js / MSE) to abort playback with a "media corruption"
+		// error around the loop boundary.
+		var maxPlausible uint64 = 1000 // 1 second hard ceiling
+		if mp4.LastVideoSampleDTS > 0 && mp4.LastVideoSampleDTS*10 < maxPlausible {
+			maxPlausible = mp4.LastVideoSampleDTS * 10
+		}
+		if duration > maxPlausible {
+			log.Log.Warning(fmt.Sprintf("mp4.flushPendingVideoSample(): video PTS jumped forward (nextPTS=%d, prevDTS=%d, gap=%d ms) - clamping to %d ms (likely source loop/stall discontinuity)", nextPTS, mp4.VideoFullSample.DecodeTime, duration, maxPlausible))
+			duration = mp4.LastVideoSampleDTS
+			if duration == 0 {
+				duration = 33
+			}
+		}
 	} else {
 		// No valid nextPTS (Close case) or PTS went backwards (jitter/discontinuity)
 		if nextPTS > 0 {
@@ -289,6 +327,37 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
 		}
 		shouldFlush := !mp4.Start || elapsed >= FragmentDurationMs
 
+		// Detect upstream source-loop / restart discontinuity. When an MP4 is
+		// looped through virtual-rtsp (ffmpeg `-stream_loop -1 -re`) the loop
+		// seam emits a fresh IDR much sooner than a normal GOP would. PTS keeps
+		// growing monotonically, so the timing-only `elapsed` check above does
+		// not catch it and the seam IDR ends up as a mid-fragment sync sample.
+		// MSE-based players (Video.js / Chromium / Firefox) reject the resulting
+		// fragment with a "media corruption" error because the inner IDR resets
+		// frame_num/POC inside what they expect to be a single GOP. Force a
+		// fragment boundary whenever two consecutive keyframes arrive much
+		// closer than a normal GOP.
+		//
+		// We only flag this as a seam when it is a *sudden* anomaly: the
+		// previous keyframe gap must have been healthy (>= MinNormalGOPMs).
+		// This avoids false positives on cameras that legitimately emit
+		// short-interval IDRs (short GOP, motion-triggered recovery IDRs,
+		// all-intra streams) where every keyframe would otherwise be flagged
+		// in a cascade, producing many tiny fragments and log spam.
+		if trackID == uint32(mp4.VideoTrack) && mp4.Start &&
+			mp4.LastKeyframeRawPTS > 0 && pts > mp4.LastKeyframeRawPTS {
+			gap := pts - mp4.LastKeyframeRawPTS
+			if !shouldFlush && gap < MinNormalGOPMs &&
+				(mp4.LastKeyframeGapMs == 0 || mp4.LastKeyframeGapMs >= MinNormalGOPMs) {
+				log.Log.Warning(fmt.Sprintf("mp4.AddSampleToTrack(): forcing fragment flush at unexpectedly close keyframe (gap=%d ms, fragment elapsed=%d ms) - likely upstream loop/restart discontinuity", gap, elapsed))
+				shouldFlush = true
+			}
+			mp4.LastKeyframeGapMs = gap
+		}
+		if trackID == uint32(mp4.VideoTrack) {
+			mp4.LastKeyframeRawPTS = pts
+		}
+
 		if shouldFlush {
 			// Write the previous segment to the file
 			if mp4.Start {
@@ -393,6 +462,14 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
 					if started {
 						dts = 1
 					}
+					// Guard against forward PTS jumps (e.g. virtual-rtsp loop
+					// boundary or upstream stalls). Without this clamp the
+					// audio trun would carry an enormous sample duration that
+					// renders the recording unplayable in browsers.
+					if mp4.LastAudioSampleDTS > 0 && dts > mp4.LastAudioSampleDTS*10 {
+						log.Log.Warning(fmt.Sprintf("mp4.AddSampleToTrack(): audio PTS jumped forward (pts=%d, prevDTS=%d, gap=%d) - clamping to last known duration", pts, mp4.AudioFullSample.DecodeTime, dts))
+						dts = mp4.LastAudioSampleDTS
+					}
 					mp4.LastAudioSampleDTS = dts
 					//fmt.Printf("Adding sample to track %d, PTS: %d, Duration: %d, size: %d\n", trackID, pts, dts, len(aac[7:]))
 					mp4.AudioTotalDuration += dts

machinery/src/video/mp4_loopseam_test.go

@@ -0,0 +1,102 @@
+package video
+
+import (
+	"fmt"
+	"os"
+	"testing"
+
+	mp4ff "github.com/Eyevinn/mp4ff/mp4"
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+// TestMP4LoopSeamIsolation reproduces the loop-seam pattern from the
+// failing virtual-rtsp recordings: ~1s GOPs, but at the source-MP4
+// loop boundary an IDR arrives prematurely (~200-870ms after the
+// previous IDR). Without the fix this seam IDR ends up bunched into
+// the same fragment as the prior GOP's IDR which trips macOS
+// VideoToolbox (kVTVideoDecoderBadDataErr / -12909). The fix forces
+// a fragment flush whenever two IDRs arrive closer than MinNormalGOPMs.
+func TestMP4LoopSeamIsolation(t *testing.T) {
+	tmpFile := "/tmp/test_loop_seam.mp4"
+	defer os.Remove(tmpFile)
+
+	sps := []byte{0x67, 0x42, 0xc0, 0x1e, 0xd9, 0x00, 0xa0, 0x47, 0xfe, 0xc8}
+	pps := []byte{0x68, 0xce, 0x38, 0x80}
+	mp4Video := NewMP4(tmpFile, [][]byte{sps}, [][]byte{pps}, nil, 30)
+	mp4Video.SetWidth(1920)
+	mp4Video.SetHeight(1080)
+	v := mp4Video.AddVideoTrack("H264")
+
+	mk := func(k bool) []byte {
+		nt := byte(0x01)
+		if k {
+			nt = 0x65
+		}
+		f := []byte{0, 0, 0, 1, nt}
+		for i := 0; i < 200; i++ {
+			f = append(f, byte(i))
+		}
+		return f
+	}
+
+	frameDur := uint64(33)
+	pts := uint64(0)
+	emit := func(n int, gopLen int) {
+		for f := 0; f < n; f++ {
+			isKey := (f % gopLen) == 0
+			mp4Video.AddSampleToTrack(v, isKey, mk(isKey), pts)
+			pts += frameDur
+		}
+	}
+
+	// 17 seconds of normal content (last "good" IDR at sec 17).
+	emit(17*30, 30)
+	// Seam: IDR arrives ~150ms after previous (vs normal ~1000ms).
+	// This matches the realistic virtual-rtsp / ffmpeg `-stream_loop`
+	// loop boundary, where the new clip's first IDR is emitted shortly
+	// after the previous clip's final IDR.
+	pts -= 820
+	emit(13*30, 30)
+
+	mp4Video.Close(&models.Config{Signing: &models.Signing{PrivateKey: ""}})
+
+	f, _ := os.Open(tmpFile)
+	defer f.Close()
+	parsed, err := mp4ff.DecodeFile(f)
+	if err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+
+	fragIdx := 0
+	for _, seg := range parsed.Segments {
+		for _, fr := range seg.Fragments {
+			for _, traf := range fr.Moof.Trafs {
+				if traf.Tfhd.TrackID != 1 {
+					continue
+				}
+				tfdt := traf.Tfdt.BaseMediaDecodeTime()
+				offset := uint64(0)
+				var keys []uint64
+				for _, trun := range traf.Truns {
+					for _, s := range trun.Samples {
+						if (s.Flags>>24)&0x03 == 0x02 {
+							keys = append(keys, offset)
+						}
+						offset += uint64(s.Dur)
+					}
+				}
+				fmt.Printf("frag %d tfdt=%d samples_dur=%d keys@%v\n",
+					fragIdx, tfdt, offset, keys)
+				for i := 1; i < len(keys); i++ {
+					gap := keys[i] - keys[i-1]
+					if gap < MinNormalGOPMs {
+						t.Errorf("frag %d (tfdt=%d): two IDRs only %d ms apart "+
+							"in same fragment (< %d) - seam was not isolated",
+							fragIdx, tfdt, gap, MinNormalGOPMs)
+					}
+				}
+				fragIdx++
+			}
+		}
+	}
+}

machinery/src/webrtc/aac_transcoder_ffmpeg.go

@@ -0,0 +1,270 @@
+// AAC to G.711 µ-law transcoder using FFmpeg (libavcodec + libswresample).
+// Build with:  go build -tags ffmpeg ...
+//
+// Requires:  libavcodec-dev, libavutil-dev, libswresample-dev  (FFmpeg ≥ 5.x)
+// and an AAC decoder compiled into the FFmpeg build (usually the default).
+//
+//go:build ffmpeg
+
+package webrtc
+
+/*
+#cgo pkg-config: libavcodec libavutil libswresample
+#cgo CFLAGS: -Wno-deprecated-declarations
+
+#include <libavcodec/avcodec.h>
+#include <libavutil/channel_layout.h>
+#include <libavutil/frame.h>
+#include <libavutil/mem.h>
+#include <libavutil/opt.h>
+#include <libswresample/swresample.h>
+#include <stdlib.h>
+#include <string.h>
+
+// ── Transcoder handle ───────────────────────────────────────────────────
+
+typedef struct {
+    AVCodecContext       *codec_ctx;
+    AVCodecParserContext *parser;
+    SwrContext           *swr_ctx;
+    AVFrame              *frame;
+    AVPacket             *pkt;
+    int                   swr_initialized;
+    int                   in_sample_rate;
+    int                   in_channels;
+} aac_transcoder_t;
+
+// ── Create / Destroy ────────────────────────────────────────────────────
+
+static aac_transcoder_t* aac_transcoder_create(void) {
+    const AVCodec *codec = avcodec_find_decoder(AV_CODEC_ID_AAC);
+    if (!codec) return NULL;
+
+    aac_transcoder_t *t = (aac_transcoder_t*)calloc(1, sizeof(aac_transcoder_t));
+    if (!t) return NULL;
+
+    t->codec_ctx = avcodec_alloc_context3(codec);
+    if (!t->codec_ctx) { free(t); return NULL; }
+
+    if (avcodec_open2(t->codec_ctx, codec, NULL) < 0) {
+        avcodec_free_context(&t->codec_ctx);
+        free(t);
+        return NULL;
+    }
+
+    t->parser = av_parser_init(AV_CODEC_ID_AAC);
+    if (!t->parser) {
+        avcodec_free_context(&t->codec_ctx);
+        free(t);
+        return NULL;
+    }
+
+    t->frame = av_frame_alloc();
+    t->pkt   = av_packet_alloc();
+    if (!t->frame || !t->pkt) {
+        if (t->frame) av_frame_free(&t->frame);
+        if (t->pkt)   av_packet_free(&t->pkt);
+        av_parser_close(t->parser);
+        avcodec_free_context(&t->codec_ctx);
+        free(t);
+        return NULL;
+    }
+
+    return t;
+}
+
+static void aac_transcoder_destroy(aac_transcoder_t *t) {
+    if (!t) return;
+    if (t->swr_ctx)   swr_free(&t->swr_ctx);
+    if (t->frame)     av_frame_free(&t->frame);
+    if (t->pkt)       av_packet_free(&t->pkt);
+    if (t->parser)    av_parser_close(t->parser);
+    if (t->codec_ctx) avcodec_free_context(&t->codec_ctx);
+    free(t);
+}
+
+// ── Lazy resampler init (called after the first decoded frame) ──────────
+
+static int aac_init_swr(aac_transcoder_t *t) {
+    int64_t in_ch_layout = (int64_t)t->codec_ctx->channel_layout;
+    if (in_ch_layout == 0)
+        in_ch_layout = av_get_default_channel_layout(t->codec_ctx->channels);
+
+    t->swr_ctx = swr_alloc_set_opts(
+        NULL,
+        AV_CH_LAYOUT_MONO,                             // out: mono
+        AV_SAMPLE_FMT_S16,                             // out: signed 16-bit
+        8000,                                           // out: 8 kHz
+        in_ch_layout,                                   // in:  from decoder
+        t->codec_ctx->sample_fmt,                       // in:  from decoder
+        t->codec_ctx->sample_rate,                      // in:  from decoder
+        0, NULL);
+
+    if (!t->swr_ctx) return -1;
+    if (swr_init(t->swr_ctx) < 0) {
+        swr_free(&t->swr_ctx);
+        return -1;
+    }
+
+    t->in_sample_rate = t->codec_ctx->sample_rate;
+    t->in_channels    = t->codec_ctx->channels;
+    t->swr_initialized = 1;
+    return 0;
+}
+
+// ── Transcode ADTS → 8 kHz mono S16 PCM ────────────────────────────────
+// Caller must free *out_pcm with av_free() when non-NULL.
+
+static int aac_transcode_to_pcm(aac_transcoder_t *t,
+                                 const uint8_t *data, int data_size,
+                                 uint8_t **out_pcm, int *out_size) {
+    *out_pcm  = NULL;
+    *out_size = 0;
+    if (!data || data_size <= 0) return 0;
+
+    int buf_cap = 8192;
+    uint8_t *buf = (uint8_t*)av_malloc(buf_cap);
+    if (!buf) return -1;
+    int buf_len = 0;
+
+    while (data_size > 0) {
+        uint8_t *pout = NULL;
+        int      pout_size = 0;
+
+        int used = av_parser_parse2(t->parser, t->codec_ctx,
+                                    &pout, &pout_size,
+                                    data, data_size,
+                                    AV_NOPTS_VALUE, AV_NOPTS_VALUE, 0);
+        if (used < 0) break;
+        data      += used;
+        data_size -= used;
+        if (pout_size == 0) continue;
+
+        // Feed parsed frame to decoder
+        t->pkt->data = pout;
+        t->pkt->size = pout_size;
+        if (avcodec_send_packet(t->codec_ctx, t->pkt) < 0) continue;
+
+        // Pull all decoded frames
+        while (avcodec_receive_frame(t->codec_ctx, t->frame) == 0) {
+            if (!t->swr_initialized) {
+                if (aac_init_swr(t) < 0) {
+                    av_frame_unref(t->frame);
+                    av_free(buf);
+                    return -1;
+                }
+            }
+
+            int out_samples = swr_get_out_samples(t->swr_ctx,
+                                                   t->frame->nb_samples);
+            if (out_samples <= 0) out_samples = t->frame->nb_samples;
+
+            int needed = buf_len + out_samples * 2; // S16 = 2 bytes/sample
+            if (needed > buf_cap) {
+                buf_cap = needed * 2;
+                uint8_t *tmp = (uint8_t*)av_realloc(buf, buf_cap);
+                if (!tmp) { av_frame_unref(t->frame); av_free(buf); return -1; }
+                buf = tmp;
+            }
+
+            uint8_t *dst = buf + buf_len;
+            int converted = swr_convert(t->swr_ctx,
+                                        &dst, out_samples,
+                                        (const uint8_t**)t->frame->extended_data,
+                                        t->frame->nb_samples);
+            if (converted > 0)
+                buf_len += converted * 2;
+
+            av_frame_unref(t->frame);
+        }
+    }
+
+    if (buf_len == 0) {
+        av_free(buf);
+        return 0;
+    }
+
+    *out_pcm  = buf;
+    *out_size = buf_len;
+    return 0;
+}
+*/
+import "C"
+
+import (
+	"errors"
+	"fmt"
+	"unsafe"
+
+	"github.com/kerberos-io/agent/machinery/src/log"
+	"github.com/zaf/g711"
+)
+
+// AACTranscodingAvailable reports whether AAC→PCMU transcoding
+// is compiled in (requires the "ffmpeg" build tag).
+func AACTranscodingAvailable() bool { return true }
+
+// AACTranscoder decodes ADTS-wrapped AAC audio to 8 kHz mono PCM
+// and encodes it as G.711 µ-law for WebRTC transport.
+type AACTranscoder struct {
+	handle *C.aac_transcoder_t
+}
+
+// NewAACTranscoder creates a transcoder backed by FFmpeg's AAC decoder.
+func NewAACTranscoder() (*AACTranscoder, error) {
+	h := C.aac_transcoder_create()
+	if h == nil {
+		return nil, errors.New("failed to create AAC transcoder (FFmpeg AAC decoder not available?)")
+	}
+	log.Log.Info("webrtc.aac_transcoder: AAC → G.711 µ-law transcoder initialised (FFmpeg)")
+	return &AACTranscoder{handle: h}, nil
+}
+
+// Transcode converts an ADTS buffer (one or more AAC frames) into
+// G.711 µ-law encoded audio suitable for a PCMU WebRTC track.
+func (t *AACTranscoder) Transcode(adtsData []byte) ([]byte, error) {
+	if t == nil || t.handle == nil || len(adtsData) == 0 {
+		return nil, nil
+	}
+
+	var outPCM *C.uint8_t
+	var outSize C.int
+
+	ret := C.aac_transcode_to_pcm(
+		t.handle,
+		(*C.uint8_t)(unsafe.Pointer(&adtsData[0])),
+		C.int(len(adtsData)),
+		&outPCM, &outSize,
+	)
+	if ret < 0 {
+		return nil, errors.New("AAC decode/resample failed")
+	}
+	if outSize == 0 || outPCM == nil {
+		return nil, nil // decoder buffering, no output yet
+	}
+	defer C.av_free(unsafe.Pointer(outPCM))
+
+	// Copy S16LE PCM to Go slice, then encode to µ-law.
+	pcm := C.GoBytes(unsafe.Pointer(outPCM), outSize)
+	ulaw := g711.EncodeUlaw(pcm)
+
+	// Log resampler details once.
+	if t.handle.swr_initialized == 1 && t.handle.in_sample_rate != 0 {
+		log.Log.Info(fmt.Sprintf(
+			"webrtc.aac_transcoder: first output – resampling %d Hz / %d ch → 8000 Hz mono → µ-law",
+			int(t.handle.in_sample_rate), int(t.handle.in_channels)))
+		// Prevent repeated logging by zeroing the field we check.
+		t.handle.in_sample_rate = 0
+	}
+
+	return ulaw, nil
+}
+
+// Close releases all FFmpeg resources held by the transcoder.
+func (t *AACTranscoder) Close() {
+	if t != nil && t.handle != nil {
+		C.aac_transcoder_destroy(t.handle)
+		t.handle = nil
+		log.Log.Info("webrtc.aac_transcoder: transcoder closed")
+	}
+}

machinery/src/webrtc/aac_transcoder_stub.go

@@ -0,0 +1,205 @@
+// AAC transcoding fallback that uses the ffmpeg binary at runtime.
+// Build with -tags ffmpeg to use the in-process CGO implementation instead.
+//
+//go:build !ffmpeg
+
+package webrtc
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"os/exec"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/log"
+)
+
+// AACTranscodingAvailable reports whether AAC→PCMU transcoding
+// is available in the current runtime.
+func AACTranscodingAvailable() bool {
+	_, err := exec.LookPath("ffmpeg")
+	return err == nil
+}
+
+// AACTranscoder uses an ffmpeg subprocess to convert ADTS AAC to raw PCMU.
+type AACTranscoder struct {
+	cmd       *exec.Cmd
+	stdin     io.WriteCloser
+	stdout    io.ReadCloser
+	stderrBuf bytes.Buffer
+
+	mu        sync.Mutex
+	outMu     sync.Mutex
+	outBuf    bytes.Buffer
+	closed    bool
+	closeOnce sync.Once
+}
+
+// NewAACTranscoder creates a runtime ffmpeg-based transcoder.
+func NewAACTranscoder() (*AACTranscoder, error) {
+	ffmpegPath, err := exec.LookPath("ffmpeg")
+	if err != nil {
+		return nil, errors.New("AAC transcoding not available: ffmpeg binary not found in PATH")
+	}
+	log.Log.Info("webrtc.aac_transcoder: using ffmpeg binary at " + ffmpegPath)
+
+	cmd := exec.Command(
+		ffmpegPath,
+		"-hide_banner",
+		"-loglevel", "error",
+		"-fflags", "+nobuffer",
+		"-flags", "low_delay",
+		"-f", "aac",
+		"-i", "pipe:0",
+		"-vn",
+		"-ac", "1",
+		"-ar", "8000",
+		"-acodec", "pcm_mulaw",
+		"-f", "mulaw",
+		"pipe:1",
+	)
+
+	stdin, err := cmd.StdinPipe()
+	if err != nil {
+		return nil, err
+	}
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+	cmd.Stderr = &bytes.Buffer{}
+
+	if err := cmd.Start(); err != nil {
+		return nil, err
+	}
+
+	t := &AACTranscoder{
+		cmd:    cmd,
+		stdin:  stdin,
+		stdout: stdout,
+	}
+	if stderrBuf, ok := cmd.Stderr.(*bytes.Buffer); ok {
+		t.stderrBuf = *stderrBuf
+	}
+
+	go func() {
+		buf := make([]byte, 4096)
+		for {
+			n, readErr := stdout.Read(buf)
+			if n > 0 {
+				t.outMu.Lock()
+				_, _ = t.outBuf.Write(buf[:n])
+				buffered := t.outBuf.Len()
+				t.outMu.Unlock()
+				if buffered <= 8192 || buffered%16000 == 0 {
+					log.Log.Info("webrtc.aac_transcoder: ffmpeg produced PCMU bytes, buffered=" + strconv.Itoa(buffered))
+				}
+			}
+			if readErr != nil {
+				if readErr != io.EOF {
+					log.Log.Warning("webrtc.aac_transcoder: stdout reader stopped: " + readErr.Error())
+				}
+				return
+			}
+		}
+	}()
+
+	log.Log.Info("webrtc.aac_transcoder: AAC → PCMU transcoder initialised (ffmpeg process)")
+	return t, nil
+}
+
+// Transcode writes ADTS AAC to ffmpeg and returns any PCMU bytes produced.
+func (t *AACTranscoder) Transcode(adtsData []byte) ([]byte, error) {
+	if t == nil || len(adtsData) == 0 {
+		return nil, nil
+	}
+
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	if t.closed {
+		return nil, errors.New("AAC transcoder is closed")
+	}
+
+	if _, err := t.stdin.Write(adtsData); err != nil {
+		return nil, err
+	}
+	if len(adtsData) <= 512 || len(adtsData)%1024 == 0 {
+		log.Log.Info("webrtc.aac_transcoder: wrote AAC bytes to ffmpeg, input=" + strconv.Itoa(len(adtsData)))
+	}
+
+	deadline := time.Now().Add(75 * time.Millisecond)
+	for {
+		data := t.readAvailable()
+		if len(data) > 0 {
+			log.Log.Info("webrtc.aac_transcoder: returning PCMU bytes=" + strconv.Itoa(len(data)))
+			return data, nil
+		}
+
+		if time.Now().After(deadline) {
+			if stderr := t.stderrString(); stderr != "" {
+				log.Log.Warning("webrtc.aac_transcoder: no output before deadline, ffmpeg stderr: " + stderr)
+			} else {
+				log.Log.Info("webrtc.aac_transcoder: no PCMU output before deadline")
+			}
+			return nil, nil
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}
+
+func (t *AACTranscoder) readAvailable() []byte {
+	t.outMu.Lock()
+	defer t.outMu.Unlock()
+
+	if t.outBuf.Len() == 0 {
+		return nil
+	}
+
+	out := make([]byte, t.outBuf.Len())
+	copy(out, t.outBuf.Bytes())
+	t.outBuf.Reset()
+	return out
+}
+
+func (t *AACTranscoder) stderrString() string {
+	if t == nil {
+		return ""
+	}
+	if stderrBuf, ok := t.cmd.Stderr.(*bytes.Buffer); ok {
+		return strings.TrimSpace(stderrBuf.String())
+	}
+	return strings.TrimSpace(t.stderrBuf.String())
+}
+
+// Close stops the ffmpeg subprocess.
+func (t *AACTranscoder) Close() {
+	if t == nil {
+		return
+	}
+
+	t.closeOnce.Do(func() {
+		t.mu.Lock()
+		t.closed = true
+		if t.stdin != nil {
+			_ = t.stdin.Close()
+		}
+		t.mu.Unlock()
+
+		if t.stdout != nil {
+			_ = t.stdout.Close()
+		}
+
+		if t.cmd != nil {
+			_ = t.cmd.Process.Kill()
+			_, _ = t.cmd.Process.Wait()
+			if stderr := t.stderrString(); stderr != "" {
+				log.Log.Info("webrtc.aac_transcoder: ffmpeg stderr on close: " + stderr)
+			}
+		}
+	})
+}

machinery/src/webrtc/main.go

@@ -4,13 +4,14 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/json"
+	"fmt"
 	"io"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
-	//"github.com/izern/go-fdkaac/fdkaac"
 	"github.com/kerberos-io/agent/machinery/src/capture"
 	"github.com/kerberos-io/agent/machinery/src/log"
 	"github.com/kerberos-io/agent/machinery/src/models"
@@ -139,6 +140,11 @@ func (cm *ConnectionManager) GetPeerConnectionCount() int64 {
 	return atomic.LoadInt64(&cm.peerConnectionCount)
 }
 
+// GetActivePeerConnectionCount returns the current number of connected WebRTC readers.
+func GetActivePeerConnectionCount() int64 {
+	return globalConnectionManager.GetPeerConnectionCount()
+}
+
 // IncrementPeerCount atomically increments the peer connection count
 func (cm *ConnectionManager) IncrementPeerCount() int64 {
 	return atomic.AddInt64(&cm.peerConnectionCount, 1)
@@ -251,7 +257,79 @@ func RegisterDefaultInterceptors(mediaEngine *pionWebRTC.MediaEngine, intercepto
 	return nil
 }
 
-func InitializeWebRTCConnection(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, videoBroadcaster *TrackBroadcaster, audioBroadcaster *TrackBroadcaster, handshake models.RequestHDStreamPayload) {
+func publishSignalingMessageAsync(mqttClient mqtt.Client, topic string, payload []byte, description string) {
+	if mqttClient == nil {
+		log.Log.Error("webrtc.main.publishSignalingMessageAsync(): mqtt client is nil for " + description)
+		return
+	}
+
+	token := mqttClient.Publish(topic, 2, false, payload)
+	go func() {
+		if !token.WaitTimeout(5 * time.Second) {
+			log.Log.Warning("webrtc.main.publishSignalingMessageAsync(): timed out publishing " + description)
+			return
+		}
+		if err := token.Error(); err != nil {
+			log.Log.Error("webrtc.main.publishSignalingMessageAsync(): failed publishing " + description + ": " + err.Error())
+		}
+	}()
+}
+
+func sendCandidateSignal(configuration *models.Configuration, mqttClient mqtt.Client, hubKey string, handshake models.LiveHDHandshake, candidateJSON []byte) {
+	if handshake.Signaling != nil && handshake.Signaling.SendCandidate != nil {
+		if err := handshake.Signaling.SendCandidate(handshake.Payload.SessionID, string(candidateJSON)); err != nil {
+			log.Log.Error("webrtc.main.sendCandidateSignal(): " + err.Error())
+		}
+		return
+	}
+
+	message := models.Message{
+		Payload: models.Payload{
+			Action:   "receive-hd-candidates",
+			DeviceId: configuration.Config.Key,
+			Value: map[string]interface{}{
+				"candidate":  string(candidateJSON),
+				"session_id": handshake.Payload.SessionID,
+			},
+		},
+	}
+	payload, err := models.PackageMQTTMessage(configuration, message)
+	if err == nil {
+		publishSignalingMessageAsync(mqttClient, "kerberos/hub/"+hubKey, payload, "ICE candidate for session "+handshake.Payload.SessionID)
+	} else {
+		log.Log.Info("webrtc.main.sendCandidateSignal(): while packaging mqtt message: " + err.Error())
+	}
+}
+
+func sendAnswerSignal(configuration *models.Configuration, mqttClient mqtt.Client, hubKey string, handshake models.LiveHDHandshake, answer pionWebRTC.SessionDescription) {
+	encodedAnswer := base64.StdEncoding.EncodeToString([]byte(answer.SDP))
+
+	if handshake.Signaling != nil && handshake.Signaling.SendAnswer != nil {
+		if err := handshake.Signaling.SendAnswer(handshake.Payload.SessionID, encodedAnswer); err != nil {
+			log.Log.Error("webrtc.main.sendAnswerSignal(): " + err.Error())
+		}
+		return
+	}
+
+	message := models.Message{
+		Payload: models.Payload{
+			Action:   "receive-hd-answer",
+			DeviceId: configuration.Config.Key,
+			Value: map[string]interface{}{
+				"sdp":        []byte(encodedAnswer),
+				"session_id": handshake.Payload.SessionID,
+			},
+		},
+	}
+	payload, err := models.PackageMQTTMessage(configuration, message)
+	if err == nil {
+		publishSignalingMessageAsync(mqttClient, "kerberos/hub/"+hubKey, payload, "SDP answer for session "+handshake.Payload.SessionID)
+	} else {
+		log.Log.Info("webrtc.main.sendAnswerSignal(): while packaging mqtt message: " + err.Error())
+	}
+}
+
+func InitializeWebRTCConnection(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, videoBroadcaster *TrackBroadcaster, audioBroadcaster *TrackBroadcaster, handshake models.LiveHDHandshake) {
 
 	config := configuration.Config
 	deviceKey := config.Key
@@ -259,14 +337,15 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 	turnServers := []string{config.TURNURI}
 	turnServersUsername := config.TURNUsername
 	turnServersCredential := config.TURNPassword
+	handshakePayload := handshake.Payload
 
 	// We create a channel which will hold the candidates for this session.
-	sessionKey := config.Key + "/" + handshake.SessionID
+	sessionKey := config.Key + "/" + handshakePayload.SessionID
 	candidateChannel := globalConnectionManager.GetOrCreateCandidateChannel(sessionKey)
 
 	// Set variables
-	hubKey := handshake.HubKey
-	sessionDescription := handshake.SessionDescription
+	hubKey := handshakePayload.HubKey
+	sessionDescription := handshakePayload.SessionDescription
 
 	// Create WebRTC object
 	w := CreateWebRTC(deviceKey, stunServers, turnServers, turnServersUsername, turnServersCredential)
@@ -423,12 +502,12 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 			// Log ICE connection state changes for diagnostics
 			peerConnection.OnICEConnectionStateChange(func(iceState pionWebRTC.ICEConnectionState) {
 				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): ICE connection state changed to: " + iceState.String() +
-					" (session: " + handshake.SessionID + ")")
+					" (session: " + handshakePayload.SessionID + ")")
 			})
 
 			peerConnection.OnConnectionStateChange(func(connectionState pionWebRTC.PeerConnectionState) {
 				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): connection state changed to: " + connectionState.String() +
-					" (session: " + handshake.SessionID + ")")
+					" (session: " + handshakePayload.SessionID + ")")
 
 				switch connectionState {
 				case pionWebRTC.PeerConnectionStateDisconnected:
@@ -437,9 +516,9 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 					wrapper.disconnectMu.Lock()
 					if wrapper.disconnectTimer == nil {
 						log.Log.Info("webrtc.main.InitializeWebRTCConnection(): peer disconnected, waiting " +
-							disconnectGracePeriod.String() + " for recovery (session: " + handshake.SessionID + ")")
+							disconnectGracePeriod.String() + " for recovery (session: " + handshakePayload.SessionID + ")")
 						wrapper.disconnectTimer = time.AfterFunc(disconnectGracePeriod, func() {
-							log.Log.Info("webrtc.main.InitializeWebRTCConnection(): disconnect grace period expired, closing connection (session: " + handshake.SessionID + ")")
+							log.Log.Info("webrtc.main.InitializeWebRTCConnection(): disconnect grace period expired, closing connection (session: " + handshakePayload.SessionID + ")")
 							cleanupPeerConnection(sessionKey, wrapper)
 						})
 					}
@@ -471,7 +550,7 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 					if wrapper.disconnectTimer != nil {
 						wrapper.disconnectTimer.Stop()
 						wrapper.disconnectTimer = nil
-						log.Log.Info("webrtc.main.InitializeWebRTCConnection(): connection recovered from disconnected state (session: " + handshake.SessionID + ")")
+						log.Log.Info("webrtc.main.InitializeWebRTCConnection(): connection recovered from disconnected state (session: " + handshakePayload.SessionID + ")")
 					}
 					wrapper.disconnectMu.Unlock()
 
@@ -482,33 +561,6 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				}
 			})
 
-			go func() {
-				defer func() {
-					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): candidate processor stopped for session: " + handshake.SessionID)
-				}()
-
-				// Iterate over the candidates and send them to the remote client
-				for {
-					select {
-					case <-ctx.Done():
-						return
-					case candidate, ok := <-candidateChannel:
-						if !ok {
-							return
-						}
-						log.Log.Info("webrtc.main.InitializeWebRTCConnection(): Received candidate from channel: " + candidate)
-						candidateInit, decodeErr := decodeICECandidate(candidate)
-						if decodeErr != nil {
-							log.Log.Error("webrtc.main.InitializeWebRTCConnection(): error decoding candidate: " + decodeErr.Error())
-							continue
-						}
-						if candidateErr := peerConnection.AddICECandidate(candidateInit); candidateErr != nil {
-							log.Log.Error("webrtc.main.InitializeWebRTCConnection(): error adding candidate: " + candidateErr.Error())
-						}
-					}
-				}
-			}()
-
 			// When an ICE candidate is available send to the other peer using the signaling server (MQTT).
 			// The other peer will add this candidate by calling AddICECandidate.
 			// This handler must be registered before setting the local description, otherwise early candidates can be missed.
@@ -556,27 +608,13 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				candateBinary, err := json.Marshal(candidateJSON)
 				if err == nil {
 					valueMap["candidate"] = string(candateBinary)
-					valueMap["session_id"] = handshake.SessionID
+					valueMap["session_id"] = handshakePayload.SessionID
 					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): sending " + candidateType + " candidate to hub")
 				} else {
 					log.Log.Error("webrtc.main.InitializeWebRTCConnection(): failed to marshal candidate: " + err.Error())
 				}
 
-				// We'll send the candidate to the hub
-				message := models.Message{
-					Payload: models.Payload{
-						Action:   "receive-hd-candidates",
-						DeviceId: configuration.Config.Key,
-						Value:    valueMap,
-					},
-				}
-				payload, err := models.PackageMQTTMessage(configuration, message)
-				if err == nil {
-					token := mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
-					token.Wait()
-				} else {
-					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): while packaging mqtt message: " + err.Error())
-				}
+				sendCandidateSignal(configuration, mqttClient, hubKey, handshake, candateBinary)
 			})
 
 			offer := w.CreateOffer(sd)
@@ -586,6 +624,35 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				return
 			}
 
+			go func() {
+				defer func() {
+					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): candidate processor stopped for session: " + handshakePayload.SessionID)
+				}()
+
+				// Process remote candidates only after the remote description is set.
+				// MQTT can deliver candidates before the SDP offer handling completes,
+				// and Pion rejects AddICECandidate calls until SetRemoteDescription succeeds.
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case candidate, ok := <-candidateChannel:
+						if !ok {
+							return
+						}
+						log.Log.Info("webrtc.main.InitializeWebRTCConnection(): Received candidate from channel: " + candidate)
+						candidateInit, decodeErr := decodeICECandidate(candidate)
+						if decodeErr != nil {
+							log.Log.Error("webrtc.main.InitializeWebRTCConnection(): error decoding candidate: " + decodeErr.Error())
+							continue
+						}
+						if candidateErr := peerConnection.AddICECandidate(candidateInit); candidateErr != nil {
+							log.Log.Error("webrtc.main.InitializeWebRTCConnection(): error adding candidate: " + candidateErr.Error())
+						}
+					}
+				}
+			}()
+
 			answer, err := peerConnection.CreateAnswer(nil)
 			if err != nil {
 				log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while creating answer: " + err.Error())
@@ -600,27 +667,9 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 			// Store peer connection in manager
 			globalConnectionManager.AddPeerConnection(sessionKey, wrapper)
 
-			//  Create a config map
-			valueMap := make(map[string]interface{})
-			valueMap["sdp"] = []byte(base64.StdEncoding.EncodeToString([]byte(answer.SDP)))
-			valueMap["session_id"] = handshake.SessionID
 			log.Log.Info("webrtc.main.InitializeWebRTCConnection(): Send SDP answer")
 
-			// We'll send the candidate to the hub
-			message := models.Message{
-				Payload: models.Payload{
-					Action:   "receive-hd-answer",
-					DeviceId: configuration.Config.Key,
-					Value:    valueMap,
-				},
-			}
-			payload, err := models.PackageMQTTMessage(configuration, message)
-			if err == nil {
-				token := mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
-				token.Wait()
-			} else {
-				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): while packaging mqtt message: " + err.Error())
-			}
+			sendAnswerSignal(configuration, mqttClient, hubKey, handshake, answer)
 		}
 	} else {
 		globalConnectionManager.CloseCandidateChannel(sessionKey)
@@ -640,7 +689,12 @@ func NewVideoBroadcaster(streams []packets.Stream) *TrackBroadcaster {
 }
 
 func NewAudioBroadcaster(streams []packets.Stream) *TrackBroadcaster {
+	var audioCodecNames []string
+	hasAAC := false
 	for _, s := range streams {
+		if s.IsAudio {
+			audioCodecNames = append(audioCodecNames, s.Name)
+		}
 		switch s.Name {
 		case "OPUS":
 			return NewTrackBroadcaster(pionWebRTC.MimeTypeOpus, "audio", trackStreamID)
@@ -648,9 +702,18 @@ func NewAudioBroadcaster(streams []packets.Stream) *TrackBroadcaster {
 			return NewTrackBroadcaster(pionWebRTC.MimeTypePCMU, "audio", trackStreamID)
 		case "PCM_ALAW":
 			return NewTrackBroadcaster(pionWebRTC.MimeTypePCMA, "audio", trackStreamID)
+		case "AAC":
+			hasAAC = true
 		}
 	}
-	log.Log.Error("webrtc.main.NewAudioBroadcaster(): no supported audio codec found")
+	if hasAAC {
+		log.Log.Info("webrtc.main.NewAudioBroadcaster(): AAC detected, creating PCMU audio track for transcoded output")
+		return NewTrackBroadcaster(pionWebRTC.MimeTypePCMU, "audio", trackStreamID)
+	} else if len(audioCodecNames) > 0 {
+		log.Log.Error(fmt.Sprintf("webrtc.main.NewAudioBroadcaster(): no supported audio codec found (detected: %s; supported: OPUS, PCM_MULAW, PCM_ALAW)", strings.Join(audioCodecNames, ", ")))
+	} else {
+		log.Log.Info("webrtc.main.NewAudioBroadcaster(): no audio stream found in camera feed")
+	}
 	return nil
 }
 
@@ -666,18 +729,33 @@ func NewVideoTrack(streams []packets.Stream) *pionWebRTC.TrackLocalStaticSample
 
 func NewAudioTrack(streams []packets.Stream) *pionWebRTC.TrackLocalStaticSample {
 	var mimeType string
+	var audioCodecNames []string
+	hasAAC := false
 	for _, stream := range streams {
+		if stream.IsAudio {
+			audioCodecNames = append(audioCodecNames, stream.Name)
+		}
 		if stream.Name == "OPUS" {
 			mimeType = pionWebRTC.MimeTypeOpus
 		} else if stream.Name == "PCM_MULAW" {
 			mimeType = pionWebRTC.MimeTypePCMU
 		} else if stream.Name == "PCM_ALAW" {
 			mimeType = pionWebRTC.MimeTypePCMA
+		} else if stream.Name == "AAC" {
+			hasAAC = true
 		}
 	}
 	if mimeType == "" {
-		log.Log.Error("webrtc.main.NewAudioTrack(): no supported audio codec found")
-		return nil
+		if hasAAC {
+			mimeType = pionWebRTC.MimeTypePCMU
+			log.Log.Info("webrtc.main.NewAudioTrack(): AAC detected, creating PCMU audio track for transcoded output")
+		} else if len(audioCodecNames) > 0 {
+			log.Log.Error(fmt.Sprintf("webrtc.main.NewAudioTrack(): no supported audio codec found (detected: %s; supported: OPUS, PCM_MULAW, PCM_ALAW)", strings.Join(audioCodecNames, ", ")))
+			return nil
+		} else {
+			log.Log.Info("webrtc.main.NewAudioTrack(): no audio stream found in camera feed")
+			return nil
+		}
 	}
 	outboundAudioTrack, err := pionWebRTC.NewTrackLocalStaticSample(pionWebRTC.RTPCodecCapability{MimeType: mimeType}, "audio", trackStreamID)
 	if err != nil {
@@ -696,6 +774,11 @@ type streamState struct {
 	receivedKeyFrame bool
 	lastAudioSample  *pionMedia.Sample
 	lastVideoSample  *pionMedia.Sample
+	audioPacketsSeen int64
+	aacPacketsSeen   int64
+	audioSamplesSent int64
+	aacNoOutput      int64
+	aacErrors        int64
 }
 
 // codecSupport tracks which codecs are available in the stream
@@ -843,22 +926,54 @@ func processVideoPacket(pkt packets.Packet, state *streamState, videoBroadcaster
 	state.lastVideoSample = &sample
 }
 
-// processAudioPacket processes an audio packet and writes samples to the broadcaster
-func processAudioPacket(pkt packets.Packet, state *streamState, audioBroadcaster *TrackBroadcaster, hasAAC bool) {
+// processAudioPacket processes an audio packet and writes samples to the broadcaster.
+// When the packet carries AAC and a transcoder is provided, the audio is transcoded
+// to G.711 µ-law on the fly so it can be sent over a PCMU WebRTC track.
+func processAudioPacket(pkt packets.Packet, state *streamState, audioBroadcaster *TrackBroadcaster, transcoder *AACTranscoder) {
 	if audioBroadcaster == nil {
 		return
 	}
 
-	if hasAAC {
-		// AAC transcoding not yet implemented
-		// TODO: Implement AAC to PCM_MULAW transcoding
-		return
+	state.audioPacketsSeen++
+
+	audioData := pkt.Data
+
+	if pkt.Codec == "AAC" {
+		state.aacPacketsSeen++
+		if transcoder == nil {
+			state.aacErrors++
+			if state.aacErrors <= 3 || state.aacErrors%100 == 0 {
+				log.Log.Warning(fmt.Sprintf("webrtc.main.processAudioPacket(): AAC packet dropped because transcoder is nil (aac_packets=%d, input_bytes=%d)", state.aacPacketsSeen, len(pkt.Data)))
+			}
+			return // no transcoder – silently drop
+		}
+		pcmu, err := transcoder.Transcode(pkt.Data)
+		if err != nil {
+			state.aacErrors++
+			log.Log.Error("webrtc.main.processAudioPacket(): AAC transcode error: " + err.Error())
+			return
+		}
+		if len(pcmu) == 0 {
+			state.aacNoOutput++
+			if state.aacNoOutput <= 5 || state.aacNoOutput%100 == 0 {
+				log.Log.Info(fmt.Sprintf("webrtc.main.processAudioPacket(): AAC packet produced no PCMU output yet (aac_packets=%d, no_output=%d, input_bytes=%d)", state.aacPacketsSeen, state.aacNoOutput, len(pkt.Data)))
+			}
+			return // decoder still buffering
+		}
+		if state.aacPacketsSeen <= 5 || state.aacPacketsSeen%100 == 0 {
+			log.Log.Info(fmt.Sprintf("webrtc.main.processAudioPacket(): AAC transcoded to PCMU (aac_packets=%d, input_bytes=%d, output_bytes=%d, peers=%d)", state.aacPacketsSeen, len(pkt.Data), len(pcmu), audioBroadcaster.PeerCount()))
+		}
+		audioData = pcmu
 	}
 
-	sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: sampleTimestamp(pkt)}
+	sample := pionMedia.Sample{Data: audioData, PacketTimestamp: sampleTimestamp(pkt)}
 
 	if state.lastAudioSample != nil {
 		state.lastAudioSample.Duration = sampleDuration(pkt, state.lastAudioSample.PacketTimestamp, 20*time.Millisecond)
+		state.audioSamplesSent++
+		if state.audioSamplesSent <= 5 || state.audioSamplesSent%100 == 0 {
+			log.Log.Info(fmt.Sprintf("webrtc.main.processAudioPacket(): queueing audio sample (samples=%d, codec=%s, bytes=%d, duration_ms=%d, peers=%d)", state.audioSamplesSent, pkt.Codec, len(state.lastAudioSample.Data), state.lastAudioSample.Duration.Milliseconds(), audioBroadcaster.PeerCount()))
+		}
 		audioBroadcaster.WriteSample(*state.lastAudioSample)
 	}
 
@@ -892,8 +1007,22 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 		return
 	}
 
+	// Create AAC transcoder if needed (AAC → G.711 µ-law).
+	var aacTranscoder *AACTranscoder
+	if codecs.hasAAC && audioBroadcaster != nil {
+		log.Log.Info(fmt.Sprintf("webrtc.main.WriteToTrack(): AAC audio detected, creating transcoder (audio_peers=%d)", audioBroadcaster.PeerCount()))
+		t, err := NewAACTranscoder()
+		if err != nil {
+			log.Log.Error("webrtc.main.WriteToTrack(): failed to create AAC transcoder: " + err.Error())
+		} else {
+			aacTranscoder = t
+			log.Log.Info("webrtc.main.WriteToTrack(): AAC transcoder created successfully")
+			defer aacTranscoder.Close()
+		}
+	}
+
 	if config.Capture.TranscodingWebRTC == "true" {
-		log.Log.Info("webrtc.main.WriteToTrack(): transcoding enabled but not yet implemented")
+		log.Log.Info("webrtc.main.WriteToTrack(): transcoding config enabled")
 	}
 
 	// Initialize streaming state
@@ -903,6 +1032,12 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 	}
 
 	defer func() {
+		log.Log.Info(fmt.Sprintf("webrtc.main.WriteToTrack(): audio summary packets=%d aac_packets=%d sent=%d aac_no_output=%d aac_errors=%d peers=%d", state.audioPacketsSeen, state.aacPacketsSeen, state.audioSamplesSent, state.aacNoOutput, state.aacErrors, func() int {
+			if audioBroadcaster == nil {
+				return 0
+			}
+			return audioBroadcaster.PeerCount()
+		}()))
 		writeFinalSamples(state, videoBroadcaster, audioBroadcaster)
 		log.Log.Info("webrtc.main.WriteToTrack(): stopped writing to track")
 	}()
@@ -971,7 +1106,7 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 		if pkt.IsVideo {
 			processVideoPacket(pkt, state, videoBroadcaster, config)
 		} else if pkt.IsAudio {
-			processAudioPacket(pkt, state, audioBroadcaster, codecs.hasAAC)
+			processAudioPacket(pkt, state, audioBroadcaster, aacTranscoder)
 		}
 	}
 }

ui/package.json

@@ -45,9 +45,9 @@
     "crypto": false
   },
   "scripts": {
-    "start": "react-scripts start",
-    "build": "GENERATE_SOURCEMAP=false REACT_APP_ENVIRONMENT=production react-scripts build && rm -rf ../machinery/www && mv build ../machinery/www",
-    "test": "react-scripts test",
+    "start": "DISABLE_ESLINT_PLUGIN=true react-scripts start",
+    "build": "DISABLE_ESLINT_PLUGIN=true GENERATE_SOURCEMAP=false REACT_APP_ENVIRONMENT=production react-scripts build && rm -rf ../machinery/www && mv build ../machinery/www",
+    "test": "DISABLE_ESLINT_PLUGIN=true react-scripts test",
     "eject": "react-scripts eject",
     "lint": "eslint --debug 'src/**/*.{js,jsx,ts,tsx}'",
     "format": "prettier --write \"**/*.{js,jsx,json,md}\""

ui/public/locales/en/translation.json

@@ -237,4 +237,4 @@
       "remove_after_upload_enabled": "Enable delete on upload"
     }
   }
-}
+}

ui/src/actions/agent.js

@@ -194,7 +194,7 @@ export const getDashboardInformation = (onSuccess, onError) => {
   };
 };
 
-export const getEvents = (eventfilter, onSuccess, onError) => {
+export const getEvents = (eventfilter, onSuccess, onError, append = false) => {
   return (dispatch) => {
     doGetEvents(
       eventfilter,
@@ -203,6 +203,7 @@ export const getEvents = (eventfilter, onSuccess, onError) => {
           type: 'GET_EVENTS',
           events: data.events,
           filter: eventfilter,
+          append,
         });
         if (onSuccess) {
           onSuccess();

ui/src/pages/Dashboard/Dashboard.jsx

@@ -26,6 +26,23 @@ import {
 import './Dashboard.scss';
 import ReactTooltip from 'react-tooltip';
 import config from '../../config';
+import { getConfig } from '../../actions/agent';
+
+function createUUID() {
+  if (
+    typeof window !== 'undefined' &&
+    window.crypto &&
+    typeof window.crypto.randomUUID === 'function'
+  ) {
+    return window.crypto.randomUUID();
+  }
+
+  return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, (char) => {
+    const random = Math.floor(Math.random() * 16);
+    const value = char === 'x' ? random : 8 + Math.floor(random / 4);
+    return value.toString(16);
+  });
+}
 
 // eslint-disable-next-line react/prefer-stateless-function
 class Dashboard extends React.Component {
@@ -33,48 +50,55 @@ class Dashboard extends React.Component {
     super();
     this.state = {
       liveviewLoaded: false,
+      liveviewMode: 'webrtc',
       open: false,
       currentRecording: '',
       initialised: false,
     };
+    this.videoRef = React.createRef();
+    this.pendingRemoteCandidates = [];
     this.initialiseLiveview = this.initialiseLiveview.bind(this);
-    this.handleLiveviewLoad = this.handleLiveviewLoad.bind(this);
+    this.initialiseSDLiveview = this.initialiseSDLiveview.bind(this);
+    this.startWebRTCLiveview = this.startWebRTCLiveview.bind(this);
+    this.handleWebRTCSignalMessage = this.handleWebRTCSignalMessage.bind(this);
+    this.fallbackToSDLiveview = this.fallbackToSDLiveview.bind(this);
   }
 
   componentDidMount() {
-    const liveview = document.getElementsByClassName('videocard-video');
-    if (liveview && liveview.length > 0) {
-      [this.liveviewElement] = liveview;
-      this.liveviewElement.addEventListener('load', this.handleLiveviewLoad);
-    }
+    const { dispatchGetConfig } = this.props;
+    dispatchGetConfig(() => this.initialiseLiveview());
     this.initialiseLiveview();
   }
 
-  componentDidUpdate() {
-    this.initialiseLiveview();
+  componentDidUpdate(prevProps) {
+    const { images, dashboard } = this.props;
+    const { liveviewLoaded, liveviewMode } = this.state;
+    const configLoaded = this.hasAgentConfig(this.props);
+    const prevConfigLoaded = this.hasAgentConfig(prevProps);
+
+    if (!prevConfigLoaded && configLoaded) {
+      this.initialiseLiveview();
+    }
+
+    if (
+      liveviewMode === 'sd' &&
+      !liveviewLoaded &&
+      prevProps.images !== images &&
+      images.length > 0
+    ) {
+      this.setState({
+        liveviewLoaded: true,
+      });
+    }
+
+    if (!prevProps.dashboard.cameraOnline && dashboard.cameraOnline) {
+      this.initialiseLiveview();
+    }
   }
 
   componentWillUnmount() {
-    if (this.liveviewElement) {
-      this.liveviewElement.removeEventListener('load', this.handleLiveviewLoad);
-      this.liveviewElement = null;
-    }
-
-    if (this.requestStreamSubscription) {
-      this.requestStreamSubscription.unsubscribe();
-      this.requestStreamSubscription = null;
-    }
-    const { dispatchSend } = this.props;
-    const message = {
-      message_type: 'stop-sd',
-    };
-    dispatchSend(message);
-  }
-
-  handleLiveviewLoad() {
-    this.setState({
-      liveviewLoaded: true,
-    });
+    this.stopSDLiveview();
+    this.stopWebRTCLiveview();
   }
 
   handleClose() {
@@ -84,32 +108,378 @@ class Dashboard extends React.Component {
     });
   }
 
-  getCurrentTimestamp() {
-    return Math.round(Date.now() / 1000);
+  // eslint-disable-next-line react/sort-comp
+  hasAgentConfig(props) {
+    const currentProps = props || this.props;
+    const { config: configResponse } = currentProps;
+    return !!(configResponse && configResponse.config);
+  }
+
+  browserSupportsWebRTC() {
+    return (
+      typeof window !== 'undefined' &&
+      typeof window.RTCPeerConnection !== 'undefined'
+    );
+  }
+
+  buildPeerConnectionConfig() {
+    const { config: configResponse } = this.props;
+    const agentConfig =
+      configResponse && configResponse.config ? configResponse.config : {};
+    const iceServers = [];
+
+    if (agentConfig.stunuri) {
+      iceServers.push({
+        urls: [agentConfig.stunuri],
+      });
+    }
+
+    if (agentConfig.turnuri) {
+      const turnServer = {
+        urls: [agentConfig.turnuri],
+      };
+
+      if (agentConfig.turn_username) {
+        turnServer.username = agentConfig.turn_username;
+      }
+
+      if (agentConfig.turn_password) {
+        turnServer.credential = agentConfig.turn_password;
+      }
+
+      iceServers.push(turnServer);
+    }
+
+    return {
+      iceServers,
+      iceTransportPolicy: agentConfig.turn_force === 'true' ? 'relay' : 'all',
+    };
   }
 
   initialiseLiveview() {
     const { initialised } = this.state;
-    if (!initialised) {
-      const message = {
-        message_type: 'stream-sd',
-      };
-      const { connected, dispatchSend } = this.props;
-      if (connected) {
+    const { dashboard } = this.props;
+
+    if (initialised || !dashboard.cameraOnline) {
+      return;
+    }
+
+    if (!this.hasAgentConfig()) {
+      return;
+    }
+
+    if (this.browserSupportsWebRTC()) {
+      this.startWebRTCLiveview();
+    } else {
+      this.fallbackToSDLiveview('WebRTC is not supported in this browser.');
+    }
+  }
+
+  initialiseSDLiveview() {
+    if (this.requestStreamSubscription) {
+      return;
+    }
+
+    const message = {
+      message_type: 'stream-sd',
+    };
+    const { connected, dispatchSend } = this.props;
+    if (connected) {
+      dispatchSend(message);
+    }
+
+    const requestStreamInterval = interval(2000);
+    this.requestStreamSubscription = requestStreamInterval.subscribe(() => {
+      const { connected: isConnected } = this.props;
+      if (isConnected) {
         dispatchSend(message);
       }
+    });
+  }
 
-      const requestStreamInterval = interval(2000);
-      this.requestStreamSubscription = requestStreamInterval.subscribe(() => {
-        const { connected: isConnected } = this.props;
-        if (isConnected) {
-          dispatchSend(message);
-        }
-      });
-      this.setState({
-        initialised: true,
-      });
+  stopSDLiveview() {
+    if (this.requestStreamSubscription) {
+      this.requestStreamSubscription.unsubscribe();
+      this.requestStreamSubscription = null;
     }
+
+    const { dispatchSend } = this.props;
+    dispatchSend({
+      message_type: 'stop-sd',
+    });
+  }
+
+  stopWebRTCLiveview() {
+    if (this.webrtcTimeout) {
+      window.clearTimeout(this.webrtcTimeout);
+      this.webrtcTimeout = null;
+    }
+
+    if (this.webrtcSocket) {
+      this.webrtcSocket.onopen = null;
+      this.webrtcSocket.onmessage = null;
+      this.webrtcSocket.onerror = null;
+      this.webrtcSocket.onclose = null;
+      this.webrtcSocket.close();
+      this.webrtcSocket = null;
+    }
+
+    if (this.webrtcPeerConnection) {
+      this.webrtcPeerConnection.ontrack = null;
+      this.webrtcPeerConnection.onicecandidate = null;
+      this.webrtcPeerConnection.onconnectionstatechange = null;
+      this.webrtcPeerConnection.close();
+      this.webrtcPeerConnection = null;
+    }
+
+    this.pendingRemoteCandidates = [];
+    this.webrtcOfferStarted = false;
+    this.webrtcSessionId = null;
+    this.webrtcClientId = null;
+
+    if (this.videoRef.current) {
+      this.videoRef.current.srcObject = null;
+    }
+  }
+
+  sendWebRTCMessage(messageType, message = {}) {
+    if (!this.webrtcSocket || this.webrtcSocket.readyState !== WebSocket.OPEN) {
+      return;
+    }
+
+    this.webrtcSocket.send(
+      JSON.stringify({
+        client_id: this.webrtcClientId,
+        message_type: messageType,
+        message,
+      })
+    );
+  }
+
+  async handleWebRTCSignalMessage(event) {
+    let data;
+    try {
+      data = JSON.parse(event.data);
+    } catch (error) {
+      return;
+    }
+
+    const { message_type: messageType, message = {} } = data;
+    const { session_id: sessionID, sdp, candidate } = message;
+
+    if (messageType === 'hello-back') {
+      await this.beginWebRTCLiveview();
+      return;
+    }
+
+    if (sessionID && sessionID !== this.webrtcSessionId) {
+      return;
+    }
+
+    switch (messageType) {
+      case 'webrtc-answer':
+        try {
+          await this.webrtcPeerConnection.setRemoteDescription({
+            type: 'answer',
+            sdp: window.atob(sdp),
+          });
+          await this.flushPendingRemoteCandidates();
+        } catch (error) {
+          this.fallbackToSDLiveview(
+            `Unable to apply WebRTC answer: ${error.message}`
+          );
+        }
+        break;
+
+      case 'webrtc-candidate': {
+        try {
+          const candidateInit = JSON.parse(candidate);
+          if (
+            this.webrtcPeerConnection.remoteDescription &&
+            this.webrtcPeerConnection.remoteDescription.type
+          ) {
+            await this.webrtcPeerConnection.addIceCandidate(candidateInit);
+          } else {
+            this.pendingRemoteCandidates.push(candidateInit);
+          }
+        } catch (error) {
+          this.fallbackToSDLiveview(
+            `Unable to apply WebRTC candidate: ${error.message}`
+          );
+        }
+        break;
+      }
+
+      case 'webrtc-error':
+        this.fallbackToSDLiveview(
+          message.message || 'The agent could not start the WebRTC liveview.'
+        );
+        break;
+
+      default:
+        break;
+    }
+  }
+
+  async beginWebRTCLiveview() {
+    if (!this.webrtcPeerConnection || this.webrtcOfferStarted) {
+      return;
+    }
+
+    try {
+      this.webrtcOfferStarted = true;
+      const offer = await this.webrtcPeerConnection.createOffer({
+        offerToReceiveAudio: true,
+        offerToReceiveVideo: true,
+      });
+      await this.webrtcPeerConnection.setLocalDescription(offer);
+      this.sendWebRTCMessage('stream-hd', {
+        session_id: this.webrtcSessionId,
+        sdp: window.btoa(this.webrtcPeerConnection.localDescription.sdp),
+      });
+    } catch (error) {
+      this.fallbackToSDLiveview(
+        `Unable to initialise WebRTC liveview: ${error.message}`,
+      );
+    }
+  }
+
+  async flushPendingRemoteCandidates() {
+    if (
+      !this.webrtcPeerConnection ||
+      !this.webrtcPeerConnection.remoteDescription
+    ) {
+      return;
+    }
+
+    while (this.pendingRemoteCandidates.length > 0) {
+      const candidateInit = this.pendingRemoteCandidates.shift();
+      try {
+        // eslint-disable-next-line no-await-in-loop
+        await this.webrtcPeerConnection.addIceCandidate(candidateInit);
+      } catch (error) {
+        this.fallbackToSDLiveview(
+          `Unable to add remote ICE candidate: ${error.message}`,
+        );
+        return;
+      }
+    }
+  }
+
+  startWebRTCLiveview() {
+    if (this.webrtcPeerConnection || this.webrtcSocket) {
+      return;
+    }
+
+    this.stopSDLiveview();
+
+    this.webrtcClientId = createUUID();
+    this.webrtcSessionId = createUUID();
+    this.pendingRemoteCandidates = [];
+
+    this.webrtcPeerConnection = new window.RTCPeerConnection(
+      this.buildPeerConnectionConfig()
+    );
+
+    this.webrtcPeerConnection.ontrack = (event) => {
+      const [remoteStream] = event.streams;
+      if (this.videoRef.current && remoteStream) {
+        this.videoRef.current.srcObject = remoteStream;
+        const playPromise = this.videoRef.current.play();
+        if (playPromise && playPromise.catch) {
+          playPromise.catch(() => {});
+        }
+      }
+
+      this.setState({
+        liveviewLoaded: true,
+      });
+    };
+
+    this.webrtcPeerConnection.onicecandidate = (event) => {
+      if (!event.candidate) {
+        return;
+      }
+
+      this.sendWebRTCMessage('webrtc-candidate', {
+        session_id: this.webrtcSessionId,
+        candidate: JSON.stringify(event.candidate.toJSON()),
+      });
+    };
+
+    this.webrtcPeerConnection.onconnectionstatechange = () => {
+      const { connectionState } = this.webrtcPeerConnection;
+      if (connectionState === 'connected') {
+        this.setState({
+          liveviewLoaded: true,
+        });
+      }
+
+      if (
+        connectionState === 'failed' ||
+        connectionState === 'disconnected' ||
+        connectionState === 'closed'
+      ) {
+        this.fallbackToSDLiveview(
+          `WebRTC connection ${connectionState}, falling back to SD liveview.`,
+        );
+      }
+    };
+
+    this.webrtcSocket = new WebSocket(config.WS_URL);
+    this.webrtcSocket.onopen = () => {
+      this.sendWebRTCMessage('hello', {});
+    };
+    this.webrtcSocket.onmessage = this.handleWebRTCSignalMessage;
+    this.webrtcSocket.onerror = () => {
+      this.fallbackToSDLiveview('Unable to open the WebRTC signaling channel.');
+    };
+    this.webrtcSocket.onclose = () => {
+      const { liveviewLoaded } = this.state;
+      if (!liveviewLoaded) {
+        this.fallbackToSDLiveview('WebRTC signaling channel closed early.');
+      }
+    };
+
+    this.webrtcTimeout = window.setTimeout(() => {
+      const { liveviewLoaded } = this.state;
+      if (!liveviewLoaded) {
+        this.fallbackToSDLiveview(
+          'WebRTC connection timed out, falling back to SD liveview.'
+        );
+      }
+    }, 10000);
+
+    this.setState({
+      initialised: true,
+      liveviewLoaded: false,
+      liveviewMode: 'webrtc',
+    });
+  }
+
+  fallbackToSDLiveview(errorMessage) {
+    const { liveviewMode } = this.state;
+
+    if (liveviewMode === 'sd' && this.requestStreamSubscription) {
+      return;
+    }
+
+    this.stopWebRTCLiveview();
+    if (errorMessage) {
+      // eslint-disable-next-line no-console
+      console.warn(errorMessage);
+    }
+
+    this.setState(
+      {
+        initialised: true,
+        liveviewLoaded: false,
+        liveviewMode: 'sd',
+      },
+      () => {
+        this.initialiseSDLiveview();
+      }
+    );
   }
 
   openModal(file) {
@@ -121,7 +491,8 @@ class Dashboard extends React.Component {
 
   render() {
     const { dashboard, t, images } = this.props;
-    const { liveviewLoaded, open, currentRecording } = this.state;
+    const { liveviewLoaded, liveviewMode, open, currentRecording } = this.state;
+    const listenerCount = dashboard.webrtcReaders ? dashboard.webrtcReaders : 0;
 
     // We check if the camera was getting a valid frame
     // during the last 5 seconds, otherwise we assume the camera is offline.
@@ -175,7 +546,6 @@ class Dashboard extends React.Component {
             divider="0"
             footer={t('dashboard.total_recordings')}
           />
-
           <Link to="/settings">
             <Card
               title="IP Camera"
@@ -314,7 +684,9 @@ class Dashboard extends React.Component {
             )}
           </div>
           <div>
-            <h2>{t('dashboard.live_view')}</h2>
+            <h2>
+              {t('dashboard.live_view')} ({listenerCount})
+            </h2>
             {(!liveviewLoaded || !isCameraOnline) && (
               <SetupBox
                 btnicon="preferences"
@@ -331,12 +703,16 @@ class Dashboard extends React.Component {
                   liveviewLoaded && isCameraOnline ? 'visible' : 'hidden',
               }}
             >
-              <ImageCard
-                imageSrc={`data:image/png;base64, ${
-                  images.length ? images[0] : ''
-                }`}
-                onerror=""
-              />
+              {liveviewMode === 'webrtc' ? (
+                <video ref={this.videoRef} autoPlay muted playsInline />
+              ) : (
+                <ImageCard
+                  imageSrc={`data:image/png;base64, ${
+                    images.length ? images[0] : ''
+                  }`}
+                  onerror=""
+                />
+              )}
             </div>
           </div>
         </div>
@@ -348,20 +724,25 @@ class Dashboard extends React.Component {
 
 const mapStateToProps = (state /* , ownProps */) => ({
   dashboard: state.agent.dashboard,
+  config: state.agent.config,
   connected: state.wss.connected,
   images: state.wss.images,
 });
 
 const mapDispatchToProps = (dispatch) => ({
   dispatchSend: (message) => dispatch(send(message)),
+  dispatchGetConfig: (onSuccess, onError) =>
+    dispatch(getConfig(onSuccess, onError)),
 });
 
 Dashboard.propTypes = {
   dashboard: PropTypes.object.isRequired,
+  config: PropTypes.object.isRequired,
   connected: PropTypes.bool.isRequired,
   images: PropTypes.array.isRequired,
   t: PropTypes.func.isRequired,
   dispatchSend: PropTypes.func.isRequired,
+  dispatchGetConfig: PropTypes.func.isRequired,
 };
 
 export default withTranslation()(

ui/src/pages/Media/Media.jsx

@@ -3,6 +3,7 @@ import PropTypes from 'prop-types';
 import { withTranslation } from 'react-i18next';
 import {
   Breadcrumb,
+  ControlBar,
   VideoCard,
   Button,
   Modal,
@@ -16,14 +17,53 @@ import { getEvents } from '../../actions/agent';
 import config from '../../config';
 import './Media.scss';
 
+function formatDateTimeLocal(date) {
+  const year = date.getFullYear();
+  const month = String(date.getMonth() + 1).padStart(2, '0');
+  const day = String(date.getDate()).padStart(2, '0');
+  const hours = String(date.getHours()).padStart(2, '0');
+  const minutes = String(date.getMinutes()).padStart(2, '0');
+
+  return `${year}-${month}-${day}T${hours}:${minutes}`;
+}
+
+function getDefaultTimeWindow() {
+  const endDate = new Date();
+  const startDate = new Date(endDate.getTime() - 60 * 60 * 1000);
+
+  return {
+    startDateTime: formatDateTimeLocal(startDate),
+    endDateTime: formatDateTimeLocal(endDate),
+    timestamp_offset_start: Math.floor(startDate.getTime() / 1000),
+    timestamp_offset_end: Math.floor(endDate.getTime() / 1000) + 59,
+  };
+}
+
+function normalizeInputValue(valueOrEvent) {
+  if (valueOrEvent && valueOrEvent.target) {
+    return valueOrEvent.target.value;
+  }
+
+  return valueOrEvent;
+}
+
 // eslint-disable-next-line react/prefer-stateless-function
 class Media extends React.Component {
   constructor() {
     super();
-    this.state = {
-      timestamp_offset_start: 0,
-      timestamp_offset_end: 0,
+
+    const defaultTimeWindow = getDefaultTimeWindow();
+
+    const initialFilter = {
+      timestamp_offset_start: defaultTimeWindow.timestamp_offset_start,
+      timestamp_offset_end: defaultTimeWindow.timestamp_offset_end,
       number_of_elements: 12,
+    };
+
+    this.state = {
+      appliedFilter: initialFilter,
+      startDateTime: defaultTimeWindow.startDateTime,
+      endDateTime: defaultTimeWindow.endDateTime,
       isScrolling: false,
       open: false,
       currentRecording: '',
@@ -32,7 +72,8 @@ class Media extends React.Component {
 
   componentDidMount() {
     const { dispatchGetEvents } = this.props;
-    dispatchGetEvents(this.state);
+    const { appliedFilter } = this.state;
+    dispatchGetEvents(appliedFilter);
     document.addEventListener('scroll', this.trackScrolling);
   }
 
@@ -49,29 +90,107 @@ class Media extends React.Component {
 
   trackScrolling = () => {
     const { events, dispatchGetEvents } = this.props;
-    const { isScrolling } = this.state;
+    const { isScrolling, appliedFilter } = this.state;
     const wrappedElement = document.getElementById('loader');
-    if (!isScrolling && this.isBottom(wrappedElement)) {
-      this.setState({
-        isScrolling: true,
-      });
-      // Get last element
-      const lastElement = events[events.length - 1];
-      if (lastElement) {
-        this.setState({
+    if (!wrappedElement || isScrolling || !this.isBottom(wrappedElement)) {
+      return;
+    }
+
+    this.setState({
+      isScrolling: true,
+    });
+
+    // Get last element
+    const lastElement = events[events.length - 1];
+    if (lastElement) {
+      dispatchGetEvents(
+        {
+          ...appliedFilter,
           timestamp_offset_end: parseInt(lastElement.timestamp, 10),
-        });
-        dispatchGetEvents(this.state, () => {
+        },
+        () => {
           setTimeout(() => {
             this.setState({
               isScrolling: false,
             });
           }, 1000);
-        });
-      }
+        },
+        () => {
+          this.setState({
+            isScrolling: false,
+          });
+        },
+        true
+      );
+    } else {
+      this.setState({
+        isScrolling: false,
+      });
     }
   };
 
+  buildEventFilter(startDateTime, endDateTime) {
+    const { appliedFilter } = this.state;
+
+    return {
+      timestamp_offset_start: this.getTimestampFromInput(
+        startDateTime,
+        'start'
+      ),
+      timestamp_offset_end: this.getTimestampFromInput(endDateTime, 'end'),
+      number_of_elements: appliedFilter.number_of_elements,
+    };
+  }
+
+  handleDateFilterChange(field, value) {
+    const { dispatchGetEvents } = this.props;
+    const { startDateTime, endDateTime } = this.state;
+    const normalizedValue = normalizeInputValue(value);
+    const nextStartDateTime =
+      field === 'startDateTime' ? normalizedValue : startDateTime;
+    const nextEndDateTime =
+      field === 'endDateTime' ? normalizedValue : endDateTime;
+    const nextFilter = this.buildEventFilter(
+      nextStartDateTime,
+      nextEndDateTime
+    );
+    const shouldApplyFilter =
+      (nextStartDateTime === '' || nextStartDateTime.length === 16) &&
+      (nextEndDateTime === '' || nextEndDateTime.length === 16);
+
+    this.setState(
+      {
+        [field]: normalizedValue,
+        appliedFilter: shouldApplyFilter
+          ? nextFilter
+          : this.state.appliedFilter,
+        isScrolling: false,
+      },
+      () => {
+        if (shouldApplyFilter) {
+          dispatchGetEvents(nextFilter);
+        }
+      }
+    );
+  }
+
+  getTimestampFromInput(value, boundary) {
+    if (!value) {
+      return 0;
+    }
+
+    const date = new Date(value);
+    if (Number.isNaN(date.getTime())) {
+      return 0;
+    }
+
+    const seconds = Math.floor(date.getTime() / 1000);
+    if (boundary === 'end') {
+      return seconds + 59;
+    }
+    return seconds;
+  }
+
   isBottom(el) {
     return el.getBoundingClientRect().bottom + 50 <= window.innerHeight;
   }
@@ -85,7 +204,9 @@ class Media extends React.Component {
 
   render() {
     const { events, eventsLoaded, t } = this.props;
-    const { isScrolling, open, currentRecording } = this.state;
+    const { isScrolling, open, currentRecording, startDateTime, endDateTime } =
+      this.state;
+
     return (
       <div id="media">
         <Breadcrumb
@@ -102,6 +223,37 @@ class Media extends React.Component {
           </Link>
         </Breadcrumb>
 
+        <div className="media-control-bar">
+          <ControlBar>
+            <div className="media-filters">
+              <div className="media-filters__field">
+                <label htmlFor="recordings-start-time">Start time</label>
+                <input
+                  className="media-filters__input"
+                  id="recordings-start-time"
+                  type="datetime-local"
+                  value={startDateTime}
+                  onChange={(value) =>
+                    this.handleDateFilterChange('startDateTime', value)
+                  }
+                />
+              </div>
+              <div className="media-filters__field">
+                <label htmlFor="recordings-end-time">End time</label>
+                <input
+                  className="media-filters__input"
+                  id="recordings-end-time"
+                  type="datetime-local"
+                  value={endDateTime}
+                  onChange={(value) =>
+                    this.handleDateFilterChange('endDateTime', value)
+                  }
+                />
+              </div>
+            </div>
+          </ControlBar>
+        </div>
+
         <div className="stats grid-container --four-columns">
           {events.map((event) => (
             <div
@@ -123,6 +275,11 @@ class Media extends React.Component {
             </div>
           ))}
         </div>
+        {events.length === 0 && eventsLoaded === 0 && (
+          <div className="media-empty-state">
+            No recordings found in the selected time range.
+          </div>
+        )}
         {open && (
           <Modal>
             <ModalHeader
@@ -182,13 +339,13 @@ const mapStateToProps = (state /* , ownProps */) => ({
 });
 
 const mapDispatchToProps = (dispatch) => ({
-  dispatchGetEvents: (eventFilter, success, error) =>
-    dispatch(getEvents(eventFilter, success, error)),
+  dispatchGetEvents: (eventFilter, success, error, append) =>
+    dispatch(getEvents(eventFilter, success, error, append)),
 });
 
 Media.propTypes = {
   t: PropTypes.func.isRequired,
-  events: PropTypes.objectOf(PropTypes.object).isRequired,
+  events: PropTypes.arrayOf(PropTypes.object).isRequired,
   eventsLoaded: PropTypes.number.isRequired,
   dispatchGetEvents: PropTypes.func.isRequired,
 };

ui/src/pages/Media/Media.scss

@@ -1,4 +1,103 @@
 #media {
+  .media-control-bar {
+    .control-bar {
+      display: block;
+      padding: 0 var(--main-content-gutter);
+    }
+
+    .control-bar .filtering {
+      display: block;
+    }
+
+    .control-bar .filtering > * {
+      border-right: 0;
+      flex: 1 1 100% !important;
+      max-width: none;
+      width: 100%;
+    }
+  }
+
+  .media-filters {
+    align-items: stretch;
+    display: grid;
+    gap: 0;
+    grid-template-columns: repeat(2, minmax(0, 1fr));
+    min-width: 0;
+    width: 100%;
+  }
+
+  .media-filters__field {
+    border-right: 1px solid var(--bg-muted);
+    min-width: 0;
+    padding: 16px 24px;
+
+    label {
+      display: block;
+      font-size: 14px;
+      font-weight: 600;
+      margin-bottom: 8px;
+      white-space: nowrap;
+    }
+  }
+
+  .media-filters__field:first-child {
+    padding-left: 0;
+  }
+
+  .media-filters__input {
+    appearance: none;
+    background: var(--white);
+    border: 1px solid var(--grey-light);
+    border-radius: 8px;
+    box-sizing: border-box;
+    color: var(--black);
+    font-size: 16px;
+    min-height: 48px;
+    padding: 0 14px 0 0;
+    width: 100%;
+  }
+
+  .media-filters__input::-webkit-datetime-edit,
+  .media-filters__input::-webkit-datetime-edit-fields-wrapper {
+    padding: 0;
+  }
+
+  .media-filters__input:focus {
+    border-color: var(--oss);
+    outline: 0;
+  }
+
+  .media-filters__field:first-child .media-filters__input {
+    padding-left: 0;
+  }
+
+  .media-filters__field:last-child {
+    border-right: 0;
+    padding-right: 0;
+  }
+
+  @media (max-width: 700px) {
+    .media-filters {
+      grid-template-columns: 1fr;
+    }
+
+    .media-filters__field {
+      border-right: 0;
+      border-bottom: 1px solid var(--bg-muted);
+      padding-left: 0;
+      padding-right: 0;
+    }
+
+    .media-filters__field:last-child {
+      border-bottom: 0;
+    }
+  }
+
+  .media-empty-state {
+    margin: 24px 0;
+    opacity: 0.8;
+    text-align: center;
+  }
 
   #loader {
     display: flex;

ui/src/reducers/agent.js

@@ -123,16 +123,12 @@ const agent = (
       };
 
     case 'GET_EVENTS':
-      const { timestamp_offset_end } = action.filter;
       const { events } = action;
       return {
         ...state,
         eventsLoaded: events.length,
-        events:
-          timestamp_offset_end === 0
-            ? [...events]
-            : [...state.events, ...events],
-        eventfilter: action.eventfilter,
+        events: action.append ? [...state.events, ...events] : [...events],
+        eventfilter: action.filter,
       };
 
     default: