diff --git a/Makefile b/Makefile index d2ebf056..d24c904a 100644 --- a/Makefile +++ b/Makefile @@ -696,3 +696,7 @@ precommit: dockerfile-hash-precommit dockerfile-hash-precommit: @bash -c 'printf "variables:\n DOCKERFILE_HASH: `git hash-object docker/build.Dockerfile`" > .ci/dockerfile-hash.yaml' \ && git add .ci/dockerfile-hash.yaml + +.PHONY: rtcrec +rtcrec: + go build -o $(BUILDDIR)/rtcrec ./pkg/rtcrec/cmd/... diff --git a/go.mod b/go.mod index 1c2c0af0..af99c476 100644 --- a/go.mod +++ b/go.mod @@ -21,6 +21,7 @@ require ( github.com/decred/dcrd/dcrec/secp256k1 v1.0.4 github.com/dunglas/httpsfv v1.0.2 github.com/ethereum/go-ethereum v1.14.7 + github.com/fxamacker/cbor/v2 v2.8.0 github.com/go-git/go-git/v5 v5.12.0 github.com/go-gst/go-glib v1.4.0 github.com/go-gst/go-gst v1.4.0 @@ -387,6 +388,7 @@ require ( github.com/valyala/fasttemplate v1.2.2 // indirect github.com/vektah/gqlparser/v2 v2.5.22 // indirect github.com/wlynxg/anet v0.0.5 // indirect + github.com/x448/float16 v0.8.4 // indirect github.com/xanzy/ssh-agent v0.3.3 // indirect github.com/xen0n/gosmopolitan v1.3.0 // indirect github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect diff --git a/go.sum b/go.sum index decc0706..a578e1d8 100644 --- a/go.sum +++ b/go.sum @@ -261,6 +261,8 @@ github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHk github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY= github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw= +github.com/fxamacker/cbor/v2 v2.8.0 h1:fFtUGXUzXPHTIUdne5+zzMPTfffl3RD5qYnkY40vtxU= +github.com/fxamacker/cbor/v2 v2.8.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= github.com/fzipp/gocyclo v0.6.0 h1:lsblElZG7d3ALtGMx9fmxeTKZaLLpU8mET09yN4BBLo= github.com/fzipp/gocyclo v0.6.0/go.mod h1:rXPyn8fnlpa0R2csP/31uerbiVBugk5whMdlyaLkLoA= github.com/getsentry/sentry-go v0.27.0 h1:Pv98CIbtB3LkMWmXi4Joa5OOcwbmnX88sF5qbK3r3Ps= @@ -996,6 +998,8 @@ github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e h1:28X54c github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e/go.mod h1:pM99HXyEbSQHcosHc0iW7YFmwnscr+t9Te4ibko05so= github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA= +github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= +github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= github.com/xanzy/ssh-agent v0.3.3 h1:+/15pJfg/RsTxqYcX6fHqOXZwwMP+2VyYWJeWM2qQFM= github.com/xanzy/ssh-agent v0.3.3/go.mod h1:6dzNDKs0J9rVPHPhaGCukekBHKqfl+L3KghI1Bc68Uw= github.com/xen0n/gosmopolitan v1.3.0 h1:zAZI1zefvo7gcpbCOrPSHJZJYA9ZgLfJqtKzZ5pHqQM= diff --git a/js/app/components/livestream/livestream.tsx b/js/app/components/livestream/livestream.tsx index d2ce7b6b..b09eaff0 100644 --- a/js/app/components/livestream/livestream.tsx +++ b/js/app/components/livestream/livestream.tsx @@ -14,7 +14,6 @@ import Avatar from "components/home/avatar"; import Loading from "components/loading/loading"; import { Player } from "components/player/player"; import { PlayerProps } from "components/player/props"; -import Popup from "components/popup"; import Timer from "components/timer"; import Viewers from "components/viewers"; import { useFullscreen } from "contexts/FullscreenContext"; @@ -23,10 +22,6 @@ import { setSidebarUnhidden, } from "features/base/sidebarSlice"; import { getProfile } from "features/bluesky/blueskySlice"; -import { - selectTelemetry, - telemetryOpt, -} from "features/streamplace/streamplaceSlice"; import useAvatars from "hooks/useAvatars"; import { useKeyboard } from "hooks/useKeyboard"; import usePlatform from "hooks/usePlatform"; @@ -38,10 +33,9 @@ import { SafeAreaView, } from "react-native"; import storage from "storage"; -import { useAppDispatch, useAppSelector } from "store/hooks"; +import { useAppDispatch } from "store/hooks"; import { Button, - H3, isWeb, ScrollView, Text, @@ -62,7 +56,6 @@ export default function Livestream(props: Partial) { } export function LivestreamInner(props: Partial) { - const telemetry = useAppSelector(selectTelemetry); const toast = useToastController(); const viewers = useViewers(); @@ -184,48 +177,6 @@ export function LivestreamInner(props: Partial) { )} - {telemetry === null && ( - { - dispatch(telemetryOpt(false)); - }} - containerProps={{ - bottom: "$8", - zIndex: 1000, - }} - bubbleProps={{ - backgroundColor: "$accentBackground", - gap: "$3", - maxWidth: 400, - }} - > -

Player Telemetry

- - Streamplace is beta software and it helps us out to have the - player report back on how playback is working. Would you like to - opt in to optional player telemetry? - - - - - - - )} ) { }} > { if (onPress) { onPress(); diff --git a/js/app/components/settings/settings.tsx b/js/app/components/settings/settings.tsx index 4d949f65..c106f791 100644 --- a/js/app/components/settings/settings.tsx +++ b/js/app/components/settings/settings.tsx @@ -3,15 +3,17 @@ import { ArrowRight } from "@tamagui/lucide-icons"; import AQLink from "components/aqlink"; import Container from "components/container"; import { - DEFAULT_URL, - selectTelemetry, - setURL, - telemetryOpt, -} from "features/streamplace/streamplaceSlice"; + createServerSettingsRecord, + getServerSettingsFromPDS, + selectIsReady, + selectServerSettings, +} from "features/bluesky/blueskySlice"; +import { DEFAULT_URL, setURL } from "features/streamplace/streamplaceSlice"; import useStreamplaceNode from "hooks/useStreamplaceNode"; import { useEffect, useState } from "react"; +import { Switch } from "react-native"; import { useAppDispatch, useAppSelector } from "store/hooks"; -import { Button, H3, H5, Input, Switch, Text, View, XStack } from "tamagui"; +import { Button, H3, H5, Input, Text, View, XStack } from "tamagui"; import { Updates } from "./updates"; export function Settings() { @@ -47,12 +49,6 @@ export function Settings() { } }; - const telemetry = useAppSelector(selectTelemetry); - - const handleTelemetryToggle = (checked: boolean) => { - dispatch(telemetryOpt(checked)); - }; - return ( - - + value={overrideEnabled} + onValueChange={handleToggleOverride} + /> @@ -126,52 +119,82 @@ export function Settings() { - - - -

Player Telemetry

- - Optional - - - - - - - - {loggedIn && ( - - + + -
Manage Keys
- - - + +
Manage Keys
+ + + + )} ); } + +const DebugRecording = () => { + const dispatch = useAppDispatch(); + const isReady = useAppSelector(selectIsReady); + const serverSettings = useAppSelector(selectServerSettings) || {}; + const { url } = useStreamplaceNode(); + + useEffect(() => { + if (isReady) { + dispatch(getServerSettingsFromPDS()); + } + }, [isReady]); + + const u = new URL(url); + return ( + + + +

+ Allow {u.host} to record your livestream for debugging and improving + the service +

+ + Optional + + + { + if (value === true) { + dispatch( + createServerSettingsRecord({ + ...serverSettings, + debugRecording: true, + }), + ); + } else { + dispatch( + createServerSettingsRecord({ + ...serverSettings, + debugRecording: false, + }), + ); + } + }} + > + + + ); +}; diff --git a/js/app/features/bluesky/blueskySlice.tsx b/js/app/features/bluesky/blueskySlice.tsx index 762c4782..f01102bc 100644 --- a/js/app/features/bluesky/blueskySlice.tsx +++ b/js/app/features/bluesky/blueskySlice.tsx @@ -22,6 +22,7 @@ import { PlaceStreamChatProfile, PlaceStreamKey, PlaceStreamLivestream, + PlaceStreamServerSettings, StreamplaceAgent, } from "streamplace"; import { isWeb } from "tamagui"; @@ -62,6 +63,7 @@ const initialState: BlueskyState = { records: null, }, newLivestream: null, + serverSettings: null, }; const uploadThumbnail = async ( @@ -1282,6 +1284,121 @@ export const blueskySlice = createAppSlice({ }, }, ), + + getServerSettingsFromPDS: create.asyncThunk( + async (_, thunkAPI) => { + const { bluesky, streamplace } = thunkAPI.getState() as { + bluesky: BlueskyState; + streamplace: StreamplaceState; + }; + const did = bluesky.oauthSession?.did; + if (!did) { + throw new Error("No DID"); + } + const profile = bluesky.profiles[did]; + if (!profile) { + throw new Error("No profile"); + } + if (!bluesky.pdsAgent) { + throw new Error("No agent"); + } + const u = new URL(streamplace.url); + const res = await bluesky.pdsAgent.com.atproto.repo.getRecord({ + repo: did, + collection: "place.stream.server.settings", + rkey: u.host, + }); + if (!res.success) { + throw new Error("Failed to get chat profile record"); + } + + if (PlaceStreamServerSettings.isRecord(res.data.value)) { + return res.data.value; + } else { + console.log("not a record", res.data.value); + } + return null; + }, + { + pending: (state) => { + return { + ...state, + }; + }, + fulfilled: (state, action) => { + if (!action.payload) { + return state; + } + return { + ...state, + serverSettings: action.payload, + }; + }, + rejected: (state, action) => { + console.error("getServerSettingsFromPDS rejected", action.error); + return { + ...state, + }; + }, + }, + ), + + createServerSettingsRecord: create.asyncThunk( + async ({ debugRecording }: { debugRecording: boolean }, thunkAPI) => { + const { bluesky, streamplace } = thunkAPI.getState() as { + bluesky: BlueskyState; + streamplace: StreamplaceState; + }; + if (!bluesky.pdsAgent) { + throw new Error("No agent"); + } + const did = bluesky.oauthSession?.did; + if (!did) { + throw new Error("No DID"); + } + const profile = bluesky.profiles[did]; + if (!profile) { + throw new Error("No profile"); + } + if (!did) { + throw new Error("No DID"); + } + const u = new URL(streamplace.url); + const serverSettings: PlaceStreamServerSettings.Record = { + debugRecording: debugRecording, + }; + + const res = await bluesky.pdsAgent.com.atproto.repo.putRecord({ + repo: did, + collection: "place.stream.server.settings", + record: serverSettings, + rkey: u.host, + }); + if (!res.success) { + throw new Error("Failed to create server settings record"); + } + return serverSettings; + }, + { + pending: (state) => { + return { + ...state, + }; + }, + fulfilled: (state, action) => { + return { + ...state, + serverSettings: action.payload, + }; + }, + rejected: (state, action) => { + console.error("createServerSettingsRecord rejected", action.error); + return { + ...state, + }; + }, + }, + ), }), // You can define your selectors here. These selectors receive the slice @@ -1293,6 +1410,7 @@ export const blueskySlice = createAppSlice({ selectProfiles: (bluesky) => bluesky.profiles, selectStoredKey: (bluesky) => bluesky.storedKey, selectKeyRecords: (bluesky) => bluesky.streamKeysResponse, + selectServerSettings: (bluesky) => bluesky.serverSettings, selectUserProfile: (bluesky) => { const did = bluesky.oauthSession?.did; if (!did) return null; @@ -1342,6 +1460,8 @@ export const { createBlockRecord, followUser, unfollowUser, + getServerSettingsFromPDS, + createServerSettingsRecord, } = blueskySlice.actions; // Selectors returned by `slice.selectors` take the root state as their first argument. @@ -1357,4 +1477,5 @@ export const { selectNewLivestream, selectChatProfile, selectCachedProfiles, + selectServerSettings, } = blueskySlice.selectors; diff --git a/js/app/features/bluesky/blueskyTypes.tsx b/js/app/features/bluesky/blueskyTypes.tsx index 9a5a81fa..c734b61a 100644 --- a/js/app/features/bluesky/blueskyTypes.tsx +++ b/js/app/features/bluesky/blueskyTypes.tsx @@ -5,6 +5,7 @@ import { StreamKey } from "features/base/baseSlice"; import { PlaceStreamChatProfile, PlaceStreamLivestream, + PlaceStreamServerSettings, StreamplaceAgent, } from "streamplace"; import { StreamplaceOAuthClient } from "./oauthClient"; @@ -48,4 +49,5 @@ export interface BlueskyState { error: null | string; profile: null | PlaceStreamChatProfile.Record; }; + serverSettings: null | PlaceStreamServerSettings.Record; } diff --git a/js/app/features/streamplace/streamplaceSlice.tsx b/js/app/features/streamplace/streamplaceSlice.tsx index bc29ef70..6c1cae6d 100644 --- a/js/app/features/streamplace/streamplaceSlice.tsx +++ b/js/app/features/streamplace/streamplaceSlice.tsx @@ -48,7 +48,6 @@ export interface StreamplaceState { firstRequest: boolean; }; mySegments: PlaceStreamSegment.SegmentView[]; - telemetry: boolean | null; userMuted: boolean | null; chatWarned: boolean; } @@ -64,13 +63,11 @@ const initialState: StreamplaceState = { firstRequest: true, }, mySegments: [], - telemetry: null, userMuted: null, chatWarned: false, }; const USER_MUTED_KEY = "streamplaceUserMuted"; -const TELEMETRY_KEY = "streamplaceTelemetry"; const URL_KEY = "streamplaceUrl"; const CHAT_WARNING_KEY = "streamplaceChatWarning2"; @@ -80,22 +77,14 @@ export const streamplaceSlice = createAppSlice({ reducers: (create) => ({ initialize: create.asyncThunk( async (_, { getState }) => { - let [url, telemetryStr, userMutedStr, chatWarningStr] = - await Promise.all([ - Storage.getItem(URL_KEY), - Storage.getItem(TELEMETRY_KEY), - Storage.getItem(USER_MUTED_KEY), - Storage.getItem(CHAT_WARNING_KEY), - ]); + let [url, userMutedStr, chatWarningStr] = await Promise.all([ + Storage.getItem(URL_KEY), + Storage.getItem(USER_MUTED_KEY), + Storage.getItem(CHAT_WARNING_KEY), + ]); if (!url) { url = DEFAULT_URL; } - let telemetry: boolean | null = null; - if (typeof telemetryStr === "string") { - telemetry = JSON.parse(telemetryStr); - } else { - telemetry = null; - } let userMuted: boolean | null = null; console.log("userMutedStr", userMutedStr); if (typeof userMutedStr === "string") { @@ -107,18 +96,17 @@ export const streamplaceSlice = createAppSlice({ if (typeof chatWarningStr === "string") { chatWarned = chatWarningStr === "true"; } - return { url, telemetry, userMuted, chatWarned }; + return { url, userMuted, chatWarned }; }, { pending: (state) => { // state.status = "loading"; }, fulfilled: (state, action) => { - const { url, telemetry, userMuted, chatWarned } = action.payload; + const { url, userMuted, chatWarned } = action.payload; return { ...state, url, - telemetry, userMuted, initialized: true, chatWarned, @@ -141,18 +129,6 @@ export const streamplaceSlice = createAppSlice({ }; }), - telemetryOpt: create.reducer((state, action: { payload: boolean }) => { - Storage.setItem(TELEMETRY_KEY, JSON.stringify(action.payload)).catch( - (err) => { - console.error("telemetryOpt error", err); - }, - ); - return { - ...state, - telemetry: action.payload, - }; - }), - userMute: create.reducer((state, action: { payload: boolean }) => { Storage.setItem(USER_MUTED_KEY, JSON.stringify(action.payload)).catch( (err) => { @@ -248,7 +224,6 @@ export const streamplaceSlice = createAppSlice({ selectUrl: (streamplace) => streamplace.url, selectRecentSegments: (streamplace) => streamplace.recentSegments, selectMySegments: (streamplace) => streamplace.mySegments, - selectTelemetry: (streamplace) => streamplace.telemetry, selectUserMuted: (streamplace) => streamplace.userMuted, selectChatWarned: (streamplace) => streamplace.chatWarned, }, @@ -260,14 +235,12 @@ export const { setURL, initialize, pollMySegments, - telemetryOpt, userMute, chatWarn, } = streamplaceSlice.actions; export const { selectStreamplace, selectMySegments, - selectTelemetry, selectUserMuted, selectChatWarned, selectUrl, diff --git a/js/app/src/router.tsx b/js/app/src/router.tsx index 58968aea..3b4de890 100644 --- a/js/app/src/router.tsx +++ b/js/app/src/router.tsx @@ -12,6 +12,7 @@ import { NavigatorScreenParams, useLinkTo, useNavigation, + useRoute, } from "@react-navigation/native"; import { createNativeStackNavigator } from "@react-navigation/native-stack"; import { @@ -49,7 +50,7 @@ import { pollMySegments } from "features/streamplace/streamplaceSlice"; import { useLiveUser } from "hooks/useLiveUser"; import usePlatform from "hooks/usePlatform"; import { useSidebarControl } from "hooks/useSidebarControl"; -import { ReactElement, useEffect, useState } from "react"; +import { Fragment, ReactElement, useEffect, useState } from "react"; import { ImageBackground, ImageSourcePropType, @@ -332,8 +333,9 @@ export function StreamplaceDrawer() { const userIsLive = useLiveUser(); const toast = useToastController(); + const [isLiveDashboard, setIsLiveDashboard] = useState(true); useEffect(() => { - if (userIsLive && !poppedUp) { + if (!isLiveDashboard && userIsLive && !poppedUp) { setPoppedUp(true); setLivePopup(true); } @@ -358,7 +360,13 @@ export function StreamplaceDrawer() { : undefined, }, // rest - headerLeft: () => , + headerLeft: () => ( + <> + {/* this is a hack to give the popup the navigator context */} + + + + ), headerRight: () => , drawerActiveTintColor: theme.accentColor.val, unmountOnBlur: true, @@ -548,6 +556,22 @@ export function StreamplaceDrawer() { ); } +export const PopupChecker = ({ + setIsLiveDashboard, +}: { + setIsLiveDashboard: (isLiveDashboard: boolean) => void; +}) => { + const route = useRoute(); + useEffect(() => { + if (route.name === "LiveDashboard") { + setIsLiveDashboard(true); + } else { + setIsLiveDashboard(false); + } + }, [route.name]); + return ; +}; + const MainTab = () => { const theme = useTheme(); const { isWeb } = usePlatform(); diff --git a/js/app/src/screens/live-dashboard.tsx b/js/app/src/screens/live-dashboard.tsx index 2505be9a..996119eb 100644 --- a/js/app/src/screens/live-dashboard.tsx +++ b/js/app/src/screens/live-dashboard.tsx @@ -7,17 +7,20 @@ import StreamKeyScreen from "components/live-dashboard/stream-key"; import Waiting from "components/live-dashboard/waiting"; import Loading from "components/loading/loading"; import { Player } from "components/player/player"; +import Popup from "components/popup"; import ButtonSelector from "components/ui/button-selector"; import { VideoElementProvider } from "contexts/VideoElementContext"; import { + createServerSettingsRecord, + getServerSettingsFromPDS, selectIsReady, + selectServerSettings, selectUserProfile, } from "features/bluesky/blueskySlice"; -import { selectTelemetry } from "features/streamplace/streamplaceSlice"; import { useLiveUser } from "hooks/useLiveUser"; -import React, { useCallback, useState } from "react"; -import { useAppSelector } from "store/hooks"; -import { Button, H6, isWeb, Text, View } from "tamagui"; +import React, { useCallback, useEffect, useState } from "react"; +import { useAppDispatch, useAppSelector } from "store/hooks"; +import { Button, H3, H6, isWeb, Text, View } from "tamagui"; enum StreamSource { Start, @@ -29,11 +32,26 @@ export default function LiveDashboard() { const isReady = useAppSelector(selectIsReady); const userProfile = useAppSelector(selectUserProfile); const [streamSource, setStreamSource] = useState(StreamSource.Start); + const serverSettings = useAppSelector(selectServerSettings); const isLive = useLiveUser(); - const telemetry = useAppSelector(selectTelemetry); const [videoElement, setVideoElement] = useState( null, ); + const [gotSettings, setGotSettings] = useState(false); + const dispatch = useAppDispatch(); + useEffect(() => { + if (isReady) { + (async () => { + await dispatch(getServerSettingsFromPDS()); + setGotSettings(true); + })(); + } + }, [isReady]); + + let madeChoiceAboutDebugRecording = true; + if (gotSettings && serverSettings?.debugRecording === undefined) { + madeChoiceAboutDebugRecording = false; + } const [page, setPage] = useState<"update" | "create">("create"); @@ -57,7 +75,6 @@ export default function LiveDashboard() { if (isLive && streamSource !== StreamSource.Camera) { topPane = ( ); } + return ( @@ -116,6 +134,7 @@ export default function LiveDashboard() { {page === "update" && isLive ? : null} {page === "create" ? : null} + {madeChoiceAboutDebugRecording ? null : } @@ -135,6 +154,47 @@ const elems = [ }, ]; +export function DebugRecordingPopup() { + const dispatch = useAppDispatch(); + const serverSettings = useAppSelector(selectServerSettings) || {}; + const opt = (choice) => () => + dispatch( + createServerSettingsRecord({ + ...serverSettings, + debugRecording: choice, + }), + ); + return ( + +

Debug Recording

+ + Streamplace is beta software and it helps us to archive livestreams so + we can later use them for debugging. Would you like to opt in to debug + recording? + + + + + + + ); +} + export function StreamSourcePicker({ onPick, }: { diff --git a/js/docs/src/content/docs/lex-reference/server/place-stream-server-settings.md b/js/docs/src/content/docs/lex-reference/server/place-stream-server-settings.md new file mode 100644 index 00000000..75858378 --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/server/place-stream-server-settings.md @@ -0,0 +1,52 @@ +--- +title: place.stream.server.settings +description: Reference for the place.stream.server.settings lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `record` + +Record containing user settings for a particular Streamplace node + +**Record Key:** `any` + +**Record Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ---------------- | --------- | ----- | ----------------------------------------------------------------------- | ----------- | +| `debugRecording` | `boolean` | ❌ | Whether this node may archive your livestream for improving the service | | + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.server.settings", + "defs": { + "main": { + "type": "record", + "description": "Record containing user settings for a particular Streamplace node", + "key": "any", + "record": { + "type": "object", + "required": [], + "properties": { + "debugRecording": { + "type": "boolean", + "description": "Whether this node may archive your livestream for improving the service" + } + } + } + } + } +} +``` diff --git a/lexicons/place/stream/server/settings.json b/lexicons/place/stream/server/settings.json new file mode 100644 index 00000000..2b9e3a79 --- /dev/null +++ b/lexicons/place/stream/server/settings.json @@ -0,0 +1,21 @@ +{ + "lexicon": 1, + "id": "place.stream.server.settings", + "defs": { + "main": { + "type": "record", + "description": "Record containing user settings for a particular Streamplace node", + "key": "any", + "record": { + "type": "object", + "required": [], + "properties": { + "debugRecording": { + "type": "boolean", + "description": "Whether this node may archive your livestream for improving the service" + } + } + } + } + } +} diff --git a/pkg/api/api_internal.go b/pkg/api/api_internal.go index 43bf48ef..d5eb8b65 100644 --- a/pkg/api/api_internal.go +++ b/pkg/api/api_internal.go @@ -20,6 +20,7 @@ import ( "time" "github.com/julienschmidt/httprouter" + "github.com/pion/webrtc/v4" "github.com/prometheus/client_golang/prometheus/promhttp" sloghttp "github.com/samber/slog-http" "golang.org/x/sync/errgroup" @@ -31,6 +32,7 @@ import ( "stream.place/streamplace/pkg/model" notificationpkg "stream.place/streamplace/pkg/notifications" "stream.place/streamplace/pkg/renditions" + "stream.place/streamplace/pkg/rtcrec" v0 "stream.place/streamplace/pkg/schema/v0" ) @@ -603,6 +605,34 @@ func (a *StreamplaceAPI) InternalHandler(ctx context.Context) (http.Handler, err } }) + router.POST("/replay/:streamKey", func(w http.ResponseWriter, r *http.Request, p httprouter.Params) { + key := p.ByName("streamKey") + if key == "" { + errors.WriteHTTPBadRequest(w, "streamKey required", nil) + return + } + mediaSigner, err := a.MakeMediaSigner(ctx, key) + if err != nil { + errors.WriteHTTPUnauthorized(w, "invalid authorization key", err) + return + } + pc, err := rtcrec.NewReplayPeerConnection(ctx, r.Body) + if err != nil { + errors.WriteHTTPInternalServerError(w, "unable to create replay peer connection", err) + return + } + answer, err := a.MediaManager.WebRTCIngest(ctx, &webrtc.SessionDescription{SDP: "placeholder"}, mediaSigner, pc) + if err != nil { + errors.WriteHTTPInternalServerError(w, "unable to ingest web rtc", err) + return + } + w.WriteHeader(200) + if _, err := w.Write([]byte(answer.SDP)); err != nil { + errors.WriteHTTPInternalServerError(w, "unable to write response", err) + log.Error(ctx, "error writing response", "error", err) + } + }) + handler := sloghttp.Recovery(router) if log.Level(4) { handler = sloghttp.New(slog.Default())(handler) diff --git a/pkg/api/playback.go b/pkg/api/playback.go index 50fa0a7e..ddb063dd 100644 --- a/pkg/api/playback.go +++ b/pkg/api/playback.go @@ -219,7 +219,12 @@ func (a *StreamplaceAPI) HandleWebRTCIngest(ctx context.Context) httprouter.Hand return } offer := webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(body)} - answer, err := a.MediaManager.WebRTCIngest(ctx, &offer, mediaSigner) + pc, err := a.MediaManager.NewPeerConnection(ctx, mediaSigner.Streamer()) + if err != nil { + errors.WriteHTTPInternalServerError(w, "unable to create peer connection", err) + return + } + answer, err := a.MediaManager.WebRTCIngest(ctx, &offer, mediaSigner, pc) if err != nil { errors.WriteHTTPInternalServerError(w, "error playing back", err) return diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go index dbe1c382..99e157c9 100644 --- a/pkg/atproto/firehose.go +++ b/pkg/atproto/firehose.go @@ -148,6 +148,7 @@ var CollectionFilter = []string{ constants.APP_BSKY_GRAPH_FOLLOW, constants.APP_BSKY_FEED_POST, constants.APP_BSKY_GRAPH_BLOCK, + constants.PLACE_STREAM_SERVER_SETTINGS, } func (atsync *ATProtoSynchronizer) handleCommitEventOps(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) { diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index 588197c5..efa11549 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -163,6 +163,21 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD log.Error(ctx, "failed to create chat profile", "err", err) } + case *streamplace.ServerSettings: + _, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) + if err != nil { + return fmt.Errorf("failed to sync bluesky repo: %w", err) + } + settings := &model.ServerSettings{ + Server: rkey.String(), + RepoDID: userDID, + Record: recCBOR, + } + err = atsync.Model.UpdateServerSettings(ctx, settings) + if err != nil { + log.Error(ctx, "failed to create server settings", "err", err) + } + case *bsky.FeedPost: // jsonData, err := json.Marshal(d) // if err != nil { diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go index d6394d40..3352c51f 100644 --- a/pkg/constants/constants.go +++ b/pkg/constants/constants.go @@ -1,13 +1,14 @@ package constants -var PLACE_STREAM_KEY = "place.stream.key" //nolint:all -var PLACE_STREAM_LIVESTREAM = "place.stream.livestream" //nolint:all -var PLACE_STREAM_CHAT_MESSAGE = "place.stream.chat.message" //nolint:all -var PLACE_STREAM_CHAT_PROFILE = "place.stream.chat.profile" //nolint:all -var STREAMPLACE_SIGNING_KEY = "signingKey" //nolint:all -var APP_BSKY_GRAPH_FOLLOW = "app.bsky.graph.follow" //nolint:all -var APP_BSKY_FEED_POST = "app.bsky.feed.post" //nolint:all -var APP_BSKY_GRAPH_BLOCK = "app.bsky.graph.block" //nolint:all +var PLACE_STREAM_KEY = "place.stream.key" //nolint:all +var PLACE_STREAM_LIVESTREAM = "place.stream.livestream" //nolint:all +var PLACE_STREAM_CHAT_MESSAGE = "place.stream.chat.message" //nolint:all +var PLACE_STREAM_CHAT_PROFILE = "place.stream.chat.profile" //nolint:all +var PLACE_STREAM_SERVER_SETTINGS = "place.stream.server.settings" //nolint:all +var STREAMPLACE_SIGNING_KEY = "signingKey" //nolint:all +var APP_BSKY_GRAPH_FOLLOW = "app.bsky.graph.follow" //nolint:all +var APP_BSKY_FEED_POST = "app.bsky.feed.post" //nolint:all +var APP_BSKY_GRAPH_BLOCK = "app.bsky.graph.block" //nolint:all const DID_KEY_PREFIX = "did:key" //nolint:all const ADDRESS_KEY_PREFIX = "0x" //nolint:all diff --git a/pkg/gen/gen.go b/pkg/gen/gen.go index 2beb194d..9dbd6b2a 100644 --- a/pkg/gen/gen.go +++ b/pkg/gen/gen.go @@ -23,6 +23,7 @@ func main() { streamplace.ChatProfile{}, streamplace.ChatProfile_Color{}, streamplace.ChatMessage_ReplyRef{}, + streamplace.ServerSettings{}, ); err != nil { panic(err) } diff --git a/pkg/media/peer_connection.go b/pkg/media/peer_connection.go new file mode 100644 index 00000000..35b2dab3 --- /dev/null +++ b/pkg/media/peer_connection.go @@ -0,0 +1,33 @@ +package media + +import ( + "context" + + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/rtcrec" +) + +func (mm *MediaManager) NewPeerConnection(ctx context.Context, user string) (rtcrec.PeerConnection, error) { + shouldRecord := false + settings, err := mm.model.GetServerSettings(ctx, mm.cli.PublicHost, user) + if err != nil { + return nil, err + } + if settings != nil { + spsettings, err := settings.ToStreamplaceServerSettings() + if err != nil { + return nil, err + } + if spsettings.DebugRecording != nil { + shouldRecord = *spsettings.DebugRecording + } + } + if !shouldRecord { + log.Warn(ctx, "no server settings found, will not record") + } + pionpc, err := mm.webrtcAPI.NewPeerConnection(mm.webrtcConfig) + if err != nil { + return nil, err + } + return rtcrec.NewRecordingPeerConnection(ctx, *mm.cli, user, pionpc, shouldRecord) +} diff --git a/pkg/media/webrtc_ingest.go b/pkg/media/webrtc_ingest.go index a73cb87f..200bfbc1 100644 --- a/pkg/media/webrtc_ingest.go +++ b/pkg/media/webrtc_ingest.go @@ -12,22 +12,17 @@ import ( "github.com/pion/rtcp" "github.com/pion/webrtc/v4" "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/rtcrec" ) // This function remains in scope for the duration of a single users' playback -func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionDescription, signer MediaSigner) (*webrtc.SessionDescription, error) { +func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionDescription, signer MediaSigner, peerConnection rtcrec.PeerConnection) (*webrtc.SessionDescription, error) { uu, err := uuid.NewV7() if err != nil { return nil, err } - ctx = log.WithLogValues(ctx, "webrtcID", uu.String(), "mediafunc", "WebRTCIngest") - - // Create a new RTCPeerConnection - peerConnection, err := mm.webrtcAPI.NewPeerConnection(mm.webrtcConfig) - if err != nil { - return nil, fmt.Errorf("failed to create WebRTC peer connection: %w", err) - } + ctx = log.WithLogValues(ctx, "webrtcID", uu.String(), "mediafunc", "WebRTCIngest", "streamer", signer.Streamer()) // Allow us to receive 1 audio track, and 1 video track if _, err = peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio); err != nil { @@ -116,7 +111,7 @@ func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionD } // Create channel that is blocked until ICE Gathering is complete - gatherComplete := webrtc.GatheringCompletePromise(peerConnection) + gatherComplete := rtcrec.GatheringCompletePromise(peerConnection) go func() { ticker := time.NewTicker(time.Second * 1) @@ -186,7 +181,7 @@ func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionD audioFirst := false log.Warn(ctx, "setting OnTrack") - peerConnection.OnTrack(func(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) { + peerConnection.OnTrack(func(track rtcrec.TrackRemote, _ rtcrec.RTPReceiver) { log.Warn(ctx, "OnTrack") if track.Kind() == webrtc.RTPCodecTypeVideo { // Send a PLI on an interval so that the publisher is pushing a keyframe every rtcpPLIInterval diff --git a/pkg/model/model.go b/pkg/model/model.go index ed3f3311..378c6921 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -92,6 +92,10 @@ type Model interface { UpdateOAuthSession(id string, session *oatproxy.OAuthSession) error ListOAuthSessions() ([]oatproxy.OAuthSession, error) GetSessionByDID(did string) (*oatproxy.OAuthSession, error) + + UpdateServerSettings(ctx context.Context, settings *ServerSettings) error + GetServerSettings(ctx context.Context, server string, repoDID string) (*ServerSettings, error) + DeleteServerSettings(ctx context.Context, server string, repoDID string) error } func MakeDB(dbURL string) (Model, error) { @@ -151,6 +155,7 @@ func MakeDB(dbURL string) (Model, error) { ChatMessage{}, ChatProfile{}, oatproxy.OAuthSession{}, + ServerSettings{}, } { err = db.AutoMigrate(model) if err != nil { diff --git a/pkg/model/server_settings.go b/pkg/model/server_settings.go new file mode 100644 index 00000000..154fae4e --- /dev/null +++ b/pkg/model/server_settings.go @@ -0,0 +1,70 @@ +package model + +import ( + "context" + "errors" + "fmt" + "time" + + lexutil "github.com/bluesky-social/indigo/lex/util" + "gorm.io/gorm" + "stream.place/streamplace/pkg/streamplace" +) + +// ServerSettings represents a user's settings for a particular Streamplace node +type ServerSettings struct { + Server string `gorm:"primaryKey;column:server"` + RepoDID string `gorm:"primaryKey;column:repo_did"` + Record *[]byte `gorm:"column:record"` + Created time.Time `gorm:"column:created;not null"` + Updated time.Time `gorm:"column:updated;not null"` +} + +// TableName specifies the table name for the ServerSettings model +func (ServerSettings) TableName() string { + return "server_settings" +} + +// ToStreamplaceServerSettings converts the model to a streamplace ServerSettings +func (m *ServerSettings) ToStreamplaceServerSettings() (*streamplace.ServerSettings, error) { + if m.Record == nil { + return nil, fmt.Errorf("no record data") + } + rec, err := lexutil.CborDecodeValue(*m.Record) + if err != nil { + return nil, fmt.Errorf("error decoding server settings: %w", err) + } + ss, ok := rec.(*streamplace.ServerSettings) + if !ok { + return nil, fmt.Errorf("invalid server settings") + } + return ss, nil +} + +// UpdateServerSettings creates or updates a server settings record +func (m *DBModel) UpdateServerSettings(ctx context.Context, settings *ServerSettings) error { + now := time.Now() + if settings.Created.IsZero() { + settings.Created = now + } + settings.Updated = now + return m.DB.Save(settings).Error +} + +// GetServerSettings retrieves server settings for a given server and repoDID +func (m *DBModel) GetServerSettings(ctx context.Context, server string, repoDID string) (*ServerSettings, error) { + var settings ServerSettings + err := m.DB.Where("server = ? AND repo_did = ?", server, repoDID).First(&settings).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &settings, nil +} + +// DeleteServerSettings deletes server settings for a given server and repoDID +func (m *DBModel) DeleteServerSettings(ctx context.Context, server string, repoDID string) error { + return m.DB.Where("server = ? AND repo_did = ?", server, repoDID).Delete(&ServerSettings{}).Error +} diff --git a/pkg/rtcrec/cbor.go b/pkg/rtcrec/cbor.go new file mode 100644 index 00000000..007505d1 --- /dev/null +++ b/pkg/rtcrec/cbor.go @@ -0,0 +1,221 @@ +package rtcrec + +import ( + "fmt" + "io" + "sort" + "time" + + "github.com/fxamacker/cbor/v2" + "github.com/pion/webrtc/v4" +) + +type WebRTCEventDecoder struct { + dec *cbor.Decoder +} + +func Opts() (cbor.EncMode, error) { + opts := cbor.CoreDetEncOptions() + opts.Time = cbor.TimeRFC3339Nano + em, err := opts.EncMode() + if err != nil { + return nil, fmt.Errorf("failed to create encoder mode: %w", err) + } + return em, nil +} + +func MakeWebRTCDecoder(r io.Reader) (*WebRTCEventDecoder, error) { + dec := cbor.NewDecoder(r) + return &WebRTCEventDecoder{dec: dec}, nil +} + +func (d *WebRTCEventDecoder) Next() (*WebRTCEvent, error) { + var ev WebRTCEvent + err := d.dec.Decode(&ev) + if err != nil { + return nil, err + } + return &ev, err +} + +type WebRTCEventGroup struct { + Events map[string][]*WebRTCEvent + Tracks map[webrtc.SSRC]map[string][]*WebRTCEvent + FirstTime time.Time +} + +const ( + EventTypeOffer = "Offer" + EventTypeCreateAnswer = "CreateAnswer" + EventTypeSetRemoteDescription = "SetRemoteDescription" + EventTypeSetLocalDescription = "SetLocalDescription" + EventTypeLocalDescription = "LocalDescription" + EventTypeICEConnectionState = "ICEConnectionStateChange" + EventTypeConnectionState = "ConnectionStateChange" + EventTypeTrack = "Track" + EventTypeAddTransceiverFromKind = "AddTransceiverFromKind" + EventTypeICEGatheringState = "ICEGatheringState" + EventTypeDataChannel = "DataChannel" + EventTypeNegotiationNeeded = "NegotiationNeeded" + EventTypeTrackRead = "TrackRead" + EventTypeTrackCodec = "TrackCodec" + EventTypeTrackKind = "TrackKind" + EventTypeTrackPayloadType = "TrackPayloadType" + EventTypeTrackSSRC = "TrackSSRC" + EventTypeUnknown = "Unknown" +) + +// ReadAllEvents reads all WebRTC events from a CBOR reader and organizes them by type. +// Returns a map where keys are event types and values are slices of events of that type. +func ReadAllEvents(r io.Reader) (*WebRTCEventGroup, error) { + dec, err := MakeWebRTCDecoder(r) + if err != nil { + return nil, err + } + + eventList := []*WebRTCEvent{} + events := make(map[string][]*WebRTCEvent) + tracks := make(map[webrtc.SSRC]map[string][]*WebRTCEvent) + var firstTime time.Time + for { + ev, err := dec.Next() + if err == io.EOF { + break + } + if err != nil { + return nil, err + } + eventList = append(eventList, ev) + } + + sort.Slice(eventList, func(i, j int) bool { + return eventList[i].Time.Before(eventList[j].Time) + }) + + for _, ev := range eventList { + if firstTime.IsZero() { + firstTime = ev.Time + } + + // Determine the event type based on which field is non-nil + var eventType string + var trackSSRC *webrtc.SSRC + switch { + case ev.Offer != nil: + eventType = EventTypeOffer + case ev.CreateAnswer != nil: + eventType = EventTypeCreateAnswer + case ev.SetRemoteDescription != nil: + eventType = EventTypeSetRemoteDescription + case ev.SetLocalDescription != nil: + eventType = EventTypeSetLocalDescription + case ev.LocalDescription != nil: + eventType = EventTypeLocalDescription + case ev.ICEConnectionStateChange != nil: + eventType = EventTypeICEConnectionState + case ev.ConnectionStateChange != nil: + eventType = EventTypeConnectionState + case ev.Track != nil: + eventType = EventTypeTrack + case ev.AddTransceiverFromKind != nil: + eventType = EventTypeAddTransceiverFromKind + case ev.ICEGatheringState != nil: + eventType = EventTypeICEGatheringState + case ev.DataChannel != nil: + eventType = EventTypeDataChannel + case ev.NegotiationNeeded != nil: + eventType = EventTypeNegotiationNeeded + + case ev.TrackRead != nil: + trackSSRC = &ev.TrackRead.SSRC + eventType = EventTypeTrackRead + case ev.TrackCodec != nil: + trackSSRC = &ev.TrackCodec.SSRC + eventType = EventTypeTrackCodec + case ev.TrackKind != nil: + trackSSRC = &ev.TrackKind.SSRC + eventType = EventTypeTrackKind + case ev.TrackPayloadType != nil: + trackSSRC = &ev.TrackPayloadType.SSRC + eventType = EventTypeTrackPayloadType + case ev.TrackSSRC != nil: + trackSSRC = &ev.TrackSSRC.SSRC + eventType = EventTypeTrackSSRC + default: + eventType = EventTypeUnknown + panic(fmt.Sprintf("unknown event type: %+v", ev)) + } + + if trackSSRC != nil { + if tracks[*trackSSRC] == nil { + tracks[*trackSSRC] = make(map[string][]*WebRTCEvent) + } + if tracks[*trackSSRC][eventType] == nil { + tracks[*trackSSRC][eventType] = []*WebRTCEvent{} + } + tracks[*trackSSRC][eventType] = append(tracks[*trackSSRC][eventType], ev) + } + + if events[eventType] == nil { + events[eventType] = []*WebRTCEvent{} + } + + events[eventType] = append(events[eventType], ev) + } + + return &WebRTCEventGroup{ + Events: events, + Tracks: tracks, + FirstTime: firstTime, + }, nil +} + +func (g *WebRTCEventGroup) Peek(eventType string) *WebRTCEvent { + if g.Events[eventType] == nil { + panic(fmt.Sprintf("no events of type %s", eventType)) + } + if len(g.Events[eventType]) == 0 { + return nil + } + return g.Events[eventType][0] +} + +func (g *WebRTCEventGroup) PeekTrack(ssrc webrtc.SSRC, eventType string) *WebRTCEvent { + if g.Tracks[ssrc] == nil { + panic(fmt.Sprintf("no tracks for ssrc %d", ssrc)) + } + if g.Tracks[ssrc][eventType] == nil { + panic(fmt.Sprintf("no events of type %s for ssrc %d", eventType, ssrc)) + } + if len(g.Tracks[ssrc][eventType]) == 0 { + return nil + } + return g.Tracks[ssrc][eventType][0] +} + +func (g *WebRTCEventGroup) Next(eventType string) *WebRTCEvent { + if g.Events[eventType] == nil { + panic(fmt.Sprintf("no events of type %s", eventType)) + } + if len(g.Events[eventType]) == 0 { + return nil + } + ev := g.Events[eventType][0] + g.Events[eventType] = g.Events[eventType][1:] + return ev +} + +func (g *WebRTCEventGroup) NextTrack(ssrc webrtc.SSRC, eventType string) *WebRTCEvent { + if g.Tracks[ssrc] == nil { + panic(fmt.Sprintf("no tracks for ssrc %d", ssrc)) + } + if g.Tracks[ssrc][eventType] == nil { + panic(fmt.Sprintf("no events of type %s for ssrc %d", eventType, ssrc)) + } + if len(g.Tracks[ssrc][eventType]) == 0 { + return nil + } + ev := g.Tracks[ssrc][eventType][0] + g.Tracks[ssrc][eventType] = g.Tracks[ssrc][eventType][1:] + return ev +} diff --git a/pkg/rtcrec/cmd/rtcrec.go b/pkg/rtcrec/cmd/rtcrec.go new file mode 100644 index 00000000..96efa30d --- /dev/null +++ b/pkg/rtcrec/cmd/rtcrec.go @@ -0,0 +1,77 @@ +package main + +import ( + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "log" + "os" + + "stream.place/streamplace/pkg/rtcrec" +) + +func main() { + err := Start() + if err != nil { + log.Fatal(err) + } +} + +func Start() error { + var path string + flag.StringVar(&path, "path", "", "path to the file to decode") + flag.Parse() + if path == "" { + return fmt.Errorf("path is required") + } + return DecodeFile(path) +} + +func DecodeFile(path string) error { + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + dec, err := rtcrec.MakeWebRTCDecoder(f) + if err != nil { + return err + } + for { + ev, err := dec.Next() + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return err + } + if ev.TrackRead != nil { + // spitting out the data as base64 is pointless, replace with a label + n := len(ev.TrackRead.Data) + byteString := fmt.Sprintf("%d bytes", n) + bs, err := json.Marshal(ev) + if err != nil { + return err + } + var m map[string]any + err = json.Unmarshal(bs, &m) + if err != nil { + return err + } + m["trackRead"].(map[string]any)["data"] = byteString + bs, err = json.Marshal(m) + if err != nil { + return err + } + fmt.Println(string(bs)) + } else { + bs, err := json.Marshal(ev) + if err != nil { + return err + } + fmt.Println(string(bs)) + } + } +} diff --git a/pkg/rtcrec/recording_peerconnection.go b/pkg/rtcrec/recording_peerconnection.go new file mode 100644 index 00000000..f5230b78 --- /dev/null +++ b/pkg/rtcrec/recording_peerconnection.go @@ -0,0 +1,236 @@ +package rtcrec + +import ( + "context" + "fmt" + "os" + "time" + + "github.com/pion/rtcp" + "github.com/pion/webrtc/v4" + "stream.place/streamplace/pkg/aqtime" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/log" +) + +type RecordingPeerConnection struct { + enabled bool + pionpc *webrtc.PeerConnection + file *os.File + stream *RecorderStream +} + +func NewRecordingPeerConnection(ctx context.Context, cli config.CLI, user string, pionpc *webrtc.PeerConnection, enabled bool) (PeerConnection, error) { + if !enabled { + return &RecordingPeerConnection{ + pionpc: pionpc, + enabled: enabled, + }, nil + } + aqt := aqtime.FromTime(time.Now()) + f, err := cli.DataFileCreate([]string{user, "rtcrec", fmt.Sprintf("%s.cbor", aqt.FileSafeString())}, true) + if err != nil { + return nil, fmt.Errorf("failed to create data file: %w", err) + } + log.Log(ctx, "logging webrtc session to file", "file", f.Name()) + stream, err := NewRecorderStream(f) + if err != nil { + return nil, fmt.Errorf("failed to create recorder stream: %w", err) + } + return &RecordingPeerConnection{ + pionpc: pionpc, + file: f, + stream: stream, + enabled: enabled, + }, nil +} + +func (pc *RecordingPeerConnection) Do(f func()) { + if pc.enabled { + go f() + } +} + +func (pc *RecordingPeerConnection) Close() error { + return pc.pionpc.Close() +} + +func (pc *RecordingPeerConnection) CreateAnswer(options *webrtc.AnswerOptions) (webrtc.SessionDescription, error) { + now := time.Now() + ret, err := pc.pionpc.CreateAnswer(options) + if err != nil { + return ret, err + } + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + CreateAnswer: &CreateAnswer{ + SDPAnswer: ret.SDP, + }, + Time: now, + }) + }) + return ret, nil +} + +func (pc *RecordingPeerConnection) SetLocalDescription(desc webrtc.SessionDescription) error { + now := time.Now() + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + SetRemoteDescription: &SetRemoteDescription{ + SDPRemoteDescription: desc.SDP, + }, + Time: now, + }) + }) + return pc.pionpc.SetLocalDescription(desc) +} + +func (pc *RecordingPeerConnection) SetRemoteDescription(desc webrtc.SessionDescription) error { + now := time.Now() + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + SetRemoteDescription: &SetRemoteDescription{ + SDPRemoteDescription: desc.SDP, + }, + Time: now, + }) + }) + return pc.pionpc.SetRemoteDescription(desc) +} + +func (pc *RecordingPeerConnection) LocalDescription() *webrtc.SessionDescription { + now := time.Now() + desc := pc.pionpc.LocalDescription() + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + LocalDescription: &LocalDescription{ + SDPLocalDescription: pc.pionpc.LocalDescription().SDP, + }, + Time: now, + }) + }) + return desc +} + +// func (pc *RecorderPeerConnection) RemoteDescription() *webrtc.SessionDescription { +// return pc.pionpc.RemoteDescription() +// } + +func (pc *RecordingPeerConnection) OnICEConnectionStateChange(f func(webrtc.ICEConnectionState)) { + pc.pionpc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) { + now := time.Now() + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + ICEConnectionStateChange: &ICEConnectionStateChange{ + ICEConnectionState: state, + }, + Time: now, + }) + }) + f(state) + }) +} + +func (pc *RecordingPeerConnection) OnConnectionStateChange(f func(webrtc.PeerConnectionState)) { + pc.pionpc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { + now := time.Now() + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + ConnectionStateChange: &ConnectionStateChange{ + ConnectionState: state, + }, + Time: now, + }) + }) + f(state) + }) +} + +func (pc *RecordingPeerConnection) OnTrack(f func(TrackRemote, RTPReceiver)) { + pc.pionpc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) { + now := time.Now() + wrappedTrack := &RecordingTrackRemote{track: track, stream: pc.stream, pc: pc} + id := track.ID() + kind := track.Kind() + ssrc := track.SSRC() + payloadType := track.PayloadType() + streamID := track.StreamID() + msid := track.Msid() + rid := track.RID() + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + Track: &Track{ + ID: id, + Kind: kind, + SSRC: ssrc, + PayloadType: payloadType, + StreamID: streamID, + Msid: msid, + RID: rid, + }, + Time: now, + }) + }) + f(wrappedTrack, receiver) + }) +} + +func (pc *RecordingPeerConnection) WriteRTCP(pkts []rtcp.Packet) error { + return pc.pionpc.WriteRTCP(pkts) +} + +func (pc *RecordingPeerConnection) AddTransceiverFromKind(kind webrtc.RTPCodecType, init ...webrtc.RTPTransceiverInit) (RTPTransceiver, error) { + now := time.Now() + ret, err := pc.pionpc.AddTransceiverFromKind(kind, init...) + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + AddTransceiverFromKind: &AddTransceiverFromKind{ + Kind: kind, + }, + Time: now, + }) + }) + return ret, err +} + +func (pc *RecordingPeerConnection) ICEGatheringState() webrtc.ICEGatheringState { + now := time.Now() + state := pc.pionpc.ICEGatheringState() + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + ICEGatheringState: &ICEGatheringState{ + State: state, + }, + Time: now, + }) + }) + return state +} + +func (pc *RecordingPeerConnection) OnDataChannel(f func(*webrtc.DataChannel)) { + pc.pionpc.OnDataChannel(func(dc *webrtc.DataChannel) { + now := time.Now() + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + DataChannel: &DataChannel{ + Label: dc.Label(), + }, + Time: now, + }) + }) + f(dc) + }) +} + +func (pc *RecordingPeerConnection) OnNegotiationNeeded(f func()) { + pc.pionpc.OnNegotiationNeeded(func() { + now := time.Now() + pc.Do(func() { + pc.stream.Event(WebRTCEvent{ + NegotiationNeeded: &NegotiationNeeded{}, + Time: now, + }) + }) + f() + }) +} diff --git a/pkg/rtcrec/recording_track.go b/pkg/rtcrec/recording_track.go new file mode 100644 index 00000000..a6730d1f --- /dev/null +++ b/pkg/rtcrec/recording_track.go @@ -0,0 +1,105 @@ +package rtcrec + +import ( + "time" + + "github.com/pion/interceptor" + "github.com/pion/webrtc/v4" +) + +type RecordingTrackRemote struct { + track *webrtc.TrackRemote + stream *RecorderStream + pc *RecordingPeerConnection +} + +func (t *RecordingTrackRemote) do(f func()) { + go f() +} + +func (t *RecordingTrackRemote) Read(p []byte) (n int, attrs interceptor.Attributes, err error) { + n, attrs, err = t.track.Read(p) + now := time.Now() + b2 := make([]byte, n) + copy(b2, p) + t.pc.Do(func() { + errString := "" + if err != nil { + errString = err.Error() + } + t.stream.Event(WebRTCEvent{ + TrackRead: &TrackRead{ + Data: b2, + SSRC: t.track.SSRC(), + Count: n, + // Attrs: attrs, + Err: errString, + }, + Time: now, + }) + }) + return n, attrs, err +} + +func (t *RecordingTrackRemote) Codec() webrtc.RTPCodecParameters { + now := time.Now() + codec := t.track.Codec() + t.pc.Do(func() { + t.stream.Event(WebRTCEvent{ + TrackCodec: &TrackCodec{ + SSRC: t.track.SSRC(), + Codec: codec, + }, + Time: now, + }) + }) + return codec +} + +func (t *RecordingTrackRemote) ID() string { + return t.track.ID() +} + +func (t *RecordingTrackRemote) Kind() webrtc.RTPCodecType { + now := time.Now() + kind := t.track.Kind() + t.pc.Do(func() { + t.stream.Event(WebRTCEvent{ + TrackKind: &TrackKind{ + SSRC: t.track.SSRC(), + Kind: kind, + }, + Time: now, + }) + }) + return kind +} + +func (t *RecordingTrackRemote) PayloadType() webrtc.PayloadType { + now := time.Now() + payloadType := t.track.PayloadType() + t.pc.Do(func() { + t.stream.Event(WebRTCEvent{ + TrackPayloadType: &TrackPayloadType{ + SSRC: t.track.SSRC(), + PayloadType: payloadType, + }, + Time: now, + }) + }) + return payloadType +} + +func (t *RecordingTrackRemote) SSRC() webrtc.SSRC { + now := time.Now() + ssrc := t.track.SSRC() + t.pc.Do(func() { + t.stream.Event(WebRTCEvent{ + Time: now, + TrackSSRC: &TrackSSRC{ + SSRC: ssrc, + }, + }) + }) + return ssrc +} diff --git a/pkg/rtcrec/replay_peerconnection.go b/pkg/rtcrec/replay_peerconnection.go new file mode 100644 index 00000000..835f31ed --- /dev/null +++ b/pkg/rtcrec/replay_peerconnection.go @@ -0,0 +1,148 @@ +package rtcrec + +import ( + "context" + "fmt" + "io" + "time" + + "github.com/pion/rtcp" + "github.com/pion/webrtc/v4" + "stream.place/streamplace/pkg/log" +) + +type ReplayPeerConnection struct { + startTime time.Time + group *WebRTCEventGroup + ctx context.Context +} + +func NewReplayPeerConnection(ctx context.Context, r io.Reader) (PeerConnection, error) { + group, err := ReadAllEvents(r) + if err != nil { + return nil, fmt.Errorf("failed to create web rtc decoder: %w", err) + } + + return &ReplayPeerConnection{ + startTime: time.Now(), + group: group, + ctx: context.Background(), + }, nil +} + +func (pc *ReplayPeerConnection) wait(label string, t time.Time) <-chan time.Time { + now := time.Now() + historicalDiff := t.Sub(pc.group.FirstTime) + currentDiff := time.Since(pc.startTime) + diff := historicalDiff - currentDiff + log.Debug(pc.ctx, "waiting for event", "event", label, "diff", diff, "t", t, "first", pc.group.FirstTime, "startTime", pc.startTime, "now", now) + return time.After(diff) +} + +func (pc *ReplayPeerConnection) Close() error { + // todo: implment stopping here + return nil +} + +func (pc *ReplayPeerConnection) CreateAnswer(options *webrtc.AnswerOptions) (webrtc.SessionDescription, error) { + ev := pc.group.Peek(EventTypeCreateAnswer) + if ev == nil { + return webrtc.SessionDescription{}, fmt.Errorf("no create answer event found") + } + return webrtc.SessionDescription{SDP: ev.CreateAnswer.SDPAnswer}, nil +} + +func (pc *ReplayPeerConnection) SetLocalDescription(desc webrtc.SessionDescription) error { + return nil +} + +func (pc *ReplayPeerConnection) SetRemoteDescription(desc webrtc.SessionDescription) error { + return nil +} + +func (pc *ReplayPeerConnection) LocalDescription() *webrtc.SessionDescription { + ev := pc.group.Peek(EventTypeLocalDescription) + if ev == nil { + return nil + } + return &webrtc.SessionDescription{SDP: ev.LocalDescription.SDPLocalDescription} +} + +// func (pc *ReplayPeerConnection) RemoteDescription() *webrtc.SessionDescription { +// return pc.pionpc.RemoteDescription() +// } + +func (pc *ReplayPeerConnection) OnICEConnectionStateChange(f func(webrtc.ICEConnectionState)) { + go func() { + for { + ev := pc.group.Next(EventTypeICEConnectionState) + if ev == nil { + return + } + select { + case <-pc.wait("OnICEConnectionStateChange", ev.Time): + f(ev.ICEConnectionStateChange.ICEConnectionState) + case <-pc.ctx.Done(): + return + } + } + }() +} + +func (pc *ReplayPeerConnection) OnConnectionStateChange(f func(webrtc.PeerConnectionState)) { + go func() { + for { + ev := pc.group.Next(EventTypeConnectionState) + if ev == nil { + return + } + select { + case <-pc.wait("OnConnectionStateChange", ev.Time): + f(ev.ConnectionStateChange.ConnectionState) + case <-pc.ctx.Done(): + return + } + } + }() +} + +func (pc *ReplayPeerConnection) OnTrack(f func(TrackRemote, RTPReceiver)) { + go func() { + for { + ev := pc.group.Next(EventTypeTrack) + if ev == nil { + return + } + select { + case <-pc.wait("OnTrack", ev.Time): + track := &ReplayTrackRemote{ + ssrc: ev.Track.SSRC, + trackEvent: ev, + events: pc.group.Tracks[ev.Track.SSRC], + pc: pc, + } + go func() { + f(track, nil) + }() + case <-pc.ctx.Done(): + return + } + } + }() +} + +func (pc *ReplayPeerConnection) WriteRTCP(pkts []rtcp.Packet) error { + return nil +} + +func (pc *ReplayPeerConnection) AddTransceiverFromKind(kind webrtc.RTPCodecType, init ...webrtc.RTPTransceiverInit) (RTPTransceiver, error) { + return nil, nil +} + +func (pc *ReplayPeerConnection) ICEGatheringState() webrtc.ICEGatheringState { + ev := pc.group.Peek(EventTypeICEGatheringState) + if ev == nil { + return webrtc.ICEGatheringStateNew + } + return ev.ICEGatheringState.State +} diff --git a/pkg/rtcrec/replay_track.go b/pkg/rtcrec/replay_track.go new file mode 100644 index 00000000..ee529711 --- /dev/null +++ b/pkg/rtcrec/replay_track.go @@ -0,0 +1,64 @@ +package rtcrec + +import ( + "errors" + "fmt" + + "github.com/pion/interceptor" + "github.com/pion/webrtc/v4" +) + +type ReplayTrackRemote struct { + ssrc webrtc.SSRC + trackEvent *WebRTCEvent + events map[string][]*WebRTCEvent + pc *ReplayPeerConnection +} + +func (t *ReplayTrackRemote) Read(p []byte) (n int, attrs interceptor.Attributes, err error) { + ev := t.pc.group.NextTrack(t.ssrc, EventTypeTrackRead) + if ev == nil { + return 0, nil, nil + } + select { + case <-t.pc.wait(fmt.Sprintf("TrackRead %s", t.trackEvent.Track.ID), ev.Time): + var err error + if ev.TrackRead.Err != "" { + err = errors.New(ev.TrackRead.Err) + } + if ev.TrackRead.Data != nil { + copied := copy(p, ev.TrackRead.Data) + if copied != len(ev.TrackRead.Data) { + panic("copied != len(ev.TrackRead.Data)") + } + } + // log.Log(t.pc.ctx, "TrackRead", "trackId", t.trackEvent.Track.ID, "count", ev.TrackRead.Count, "err", err) + return ev.TrackRead.Count, nil, err + case <-t.pc.ctx.Done(): + return 0, nil, t.pc.ctx.Err() + } +} + +func (t *ReplayTrackRemote) Codec() webrtc.RTPCodecParameters { + ev := t.pc.group.PeekTrack(t.ssrc, EventTypeTrackCodec) + if ev == nil { + panic("no codec found") + } + return ev.TrackCodec.Codec +} + +func (t *ReplayTrackRemote) ID() string { + return t.trackEvent.Track.ID +} + +func (t *ReplayTrackRemote) Kind() webrtc.RTPCodecType { + return t.trackEvent.Track.Kind +} + +func (t *ReplayTrackRemote) PayloadType() webrtc.PayloadType { + return t.trackEvent.Track.PayloadType +} + +func (t *ReplayTrackRemote) SSRC() webrtc.SSRC { + return t.ssrc +} diff --git a/pkg/rtcrec/rtcrec.go b/pkg/rtcrec/rtcrec.go new file mode 100644 index 00000000..1b63079d --- /dev/null +++ b/pkg/rtcrec/rtcrec.go @@ -0,0 +1,56 @@ +package rtcrec + +import ( + "time" + + "github.com/pion/interceptor" + "github.com/pion/rtcp" + "github.com/pion/webrtc/v4" +) + +type PeerConnection interface { + AddTransceiverFromKind(kind webrtc.RTPCodecType, init ...webrtc.RTPTransceiverInit) (RTPTransceiver, error) + Close() error + SetRemoteDescription(description webrtc.SessionDescription) error + CreateAnswer(options *webrtc.AnswerOptions) (webrtc.SessionDescription, error) + SetLocalDescription(description webrtc.SessionDescription) error + OnICEConnectionStateChange(func(webrtc.ICEConnectionState)) + OnConnectionStateChange(func(webrtc.PeerConnectionState)) + OnTrack(func(TrackRemote, RTPReceiver)) + // OnDataChannel(func(*webrtc.DataChannel)) + // OnNegotiationNeeded(func()) + WriteRTCP(pkts []rtcp.Packet) error + ICEGatheringState() webrtc.ICEGatheringState + LocalDescription() *webrtc.SessionDescription +} + +type RTPTransceiver interface { +} + +type TrackRemote interface { + Read(p []byte) (n int, attrs interceptor.Attributes, err error) + Kind() webrtc.RTPCodecType + PayloadType() webrtc.PayloadType + Codec() webrtc.RTPCodecParameters + SSRC() webrtc.SSRC +} + +type RTPReceiver interface { +} + +func GatheringCompletePromise(pc PeerConnection) <-chan struct{} { + recorder, ok := pc.(*RecordingPeerConnection) + if ok { + return webrtc.GatheringCompletePromise(recorder.pionpc) + } + _, ok = pc.(*ReplayPeerConnection) + if ok { + ch := make(chan struct{}) + go func() { + <-time.After(100 * time.Millisecond) + ch <- struct{}{} + }() + return ch + } + panic("unknown peer connection type") +} diff --git a/pkg/rtcrec/webrtc_recording.go b/pkg/rtcrec/webrtc_recording.go new file mode 100644 index 00000000..f477611c --- /dev/null +++ b/pkg/rtcrec/webrtc_recording.go @@ -0,0 +1,152 @@ +package rtcrec + +import ( + "context" + "fmt" + "io" + "time" + + "github.com/fxamacker/cbor/v2" + "github.com/pion/webrtc/v4" + "stream.place/streamplace/pkg/log" +) + +type WebRTCRecording struct { + Events []WebRTCEvent `json:"events,omitempty"` +} + +type WebRTCEvent struct { + Offer *Offer `json:"offer,omitempty"` + CreateAnswer *CreateAnswer `json:"answer,omitempty"` + SetRemoteDescription *SetRemoteDescription `json:"setRemoteDescription,omitempty"` + SetLocalDescription *SetLocalDescription `json:"setLocalDescription,omitempty"` + LocalDescription *LocalDescription `json:"localDescription,omitempty"` + ICEConnectionStateChange *ICEConnectionStateChange `json:"iceConnectionStateChange,omitempty"` + ConnectionStateChange *ConnectionStateChange `json:"connectionStateChange,omitempty"` + Track *Track `json:"track,omitempty"` + TrackRead *TrackRead `json:"trackRead,omitempty"` + TrackCodec *TrackCodec `json:"trackCodec,omitempty"` + TrackKind *TrackKind `json:"trackKind,omitempty"` + TrackPayloadType *TrackPayloadType `json:"trackPayloadType,omitempty"` + TrackSSRC *TrackSSRC `json:"trackSSRC,omitempty"` + AddTransceiverFromKind *AddTransceiverFromKind `json:"addTransceiverFromKind,omitempty"` + ICEGatheringState *ICEGatheringState `json:"iceGatheringState,omitempty"` + DataChannel *DataChannel `json:"dataChannel,omitempty"` + NegotiationNeeded *NegotiationNeeded `json:"negotiationNeeded,omitempty"` + Time time.Time `json:"time,omitempty"` +} + +func (e *WebRTCEvent) Detail() WebRTCEventDetail { + if e.Offer != nil { + return e.Offer + } + if e.CreateAnswer != nil { + return e.CreateAnswer + } + return nil +} + +type WebRTCEventDetail interface{} + +type Offer struct { + SDPOffer string `json:"sdpOffer,omitempty"` +} + +type CreateAnswer struct { + SDPAnswer string `json:"sdpAnswer,omitempty"` +} + +type SetRemoteDescription struct { + SDPRemoteDescription string `json:"sdpRemoteDescription,omitempty"` +} + +type SetLocalDescription struct { + SDPLocalDescription string `json:"sdpRemoteDescription,omitempty"` +} + +type LocalDescription struct { + SDPLocalDescription string `json:"sdpLocalDescription,omitempty"` +} + +type ICEConnectionStateChange struct { + ICEConnectionState webrtc.ICEConnectionState `json:"iceConnectionState,omitempty"` +} + +type ConnectionStateChange struct { + ConnectionState webrtc.PeerConnectionState `json:"connectionState,omitempty"` +} + +type Track struct { + ID string `json:"id,omitempty"` + Kind webrtc.RTPCodecType `json:"kind,omitempty"` + SSRC webrtc.SSRC `json:"ssrc,omitempty"` + PayloadType webrtc.PayloadType `json:"payloadType,omitempty"` + StreamID string `json:"streamId,omitempty"` + Msid string `json:"msid,omitempty"` + RID string `json:"rid,omitempty"` +} + +type TrackRead struct { + SSRC webrtc.SSRC `json:"ssrc,omitempty"` + Data []byte `json:"data,omitempty"` + Count int `json:"count,omitempty"` + Err string `json:"err,omitempty"` +} + +type TrackCodec struct { + SSRC webrtc.SSRC `json:"ssrc,omitempty"` + Codec webrtc.RTPCodecParameters `json:"codec,omitempty"` +} + +type TrackKind struct { + SSRC webrtc.SSRC `json:"ssrc,omitempty"` + Kind webrtc.RTPCodecType `json:"kind,omitempty"` +} + +type TrackPayloadType struct { + SSRC webrtc.SSRC `json:"ssrc,omitempty"` + PayloadType webrtc.PayloadType `json:"payloadType,omitempty"` +} + +type TrackSSRC struct { + SSRC webrtc.SSRC `json:"ssrc,omitempty"` +} + +type RecorderStream struct { + encoder *cbor.Encoder +} + +func NewRecorderStream(w io.Writer) (*RecorderStream, error) { + opts := cbor.CoreDetEncOptions() + opts.Time = cbor.TimeRFC3339Nano + em, err := opts.EncMode() + if err != nil { + return nil, fmt.Errorf("failed to create encoder mode: %w", err) + } + encoder := em.NewEncoder(w) + + return &RecorderStream{ + encoder: encoder, + }, nil +} + +func (s *RecorderStream) Event(event WebRTCEvent) { + err := s.encoder.Encode(event) + if err != nil { + log.Log(context.Background(), "error encoding event", "error", err) + } +} + +type AddTransceiverFromKind struct { + Kind webrtc.RTPCodecType `json:"kind,omitempty"` +} + +type ICEGatheringState struct { + State webrtc.ICEGatheringState `json:"state,omitempty"` +} + +type DataChannel struct { + Label string `json:"label,omitempty"` +} + +type NegotiationNeeded struct{} diff --git a/pkg/rtcrec/webrtc_recording_test.go b/pkg/rtcrec/webrtc_recording_test.go new file mode 100644 index 00000000..b8844076 --- /dev/null +++ b/pkg/rtcrec/webrtc_recording_test.go @@ -0,0 +1,72 @@ +package rtcrec + +import ( + "io" + "os" + "testing" + "time" + + "github.com/fxamacker/cbor/v2" + "github.com/stretchr/testify/require" +) + +func TestWebRTCRecording(t *testing.T) { + // Create a temporary file for testing + tmpfile, err := os.CreateTemp("", "webrtc-recording-test-*") + require.NoError(t, err) + + // Create recorder stream writing to temp file + recorder, err := NewRecorderStream(tmpfile) + require.NoError(t, err) + + // Test recording an offer event + offerEvent := WebRTCEvent{ + Offer: &Offer{ + SDPOffer: "test-offer", + }, + Time: time.Now().UTC(), + } + recorder.Event(offerEvent) + + // Test recording an answer event + answerEvent := WebRTCEvent{ + CreateAnswer: &CreateAnswer{ + SDPAnswer: "test-answer", + }, + Time: time.Now().UTC(), + } + recorder.Event(answerEvent) + + // err = recorder.Close() + // require.NoError(t, err) + err = tmpfile.Close() + require.NoError(t, err) + + tmpfile, err = os.Open(tmpfile.Name()) + require.NoError(t, err) + defer tmpfile.Close() + + dec := cbor.NewDecoder(tmpfile) + + evs := []WebRTCEvent{} + err = nil + for err == nil { + var ev WebRTCEvent + err = dec.Decode(&ev) + if err == nil { + evs = append(evs, ev) + } + } + + require.ErrorIs(t, err, io.EOF) + + off, ok := evs[0].Detail().(*Offer) + require.True(t, ok) + ans, ok := evs[1].Detail().(*CreateAnswer) + require.True(t, ok) + + require.Equal(t, 2, len(evs)) + require.Equal(t, off.SDPOffer, offerEvent.Offer.SDPOffer) + require.Equal(t, ans.SDPAnswer, answerEvent.CreateAnswer.SDPAnswer) + +} diff --git a/pkg/rtcrec/wrapped_peerconnection.go b/pkg/rtcrec/wrapped_peerconnection.go new file mode 100644 index 00000000..a9d678f6 --- /dev/null +++ b/pkg/rtcrec/wrapped_peerconnection.go @@ -0,0 +1 @@ +package rtcrec diff --git a/pkg/streamplace/cbor_gen.go b/pkg/streamplace/cbor_gen.go index 71988462..72a5948a 100644 --- a/pkg/streamplace/cbor_gen.go +++ b/pkg/streamplace/cbor_gen.go @@ -2610,3 +2610,162 @@ func (t *ChatMessage_ReplyRef) UnmarshalCBOR(r io.Reader) (err error) { return nil } +func (t *ServerSettings) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + + cw := cbg.NewCborWriter(w) + fieldCount := 2 + + if t.DebugRecording == nil { + fieldCount-- + } + + if _, err := cw.Write(cbg.CborEncodeMajorType(cbg.MajMap, uint64(fieldCount))); err != nil { + return err + } + + // t.LexiconTypeID (string) (string) + if len("$type") > 1000000 { + return xerrors.Errorf("Value in field \"$type\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("$type"))); err != nil { + return err + } + if _, err := cw.WriteString(string("$type")); err != nil { + return err + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("place.stream.server.settings"))); err != nil { + return err + } + if _, err := cw.WriteString(string("place.stream.server.settings")); err != nil { + return err + } + + // t.DebugRecording (bool) (bool) + if t.DebugRecording != nil { + + if len("debugRecording") > 1000000 { + return xerrors.Errorf("Value in field \"debugRecording\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("debugRecording"))); err != nil { + return err + } + if _, err := cw.WriteString(string("debugRecording")); err != nil { + return err + } + + if t.DebugRecording == nil { + if _, err := cw.Write(cbg.CborNull); err != nil { + return err + } + } else { + if err := cbg.WriteBool(w, *t.DebugRecording); err != nil { + return err + } + } + } + return nil +} + +func (t *ServerSettings) UnmarshalCBOR(r io.Reader) (err error) { + *t = ServerSettings{} + + cr := cbg.NewCborReader(r) + + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + + if maj != cbg.MajMap { + return fmt.Errorf("cbor input should be of type map") + } + + if extra > cbg.MaxLength { + return fmt.Errorf("ServerSettings: map struct too large (%d)", extra) + } + + n := extra + + nameBuf := make([]byte, 14) + for i := uint64(0); i < n; i++ { + nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 1000000) + if err != nil { + return err + } + + if !ok { + // Field doesn't exist on this type, so ignore it + if err := cbg.ScanForLinks(cr, func(cid.Cid) {}); err != nil { + return err + } + continue + } + + switch string(nameBuf[:nameLen]) { + // t.LexiconTypeID (string) (string) + case "$type": + + { + sval, err := cbg.ReadStringWithMax(cr, 1000000) + if err != nil { + return err + } + + t.LexiconTypeID = string(sval) + } + // t.DebugRecording (bool) (bool) + case "debugRecording": + + { + 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.DebugRecording = &val + } + } + + default: + // Field doesn't exist on this type, so ignore it + if err := cbg.ScanForLinks(r, func(cid.Cid) {}); err != nil { + return err + } + } + } + + return nil +} diff --git a/pkg/streamplace/serversettings.go b/pkg/streamplace/serversettings.go new file mode 100644 index 00000000..a9e262e0 --- /dev/null +++ b/pkg/streamplace/serversettings.go @@ -0,0 +1,19 @@ +// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. + +package streamplace + +// schema: place.stream.server.settings + +import ( + "github.com/bluesky-social/indigo/lex/util" +) + +func init() { + util.RegisterType("place.stream.server.settings", &ServerSettings{}) +} // +// RECORDTYPE: ServerSettings +type ServerSettings struct { + LexiconTypeID string `json:"$type,const=place.stream.server.settings" cborgen:"$type,const=place.stream.server.settings"` + // debugRecording: Whether this node may archive your livestream for improving the service + DebugRecording *bool `json:"debugRecording,omitempty" cborgen:"debugRecording,omitempty"` +}