Merge pull request #230 from kerberos-io/feature/improve-video-format

feature/improve-video-format

.github/workflows/docker.yml

@@ -1,120 +0,0 @@
-name: Create a new release
-on:
-  release:
-    types: [created]
-  workflow_dispatch:
-    inputs:
-      tag:
-        description: "Tag for the Docker image"
-        required: true
-        default: "test"
-
-env:
-  REPO: kerberos/agent
-
-jobs:
-  build-amd64:
-    runs-on: ubuntu-latest
-    permissions:
-      contents: write
-    strategy:
-      matrix:
-        architecture: [amd64]
-    steps:
-      - name: Login to DockerHub
-        uses: docker/login-action@v2
-        with:
-          username: ${{ secrets.DOCKER_USERNAME }}
-          password: ${{ secrets.DOCKER_PASSWORD }}
-      - name: Checkout
-        uses: actions/checkout@v3
-      - uses: benjlevesque/short-sha@v2.1
-        id: short-sha
-        with:
-          length: 7
-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@v2
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v2
-      - name: Available platforms
-        run: echo ${{ steps.buildx.outputs.platforms }}
-      - name: Run Buildx
-        run: docker buildx build --platform linux/$(echo ${{matrix.architecture}} | tr - /) -t $REPO-arch:arch-${{matrix.architecture}}-${{github.event.inputs.tag || github.ref_name}} --push .
-      - name: Create new and append to manifest
-        run: docker buildx imagetools create -t $REPO:${{ github.event.inputs.tag || github.ref_name }} $REPO-arch:arch-${{matrix.architecture}}-${{github.event.inputs.tag || github.ref_name}}
-      - name: Create new and append to manifest latest
-        run: docker buildx imagetools create -t $REPO:latest $REPO-arch:arch-${{matrix.architecture}}-${{github.event.inputs.tag || github.ref_name}}
-        if: github.event.inputs.tag == 'test'
-      - name: Run Buildx with output
-        run: docker buildx build --platform linux/$(echo ${{matrix.architecture}} | tr - /) -t $REPO-arch:arch-$(echo ${{matrix.architecture}} | tr / -)-${{github.event.inputs.tag || github.ref_name}} --output type=tar,dest=output-${{matrix.architecture}}.tar .
-      - name: Strip binary
-        run: mkdir -p output/ && tar -xf output-${{matrix.architecture}}.tar -C output && rm output-${{matrix.architecture}}.tar && cd output/ && tar -cf ../agent-${{matrix.architecture}}.tar -C home/agent . && rm -rf output
-      - name: Create a release
-        uses: ncipollo/release-action@v1
-        with:
-          latest: true
-          allowUpdates: true
-          name: ${{ github.event.inputs.tag || github.ref_name }}
-          tag: ${{ github.event.inputs.tag || github.ref_name }}
-          generateReleaseNotes: false
-          omitBodyDuringUpdate: true
-          artifacts: "agent-${{matrix.architecture}}.tar"
-      # Taken from GoReleaser's own release workflow.
-      # The available Snapcraft Action has some bugs described in the issue below.
-      # The mkdirs are a hack for https://github.com/goreleaser/goreleaser/issues/1715.
-      #- name: Setup Snapcraft
-      #  run: |
-      #    sudo apt-get update
-      #    sudo apt-get -yq --no-install-suggests --no-install-recommends install snapcraft
-      #    mkdir -p $HOME/.cache/snapcraft/download
-      #    mkdir -p $HOME/.cache/snapcraft/stage-packages
-      #- name: Use Snapcraft
-      #  run: tar -xf agent-${{matrix.architecture}}.tar && snapcraft
-  build-other:
-    runs-on: ubuntu-latest
-    permissions:
-      contents: write
-    needs: build-amd64
-    strategy:
-      matrix:
-        architecture: [arm64, arm-v7, arm-v6]
-        #architecture: [arm64, arm-v7]
-    steps:
-      - name: Login to DockerHub
-        uses: docker/login-action@v2
-        with:
-          username: ${{ secrets.DOCKER_USERNAME }}
-          password: ${{ secrets.DOCKER_PASSWORD }}
-      - name: Checkout
-        uses: actions/checkout@v3
-      - uses: benjlevesque/short-sha@v2.1
-        id: short-sha
-        with:
-          length: 7
-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@v2
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v2
-      - name: Available platforms
-        run: echo ${{ steps.buildx.outputs.platforms }}
-      - name: Run Buildx
-        run: docker buildx build --platform linux/$(echo ${{matrix.architecture}} | tr - /) -t $REPO-arch:arch-${{matrix.architecture}}-${{github.event.inputs.tag || github.ref_name}} --push .
-      - name: Create new and append to manifest
-        run: docker buildx imagetools create --append -t $REPO:${{ github.event.inputs.tag || github.ref_name }} $REPO-arch:arch-${{matrix.architecture}}-${{github.event.inputs.tag || github.ref_name}}
-      - name: Create new and append to manifest latest
-        run: docker buildx imagetools create --append -t $REPO:latest $REPO-arch:arch-${{matrix.architecture}}-${{github.event.inputs.tag || github.ref_name}}
-        if: github.event.inputs.tag == 'test'
-      - name: Run Buildx with output
-        run: docker buildx build --platform linux/$(echo ${{matrix.architecture}} | tr - /) -t $REPO-arch:arch-$(echo ${{matrix.architecture}} | tr / -)-${{github.event.inputs.tag || github.ref_name}} --output type=tar,dest=output-${{matrix.architecture}}.tar .
-      - name: Strip binary
-        run: mkdir -p output/ && tar -xf output-${{matrix.architecture}}.tar -C output && rm output-${{matrix.architecture}}.tar && cd output/ && tar -cf ../agent-${{matrix.architecture}}.tar -C home/agent . && rm -rf output
-      - name: Create a release
-        uses: ncipollo/release-action@v1
-        with:
-          latest: true
-          allowUpdates: true
-          name: ${{ github.event.inputs.tag || github.ref_name }}
-          tag: ${{ github.event.inputs.tag || github.ref_name }}
-          generateReleaseNotes: false
-          omitBodyDuringUpdate: true
-          artifacts: "agent-${{matrix.architecture}}.tar"

.github/workflows/pr-build.yml

@@ -0,0 +1,75 @@
+name: Build pull request
+on:
+  pull_request:
+    types: [opened, synchronize]
+
+env:
+  REPO: kerberos/agent
+
+jobs:
+  build-amd64:
+    runs-on: ubuntu-24.04
+    permissions:
+      contents: write
+    strategy:
+      matrix:
+        architecture: [amd64]
+    steps:
+      - name: Login to DockerHub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+      - name: Checkout
+        uses: actions/checkout@v3
+      - uses: benjlevesque/short-sha@v2.1
+        id: short-sha
+        with:
+          length: 7
+      - name: Run Build
+        run: |
+          docker build -t ${{matrix.architecture}} .
+          CID=$(docker create ${{matrix.architecture}})
+          docker cp ${CID}:/home/agent ./output-${{matrix.architecture}}
+          docker rm ${CID}
+      - name: Strip binary
+        run: tar -cf agent-${{matrix.architecture}}.tar -C output-${{matrix.architecture}} . && rm -rf output-${{matrix.architecture}}
+      - name: Upload artifact
+        uses: actions/upload-artifact@v4
+        with:
+          name: agent-${{matrix.architecture}}.tar
+          path: agent-${{matrix.architecture}}.tar
+          
+  build-arm64:
+    runs-on: ubuntu-24.04-arm 
+    permissions:
+      contents: write
+    strategy:
+      matrix:
+        architecture: [arm64]
+    steps:
+      - name: Login to DockerHub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+      - name: Checkout
+        uses: actions/checkout@v3
+      - uses: benjlevesque/short-sha@v2.1
+        id: short-sha
+        with:
+          length: 7
+      - name: Run Build
+        run: |
+          docker build -t ${{matrix.architecture}} -f Dockerfile.arm64 .
+          CID=$(docker create ${{matrix.architecture}})
+          docker cp ${CID}:/home/agent ./output-${{matrix.architecture}}
+          docker rm ${CID}
+      - name: Strip binary
+        run: tar -cf agent-${{matrix.architecture}}.tar -C output-${{matrix.architecture}} . && rm -rf output-${{matrix.architecture}}
+      - name: Upload artifact
+        uses: actions/upload-artifact@v4
+        with:
+          name: agent-${{matrix.architecture}}.tar
+          path: agent-${{matrix.architecture}}.tar
+

.github/workflows/pr-description.yaml

@@ -2,6 +2,11 @@ name: Autofill PR description
 
 on: pull_request
 
+env:
+  ORGANIZATION: uugai
+  PROJECT: ${{ github.event.repository.name }}
+  PR_NUMBER: ${{ github.event.number }}
+  
 jobs:
   openai-pr-description:
     runs-on: ubuntu-22.04
@@ -16,4 +21,6 @@ jobs:
           azure_openai_api_key: ${{ secrets.AZURE_OPENAI_API_KEY }}
           azure_openai_endpoint: ${{ secrets.AZURE_OPENAI_ENDPOINT }}
           azure_openai_version: ${{ secrets.AZURE_OPENAI_VERSION }}
+          openai_model: ${{ secrets.OPENAI_MODEL }}
+          pull_request_url: https://pr${{ env.PR_NUMBER }}.api.kerberos.lol
           overwrite_description: true

.github/workflows/release-create.yml

@@ -0,0 +1,130 @@
+name: Create a new release
+on:
+  release:
+    types: [created]
+  workflow_dispatch:
+    inputs:
+      tag:
+        description: "Tag for the Docker image"
+        required: true
+        default: "test"
+
+env:
+  REPO: kerberos/agent
+
+jobs:
+  build-amd64:
+    runs-on: ubuntu-24.04
+    permissions:
+      contents: write
+    strategy:
+      matrix:
+        architecture: [amd64]
+    steps:
+      - name: Login to DockerHub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+      - name: Checkout
+        uses: actions/checkout@v3
+      - uses: benjlevesque/short-sha@v2.1
+        id: short-sha
+        with:
+          length: 7
+      - name: Run Build
+        run: |
+          docker build -t ${{matrix.architecture}} .
+          CID=$(docker create ${{matrix.architecture}})
+          docker cp ${CID}:/home/agent ./output-${{matrix.architecture}}
+          docker rm ${CID}
+      - name: Strip binary
+        run: tar -cf agent-${{matrix.architecture}}.tar -C output-${{matrix.architecture}} . && rm -rf output-${{matrix.architecture}}
+      - name: Build and push Docker image
+        run: |
+          docker tag ${{matrix.architecture}} $REPO-arch:arch-${{matrix.architecture}}-${{github.event.inputs.tag || github.ref_name}}
+          docker push $REPO-arch:arch-${{matrix.architecture}}-${{github.event.inputs.tag || github.ref_name}}
+      - name: Upload artifact
+        uses: actions/upload-artifact@v4
+        with:
+          name: agent-${{matrix.architecture}}.tar
+          path: agent-${{matrix.architecture}}.tar
+          
+  build-arm64:
+    runs-on: ubuntu-24.04-arm 
+    permissions:
+      contents: write
+    strategy:
+      matrix:
+        architecture: [arm64]
+    steps:
+      - name: Login to DockerHub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+      - name: Checkout
+        uses: actions/checkout@v3
+      - uses: benjlevesque/short-sha@v2.1
+        id: short-sha
+        with:
+          length: 7
+      - name: Run Build
+        run: |
+          docker build -t ${{matrix.architecture}} -f Dockerfile.arm64 .
+          CID=$(docker create ${{matrix.architecture}})
+          docker cp ${CID}:/home/agent ./output-${{matrix.architecture}}
+          docker rm ${CID}
+      - name: Strip binary
+        run: tar -cf agent-${{matrix.architecture}}.tar -C output-${{matrix.architecture}} . && rm -rf output-${{matrix.architecture}}
+      - name: Build and push Docker image
+        run: |
+          docker tag ${{matrix.architecture}} $REPO-arch:arch-${{matrix.architecture}}-${{github.event.inputs.tag || github.ref_name}}
+          docker push $REPO-arch:arch-${{matrix.architecture}}-${{github.event.inputs.tag || github.ref_name}}
+      - name: Upload artifact
+        uses: actions/upload-artifact@v4
+        with:
+          name: agent-${{matrix.architecture}}.tar
+          path: agent-${{matrix.architecture}}.tar
+
+  create-manifest:
+    runs-on: ubuntu-24.04
+    needs: [build-amd64, build-arm64]
+    steps:
+      - name: Login to DockerHub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+      - name: Create and push multi-arch manifest
+        run: |
+          docker manifest create $REPO:${{ github.event.inputs.tag || github.ref_name }} \
+            $REPO-arch:arch-amd64-${{github.event.inputs.tag || github.ref_name}} \
+            $REPO-arch:arch-arm64-${{github.event.inputs.tag || github.ref_name}}
+          docker manifest push $REPO:${{ github.event.inputs.tag || github.ref_name }}
+      - name: Create and push latest manifest
+        run: |
+          docker manifest create $REPO:latest \
+            $REPO-arch:arch-amd64-${{github.event.inputs.tag || github.ref_name}} \
+            $REPO-arch:arch-arm64-${{github.event.inputs.tag || github.ref_name}}
+          docker manifest push $REPO:latest
+        if: github.event.inputs.tag == 'test'
+
+  create-release:
+    runs-on: ubuntu-24.04
+    needs: [build-amd64, build-arm64]
+    permissions:
+      contents: write
+    steps:
+      - name: Download all artifacts
+        uses: actions/download-artifact@v4
+      - name: Create a release
+        uses: ncipollo/release-action@v1
+        with:
+          latest: true
+          allowUpdates: true
+          name: ${{ github.event.inputs.tag || github.ref_name }}
+          tag: ${{ github.event.inputs.tag || github.ref_name }}
+          generateReleaseNotes: false
+          omitBodyDuringUpdate: true
+          artifacts: "agent-*.tar/agent-*.tar"

.gitignore

@@ -1,6 +1,8 @@
 ui/node_modules
 ui/build
 ui/public/assets/env.js
+.DS_Store
+__debug*
 .idea
 machinery/www
 yarn.lock

Dockerfile

@@ -1,7 +1,7 @@
 
-ARG BASE_IMAGE_VERSION=70ec57e
+ARG BASE_IMAGE_VERSION=amd64-ddbe40e
 FROM kerberos/base:${BASE_IMAGE_VERSION} AS build-machinery
-LABEL AUTHOR=Kerberos.io
+LABEL AUTHOR=uug.ai
 
 ENV GOROOT=/usr/local/go
 ENV GOPATH=/go

Dockerfile.arm64

@@ -0,0 +1,138 @@
+
+ARG BASE_IMAGE_VERSION=arm64-ddbe40e
+FROM kerberos/base:${BASE_IMAGE_VERSION} AS build-machinery
+LABEL AUTHOR=uug.ai
+
+ENV GOROOT=/usr/local/go
+ENV GOPATH=/go
+ENV PATH=$GOPATH/bin:$GOROOT/bin:/usr/local/lib:$PATH
+ENV GOSUMDB=off
+
+##########################################
+# Installing some additional dependencies.
+
+RUN apt-get upgrade -y && apt-get update && apt-get install -y --fix-missing --no-install-recommends \
+	git build-essential cmake pkg-config unzip libgtk2.0-dev \
+	curl ca-certificates libcurl4-openssl-dev libssl-dev libjpeg62-turbo-dev && \
+	rm -rf /var/lib/apt/lists/*
+
+##############################################################################
+# Copy all the relevant source code in the Docker image, so we can build this.
+
+RUN mkdir -p /go/src/github.com/kerberos-io/agent
+COPY machinery /go/src/github.com/kerberos-io/agent/machinery
+RUN rm -rf /go/src/github.com/kerberos-io/agent/machinery/.env
+
+##################################################################
+# Get the latest commit hash, so we know which version we're running
+COPY .git /go/src/github.com/kerberos-io/agent/.git
+RUN cd /go/src/github.com/kerberos-io/agent/.git && git log --format="%H" -n 1 | head -c7 > /go/src/github.com/kerberos-io/agent/machinery/version
+RUN cat /go/src/github.com/kerberos-io/agent/machinery/version
+
+##################
+# Build Machinery
+
+RUN cd /go/src/github.com/kerberos-io/agent/machinery && \
+	go mod download && \
+	go build -tags timetzdata,netgo,osusergo --ldflags '-s -w -extldflags "-static -latomic"' main.go && \
+	mkdir -p /agent && \
+	mv main /agent && \
+	mv version /agent && \
+	mv data /agent && \
+	mkdir -p /agent/data/cloud && \
+	mkdir -p /agent/data/snapshots && \
+	mkdir -p /agent/data/log && \
+	mkdir -p /agent/data/recordings && \
+	mkdir -p /agent/data/capture-test && \
+	mkdir -p /agent/data/config
+
+####################################
+# Let's create a /dist folder containing just the files necessary for runtime.
+# Later, it will be copied as the / (root) of the output image.
+
+WORKDIR /dist
+RUN cp -r /agent ./
+
+####################################################################################
+# This will collect dependent libraries so they're later copied to the final image.
+
+RUN /dist/agent/main version
+
+FROM node:18.14.0-alpine3.16 AS build-ui
+
+RUN apk update && apk upgrade --available && sync
+
+########################
+# Build Web (React app)
+
+RUN mkdir -p /go/src/github.com/kerberos-io/agent/machinery/www
+COPY ui /go/src/github.com/kerberos-io/agent/ui
+RUN cd /go/src/github.com/kerberos-io/agent/ui && rm -rf yarn.lock && yarn config set network-timeout 300000 && \
+	yarn && yarn build
+
+####################################
+# Let's create a /dist folder containing just the files necessary for runtime.
+# Later, it will be copied as the / (root) of the output image.
+
+WORKDIR /dist
+RUN mkdir -p ./agent && cp -r /go/src/github.com/kerberos-io/agent/machinery/www ./agent/
+
+############################################
+# Publish main binary to GitHub release
+
+FROM alpine:latest
+
+############################
+# Protect by non-root user.
+
+RUN addgroup -S kerberosio && adduser -S agent -G kerberosio && addgroup agent video
+
+#################################
+# Copy files from previous images
+
+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/*
+
+##################
+# Try running agent
+
+RUN mv /agent/* /home/agent/
+RUN /home/agent/main version
+
+#######################
+# Make template config
+
+RUN cp /home/agent/data/config/config.json /home/agent/data/config.template.json
+
+###########################
+# Set permissions correctly
+
+RUN chown -R agent:kerberosio /home/agent/data
+RUN chown -R agent:kerberosio /home/agent/www
+
+###########################
+# Grant the necessary root capabilities to the process trying to bind to the privileged port
+RUN apk add libcap && setcap 'cap_net_bind_service=+ep' /home/agent/main
+
+###################
+# Run non-root user
+
+USER agent
+
+######################################
+# By default the app runs on port 80
+
+EXPOSE 80
+
+######################################
+# Check if agent is still running
+
+HEALTHCHECK CMD curl --fail http://localhost:80 || exit 1   
+
+###################################################
+# Leeeeettttt'ssss goooooo!!!
+# Run the shizzle from the right working directory.
+WORKDIR /home/agent
+CMD ["./main", "-action", "run", "-port", "80"]

machinery/main.go

@@ -101,31 +101,35 @@ func main() {
 	switch action {
 
 	case "version":
-		log.Log.Info("main.Main(): You are currrently running Kerberos Agent " + VERSION)
-
+		{
+			log.Log.Info("main.Main(): You are currrently running Kerberos Agent " + VERSION)
+		}
 	case "discover":
-		// Convert duration to int
-		timeout, err := time.ParseDuration(timeout + "ms")
-		if err != nil {
-			log.Log.Fatal("main.Main(): could not parse timeout: " + err.Error())
-			return
+		{
+			// Convert duration to int
+			timeout, err := time.ParseDuration(timeout + "ms")
+			if err != nil {
+				log.Log.Fatal("main.Main(): could not parse timeout: " + err.Error())
+				return
+			}
+			onvif.Discover(timeout)
 		}
-		onvif.Discover(timeout)
-
 	case "decrypt":
-		log.Log.Info("main.Main(): Decrypting: " + flag.Arg(0) + " with key: " + flag.Arg(1))
-		symmetricKey := []byte(flag.Arg(1))
+		{
+			log.Log.Info("main.Main(): Decrypting: " + flag.Arg(0) + " with key: " + flag.Arg(1))
+			symmetricKey := []byte(flag.Arg(1))
 
-		if symmetricKey == nil || len(symmetricKey) == 0 {
-			log.Log.Fatal("main.Main(): symmetric key should not be empty")
-			return
-		}
-		if len(symmetricKey) != 32 {
-			log.Log.Fatal("main.Main(): symmetric key should be 32 bytes")
-			return
-		}
+			if len(symmetricKey) == 0 {
+				log.Log.Fatal("main.Main(): symmetric key should not be empty")
+				return
+			}
+			if len(symmetricKey) != 32 {
+				log.Log.Fatal("main.Main(): symmetric key should be 32 bytes")
+				return
+			}
 
-		utils.Decrypt(flag.Arg(0), symmetricKey)
+			utils.Decrypt(flag.Arg(0), symmetricKey)
+		}
 
 	case "run":
 		{
@@ -213,6 +217,8 @@ func main() {
 			routers.StartWebserver(configDirectory, &configuration, &communication, &capture)
 		}
 	default:
-		log.Log.Error("main.Main(): Sorry I don't understand :(")
+		{
+			log.Log.Error("main.Main(): Sorry I don't understand :(")
+		}
 	}
 }

machinery/src/capture/main.go

@@ -280,7 +280,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					vpsNALUS := configuration.Config.Capture.IPCamera.VPSNALUs
 
 					// Create a video file, and set the dimensions.
-					mp4Video = video.NewMP4(fullName, spsNALUS, ppsNALUS, vpsNALUS)
+					mp4Video = video.NewMP4(fullName, spsNALUS, ppsNALUS, vpsNALUS, configuration.Config.Capture.MaxLengthRecording)
 					mp4Video.SetWidth(width)
 					mp4Video.SetHeight(height)
 
@@ -500,7 +500,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 				vpsNALUS := configuration.Config.Capture.IPCamera.VPSNALUs
 
 				// Create a video file, and set the dimensions.
-				mp4Video := video.NewMP4(fullName, spsNALUS, ppsNALUS, vpsNALUS)
+				mp4Video := video.NewMP4(fullName, spsNALUS, ppsNALUS, vpsNALUS, configuration.Config.Capture.MaxLengthRecording)
 				mp4Video.SetWidth(width)
 				mp4Video.SetHeight(height)
 

machinery/src/cloud/Cloud.go

@@ -804,6 +804,12 @@ func HandleLiveStreamHD(livestreamCursor *packets.QueueCursor, configuration *mo
 			streams, _ := rtspClient.GetStreams()
 			videoTrack := webrtc.NewVideoTrack(streams)
 			audioTrack := webrtc.NewAudioTrack(streams)
+
+			if videoTrack == nil && audioTrack == nil {
+				log.Log.Error("cloud.HandleLiveStreamHD(): failed to create both video and audio tracks")
+				return
+			}
+
 			go webrtc.WriteToTrack(livestreamCursor, configuration, communication, mqttClient, videoTrack, audioTrack, rtspClient)
 
 			if config.Capture.ForwardWebRTC == "true" {

machinery/src/routers/mqtt/main.go

@@ -123,7 +123,6 @@ func ConfigureMQTT(configDirectory string, configuration *models.Configuration,
 			opts.SetClientID(mqttClientID)
 			log.Log.Info("routers.mqtt.main.ConfigureMQTT(): Set ClientID " + mqttClientID)
 			rand.Seed(time.Now().UnixNano())
-			webrtc.CandidateArrays = make(map[string](chan string))
 
 			opts.OnConnect = func(c mqtt.Client) {
 				// We managed to connect to the MQTT broker, hurray!

machinery/src/video/mp4.go

@@ -22,52 +22,73 @@ import (
 
 var LastPTS uint64 = 0 // Last PTS for the current segment
 
+// FragmentDurationMs is the target duration for each fragment in milliseconds.
+// Fragments will be flushed at the first keyframe after this duration has elapsed,
+// resulting in ~3 second fragments (assuming a typical GOP interval).
+const FragmentDurationMs = 3000
+
 type MP4 struct {
 	// FileName is the name of the file
-	FileName           string
-	width              int
-	height             int
-	Segments           []*mp4ff.MediaSegment // List of media segments
-	Segment            *mp4ff.MediaSegment
-	MultiTrackFragment *mp4ff.Fragment
-	TrackIDs           []uint32
-	FileWriter         *os.File
-	Writer             *bufio.Writer
-	SegmentCount       int
-	SampleCount        int
-	StartPTS           uint64
-	VideoTotalDuration uint64
-	AudioTotalDuration uint64
-	AudioPTS           uint64
-	Start              bool
-	SPSNALUs           [][]byte // SPS NALUs for H264
-	PPSNALUs           [][]byte // PPS NALUs for H264
-	VPSNALUs           [][]byte // VPS NALUs for H264
-	FreeBoxSize        int64
-	MoofBoxes          int64             // Number of moof boxes in the file
-	MoofBoxSizes       []int64           // Sizes of each moof box
-	StartTime          uint64            // Start time of the MP4 file
-	VideoTrackName     string            // Name of the video track
-	VideoTrack         int               // Track ID for the video track
-	AudioTrackName     string            // Name of the audio track
-	AudioTrack         int               // Track ID for the audio track
-	VideoFullSample    *mp4ff.FullSample // Full sample for video track
-	AudioFullSample    *mp4ff.FullSample // Full sample for audio track
-	LastAudioSampleDTS uint64            // Last PTS for audio sample
-	LastVideoSampleDTS uint64            // Last PTS for video sample
-	SampleType         string            // Type of the sample (e.g., "video", "audio", "subtitle")
+	FileName            string
+	width               int
+	height              int
+	Segments            []*mp4ff.MediaSegment // List of media segments
+	Segment             *mp4ff.MediaSegment
+	MultiTrackFragment  *mp4ff.Fragment
+	TrackIDs            []uint32
+	FileWriter          *os.File
+	Writer              *bufio.Writer
+	SegmentCount        int
+	SampleCount         int
+	StartPTS            uint64
+	VideoTotalDuration  uint64
+	AudioTotalDuration  uint64
+	AudioPTS            uint64
+	Start               bool
+	SPSNALUs            [][]byte // SPS NALUs for H264
+	PPSNALUs            [][]byte // PPS NALUs for H264
+	VPSNALUs            [][]byte // VPS NALUs for H264
+	FreeBoxSize         int64
+	FragmentStartRawPTS uint64            // Raw PTS for timing when to flush fragments
+	FragmentStartDTS    uint64            // Accumulated VideoTotalDuration at fragment start (matches tfdt)
+	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
+	SegmentBaseDecTimes []uint64          // Base decode time of each segment
+	StartTime           uint64            // Start time of the MP4 file
+	VideoTrackName      string            // Name of the video track
+	VideoTrack          int               // Track ID for the video track
+	AudioTrackName      string            // Name of the audio track
+	AudioTrack          int               // Track ID for the audio track
+	VideoFullSample     *mp4ff.FullSample // Full sample for video track
+	AudioFullSample     *mp4ff.FullSample // Full sample for audio track
+	LastAudioSampleDTS  uint64            // Last PTS for audio sample
+	LastVideoSampleDTS  uint64            // Last PTS for video sample
+	SampleType          string            // Type of the sample (e.g., "video", "audio", "subtitle")
 }
 
-// NewMP4 creates a new MP4 object
-func NewMP4(fileName string, spsNALUs [][]byte, ppsNALUs [][]byte, vpsNALUs [][]byte) *MP4 {
+// NewMP4 creates a new MP4 object.
+// maxDurationSec is the maximum expected recording duration in seconds,
+// used to calculate the free-box placeholder size for ftyp+moov+sidx.
+func NewMP4(fileName string, spsNALUs [][]byte, ppsNALUs [][]byte, vpsNALUs [][]byte, maxDurationSec int64) *MP4 {
 
-	init := mp4ff.NewMP4Init()
-
-	// Add a free box to the init segment
-	// Prepend a free box to the init segment with a size of 4096 bytes, so we can overwrite it later with the actual init segment.
-	freeBoxSize := 4096
+	// Calculate the placeholder size needed at the start of the file.
+	// Components:
+	//   ftyp:  ~32 bytes
+	//   moov:  ~1500 bytes (mvhd + mvex + video trak + audio trak + UUID)
+	//   sidx:  24 bytes fixed + 12 bytes per segment reference
+	// Segments are ~FragmentDurationMs each, so:
+	//   numSegments = ceil(maxDurationSec * 1000 / FragmentDurationMs) + 1 (safety margin)
+	//   sidxSize    = 24 + 12 * numSegments
+	baseSize := int64(2560) // ftyp + moov + extra headroom for large UUID signatures
+	numSegments := int64(0)
+	if maxDurationSec > 0 {
+		// Use integer ceiling division to avoid underestimating the number of segments.
+		numSegments = ((maxDurationSec*1000)+FragmentDurationMs-1)/FragmentDurationMs + 1
+	}
+	sidxSize := int64(24 + 12*numSegments)
+	freeBoxSize := int(baseSize + sidxSize)
 	free := mp4ff.NewFreeBox(make([]byte, freeBoxSize))
-	init.AddChild(free)
 
 	// Create a writer
 	ofd, err := os.Create(fileName)
@@ -77,16 +98,15 @@ func NewMP4(fileName string, spsNALUs [][]byte, ppsNALUs [][]byte, vpsNALUs [][]
 	// Create a buffered writer
 	bufferedWriter := bufio.NewWriterSize(ofd, 64*1024) // 64KB buffer
 
-	// We will write the empty init segment to the file
-	// so we can overwrite it later with the actual init segment.
-	err = init.Encode(bufferedWriter)
+	// Write the free box placeholder at the start of the file
+	err = free.Encode(bufferedWriter)
 	if err != nil {
 	}
 
 	return &MP4{
 		FileName:    fileName,
 		StartTime:   uint64(time.Now().Unix()),
-		FreeBoxSize: int64(freeBoxSize),
+		FreeBoxSize: int64(freeBoxSize) + 8, // payload + 8 byte box header
 		FileWriter:  ofd,
 		Writer:      bufferedWriter,
 		SPSNALUs:    spsNALUs,
@@ -134,38 +154,57 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
 
 	if isKeyframe {
 
-		// Write the segment to the file
+		// Determine whether to start a new fragment.
+		// We only flush at a keyframe boundary once at least FragmentDurationMs
+		// of content has been accumulated, resulting in ~3 second fragments.
+		elapsed := uint64(0)
 		if mp4.Start {
-			mp4.MoofBoxes = mp4.MoofBoxes + 1
-			mp4.MoofBoxSizes = append(mp4.MoofBoxSizes, int64(mp4.Segment.Size()))
-			err := mp4.Segment.Encode(mp4.Writer)
-			if err != nil {
-				log.Log.Error("mp4.AddSampleToTrack(): error encoding segment: " + err.Error())
+			elapsed = pts - mp4.FragmentStartRawPTS
+		}
+		shouldFlush := !mp4.Start || elapsed >= FragmentDurationMs
+
+		if shouldFlush {
+			// Write the previous segment to the file
+			if mp4.Start {
+				mp4.MoofBoxes = mp4.MoofBoxes + 1
+				mp4.MoofBoxSizes = append(mp4.MoofBoxSizes, int64(mp4.Segment.Size()))
+				// Track the segment's duration and base decode time for sidx.
+				// Use accumulated VideoTotalDuration which matches the tfdt values
+				// in the trun boxes, NOT raw PTS from the camera.
+				segDuration := mp4.VideoTotalDuration - mp4.FragmentStartDTS
+				mp4.SegmentDurations = append(mp4.SegmentDurations, segDuration)
+				mp4.SegmentBaseDecTimes = append(mp4.SegmentBaseDecTimes, mp4.FragmentStartDTS)
+				err := mp4.Segment.Encode(mp4.Writer)
+				if err != nil {
+					log.Log.Error("mp4.AddSampleToTrack(): error encoding segment: " + err.Error())
+				}
+				mp4.Segments = append(mp4.Segments, mp4.Segment)
 			}
-			mp4.Segments = append(mp4.Segments, mp4.Segment)
+
+			mp4.Start = true
+
+			// Increment the segment count
+			mp4.SegmentCount = mp4.SegmentCount + 1
+
+			// Create a new media segment
+			seg := mp4ff.NewMediaSegment()
+
+			// Create a video fragment
+			multiTrackFragment, err := mp4ff.CreateMultiTrackFragment(uint32(mp4.SegmentCount), mp4.TrackIDs)
+			if err != nil {
+				log.Log.Error("mp4.AddSampleToTrack(): error creating multi track fragment: " + err.Error())
+			}
+			mp4.MultiTrackFragment = multiTrackFragment
+			seg.AddFragment(multiTrackFragment)
+
+			// Set to MP4 struct
+			mp4.Segment = seg
+
+			// Set the start PTS for the next segment
+			mp4.StartPTS = pts
+			mp4.FragmentStartRawPTS = pts
+			mp4.FragmentStartDTS = mp4.VideoTotalDuration
 		}
-
-		mp4.Start = true
-
-		// Increment the segment count
-		mp4.SegmentCount = mp4.SegmentCount + 1
-
-		// Create a new media segment
-		seg := mp4ff.NewMediaSegment()
-
-		// Create a video fragment
-		multiTrackFragment, err := mp4ff.CreateMultiTrackFragment(uint32(mp4.SegmentCount), mp4.TrackIDs) // Assuming 1 for video track and 2 for audio track
-		if err != nil {
-			log.Log.Error("mp4.AddSampleToTrack(): error creating multi track fragment: " + err.Error())
-		}
-		mp4.MultiTrackFragment = multiTrackFragment
-		seg.AddFragment(multiTrackFragment)
-
-		// Set to MP4 struct
-		mp4.Segment = seg
-
-		// Set the start PTS for the next segment
-		mp4.StartPTS = pts
 	}
 
 	if mp4.Start {
@@ -263,8 +302,60 @@ func (mp4 *MP4) Close(config *models.Config) {
 		log.Log.Error("mp4.Close(): no video or audio samples added, cannot create MP4 file")
 	}
 
+	// Add final pending samples before closing
+	if mp4.Segment != nil {
+		// Add final video sample if pending
+		if mp4.VideoFullSample != nil {
+			duration := mp4.LastVideoSampleDTS
+			if duration == 0 {
+				duration = 33 // Default ~30fps frame duration
+			}
+			mp4.VideoTotalDuration += duration
+			mp4.VideoFullSample.DecodeTime = mp4.VideoTotalDuration - duration
+			mp4.VideoFullSample.Sample.Dur = uint32(duration)
+			err := mp4.MultiTrackFragment.AddFullSampleToTrack(*mp4.VideoFullSample, uint32(mp4.VideoTrack))
+			if err != nil {
+				log.Log.Error("mp4.Close(): error adding final video sample: " + err.Error())
+			}
+			mp4.VideoFullSample = nil
+		}
+
+		// Add final audio sample if pending
+		if mp4.AudioFullSample != nil && mp4.AudioTrack > 0 {
+			SplitAACFrame(mp4.AudioFullSample.Data, func(started bool, aac []byte) {
+				sampleToAdd := *mp4.AudioFullSample
+				dts := mp4.LastAudioSampleDTS
+				if dts == 0 {
+					dts = 1024 // Default AAC frame duration
+				}
+				mp4.AudioTotalDuration += dts
+				mp4.AudioPTS += dts
+				sampleToAdd.Data = aac[7:]
+				sampleToAdd.DecodeTime = mp4.AudioPTS - dts
+				sampleToAdd.Sample.Dur = uint32(dts)
+				sampleToAdd.Sample.Size = uint32(len(aac[7:]))
+				err := mp4.MultiTrackFragment.AddFullSampleToTrack(sampleToAdd, uint32(mp4.AudioTrack))
+				if err != nil {
+					log.Log.Error("mp4.Close(): error adding final audio sample: " + err.Error())
+				}
+			})
+			mp4.AudioFullSample = nil
+		}
+	}
+
 	// Encode the last segment
 	if mp4.Segment != nil {
+		// Track the last segment's size, duration and base decode time.
+		// Use accumulated VideoTotalDuration which matches tfdt values.
+		mp4.MoofBoxes = mp4.MoofBoxes + 1
+		mp4.MoofBoxSizes = append(mp4.MoofBoxSizes, int64(mp4.Segment.Size()))
+		lastSegDuration := mp4.VideoTotalDuration - mp4.FragmentStartDTS
+		if lastSegDuration == 0 {
+			lastSegDuration = mp4.LastVideoSampleDTS
+		}
+		mp4.SegmentDurations = append(mp4.SegmentDurations, lastSegDuration)
+		mp4.SegmentBaseDecTimes = append(mp4.SegmentBaseDecTimes, mp4.FragmentStartDTS)
+
 		err := mp4.Segment.Encode(mp4.Writer)
 		if err != nil {
 			log.Log.Error("mp4.Close(): error encoding last segment: " + err.Error())
@@ -272,10 +363,14 @@ func (mp4 *MP4) Close(config *models.Config) {
 	}
 
 	mp4.Writer.Flush()
-	defer mp4.FileWriter.Close()
+	// Ensure all segment data is on disk before we overwrite the placeholder at offset 0.
+	if err := mp4.FileWriter.Sync(); err != nil {
+		log.Log.Error("mp4.Close(): error syncing file: " + err.Error())
+	}
 
 	// Now we have all the moof and mdat boxes written to the file.
-	// We can now generate the ftyp and moov boxes, and replace it with the free box we added earlier (size of 2048 bytes).
+	// We build the ftyp + moov init segment and write it at the start,
+	// overwriting the free box placeholder we reserved in NewMP4.
 	init := mp4ff.NewMP4Init()
 
 	// Create a new ftyp box
@@ -316,8 +411,10 @@ func (mp4 *MP4) Close(config *models.Config) {
 		if err != nil {
 		}
 		init.Moov.Traks[0].Tkhd.Duration = mp4.VideoTotalDuration
+		init.Moov.Traks[0].Tkhd.Width = mp4ff.Fixed32(uint32(mp4.width) << 16)
+		init.Moov.Traks[0].Tkhd.Height = mp4ff.Fixed32(uint32(mp4.height) << 16)
 		init.Moov.Traks[0].Mdia.Hdlr.Name = "agent " + utils.VERSION
-		//init.Moov.Traks[0].Mdia.Mdhd.Duration = mp4.VideoTotalDuration
+		init.Moov.Traks[0].Mdia.Mdhd.Duration = mp4.VideoTotalDuration
 	case "H265", "HVC1":
 		init.AddEmptyTrack(videoTimescale, "video", "und")
 		includePS := true
@@ -325,8 +422,10 @@ func (mp4 *MP4) Close(config *models.Config) {
 		if err != nil {
 		}
 		init.Moov.Traks[0].Tkhd.Duration = mp4.VideoTotalDuration
+		init.Moov.Traks[0].Tkhd.Width = mp4ff.Fixed32(uint32(mp4.width) << 16)
+		init.Moov.Traks[0].Tkhd.Height = mp4ff.Fixed32(uint32(mp4.height) << 16)
 		init.Moov.Traks[0].Mdia.Hdlr.Name = "agent " + utils.VERSION
-		//init.Moov.Traks[0].Mdia.Mdhd.Duration = mp4.VideoTotalDuration
+		init.Moov.Traks[0].Mdia.Mdhd.Duration = mp4.VideoTotalDuration
 	}
 
 	// Try adding audio track if available
@@ -345,7 +444,7 @@ func (mp4 *MP4) Close(config *models.Config) {
 		}
 		init.Moov.Traks[1].Tkhd.Duration = mp4.AudioTotalDuration
 		init.Moov.Traks[1].Mdia.Hdlr.Name = "agent " + utils.VERSION
-		//init.Moov.Traks[1].Mdia.Mdhd.Duration = mp4.AudioTotalDuration
+		init.Moov.Traks[1].Mdia.Mdhd.Duration = mp4.AudioTotalDuration
 	}
 
 	// Try adding subtitle track if available
@@ -423,126 +522,94 @@ func (mp4 *MP4) Close(config *models.Config) {
 		}
 	}
 
-	// We will also calculate the SIDX box, which is a segment index box that contains information about the segments in the file.
-	// This is useful for seeking in the file, and for streaming the file.
-	/*sidx := &mp4ff.SidxBox{
-		Version:                  0,
-		Flags:                    0,
-		ReferenceID:              0,
-		Timescale:                videoTimescale,
-		EarliestPresentationTime: 0,
-		FirstOffset:              0,
-		SidxRefs:                 make([]mp4ff.SidxRef, 0),
-	}
-	referenceTrak := init.Moov.Trak
-	trex, ok := init.Moov.Mvex.GetTrex(referenceTrak.Tkhd.TrackID)
-	if !ok {
-		// We have an issue.
+	// Build a Segment Index (sidx) box so players can seek directly to any
+	// fragment without scanning the entire file.
+	if len(mp4.SegmentDurations) > 0 {
+		sidx := &mp4ff.SidxBox{
+			Version:                  1,
+			Flags:                    0,
+			ReferenceID:              uint32(mp4.VideoTrack),
+			Timescale:                videoTimescale,
+			EarliestPresentationTime: 0,
+			FirstOffset:              0,
+			SidxRefs:                 make([]mp4ff.SidxRef, 0, len(mp4.SegmentDurations)),
+		}
+		for i, dur := range mp4.SegmentDurations {
+			sidx.SidxRefs = append(sidx.SidxRefs, mp4ff.SidxRef{
+				ReferenceType:      0, // media reference
+				ReferencedSize:     uint32(mp4.MoofBoxSizes[i]),
+				SubSegmentDuration: uint32(dur),
+				StartsWithSAP:      1,
+				SAPType:            1,
+			})
+		}
+		init.AddChild(sidx)
 	}
 
-	segDatas, err := findSegmentData(mp4.Segments, referenceTrak, trex)
-	if err != nil {
-		// We have an issue.
-	}
-	fillSidx(sidx, referenceTrak, segDatas, true)
-
-	// Add the SIDX box to the moov box
-	init.AddChild(sidx)*/
-
-	// Get a bit slice writer for the init segment
-	// Get a byte buffer of FreeBoxSize bytes to write the init segment
-	buffer := bytes.NewBuffer(make([]byte, 0))
-	init.Encode(buffer)
-
-	// The first FreeBoxSize bytes of the file is a free box, so we can read it and replace it with the moov box.
-	// The init box might not be FreeBoxSize bytes, so we need to read the first FreeBoxSize bytes and then replace it with the moov box.
-	// while the remaining bytes are for a new free box.
-	// Write the init segment at the beginning of the file, replacing the free box
-	if _, err := mp4.FileWriter.WriteAt(buffer.Bytes(), 0); err != nil {
+	// Encode the ftyp + moov + sidx into a buffer to measure the total size.
+	// Then compute the correct sidx.FirstOffset (the gap between the end of
+	// the sidx box and the first moof, occupied by the trailing free box)
+	// and re-encode with the corrected value.
+	var initBuf bytes.Buffer
+	if err := init.Encode(&initBuf); err != nil {
+		log.Log.Error("mp4.Close(): error encoding init segment: " + err.Error())
 	}
 
-	// Calculate the remaining size for the free box
-	remainingSize := mp4.FreeBoxSize - int64(buffer.Len())
-	if remainingSize > 0 {
-		newFreeBox := mp4ff.NewFreeBox(make([]byte, remainingSize))
+	initSize := int64(initBuf.Len())
+
+	// The sidx.FirstOffset is defined as the distance (in bytes) from the
+	// anchor point (first byte after the sidx box) to the first byte of
+	// the first referenced moof/mdat. Since sidx is the last box in init,
+	// the anchor point is at initSize, and the first moof is at FreeBoxSize.
+	if len(mp4.SegmentDurations) > 0 {
+		if mp4.FreeBoxSize < initSize {
+			// Avoid computing a negative offset and wrapping it to uint64.
+			log.Log.Error("mp4.Close(): FreeBoxSize is smaller than initSize; skipping sidx FirstOffset adjustment")
+		} else {
+			firstOffset := uint64(mp4.FreeBoxSize - initSize)
+			// Find the sidx we added and update its FirstOffset
+			for _, child := range init.Children {
+				if sidxBox, ok := child.(*mp4ff.SidxBox); ok {
+					sidxBox.FirstOffset = firstOffset
+					break
+				}
+			}
+			// Re-encode with the corrected FirstOffset (same size, no layout change)
+			initBuf.Reset()
+			if err := init.Encode(&initBuf); err != nil {
+				log.Log.Error("mp4.Close(): error re-encoding init segment: " + err.Error())
+			}
+			initSize = int64(initBuf.Len())
+		}
+	}
+
+	if initSize > mp4.FreeBoxSize {
+		log.Log.Error(fmt.Sprintf("mp4.Close(): init segment (%d bytes) exceeds reserved space (%d bytes), file may be corrupt", initSize, mp4.FreeBoxSize))
+	}
+
+	// Write the init segment at the beginning of the file, overwriting the free box placeholder.
+	if _, err := mp4.FileWriter.WriteAt(initBuf.Bytes(), 0); err != nil {
+		log.Log.Error("mp4.Close(): error writing init segment: " + err.Error())
+	}
+
+	// Fill any remaining reserved space with a new (smaller) free box so
+	// the byte offsets of the moof/mdat boxes that follow are preserved.
+	remainingSize := mp4.FreeBoxSize - initSize
+	if remainingSize >= 8 { // minimum box size is 8 bytes (header only)
+		newFree := mp4ff.NewFreeBox(make([]byte, remainingSize-8))
 		var freeBuf bytes.Buffer
-		if err := newFreeBox.Encode(&freeBuf); err != nil {
+		if err := newFree.Encode(&freeBuf); err != nil {
 			log.Log.Error("mp4.Close(): error encoding free box: " + err.Error())
 		}
-		if _, err := mp4.FileWriter.WriteAt(freeBuf.Bytes(), int64(buffer.Len())); err != nil {
+		if _, err := mp4.FileWriter.WriteAt(freeBuf.Bytes(), initSize); err != nil {
 			log.Log.Error("mp4.Close(): error writing free box: " + err.Error())
 		}
 	}
-}
 
-type segData struct {
-	startPos         uint64
-	presentationTime uint64
-	baseDecodeTime   uint64
-	dur              uint32
-	size             uint32
-}
-
-func fillSidx(sidx *mp4ff.SidxBox, refTrak *mp4ff.TrakBox, segDatas []segData, nonZeroEPT bool) {
-	ept := uint64(0)
-	if nonZeroEPT {
-		ept = segDatas[0].presentationTime
+	if err := mp4.FileWriter.Sync(); err != nil {
+		log.Log.Error("mp4.Close(): error syncing file: " + err.Error())
 	}
-	sidx.Version = 1
-	sidx.Timescale = refTrak.Mdia.Mdhd.Timescale
-	sidx.ReferenceID = 1
-	sidx.EarliestPresentationTime = ept
-	sidx.FirstOffset = 0
-	sidx.SidxRefs = make([]mp4ff.SidxRef, 0, len(segDatas))
-
-	for _, segData := range segDatas {
-		size := segData.size
-		sidx.SidxRefs = append(sidx.SidxRefs, mp4ff.SidxRef{
-			ReferencedSize:     size,
-			SubSegmentDuration: segData.dur,
-			StartsWithSAP:      1,
-			SAPType:            1,
-		})
-	}
-}
-
-// findSegmentData returns a slice of segment media data using a reference track.
-func findSegmentData(segs []*mp4ff.MediaSegment, refTrak *mp4ff.TrakBox, trex *mp4ff.TrexBox) ([]segData, error) {
-	segDatas := make([]segData, 0, len(segs))
-	for _, seg := range segs {
-		var firstCompositionTimeOffest int64
-		dur := uint32(0)
-		var baseTime uint64
-		for fIdx, frag := range seg.Fragments {
-			for _, traf := range frag.Moof.Trafs {
-				tfhd := traf.Tfhd
-				if tfhd.TrackID == refTrak.Tkhd.TrackID { // Find track that gives sidx time values
-					if fIdx == 0 {
-						baseTime = traf.Tfdt.BaseMediaDecodeTime()
-					}
-					for i, trun := range traf.Truns {
-						trun.AddSampleDefaultValues(tfhd, trex)
-						samples := trun.GetSamples()
-						for j, sample := range samples {
-							if fIdx == 0 && i == 0 && j == 0 {
-								firstCompositionTimeOffest = int64(sample.CompositionTimeOffset)
-							}
-							dur += sample.Dur
-						}
-					}
-				}
-			}
-		}
-		sd := segData{
-			startPos:         seg.StartPos,
-			presentationTime: uint64(int64(baseTime) + firstCompositionTimeOffest),
-			baseDecodeTime:   baseTime,
-			dur:              dur,
-			size:             uint32(seg.Size()),
-		}
-		segDatas = append(segDatas, sd)
-	}
-	return segDatas, nil
+	mp4.FileWriter.Close()
 }
 
 // annexBToLengthPrefixed converts Annex B formatted H264 data (with start codes)

machinery/src/webrtc/main.go

@@ -1,6 +1,7 @@
 package webrtc
 
 import (
+	"context"
 	"encoding/base64"
 	"encoding/json"
 	"io"
@@ -22,13 +23,104 @@ import (
 	pionMedia "github.com/pion/webrtc/v4/pkg/media"
 )
 
-var (
-	CandidatesMutex     sync.Mutex
-	CandidateArrays     map[string](chan string)
-	peerConnectionCount int64
-	peerConnections     map[string]*pionWebRTC.PeerConnection
+const (
+	// Channel buffer sizes
+	candidateChannelBuffer = 100
+	rtcpBufferSize         = 1500
+
+	// Timeouts and intervals
+	keepAliveTimeout = 15 * time.Second
+	defaultTimeout   = 10 * time.Second
+
+	// Track identifiers
+	trackStreamID = "kerberos-stream"
 )
 
+// ConnectionManager manages WebRTC peer connections in a thread-safe manner
+type ConnectionManager struct {
+	mu                  sync.RWMutex
+	candidateChannels   map[string]chan string
+	peerConnections     map[string]*peerConnectionWrapper
+	peerConnectionCount int64
+}
+
+// peerConnectionWrapper wraps a peer connection with additional metadata
+type peerConnectionWrapper struct {
+	conn      *pionWebRTC.PeerConnection
+	cancelCtx context.CancelFunc
+	done      chan struct{}
+}
+
+var globalConnectionManager = NewConnectionManager()
+
+// NewConnectionManager creates a new connection manager
+func NewConnectionManager() *ConnectionManager {
+	return &ConnectionManager{
+		candidateChannels: make(map[string]chan string),
+		peerConnections:   make(map[string]*peerConnectionWrapper),
+	}
+}
+
+// GetOrCreateCandidateChannel gets or creates a candidate channel for a session
+func (cm *ConnectionManager) GetOrCreateCandidateChannel(sessionKey string) chan string {
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+
+	if ch, exists := cm.candidateChannels[sessionKey]; exists {
+		return ch
+	}
+
+	ch := make(chan string, candidateChannelBuffer)
+	cm.candidateChannels[sessionKey] = ch
+	return ch
+}
+
+// CloseCandidateChannel safely closes and removes a candidate channel
+func (cm *ConnectionManager) CloseCandidateChannel(sessionKey string) {
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+
+	if ch, exists := cm.candidateChannels[sessionKey]; exists {
+		close(ch)
+		delete(cm.candidateChannels, sessionKey)
+	}
+}
+
+// AddPeerConnection adds a peer connection to the manager
+func (cm *ConnectionManager) AddPeerConnection(sessionID string, wrapper *peerConnectionWrapper) {
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+	cm.peerConnections[sessionID] = wrapper
+}
+
+// RemovePeerConnection removes a peer connection from the manager
+func (cm *ConnectionManager) RemovePeerConnection(sessionID string) {
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+
+	if wrapper, exists := cm.peerConnections[sessionID]; exists {
+		if wrapper.cancelCtx != nil {
+			wrapper.cancelCtx()
+		}
+		delete(cm.peerConnections, sessionID)
+	}
+}
+
+// GetPeerConnectionCount returns the current count of active peer connections
+func (cm *ConnectionManager) GetPeerConnectionCount() int64 {
+	return atomic.LoadInt64(&cm.peerConnectionCount)
+}
+
+// IncrementPeerCount atomically increments the peer connection count
+func (cm *ConnectionManager) IncrementPeerCount() int64 {
+	return atomic.AddInt64(&cm.peerConnectionCount, 1)
+}
+
+// DecrementPeerCount atomically decrements the peer connection count
+func (cm *ConnectionManager) DecrementPeerCount() int64 {
+	return atomic.AddInt64(&cm.peerConnectionCount, -1)
+}
+
 type WebRTC struct {
 	Name                  string
 	StunServers           []string
@@ -46,7 +138,7 @@ func CreateWebRTC(name string, stunServers []string, turnServers []string, turnS
 		TurnServers:           turnServers,
 		TurnServersUsername:   turnServersUsername,
 		TurnServersCredential: turnServersCredential,
-		Timer:                 time.NewTimer(time.Second * 10),
+		Timer:                 time.NewTimer(defaultTimeout),
 	}
 }
 
@@ -68,19 +160,14 @@ func (w WebRTC) CreateOffer(sd []byte) pionWebRTC.SessionDescription {
 }
 
 func RegisterCandidates(key string, candidate models.ReceiveHDCandidatesPayload) {
-	// Set lock
-	CandidatesMutex.Lock()
-	_, ok := CandidateArrays[key]
-	if !ok {
-		CandidateArrays[key] = make(chan string, 100)
-	}
-	log.Log.Info("webrtc.main.HandleReceiveHDCandidates(): " + candidate.Candidate)
+	ch := globalConnectionManager.GetOrCreateCandidateChannel(key)
+
+	log.Log.Info("webrtc.main.RegisterCandidates(): " + candidate.Candidate)
 	select {
-	case CandidateArrays[key] <- candidate.Candidate:
+	case ch <- candidate.Candidate:
 	default:
-		log.Log.Info("webrtc.main.HandleReceiveHDCandidates(): channel is full.")
+		log.Log.Info("webrtc.main.RegisterCandidates(): channel is full, dropping candidate")
 	}
-	CandidatesMutex.Unlock()
 }
 
 func RegisterDefaultInterceptors(mediaEngine *pionWebRTC.MediaEngine, interceptorRegistry *interceptor.Registry) error {
@@ -107,12 +194,7 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 
 	// We create a channel which will hold the candidates for this session.
 	sessionKey := config.Key + "/" + handshake.SessionID
-	CandidatesMutex.Lock()
-	_, ok := CandidateArrays[sessionKey]
-	if !ok {
-		CandidateArrays[sessionKey] = make(chan string, 100)
-	}
-	CandidatesMutex.Unlock()
+	candidateChannel := globalConnectionManager.GetOrCreateCandidateChannel(sessionKey)
 
 	// Set variables
 	hubKey := handshake.HubKey
@@ -178,81 +260,126 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 
 		if err == nil && peerConnection != nil {
 
-			var videoSender *pionWebRTC.RTPSender = nil
-			if videoSender, err = peerConnection.AddTrack(videoTrack); err != nil {
-				log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while adding video track: " + err.Error())
+			// Create context for this connection
+			ctx, cancel := context.WithCancel(context.Background())
+			wrapper := &peerConnectionWrapper{
+				conn:      peerConnection,
+				cancelCtx: cancel,
+				done:      make(chan struct{}),
 			}
+
+			var videoSender *pionWebRTC.RTPSender = nil
+			if videoTrack != nil {
+				if videoSender, err = peerConnection.AddTrack(videoTrack); err != nil {
+					log.Log.Error("webrtc.main.InitializeWebRTCConnection(): error adding video track: " + err.Error())
+					cancel()
+					return
+				}
+			} else {
+				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): video track is nil, skipping video")
+			}
+
 			// Read incoming RTCP packets
 			// Before these packets are returned they are processed by interceptors. For things
 			// like NACK this needs to be called.
-			go func() {
-				rtcpBuf := make([]byte, 1500)
-				for {
-					if _, _, rtcpErr := videoSender.Read(rtcpBuf); rtcpErr != nil {
-						return
+			if videoSender != nil {
+				go func() {
+					defer func() {
+						log.Log.Info("webrtc.main.InitializeWebRTCConnection(): video RTCP reader stopped")
+					}()
+					rtcpBuf := make([]byte, rtcpBufferSize)
+					for {
+						select {
+						case <-ctx.Done():
+							return
+						default:
+							if _, _, rtcpErr := videoSender.Read(rtcpBuf); rtcpErr != nil {
+								return
+							}
+						}
 					}
-				}
-			}()
+				}()
+			}
 
 			var audioSender *pionWebRTC.RTPSender = nil
-			if audioSender, err = peerConnection.AddTrack(audioTrack); err != nil {
-				log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while adding audio track: " + err.Error())
-			} // Read incoming RTCP packets
+			if audioTrack != nil {
+				if audioSender, err = peerConnection.AddTrack(audioTrack); err != nil {
+					log.Log.Error("webrtc.main.InitializeWebRTCConnection(): error adding audio track: " + err.Error())
+					cancel()
+					return
+				}
+			} else {
+				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): audio track is nil, skipping audio")
+			}
+
+			// Read incoming RTCP packets
 			// Before these packets are returned they are processed by interceptors. For things
 			// like NACK this needs to be called.
-			go func() {
-				rtcpBuf := make([]byte, 1500)
-				for {
-					if _, _, rtcpErr := audioSender.Read(rtcpBuf); rtcpErr != nil {
-						return
+			if audioSender != nil {
+				go func() {
+					defer func() {
+						log.Log.Info("webrtc.main.InitializeWebRTCConnection(): audio RTCP reader stopped")
+					}()
+					rtcpBuf := make([]byte, rtcpBufferSize)
+					for {
+						select {
+						case <-ctx.Done():
+							return
+						default:
+							if _, _, rtcpErr := audioSender.Read(rtcpBuf); rtcpErr != nil {
+								return
+							}
+						}
 					}
-				}
-			}()
+				}()
+			}
 
 			peerConnection.OnConnectionStateChange(func(connectionState pionWebRTC.PeerConnectionState) {
-				if connectionState == pionWebRTC.PeerConnectionStateDisconnected || connectionState == pionWebRTC.PeerConnectionStateClosed {
-					// Set lock
-					CandidatesMutex.Lock()
-					atomic.AddInt64(&peerConnectionCount, -1)
-					_, ok := CandidateArrays[sessionKey]
-					if ok {
-						close(CandidateArrays[sessionKey])
-						delete(CandidateArrays, sessionKey)
-					}
-					// Not really needed.
-					//senders := peerConnection.GetSenders()
-					//for _, sender := range senders {
-					//	if err := peerConnection.RemoveTrack(sender); err != nil {
-					//		log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while removing track: " + err.Error())
-					//	}
-					//}
-					if err := peerConnection.Close(); err != nil {
-						log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while closing peer connection: " + err.Error())
-					}
-					peerConnections[handshake.SessionID] = nil
-					delete(peerConnections, handshake.SessionID)
-					CandidatesMutex.Unlock()
-				} else if connectionState == pionWebRTC.PeerConnectionStateConnected {
-					CandidatesMutex.Lock()
-					atomic.AddInt64(&peerConnectionCount, 1)
-					CandidatesMutex.Unlock()
-				} else if connectionState == pionWebRTC.PeerConnectionStateFailed {
-					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): ICEConnectionStateFailed")
-				}
 				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): connection state changed to: " + connectionState.String())
-				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): Number of peers connected (" + strconv.FormatInt(peerConnectionCount, 10) + ")")
+
+				switch connectionState {
+				case pionWebRTC.PeerConnectionStateDisconnected, pionWebRTC.PeerConnectionStateClosed:
+					count := globalConnectionManager.DecrementPeerCount()
+					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): Peer disconnected. Active peers: " + string(rune(count)))
+
+					// Clean up resources
+					globalConnectionManager.CloseCandidateChannel(sessionKey)
+
+					if err := peerConnection.Close(); err != nil {
+						log.Log.Error("webrtc.main.InitializeWebRTCConnection(): error closing peer connection: " + err.Error())
+					}
+
+					globalConnectionManager.RemovePeerConnection(handshake.SessionID)
+					close(wrapper.done)
+
+				case pionWebRTC.PeerConnectionStateConnected:
+					count := globalConnectionManager.IncrementPeerCount()
+					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): Peer connected. Active peers: " + string(rune(count)))
+
+				case pionWebRTC.PeerConnectionStateFailed:
+					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): ICE connection failed")
+				}
 			})
 
 			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
-				// Non blocking channe
-				for candidate := range CandidateArrays[sessionKey] {
-					CandidatesMutex.Lock()
-					log.Log.Info(">>>> webrtc.main.InitializeWebRTCConnection(): Received candidate from channel: " + candidate)
-					if candidateErr := peerConnection.AddICECandidate(pionWebRTC.ICECandidateInit{Candidate: string(candidate)}); candidateErr != nil {
-						log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while adding candidate: " + candidateErr.Error())
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case candidate, ok := <-candidateChannel:
+						if !ok {
+							return
+						}
+						log.Log.Info("webrtc.main.InitializeWebRTCConnection(): Received candidate from channel: " + candidate)
+						if candidateErr := peerConnection.AddICECandidate(pionWebRTC.ICECandidateInit{Candidate: candidate}); candidateErr != nil {
+							log.Log.Error("webrtc.main.InitializeWebRTCConnection(): error adding candidate: " + candidateErr.Error())
+						}
 					}
-					CandidatesMutex.Unlock()
 				}
 			}()
 
@@ -270,22 +397,56 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 
 			// 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
+			var hasRelayCandidates bool
 			peerConnection.OnICECandidate(func(candidate *pionWebRTC.ICECandidate) {
+
 				if candidate == nil {
+					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): ICE gathering complete (candidate is nil)")
+					if !hasRelayCandidates {
+						log.Log.Error("webrtc.main.InitializeWebRTCConnection(): WARNING - No TURN (relay) candidates were gathered! TURN servers: " +
+							config.TURNURI + ", Username: " + config.TURNUsername + ", ForceTurn: " + config.ForceTurn)
+					}
 					return
 				}
 
+				// Log candidate details for debugging
+				candidateJSON := candidate.ToJSON()
+				candidateStr := candidateJSON.Candidate
+
+				// Determine candidate type from the candidate string
+				candidateType := "unknown"
+				if candidateJSON.Candidate != "" {
+					switch candidate.Typ {
+					case pionWebRTC.ICECandidateTypeRelay:
+						candidateType = "relay"
+					case pionWebRTC.ICECandidateTypeSrflx:
+						candidateType = "srflx"
+					case pionWebRTC.ICECandidateTypeHost:
+						candidateType = "host"
+					case pionWebRTC.ICECandidateTypePrflx:
+						candidateType = "prflx"
+					}
+				}
+
+				// Track if we received any relay (TURN) candidates
+				if candidateType == "relay" {
+					hasRelayCandidates = true
+				}
+
+				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): ICE candidate received - Type: " + candidateType +
+					", Candidate: " + candidateStr)
+
 				//  Create a config map
 				valueMap := make(map[string]interface{})
-				candateJSON := candidate.ToJSON()
-				candateBinary, err := json.Marshal(candateJSON)
+				candateBinary, err := json.Marshal(candidateJSON)
 				if err == nil {
 					valueMap["candidate"] = string(candateBinary)
 					// SDP is not needed to be send..
 					//valueMap["sdp"] = []byte(base64.StdEncoding.EncodeToString([]byte(answer.SDP)))
 					valueMap["session_id"] = handshake.SessionID
+					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): sending " + candidateType + " candidate to hub")
 				} else {
-					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): something went wrong while marshalling candidate: " + err.Error())
+					log.Log.Error("webrtc.main.InitializeWebRTCConnection(): failed to marshal candidate: " + err.Error())
 				}
 
 				// We'll send the candidate to the hub
@@ -305,8 +466,8 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				}
 			})
 
-			// Create a channel which will be used to send candidates to the other peer
-			peerConnections[handshake.SessionID] = peerConnection
+			// Store peer connection in manager
+			globalConnectionManager.AddPeerConnection(handshake.SessionID, wrapper)
 
 			if err == nil {
 				//  Create a config map
@@ -339,7 +500,11 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 
 func NewVideoTrack(streams []packets.Stream) *pionWebRTC.TrackLocalStaticSample {
 	mimeType := pionWebRTC.MimeTypeH264
-	outboundVideoTrack, _ := pionWebRTC.NewTrackLocalStaticSample(pionWebRTC.RTPCodecCapability{MimeType: mimeType}, "video", "pion124")
+	outboundVideoTrack, err := pionWebRTC.NewTrackLocalStaticSample(pionWebRTC.RTPCodecCapability{MimeType: mimeType}, "video", trackStreamID)
+	if err != nil {
+		log.Log.Error("webrtc.main.NewVideoTrack(): error creating video track: " + err.Error())
+		return nil
+	}
 	return outboundVideoTrack
 }
 
@@ -354,161 +519,245 @@ func NewAudioTrack(streams []packets.Stream) *pionWebRTC.TrackLocalStaticSample
 			mimeType = pionWebRTC.MimeTypePCMA
 		}
 	}
-	outboundAudioTrack, _ := pionWebRTC.NewTrackLocalStaticSample(pionWebRTC.RTPCodecCapability{MimeType: mimeType}, "audio", "pion124")
+	if mimeType == "" {
+		log.Log.Error("webrtc.main.NewAudioTrack(): no supported audio codec found")
+		return nil
+	}
+	outboundAudioTrack, err := pionWebRTC.NewTrackLocalStaticSample(pionWebRTC.RTPCodecCapability{MimeType: mimeType}, "audio", trackStreamID)
+	if err != nil {
+		log.Log.Error("webrtc.main.NewAudioTrack(): error creating audio track: " + err.Error())
+		return nil
+	}
 	return outboundAudioTrack
 }
 
+// streamState holds state information for the streaming process
+type streamState struct {
+	lastKeepAlive    int64
+	peerCount        int64
+	start            bool
+	receivedKeyFrame bool
+	lastAudioSample  *pionMedia.Sample
+	lastVideoSample  *pionMedia.Sample
+}
+
+// codecSupport tracks which codecs are available in the stream
+type codecSupport struct {
+	hasH264      bool
+	hasPCM_MULAW bool
+	hasAAC       bool
+	hasOpus      bool
+}
+
+// detectCodecs examines the stream to determine which codecs are available
+func detectCodecs(rtspClient capture.RTSPClient) codecSupport {
+	support := codecSupport{}
+	streams, _ := rtspClient.GetStreams()
+
+	for _, stream := range streams {
+		switch stream.Name {
+		case "H264":
+			support.hasH264 = true
+		case "PCM_MULAW":
+			support.hasPCM_MULAW = true
+		case "AAC":
+			support.hasAAC = true
+		case "OPUS":
+			support.hasOpus = true
+		}
+	}
+
+	return support
+}
+
+// hasValidCodecs checks if at least one valid video or audio codec is present
+func (cs codecSupport) hasValidCodecs() bool {
+	hasVideo := cs.hasH264
+	hasAudio := cs.hasPCM_MULAW || cs.hasAAC || cs.hasOpus
+	return hasVideo || hasAudio
+}
+
+// shouldContinueStreaming determines if streaming should continue based on keepalive and peer count
+func shouldContinueStreaming(config models.Config, state *streamState) bool {
+	if config.Capture.ForwardWebRTC != "true" {
+		return true
+	}
+
+	now := time.Now().Unix()
+	hasTimedOut := (now - state.lastKeepAlive) > int64(keepAliveTimeout.Seconds())
+	hasNoPeers := state.peerCount == 0
+
+	return !hasTimedOut && !hasNoPeers
+}
+
+// updateStreamState updates keepalive and peer count from communication channels
+func updateStreamState(communication *models.Communication, state *streamState) {
+	select {
+	case keepAliveStr := <-communication.HandleLiveHDKeepalive:
+		if val, err := strconv.ParseInt(keepAliveStr, 10, 64); err == nil {
+			state.lastKeepAlive = val
+		}
+	default:
+	}
+
+	select {
+	case peerCountStr := <-communication.HandleLiveHDPeers:
+		if val, err := strconv.ParseInt(peerCountStr, 10, 64); err == nil {
+			state.peerCount = val
+		}
+	default:
+	}
+}
+
+// writeFinalSamples writes any remaining buffered samples
+func writeFinalSamples(state *streamState, videoTrack, audioTrack *pionWebRTC.TrackLocalStaticSample) {
+	if state.lastVideoSample != nil && videoTrack != nil {
+		if err := videoTrack.WriteSample(*state.lastVideoSample); err != nil && err != io.ErrClosedPipe {
+			log.Log.Error("webrtc.main.writeFinalSamples(): error writing final video sample: " + err.Error())
+		}
+	}
+
+	if state.lastAudioSample != nil && audioTrack != nil {
+		if err := audioTrack.WriteSample(*state.lastAudioSample); err != nil && err != io.ErrClosedPipe {
+			log.Log.Error("webrtc.main.writeFinalSamples(): error writing final audio sample: " + err.Error())
+		}
+	}
+}
+
+// processVideoPacket processes a video packet and writes samples to the track
+func processVideoPacket(pkt packets.Packet, state *streamState, videoTrack *pionWebRTC.TrackLocalStaticSample, config models.Config) {
+	if videoTrack == nil {
+		return
+	}
+
+	// Start at the first keyframe
+	if pkt.IsKeyFrame {
+		state.start = true
+	}
+
+	if !state.start {
+		return
+	}
+
+	sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: uint32(pkt.Time)}
+
+	if config.Capture.ForwardWebRTC == "true" {
+		// Remote forwarding not yet implemented
+		log.Log.Debug("webrtc.main.processVideoPacket(): remote forwarding not implemented")
+		return
+	}
+
+	if state.lastVideoSample != nil {
+		duration := sample.PacketTimestamp - state.lastVideoSample.PacketTimestamp
+		state.lastVideoSample.Duration = time.Duration(duration) * time.Millisecond
+
+		if err := videoTrack.WriteSample(*state.lastVideoSample); err != nil && err != io.ErrClosedPipe {
+			log.Log.Error("webrtc.main.processVideoPacket(): error writing video sample: " + err.Error())
+		}
+	}
+
+	state.lastVideoSample = &sample
+}
+
+// processAudioPacket processes an audio packet and writes samples to the track
+func processAudioPacket(pkt packets.Packet, state *streamState, audioTrack *pionWebRTC.TrackLocalStaticSample, hasAAC bool) {
+	if audioTrack == nil {
+		return
+	}
+
+	if hasAAC {
+		// AAC transcoding not yet implemented
+		// TODO: Implement AAC to PCM_MULAW transcoding
+		return
+	}
+
+	sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: uint32(pkt.Time)}
+
+	if state.lastAudioSample != nil {
+		duration := sample.PacketTimestamp - state.lastAudioSample.PacketTimestamp
+		state.lastAudioSample.Duration = time.Duration(duration) * time.Millisecond
+
+		if err := audioTrack.WriteSample(*state.lastAudioSample); err != nil && err != io.ErrClosedPipe {
+			log.Log.Error("webrtc.main.processAudioPacket(): error writing audio sample: " + err.Error())
+		}
+	}
+
+	state.lastAudioSample = &sample
+}
+
 func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, videoTrack *pionWebRTC.TrackLocalStaticSample, audioTrack *pionWebRTC.TrackLocalStaticSample, rtspClient capture.RTSPClient) {
 
 	config := configuration.Config
 
-	// Make peerconnection map
-	peerConnections = make(map[string]*pionWebRTC.PeerConnection)
-
-	// Set the indexes for the video & audio streams
-	// Later when we read a packet we need to figure out which track to send it to.
-	hasH264 := false
-	hasPCM_MULAW := false
-	hasAAC := false
-	hasOpus := false
-	streams, _ := rtspClient.GetStreams()
-	for _, stream := range streams {
-		if stream.Name == "H264" {
-			hasH264 = true
-		} else if stream.Name == "PCM_MULAW" {
-			hasPCM_MULAW = true
-		} else if stream.Name == "AAC" {
-			hasAAC = true
-		} else if stream.Name == "OPUS" {
-			hasOpus = true
-		}
+	// Check if at least one track is available
+	if videoTrack == nil && audioTrack == nil {
+		log.Log.Error("webrtc.main.WriteToTrack(): both video and audio tracks are nil, cannot proceed")
+		return
 	}
 
-	if !hasH264 && !hasPCM_MULAW && !hasAAC && !hasOpus {
-		log.Log.Error("webrtc.main.WriteToTrack(): no valid video codec and audio codec found.")
-	} else {
-		if config.Capture.TranscodingWebRTC == "true" {
-			// Todo..
-		} else {
-			//log.Log.Info("webrtc.main.WriteToTrack(): not using a transcoder.")
-		}
+	// Detect available codecs
+	codecs := detectCodecs(rtspClient)
 
-		var cursorError error
-		var pkt packets.Packet
-		var lastAudioSample *pionMedia.Sample = nil
-		var lastVideoSample *pionMedia.Sample = nil
-
-		start := false
-		receivedKeyFrame := false
-		lastKeepAlive := "0"
-		peerCount := "0"
-
-		for cursorError == nil {
-
-			pkt, cursorError = livestreamCursor.ReadPacket()
-
-			//if config.Capture.ForwardWebRTC != "true" && peerConnectionCount == 0 {
-			//	start = false
-			//	receivedKeyFrame = false
-			//	continue
-			//}
-
-			select {
-			case lastKeepAlive = <-communication.HandleLiveHDKeepalive:
-			default:
-			}
-
-			select {
-			case peerCount = <-communication.HandleLiveHDPeers:
-			default:
-			}
-
-			now := time.Now().Unix()
-			lastKeepAliveN, _ := strconv.ParseInt(lastKeepAlive, 10, 64)
-			hasTimedOut := (now - lastKeepAliveN) > 15 // if longer then no response in 15 sec.
-			hasNoPeers := peerCount == "0"
-
-			if config.Capture.ForwardWebRTC == "true" && (hasTimedOut || hasNoPeers) {
-				start = false
-				receivedKeyFrame = false
-				continue
-			}
-
-			if len(pkt.Data) == 0 || pkt.Data == nil {
-				receivedKeyFrame = false
-				continue
-			}
-
-			if !receivedKeyFrame {
-				if pkt.IsKeyFrame {
-					receivedKeyFrame = true
-				} else {
-					continue
-				}
-			}
-
-			//if config.Capture.TranscodingWebRTC == "true" {
-			// We will transcode the video
-			// TODO..
-			//}
-
-			if pkt.IsVideo {
-
-				// Start at the first keyframe
-				if pkt.IsKeyFrame {
-					start = true
-				}
-				if start {
-					sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: uint32(pkt.Time)}
-					//sample = pionMedia.Sample{Data: pkt.Data, Duration: time.Second}
-					if config.Capture.ForwardWebRTC == "true" {
-						// We will send the video to a remote peer
-						// TODO..
-					} else {
-						if lastVideoSample != nil {
-							duration := sample.PacketTimestamp - lastVideoSample.PacketTimestamp
-							bufferDurationCasted := time.Duration(duration) * time.Millisecond
-							lastVideoSample.Duration = bufferDurationCasted
-							if err := videoTrack.WriteSample(*lastVideoSample); err != nil && err != io.ErrClosedPipe {
-								log.Log.Error("webrtc.main.WriteToTrack(): something went wrong while writing sample: " + err.Error())
-							}
-						}
-
-						lastVideoSample = &sample
-					}
-				}
-			} else if pkt.IsAudio {
-
-				// @TODO: We need to check if the audio is PCM_MULAW or AAC
-				// If AAC we need to transcode it to PCM_MULAW
-				// If PCM_MULAW we can send it directly.
-
-				if hasAAC {
-					// We will transcode the audio from AAC to PCM_MULAW
-					// Not sure how to do this yet, but we need to use a decoder
-					// and then encode it to PCM_MULAW.
-					// TODO..
-					//d := fdkaac.NewAacDecoder()
-					continue
-				}
-
-				// We will send the audio
-				sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: uint32(pkt.Time)}
-
-				if lastAudioSample != nil {
-					duration := sample.PacketTimestamp - lastAudioSample.PacketTimestamp
-					bufferDurationCasted := time.Duration(duration) * time.Millisecond
-					lastAudioSample.Duration = bufferDurationCasted
-					if err := audioTrack.WriteSample(*lastAudioSample); err != nil && err != io.ErrClosedPipe {
-						log.Log.Error("webrtc.main.WriteToTrack(): something went wrong while writing sample: " + err.Error())
-					}
-				}
-
-				lastAudioSample = &sample
-			}
-		}
+	if !codecs.hasValidCodecs() {
+		log.Log.Error("webrtc.main.WriteToTrack(): no valid video or audio codec found")
+		return
 	}
 
-	peerConnectionCount = 0
-	log.Log.Info("webrtc.main.WriteToTrack(): stop writing to track.")
+	if config.Capture.TranscodingWebRTC == "true" {
+		log.Log.Info("webrtc.main.WriteToTrack(): transcoding enabled but not yet implemented")
+	}
+
+	// Initialize streaming state
+	state := &streamState{
+		lastKeepAlive: time.Now().Unix(),
+		peerCount:     0,
+	}
+
+	defer func() {
+		writeFinalSamples(state, videoTrack, audioTrack)
+		log.Log.Info("webrtc.main.WriteToTrack(): stopped writing to track")
+	}()
+
+	var pkt packets.Packet
+	var cursorError error
+
+	for cursorError == nil {
+		pkt, cursorError = livestreamCursor.ReadPacket()
+
+		if cursorError != nil {
+			break
+		}
+
+		// Update state from communication channels
+		updateStreamState(communication, state)
+
+		// Check if we should continue streaming
+		if !shouldContinueStreaming(config, state) {
+			state.start = false
+			state.receivedKeyFrame = false
+			continue
+		}
+
+		// Skip empty packets
+		if len(pkt.Data) == 0 || pkt.Data == nil {
+			state.receivedKeyFrame = false
+			continue
+		}
+
+		// Wait for first keyframe before processing
+		if !state.receivedKeyFrame {
+			if pkt.IsKeyFrame {
+				state.receivedKeyFrame = true
+			} else {
+				continue
+			}
+		}
+
+		// Process video or audio packets
+		if pkt.IsVideo {
+			processVideoPacket(pkt, state, videoTrack, config)
+		} else if pkt.IsAudio {
+			processAudioPacket(pkt, state, audioTrack, codecs.hasAAC)
+		}
+	}
 }