Merge pull request #420 from streamplace/eli/problem-detection

implement basic stream problem detection

js/app/components/live-dashboard/problems.tsx

···
+import { useLivestreamStore } from "@streamplace/components";
+import { LivestreamProblem } from "@streamplace/components/src/livestream-store/livestream-state";
+import { ExternalLink } from "@tamagui/lucide-icons";
+import { useState } from "react";
+import { Linking, Pressable } from "react-native";
+import { Button, H3, Text, View } from "tamagui";
+
+const Problems = ({
+  probs,
+  onIgnore,
+}: {
+  probs: LivestreamProblem[];
+  onIgnore: () => void;
+}) => {
+  return (
+    <View gap={"$3"}>
+      <View>
+        <H3>Optimize Your Stream</H3>
+        <Text>
+          We’ve found a few things that could improve your stream’s reliability.
+        </Text>
+      </View>
+      {probs.map((p) => (
+        <View>
+          <View gap="$2" key={p.message} flexDirection="row" ai="flex-start">
+            <Text
+              borderRadius="$2"
+              px="$2"
+              width={82}
+              textAlign="center"
+              bg={
+                p.severity === "error"
+                  ? "$red8Dark"
+                  : p.severity === "warning"
+                    ? "$yellow8Dark"
+                    : "$blue8Dark"
+              }
+            >
+              {p.severity}
+            </Text>
+            <View flex={1} gap="$1">
+              <Text>{p.code}</Text>
+              <Text color="$gray11Dark" fontSize="$6">
+                {p.message}
+              </Text>
+              {p.link && (
+                <Pressable onPress={() => p.link && Linking.openURL(p.link)}>
+                  <View flexDirection="row" ai="center" gap="$2">
+                    <Text color="$blue10" fontSize="$6">
+                      Learn More
+                    </Text>
+                    <ExternalLink size="$1" />
+                  </View>
+                </Pressable>
+              )}
+            </View>
+          </View>
+        </View>
+      ))}
+
+      <Button onPress={onIgnore}>
+        <Text>Ignore</Text>
+      </Button>
+    </View>
+  );
+};
+
+export const ProblemsWrapper = ({
+  children,
+}: {
+  children: React.ReactElement;
+}) => {
+  const problems = useLivestreamStore((x) => x.problems);
+  const [dismiss, setDismiss] = useState(false);
+
+  return (
+    <View position="relative" f={1} ai="center" jc="center" fb={0}>
+      {children}
+      {problems.length > 0 && !dismiss && (
+        <View
+          position="absolute"
+          top={0}
+          left={0}
+          right={0}
+          bottom={0}
+          backgroundColor="rgba(0, 0, 0, 0.8)"
+          ai="center"
+          jc="flex-start"
+          zIndex={100}
+          padding="$8"
+        >
+          <View
+            backgroundColor="$gray1"
+            borderColor="$gray5"
+            borderWidth={1}
+            borderRadius="$4"
+            padding="$4"
+            maxWidth={700}
+            width="100%"
+          >
+            <Problems probs={problems} onIgnore={() => setDismiss(true)} />
+          </View>
+        </View>
+      )}
+    </View>
+  );
+};

js/app/src/screens/live-dashboard.tsx

···
 import { useAppDispatch, useAppSelector } from "store/hooks";
 import { Button, H3, H6, isWeb, Text, View } from "tamagui";
 
+import { ProblemsWrapper } from "components/live-dashboard/problems";
+
 enum StreamSource {
   Start,
   Camera,
···
             {closeButton}
           </View>
           <View f={1} ai="center" jc="center" fb={0}>
-            <ButtonSelector
-              values={[
-                { label: "Create", value: "create" },
-                { label: "Update", value: "update" },
-              ]}
-              disabledValues={isLive ? [] : ["update"]}
-              selectedValue={page}
-              setSelectedValue={setPage}
-              maxWidth={250}
-              width="100%"
-            />
-            {page === "update" && isLive ? <UpdateLivestream /> : null}
-            {page === "create" ? <CreateLivestream /> : null}
+            <ProblemsWrapper>
+              <>
+                <ButtonSelector
+                  values={[
+                    { label: "Create", value: "create" },
+                    { label: "Update", value: "update" },
+                  ]}
+                  disabledValues={isLive ? [] : ["update"]}
+                  selectedValue={page}
+                  setSelectedValue={setPage}
+                  maxWidth={250}
+                  width="100%"
+                />
+                {page === "update" && isLive ? <UpdateLivestream /> : null}
+                {page === "create" ? <CreateLivestream /> : null}
+              </>
+            </ProblemsWrapper>
           </View>
           {madeChoiceAboutDebugRecording ? null : <DebugRecordingPopup />}
         </View>

js/components/src/livestream-store/livestream-state.tsx

···
   viewers: number | null;
   pendingHides: string[];
   segment: PlaceStreamSegment.Record | null;
+  recentSegments: PlaceStreamSegment.Record[];
+  problems: LivestreamProblem[];
   renditions: PlaceStreamDefs.Rendition[];
   replyToMessage: ChatMessageViewHydrated | null;
   streamKey: string | null;
   setStreamKey: (key: string | null) => void;
 }
+
+export interface LivestreamProblem {
+  code: string;
+  message: string;
+  severity: "error" | "warning" | "info";
+  link?: string;
+}

js/components/src/livestream-store/livestream-store.tsx

···
     streamKey: null,
     setStreamKey: (sk) => set({ streamKey: sk }),
     authors: {},
+    recentSegments: [],
+    problems: [],
   }));
 };
 

js/components/src/livestream-store/problems.tsx

···
+import { PlaceStreamSegment } from "streamplace";
+import { LivestreamProblem } from "./livestream-state";
+
+const VARIANCE_THRESHOLD = 0.5;
+const DURATION_THRESHOLD = 5000000000; // 5s in ns
+
+const detectVariableSegmentLength = (
+  segments: PlaceStreamSegment.Record[],
+): { variable: boolean; duration: boolean } => {
+  if (segments.length < 3) {
+    // Need at least 3 segments to detect variability
+    return { variable: false, duration: false };
+  }
+
+  const durations = segments
+    .map((segment) => segment.duration)
+    .filter(
+      (duration): duration is number => duration !== undefined && duration > 0,
+    );
+
+  if (durations.length < 3) {
+    return { variable: false, duration: false };
+  }
+
+  // Calculate mean
+  const mean =
+    durations.reduce((sum: number, duration: number) => sum + duration, 0) /
+    durations.length;
+
+  // Calculate standard deviation
+  const variance =
+    durations.reduce((sum: number, duration: number) => {
+      const diff = duration - mean;
+      return sum + diff * diff;
+    }, 0) / durations.length;
+  const stdDev = Math.sqrt(variance);
+
+  // Calculate coefficient of variation (CV)
+  const cv = stdDev / mean;
+
+  // CV > 0.5 indicates high variability
+  // This threshold can be adjusted based on testing
+  return {
+    variable: cv > VARIANCE_THRESHOLD,
+    duration: mean > DURATION_THRESHOLD,
+  };
+};
+
+export const findProblems = (
+  segments: PlaceStreamSegment.Record[],
+): LivestreamProblem[] => {
+  const problems: LivestreamProblem[] = [];
+  let hasBFrames = false;
+  for (const segment of segments) {
+    const video = segment.video?.[0];
+    if (!video) {
+      // i mean yes this is a problem but it can't happen yet
+      continue;
+    }
+    if (video.bframes === true) {
+      hasBFrames = true;
+      break;
+    }
+  }
+  if (hasBFrames) {
+    problems.push({
+      code: "bframes",
+      message:
+        "Your stream contains B-Frames, which are not supported in Streamplace. Your stream will stutter.",
+      severity: "error",
+      link: "https://stream.place/docs/guides/start-streaming/obs/#obs-configuration",
+    });
+  }
+
+  const { variable, duration } = detectVariableSegmentLength(segments);
+  if (variable) {
+    problems.push({
+      code: "variable_segment_length",
+      message:
+        "Your stream contains variable segment lengths, which may cause playback issues.",
+      severity: "warning",
+      link: "https://stream.place/docs/guides/start-streaming/obs/#obs-configuration",
+    });
+  }
+  if (duration) {
+    problems.push({
+      code: "long_segments",
+      message:
+        "Your stream contains long segments (>5s). This will work fine, but increases the delay of the livestream.",
+      severity: "warning",
+      link: "https://stream.place/docs/guides/start-streaming/obs/#obs-configuration",
+    });
+  }
+
+  return problems;
+};

js/components/src/livestream-store/websocket-consumer.tsx

···
 import { SystemMessages } from "../lib/system-messages";
 import { reduceChat } from "./chat";
 import { LivestreamState } from "./livestream-state";
+import { findProblems } from "./problems";
+
+const MAX_RECENT_SEGMENTS = 10;
 
 export const handleWebSocketMessages = (
   state: LivestreamState,
···
       };
       state = reduceChat(state, [hydrated], [], []);
     } else if (PlaceStreamSegment.isRecord(message)) {
+      const newRecentSegments = [...state.recentSegments];
+      newRecentSegments.unshift(message);
+      if (newRecentSegments.length > MAX_RECENT_SEGMENTS) {
+        newRecentSegments.pop();
+      }
       state = {
         ...state,
         segment: message as PlaceStreamSegment.Record,
+        recentSegments: newRecentSegments,
+        problems: findProblems(newRecentSegments),
       };
     } else if (PlaceStreamDefs.isBlockView(message)) {
       const block = message as PlaceStreamDefs.BlockView;

js/docs/src/content/docs/guides/start-streaming/obs.md

···
 6. Click "Generate Stream Key"
    - The stream key will automatically be copied to your clipboard
 
-### 2. Configure OBS Studio
+### 2. Configure OBS Studio <a name="obs-configuration"></a>
 
 #### 2a. Initial OBS Configuration
 

js/docs/src/content/docs/lex-reference/place-stream-segment.md

···
 | `width`     | `integer`                  | ✅    |             |              |
 | `height`    | `integer`                  | ✅    |             |              |
 | `framerate` | [`#framerate`](#framerate) | ❌    |             |              |
+| `bframes`   | `boolean`                  | ❌    |             |              |
 
 ---
 
···
         "framerate": {
           "type": "ref",
           "ref": "#framerate"
+        },
+        "bframes": {
+          "type": "boolean"
         }
       }
     },

lexicons/place/stream/segment.json

···
         "framerate": {
           "type": "ref",
           "ref": "#framerate"
-        }
+        },
+        "bframes": { "type": "boolean" }
       }
     },
     "framerate": {

pkg/media/audio_smear.go

···
 		NeedDataFunc: ReaderNeedData(ctx, input),
 	})
 
+	seg := SegmentData{
+		Audio: []SegmentBuffer{},
+		Video: []SegmentBuffer{},
+	}
+
 	audioSinkElem, err := pipeline.GetElementByName("audioappsink")
 	if err != nil {
 		return nil, fmt.Errorf("failed to get audioappsink element: %w", err)
···
 	audioSink := app.SinkFromElement(audioSinkElem)
 	if audioSink == nil {
 		return nil, fmt.Errorf("failed to get audioappsink element: %w", err)
-	}
-
-	seg := SegmentData{
-		Audio: []SegmentBuffer{},
-		Video: []SegmentBuffer{},
 	}
 
 	audioSink.SetCallbacks(&app.SinkCallbacks{

pkg/media/media_data_parser.go

···
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	pipelineSlice := []string{
-		"appsrc name=appsrc ! qtdemux name=demux ! fakesink sync=false",
+		"appsrc name=appsrc ! qtdemux name=demux",
+		"demux.video_0 ! queue ! h264parse name=videoparse disable-passthrough=true config-interval=-1 ! h264timestamper ! appsink sync=false name=videoappsink",
+		"demux.audio_0 ! queue ! opusparse name=audioparse ! appsink sync=false name=audioappsink",
 	}
 
 	pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n"))
···
 		return nil, fmt.Errorf("error connecting pad-add: %w", err)
 	}
 
+	audioSinkElem, err := pipeline.GetElementByName("audioappsink")
+	if err != nil {
+		return nil, fmt.Errorf("failed to get audioappsink element: %w", err)
+	}
+	audioSink := app.SinkFromElement(audioSinkElem)
+	if audioSink == nil {
+		return nil, fmt.Errorf("failed to get audioappsink element: %w", err)
+	}
+
+	audioSink.SetCallbacks(&app.SinkCallbacks{
+		NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
+			sample := sink.PullSample()
+			if sample == nil {
+				return gst.FlowOK
+			}
+
+			return gst.FlowOK
+		},
+	})
+
+	videoSinkElem, err := pipeline.GetElementByName("videoappsink")
+	if err != nil {
+		return nil, fmt.Errorf("failed to get videoappsink element: %w", err)
+	}
+	videoSink := app.SinkFromElement(videoSinkElem)
+	if videoSink == nil {
+		return nil, fmt.Errorf("failed to get videoappsink element: %w", err)
+	}
+
+	hasBFrames := false
+	videoSink.SetCallbacks(&app.SinkCallbacks{
+		NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
+			sample := sink.PullSample()
+			if sample == nil {
+				return gst.FlowOK
+			}
+
+			buf := sample.GetBuffer()
+			pts := buf.PresentationTimestamp().String()
+			dts := buf.DecodingTimestamp().String()
+
+			if pts != dts {
+				hasBFrames = true
+			}
+
+			return gst.FlowOK
+		},
+	})
+
 	go func() {
 		if err := HandleBusMessages(ctx, pipeline); err != nil {
 			log.Log(ctx, "pipeline error", "error", err)
···
 	if audioMetadata == nil {
 		return nil, fmt.Errorf("no audio metadata")
 	}
+
+	videoMetadata.BFrames = hasBFrames
 
 	meta := &model.SegmentMediaData{
 		Video: []*model.SegmentMediadataVideo{videoMetadata},

pkg/media/media_data_parser_test.go

···
 
 	"github.com/stretchr/testify/require"
 	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/test/remote"
 )
 
 func TestMediaDataParser(t *testing.T) {
···
 		mediaData, err := ParseSegmentMediaData(ctx, bs)
 		require.NoError(t, err)
 		require.NotNil(t, mediaData)
+		require.False(t, mediaData.Video[0].BFrames, "Video should not have BFrames")
+		require.Greater(t, mediaData.Duration, int64(0), "Video duration should not be empty")
+	})
+}
+
+func TestMediaDataParserBFrames(t *testing.T) {
+	withNoGSTLeaks(t, func() {
+		inputFile, err := os.Open(remote.RemoteFixture("5ea6c4491bade0cdcad3770aa0b63b2cd7a580e233ee320d5bc2282503b26491/segment-with-bframes.mp4"))
+		require.NoError(t, err)
+		defer inputFile.Close()
+		bs, err := io.ReadAll(inputFile)
+		require.NoError(t, err)
+
+		ctx := log.WithDebugValue(context.Background(), map[string]map[string]int{"GStreamerFunc": {"ParseSegmentMediaData": 9}})
+		mediaData, err := ParseSegmentMediaData(ctx, bs)
+		require.NoError(t, err)
+		require.NotNil(t, mediaData)
+		require.True(t, mediaData.Video[0].BFrames, "Video should have BFrames")
 		require.Greater(t, mediaData.Duration, int64(0), "Video duration should not be empty")
 	})
 }

pkg/model/segment.go

···
 )
 
 type SegmentMediadataVideo struct {
-	Width  int `json:"width"`
-	Height int `json:"height"`
-	FPSNum int `json:"fpsNum"`
-	FPSDen int `json:"fpsDen"`
+	Width   int  `json:"width"`
+	Height  int  `json:"height"`
+	FPSNum  int  `json:"fpsNum"`
+	FPSDen  int  `json:"fpsDen"`
+	BFrames bool `json:"bframes"`
 }
 
 type SegmentMediadataAudio struct {
···
 					Num: int64(s.MediaData.Video[0].FPSNum),
 					Den: int64(s.MediaData.Video[0].FPSDen),
 				},
+				Bframes: &s.MediaData.Video[0].BFrames,
 			},
 		},
 		Audio: []*streamplace.Segment_Audio{

pkg/streamplace/cbor_gen.go

···
 	}
 
 	cw := cbg.NewCborWriter(w)
-	fieldCount := 4
+	fieldCount := 5
+
+	if t.Bframes == nil {
+		fieldCount--
+	}
 
 	if t.Framerate == nil {
 		fieldCount--
···
 	} else {
 		if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.Height-1)); err != nil {
 			return err
+		}
+	}
+
+	// t.Bframes (bool) (bool)
+	if t.Bframes != nil {
+
+		if len("bframes") > 1000000 {
+			return xerrors.Errorf("Value in field \"bframes\" was too long")
+		}
+
+		if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("bframes"))); err != nil {
+			return err
+		}
+		if _, err := cw.WriteString(string("bframes")); err != nil {
+			return err
+		}
+
+		if t.Bframes == nil {
+			if _, err := cw.Write(cbg.CborNull); err != nil {
+				return err
+			}
+		} else {
+			if err := cbg.WriteBool(w, *t.Bframes); err != nil {
+				return err
+			}
 		}
 	}
 
···
 				}
 
 				t.Height = int64(extraI)
+			}
+			// t.Bframes (bool) (bool)
+		case "bframes":
+
+			{
+				b, err := cr.ReadByte()
+				if err != nil {
+					return err
+				}
+				if b != cbg.CborNull[0] {
+					if err := cr.UnreadByte(); err != nil {
+						return err
+					}
+
+					maj, extra, err = cr.ReadHeader()
+					if err != nil {
+						return err
+					}
+					if maj != cbg.MajOther {
+						return fmt.Errorf("booleans must be major type 7")
+					}
+
+					var val bool
+					switch extra {
+					case 20:
+						val = false
+					case 21:
+						val = true
+					default:
+						return fmt.Errorf("booleans are either major type 7, value 20 or 21 (got %d)", extra)
+					}
+					t.Bframes = &val
+				}
 			}
 			// t.Framerate (streamplace.Segment_Framerate) (struct)
 		case "framerate":

pkg/streamplace/streamsegment.go

···
 
 // Segment_Video is a "video" in the place.stream.segment schema.
 type Segment_Video struct {
+	Bframes   *bool              `json:"bframes,omitempty" cborgen:"bframes,omitempty"`
 	Codec     string             `json:"codec" cborgen:"codec"`
 	Framerate *Segment_Framerate `json:"framerate,omitempty" cborgen:"framerate,omitempty"`
 	Height    int64              `json:"height" cborgen:"height"`