+4

Makefile

··· 696 696 dockerfile-hash-precommit: 697 697 @bash -c 'printf "variables:\n DOCKERFILE_HASH: `git hash-object docker/build.Dockerfile`" > .ci/dockerfile-hash.yaml' \ 698 698 && git add .ci/dockerfile-hash.yaml 699 699 + 700 700 + .PHONY: rtcrec 701 701 + rtcrec: 702 702 + go build -o $(BUILDDIR)/rtcrec ./pkg/rtcrec/cmd/...

+2

go.mod

··· 21 21 github.com/decred/dcrd/dcrec/secp256k1 v1.0.4 22 22 github.com/dunglas/httpsfv v1.0.2 23 23 github.com/ethereum/go-ethereum v1.14.7 24 24 + github.com/fxamacker/cbor/v2 v2.8.0 24 25 github.com/go-git/go-git/v5 v5.12.0 25 26 github.com/go-gst/go-glib v1.4.0 26 27 github.com/go-gst/go-gst v1.4.0 ··· 387 388 github.com/valyala/fasttemplate v1.2.2 // indirect 388 389 github.com/vektah/gqlparser/v2 v2.5.22 // indirect 389 390 github.com/wlynxg/anet v0.0.5 // indirect 391 391 + github.com/x448/float16 v0.8.4 // indirect 390 392 github.com/xanzy/ssh-agent v0.3.3 // indirect 391 393 github.com/xen0n/gosmopolitan v1.3.0 // indirect 392 394 github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect

+4

go.sum

··· 261 261 github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= 262 262 github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY= 263 263 github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw= 264 264 + github.com/fxamacker/cbor/v2 v2.8.0 h1:fFtUGXUzXPHTIUdne5+zzMPTfffl3RD5qYnkY40vtxU= 265 265 + github.com/fxamacker/cbor/v2 v2.8.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= 264 266 github.com/fzipp/gocyclo v0.6.0 h1:lsblElZG7d3ALtGMx9fmxeTKZaLLpU8mET09yN4BBLo= 265 267 github.com/fzipp/gocyclo v0.6.0/go.mod h1:rXPyn8fnlpa0R2csP/31uerbiVBugk5whMdlyaLkLoA= 266 268 github.com/getsentry/sentry-go v0.27.0 h1:Pv98CIbtB3LkMWmXi4Joa5OOcwbmnX88sF5qbK3r3Ps= ··· 996 998 github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e/go.mod h1:pM99HXyEbSQHcosHc0iW7YFmwnscr+t9Te4ibko05so= 997 999 github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= 998 1000 github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA= 1001 1001 + github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= 1002 1002 + github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= 999 1003 github.com/xanzy/ssh-agent v0.3.3 h1:+/15pJfg/RsTxqYcX6fHqOXZwwMP+2VyYWJeWM2qQFM= 1000 1004 github.com/xanzy/ssh-agent v0.3.3/go.mod h1:6dzNDKs0J9rVPHPhaGCukekBHKqfl+L3KghI1Bc68Uw= 1001 1005 github.com/xen0n/gosmopolitan v1.3.0 h1:zAZI1zefvo7gcpbCOrPSHJZJYA9ZgLfJqtKzZ5pHqQM=

+1 -51

js/app/components/livestream/livestream.tsx

··· 14 14 import Loading from "components/loading/loading"; 15 15 import { Player } from "components/player/player"; 16 16 import { PlayerProps } from "components/player/props"; 17 17 - import Popup from "components/popup"; 18 17 import Timer from "components/timer"; 19 18 import Viewers from "components/viewers"; 20 19 import { useFullscreen } from "contexts/FullscreenContext"; ··· 23 22 setSidebarUnhidden, 24 23 } from "features/base/sidebarSlice"; 25 24 import { getProfile } from "features/bluesky/blueskySlice"; 26 26 - import { 27 27 - selectTelemetry, 28 28 - telemetryOpt, 29 29 - } from "features/streamplace/streamplaceSlice"; 30 25 import useAvatars from "hooks/useAvatars"; 31 26 import { useKeyboard } from "hooks/useKeyboard"; 32 27 import usePlatform from "hooks/usePlatform"; ··· 38 33 SafeAreaView, 39 34 } from "react-native"; 40 35 import storage from "storage"; 41 41 - import { useAppDispatch, useAppSelector } from "store/hooks"; 36 36 + import { useAppDispatch } from "store/hooks"; 42 37 import { 43 38 Button, 44 44 - H3, 45 39 isWeb, 46 40 ScrollView, 47 41 Text, ··· 62 56 } 63 57 64 58 export function LivestreamInner(props: Partial<PlayerProps>) { 65 65 - const telemetry = useAppSelector(selectTelemetry); 66 59 const toast = useToastController(); 67 60 const viewers = useViewers(); 68 61 ··· 184 177 <Loading /> 185 178 </View> 186 179 )} 187 187 - {telemetry === null && ( 188 188 - <Popup 189 189 - onClose={() => { 190 190 - dispatch(telemetryOpt(false)); 191 191 - }} 192 192 - containerProps={{ 193 193 - bottom: "$8", 194 194 - zIndex: 1000, 195 195 - }} 196 196 - bubbleProps={{ 197 197 - backgroundColor: "$accentBackground", 198 198 - gap: "$3", 199 199 - maxWidth: 400, 200 200 - }} 201 201 - > 202 202 - <H3 textAlign="center">Player Telemetry</H3> 203 203 - <Text> 204 204 - Streamplace is beta software and it helps us out to have the 205 205 - player report back on how playback is working. Would you like to 206 206 - opt in to optional player telemetry? 207 207 - </Text> 208 208 - <View flexDirection="row" gap="$2" f={1}> 209 209 - <Button 210 210 - f={3} 211 211 - backgroundColor="$accentColor" 212 212 - onPress={() => { 213 213 - dispatch(telemetryOpt(true)); 214 214 - }} 215 215 - > 216 216 - Opt in 217 217 - </Button> 218 218 - <Button 219 219 - f={3} 220 220 - onPress={() => { 221 221 - dispatch(telemetryOpt(false)); 222 222 - }} 223 223 - > 224 224 - Opt out 225 225 - </Button> 226 226 - </View> 227 227 - </Popup> 228 228 - )} 229 180 <View 230 181 f={1} 231 182 opacity={videoWidth === 0 ? 0 : 1} ··· 250 201 }} 251 202 > 252 203 <Player 253 253 - telemetry={telemetry === true} 254 204 src={src} 255 205 fullscreen={fullscreen} 256 206 setFullscreen={setFullscreen}

+2

js/app/components/popup.tsx

··· 21 21 f={1} 22 22 alignItems="center" 23 23 width="100%" 24 24 + pointerEvents="none" 24 25 {...viewProps} 25 26 > 26 27 <View ··· 28 29 alignItems="stretch" 29 30 padding="$4" 30 31 borderRadius="$4" 32 32 + pointerEvents="auto" 31 33 onPress={() => { 32 34 if (onPress) { 33 35 onPress();

+83 -60

js/app/components/settings/settings.tsx

··· 3 3 import AQLink from "components/aqlink"; 4 4 import Container from "components/container"; 5 5 import { 6 6 - DEFAULT_URL, 7 7 - selectTelemetry, 8 8 - setURL, 9 9 - telemetryOpt, 10 10 - } from "features/streamplace/streamplaceSlice"; 6 6 + createServerSettingsRecord, 7 7 + getServerSettingsFromPDS, 8 8 + selectIsReady, 9 9 + selectServerSettings, 10 10 + } from "features/bluesky/blueskySlice"; 11 11 + import { DEFAULT_URL, setURL } from "features/streamplace/streamplaceSlice"; 11 12 import useStreamplaceNode from "hooks/useStreamplaceNode"; 12 13 import { useEffect, useState } from "react"; 14 14 + import { Switch } from "react-native"; 13 15 import { useAppDispatch, useAppSelector } from "store/hooks"; 14 14 - import { Button, H3, H5, Input, Switch, Text, View, XStack } from "tamagui"; 16 16 + import { Button, H3, H5, Input, Text, View, XStack } from "tamagui"; 15 17 import { Updates } from "./updates"; 16 18 17 19 export function Settings() { ··· 47 49 } 48 50 }; 49 51 50 50 - const telemetry = useAppSelector(selectTelemetry); 51 51 - 52 52 - const handleTelemetryToggle = (checked: boolean) => { 53 53 - dispatch(telemetryOpt(checked)); 54 54 - }; 55 55 - 56 52 return ( 57 53 <Container alignItems="center" justifyContent="center"> 58 54 <View ··· 88 84 </Text> 89 85 </View> 90 86 <Switch 91 91 - size="small" 92 92 - checked={overrideEnabled} 93 93 - onCheckedChange={handleToggleOverride} 94 94 - > 95 95 - <Switch.Thumb animation="bouncy" /> 96 96 - </Switch> 87 87 + value={overrideEnabled} 88 88 + onValueChange={handleToggleOverride} 89 89 + /> 97 90 </View> 98 91 </XStack> 99 92 ··· 126 119 </XStack> 127 120 </View> 128 121 129 129 - <View alignItems="center" justifyContent="center" gap="$4"> 130 130 - <XStack 131 131 - alignItems="center" 132 132 - justifyContent="space-between" 133 133 - width="100%" 134 134 - > 135 135 - <View flex={1} pr="$3"> 136 136 - <H3 fontSize="$7">Player Telemetry</H3> 137 137 - <Text fontSize="$5" color="$gray10"> 138 138 - Optional 139 139 - </Text> 140 140 - </View> 141 141 - <Switch 142 142 - size="$3" 143 143 - checked={telemetry === true} 144 144 - onCheckedChange={handleTelemetryToggle} 145 145 - theme="purple" 146 146 - > 147 147 - <Switch.Thumb animation="bouncy" /> 148 148 - </Switch> 149 149 - </XStack> 150 150 - </View> 151 151 - 152 122 {loggedIn && ( 153 153 - <AQLink 154 154 - to={{ 155 155 - screen: "KeyManagement", 156 156 - }} 157 157 - > 158 158 - <View 159 159 - flexDirection="row" 160 160 - gap="$2" 161 161 - alignItems="center" 162 162 - justifyContent="center" 163 163 - borderWidth={1} 164 164 - borderColor="$color.gray3Dark" 165 165 - padding="$2" 166 166 - borderRadius="$4" 167 167 - backgroundColor="$color.gray1Dark" 123 123 + <> 124 124 + <DebugRecording /> 125 125 + <AQLink 126 126 + to={{ 127 127 + screen: "KeyManagement", 128 128 + }} 168 129 > 169 169 - <H5>Manage Keys</H5> 170 170 - <ArrowRight size="$1" /> 171 171 - </View> 172 172 - </AQLink> 130 130 + <View 131 131 + flexDirection="row" 132 132 + gap="$2" 133 133 + alignItems="center" 134 134 + justifyContent="center" 135 135 + borderWidth={1} 136 136 + borderColor="$color.gray3Dark" 137 137 + padding="$2" 138 138 + borderRadius="$4" 139 139 + backgroundColor="$color.gray1Dark" 140 140 + > 141 141 + <H5>Manage Keys</H5> 142 142 + <ArrowRight size="$1" /> 143 143 + </View> 144 144 + </AQLink> 145 145 + </> 173 146 )} 174 147 </View> 175 148 </Container> 176 149 ); 177 150 } 151 151 + 152 152 + const DebugRecording = () => { 153 153 + const dispatch = useAppDispatch(); 154 154 + const isReady = useAppSelector(selectIsReady); 155 155 + const serverSettings = useAppSelector(selectServerSettings) || {}; 156 156 + const { url } = useStreamplaceNode(); 157 157 + 158 158 + useEffect(() => { 159 159 + if (isReady) { 160 160 + dispatch(getServerSettingsFromPDS()); 161 161 + } 162 162 + }, [isReady]); 163 163 + 164 164 + const u = new URL(url); 165 165 + return ( 166 166 + <View alignItems="center" justifyContent="center" gap="$4"> 167 167 + <XStack alignItems="center" justifyContent="space-between" width="100%"> 168 168 + <View flex={1} pr="$3"> 169 169 + <H3 fontSize="$8"> 170 170 + Allow {u.host} to record your livestream for debugging and improving 171 171 + the service 172 172 + </H3> 173 173 + <Text fontSize="$5" color="$gray10"> 174 174 + Optional 175 175 + </Text> 176 176 + </View> 177 177 + <Switch 178 178 + value={serverSettings?.debugRecording === true} 179 179 + onValueChange={(value) => { 180 180 + if (value === true) { 181 181 + dispatch( 182 182 + createServerSettingsRecord({ 183 183 + ...serverSettings, 184 184 + debugRecording: true, 185 185 + }), 186 186 + ); 187 187 + } else { 188 188 + dispatch( 189 189 + createServerSettingsRecord({ 190 190 + ...serverSettings, 191 191 + debugRecording: false, 192 192 + }), 193 193 + ); 194 194 + } 195 195 + }} 196 196 + ></Switch> 197 197 + </XStack> 198 198 + </View> 199 199 + ); 200 200 + };

+121

js/app/features/bluesky/blueskySlice.tsx

··· 22 22 PlaceStreamChatProfile, 23 23 PlaceStreamKey, 24 24 PlaceStreamLivestream, 25 25 + PlaceStreamServerSettings, 25 26 StreamplaceAgent, 26 27 } from "streamplace"; 27 28 import { isWeb } from "tamagui"; ··· 62 63 records: null, 63 64 }, 64 65 newLivestream: null, 66 66 + serverSettings: null, 65 67 }; 66 68 67 69 const uploadThumbnail = async ( ··· 1282 1284 }, 1283 1285 }, 1284 1286 ), 1287 1287 + 1288 1288 + getServerSettingsFromPDS: create.asyncThunk( 1289 1289 + async (_, thunkAPI) => { 1290 1290 + const { bluesky, streamplace } = thunkAPI.getState() as { 1291 1291 + bluesky: BlueskyState; 1292 1292 + streamplace: StreamplaceState; 1293 1293 + }; 1294 1294 + const did = bluesky.oauthSession?.did; 1295 1295 + if (!did) { 1296 1296 + throw new Error("No DID"); 1297 1297 + } 1298 1298 + const profile = bluesky.profiles[did]; 1299 1299 + if (!profile) { 1300 1300 + throw new Error("No profile"); 1301 1301 + } 1302 1302 + if (!bluesky.pdsAgent) { 1303 1303 + throw new Error("No agent"); 1304 1304 + } 1305 1305 + const u = new URL(streamplace.url); 1306 1306 + const res = await bluesky.pdsAgent.com.atproto.repo.getRecord({ 1307 1307 + repo: did, 1308 1308 + collection: "place.stream.server.settings", 1309 1309 + rkey: u.host, 1310 1310 + }); 1311 1311 + if (!res.success) { 1312 1312 + throw new Error("Failed to get chat profile record"); 1313 1313 + } 1314 1314 + 1315 1315 + if (PlaceStreamServerSettings.isRecord(res.data.value)) { 1316 1316 + return res.data.value; 1317 1317 + } else { 1318 1318 + console.log("not a record", res.data.value); 1319 1319 + } 1320 1320 + return null; 1321 1321 + }, 1322 1322 + { 1323 1323 + pending: (state) => { 1324 1324 + return { 1325 1325 + ...state, 1326 1326 + }; 1327 1327 + }, 1328 1328 + fulfilled: (state, action) => { 1329 1329 + if (!action.payload) { 1330 1330 + return state; 1331 1331 + } 1332 1332 + return { 1333 1333 + ...state, 1334 1334 + serverSettings: action.payload, 1335 1335 + }; 1336 1336 + }, 1337 1337 + rejected: (state, action) => { 1338 1338 + console.error("getServerSettingsFromPDS rejected", action.error); 1339 1339 + return { 1340 1340 + ...state, 1341 1341 + }; 1342 1342 + }, 1343 1343 + }, 1344 1344 + ), 1345 1345 + 1346 1346 + createServerSettingsRecord: create.asyncThunk( 1347 1347 + async ({ debugRecording }: { debugRecording: boolean }, thunkAPI) => { 1348 1348 + const { bluesky, streamplace } = thunkAPI.getState() as { 1349 1349 + bluesky: BlueskyState; 1350 1350 + streamplace: StreamplaceState; 1351 1351 + }; 1352 1352 + if (!bluesky.pdsAgent) { 1353 1353 + throw new Error("No agent"); 1354 1354 + } 1355 1355 + const did = bluesky.oauthSession?.did; 1356 1356 + if (!did) { 1357 1357 + throw new Error("No DID"); 1358 1358 + } 1359 1359 + const profile = bluesky.profiles[did]; 1360 1360 + if (!profile) { 1361 1361 + throw new Error("No profile"); 1362 1362 + } 1363 1363 + if (!did) { 1364 1364 + throw new Error("No DID"); 1365 1365 + } 1366 1366 + const u = new URL(streamplace.url); 1367 1367 + const serverSettings: PlaceStreamServerSettings.Record = { 1368 1368 + debugRecording: debugRecording, 1369 1369 + }; 1370 1370 + 1371 1371 + const res = await bluesky.pdsAgent.com.atproto.repo.putRecord({ 1372 1372 + repo: did, 1373 1373 + collection: "place.stream.server.settings", 1374 1374 + record: serverSettings, 1375 1375 + rkey: u.host, 1376 1376 + }); 1377 1377 + if (!res.success) { 1378 1378 + throw new Error("Failed to create server settings record"); 1379 1379 + } 1380 1380 + return serverSettings; 1381 1381 + }, 1382 1382 + { 1383 1383 + pending: (state) => { 1384 1384 + return { 1385 1385 + ...state, 1386 1386 + }; 1387 1387 + }, 1388 1388 + fulfilled: (state, action) => { 1389 1389 + return { 1390 1390 + ...state, 1391 1391 + serverSettings: action.payload, 1392 1392 + }; 1393 1393 + }, 1394 1394 + rejected: (state, action) => { 1395 1395 + console.error("createServerSettingsRecord rejected", action.error); 1396 1396 + return { 1397 1397 + ...state, 1398 1398 + }; 1399 1399 + }, 1400 1400 + }, 1401 1401 + ), 1285 1402 }), 1286 1403 1287 1404 // You can define your selectors here. These selectors receive the slice ··· 1293 1410 selectProfiles: (bluesky) => bluesky.profiles, 1294 1411 selectStoredKey: (bluesky) => bluesky.storedKey, 1295 1412 selectKeyRecords: (bluesky) => bluesky.streamKeysResponse, 1413 1413 + selectServerSettings: (bluesky) => bluesky.serverSettings, 1296 1414 selectUserProfile: (bluesky) => { 1297 1415 const did = bluesky.oauthSession?.did; 1298 1416 if (!did) return null; ··· 1342 1460 createBlockRecord, 1343 1461 followUser, 1344 1462 unfollowUser, 1463 1463 + getServerSettingsFromPDS, 1464 1464 + createServerSettingsRecord, 1345 1465 } = blueskySlice.actions; 1346 1466 1347 1467 // Selectors returned by `slice.selectors` take the root state as their first argument. ··· 1357 1477 selectNewLivestream, 1358 1478 selectChatProfile, 1359 1479 selectCachedProfiles, 1480 1480 + selectServerSettings, 1360 1481 } = blueskySlice.selectors;

+2

js/app/features/bluesky/blueskyTypes.tsx

··· 5 5 import { 6 6 PlaceStreamChatProfile, 7 7 PlaceStreamLivestream, 8 8 + PlaceStreamServerSettings, 8 9 StreamplaceAgent, 9 10 } from "streamplace"; 10 11 import { StreamplaceOAuthClient } from "./oauthClient"; ··· 48 49 error: null | string; 49 50 profile: null | PlaceStreamChatProfile.Record; 50 51 }; 52 52 + serverSettings: null | PlaceStreamServerSettings.Record; 51 53 }

+7 -34

js/app/features/streamplace/streamplaceSlice.tsx

··· 48 48 firstRequest: boolean; 49 49 }; 50 50 mySegments: PlaceStreamSegment.SegmentView[]; 51 51 - telemetry: boolean | null; 52 51 userMuted: boolean | null; 53 52 chatWarned: boolean; 54 53 } ··· 64 63 firstRequest: true, 65 64 }, 66 65 mySegments: [], 67 67 - telemetry: null, 68 66 userMuted: null, 69 67 chatWarned: false, 70 68 }; 71 69 72 70 const USER_MUTED_KEY = "streamplaceUserMuted"; 73 73 - const TELEMETRY_KEY = "streamplaceTelemetry"; 74 71 const URL_KEY = "streamplaceUrl"; 75 72 const CHAT_WARNING_KEY = "streamplaceChatWarning2"; 76 73 ··· 80 77 reducers: (create) => ({ 81 78 initialize: create.asyncThunk( 82 79 async (_, { getState }) => { 83 83 - let [url, telemetryStr, userMutedStr, chatWarningStr] = 84 84 - await Promise.all([ 85 85 - Storage.getItem(URL_KEY), 86 86 - Storage.getItem(TELEMETRY_KEY), 87 87 - Storage.getItem(USER_MUTED_KEY), 88 88 - Storage.getItem(CHAT_WARNING_KEY), 89 89 - ]); 80 80 + let [url, userMutedStr, chatWarningStr] = await Promise.all([ 81 81 + Storage.getItem(URL_KEY), 82 82 + Storage.getItem(USER_MUTED_KEY), 83 83 + Storage.getItem(CHAT_WARNING_KEY), 84 84 + ]); 90 85 if (!url) { 91 86 url = DEFAULT_URL; 92 92 - } 93 93 - let telemetry: boolean | null = null; 94 94 - if (typeof telemetryStr === "string") { 95 95 - telemetry = JSON.parse(telemetryStr); 96 96 - } else { 97 97 - telemetry = null; 98 87 } 99 88 let userMuted: boolean | null = null; 100 89 console.log("userMutedStr", userMutedStr); ··· 107 96 if (typeof chatWarningStr === "string") { 108 97 chatWarned = chatWarningStr === "true"; 109 98 } 110 110 - return { url, telemetry, userMuted, chatWarned }; 99 99 + return { url, userMuted, chatWarned }; 111 100 }, 112 101 { 113 102 pending: (state) => { 114 103 // state.status = "loading"; 115 104 }, 116 105 fulfilled: (state, action) => { 117 117 - const { url, telemetry, userMuted, chatWarned } = action.payload; 106 106 + const { url, userMuted, chatWarned } = action.payload; 118 107 return { 119 108 ...state, 120 109 url, 121 121 - telemetry, 122 110 userMuted, 123 111 initialized: true, 124 112 chatWarned, ··· 138 126 return { 139 127 ...state, 140 128 url: action.payload, 141 141 - }; 142 142 - }), 143 143 - 144 144 - telemetryOpt: create.reducer((state, action: { payload: boolean }) => { 145 145 - Storage.setItem(TELEMETRY_KEY, JSON.stringify(action.payload)).catch( 146 146 - (err) => { 147 147 - console.error("telemetryOpt error", err); 148 148 - }, 149 149 - ); 150 150 - return { 151 151 - ...state, 152 152 - telemetry: action.payload, 153 129 }; 154 130 }), 155 131 ··· 248 224 selectUrl: (streamplace) => streamplace.url, 249 225 selectRecentSegments: (streamplace) => streamplace.recentSegments, 250 226 selectMySegments: (streamplace) => streamplace.mySegments, 251 251 - selectTelemetry: (streamplace) => streamplace.telemetry, 252 227 selectUserMuted: (streamplace) => streamplace.userMuted, 253 228 selectChatWarned: (streamplace) => streamplace.chatWarned, 254 229 }, ··· 260 235 setURL, 261 236 initialize, 262 237 pollMySegments, 263 263 - telemetryOpt, 264 238 userMute, 265 239 chatWarn, 266 240 } = streamplaceSlice.actions; 267 241 export const { 268 242 selectStreamplace, 269 243 selectMySegments, 270 270 - selectTelemetry, 271 244 selectUserMuted, 272 245 selectChatWarned, 273 246 selectUrl,

+27 -3

js/app/src/router.tsx

··· 12 12 NavigatorScreenParams, 13 13 useLinkTo, 14 14 useNavigation, 15 15 + useRoute, 15 16 } from "@react-navigation/native"; 16 17 import { createNativeStackNavigator } from "@react-navigation/native-stack"; 17 18 import { ··· 49 50 import { useLiveUser } from "hooks/useLiveUser"; 50 51 import usePlatform from "hooks/usePlatform"; 51 52 import { useSidebarControl } from "hooks/useSidebarControl"; 52 52 - import { ReactElement, useEffect, useState } from "react"; 53 53 + import { Fragment, ReactElement, useEffect, useState } from "react"; 53 54 import { 54 55 ImageBackground, 55 56 ImageSourcePropType, ··· 332 333 const userIsLive = useLiveUser(); 333 334 const toast = useToastController(); 334 335 336 336 + const [isLiveDashboard, setIsLiveDashboard] = useState(true); 335 337 useEffect(() => { 336 336 - if (userIsLive && !poppedUp) { 338 338 + if (!isLiveDashboard && userIsLive && !poppedUp) { 337 339 setPoppedUp(true); 338 340 setLivePopup(true); 339 341 } ··· 358 360 : undefined, 359 361 }, 360 362 // rest 361 361 - headerLeft: () => <NavigationButton />, 363 363 + headerLeft: () => ( 364 364 + <> 365 365 + {/* this is a hack to give the popup the navigator context */} 366 366 + <PopupChecker setIsLiveDashboard={setIsLiveDashboard} /> 367 367 + <NavigationButton /> 368 368 + </> 369 369 + ), 362 370 headerRight: () => <AvatarButton />, 363 371 drawerActiveTintColor: theme.accentColor.val, 364 372 unmountOnBlur: true, ··· 547 555 </> 548 556 ); 549 557 } 558 558 + 559 559 + export const PopupChecker = ({ 560 560 + setIsLiveDashboard, 561 561 + }: { 562 562 + setIsLiveDashboard: (isLiveDashboard: boolean) => void; 563 563 + }) => { 564 564 + const route = useRoute(); 565 565 + useEffect(() => { 566 566 + if (route.name === "LiveDashboard") { 567 567 + setIsLiveDashboard(true); 568 568 + } else { 569 569 + setIsLiveDashboard(false); 570 570 + } 571 571 + }, [route.name]); 572 572 + return <Fragment />; 573 573 + }; 550 574 551 575 const MainTab = () => { 552 576 const theme = useTheme();

+66 -6

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

··· 7 7 import Waiting from "components/live-dashboard/waiting"; 8 8 import Loading from "components/loading/loading"; 9 9 import { Player } from "components/player/player"; 10 10 + import Popup from "components/popup"; 10 11 import ButtonSelector from "components/ui/button-selector"; 11 12 import { VideoElementProvider } from "contexts/VideoElementContext"; 12 13 import { 14 14 + createServerSettingsRecord, 15 15 + getServerSettingsFromPDS, 13 16 selectIsReady, 17 17 + selectServerSettings, 14 18 selectUserProfile, 15 19 } from "features/bluesky/blueskySlice"; 16 16 - import { selectTelemetry } from "features/streamplace/streamplaceSlice"; 17 20 import { useLiveUser } from "hooks/useLiveUser"; 18 18 - import React, { useCallback, useState } from "react"; 19 19 - import { useAppSelector } from "store/hooks"; 20 20 - import { Button, H6, isWeb, Text, View } from "tamagui"; 21 21 + import React, { useCallback, useEffect, useState } from "react"; 22 22 + import { useAppDispatch, useAppSelector } from "store/hooks"; 23 23 + import { Button, H3, H6, isWeb, Text, View } from "tamagui"; 21 24 22 25 enum StreamSource { 23 26 Start, ··· 29 32 const isReady = useAppSelector(selectIsReady); 30 33 const userProfile = useAppSelector(selectUserProfile); 31 34 const [streamSource, setStreamSource] = useState(StreamSource.Start); 35 35 + const serverSettings = useAppSelector(selectServerSettings); 32 36 const isLive = useLiveUser(); 33 33 - const telemetry = useAppSelector(selectTelemetry); 34 37 const [videoElement, setVideoElement] = useState<HTMLVideoElement | null>( 35 38 null, 36 39 ); 40 40 + const [gotSettings, setGotSettings] = useState(false); 41 41 + const dispatch = useAppDispatch(); 42 42 + useEffect(() => { 43 43 + if (isReady) { 44 44 + (async () => { 45 45 + await dispatch(getServerSettingsFromPDS()); 46 46 + setGotSettings(true); 47 47 + })(); 48 48 + } 49 49 + }, [isReady]); 50 50 + 51 51 + let madeChoiceAboutDebugRecording = true; 52 52 + if (gotSettings && serverSettings?.debugRecording === undefined) { 53 53 + madeChoiceAboutDebugRecording = false; 54 54 + } 37 55 38 56 const [page, setPage] = useState<"update" | "create">("create"); 39 57 ··· 57 75 if (isLive && streamSource !== StreamSource.Camera) { 58 76 topPane = ( 59 77 <Player 60 60 - telemetry={telemetry === true} 61 78 src={userProfile.did} 62 79 name={userProfile.handle} 63 80 videoRef={videoRef} ··· 93 110 </Button> 94 111 ); 95 112 } 113 113 + 96 114 return ( 97 115 <LivestreamProvider src={userProfile.did}> 98 116 <VideoElementProvider videoElement={videoElement}> ··· 116 134 {page === "update" && isLive ? <UpdateLivestream /> : null} 117 135 {page === "create" ? <CreateLivestream /> : null} 118 136 </View> 137 137 + {madeChoiceAboutDebugRecording ? null : <DebugRecordingPopup />} 119 138 </View> 120 139 </VideoElementProvider> 121 140 </LivestreamProvider> ··· 134 153 to: StreamSource.StreamKey, 135 154 }, 136 155 ]; 156 156 + 157 157 + export function DebugRecordingPopup() { 158 158 + const dispatch = useAppDispatch(); 159 159 + const serverSettings = useAppSelector(selectServerSettings) || {}; 160 160 + const opt = (choice) => () => 161 161 + dispatch( 162 162 + createServerSettingsRecord({ 163 163 + ...serverSettings, 164 164 + debugRecording: choice, 165 165 + }), 166 166 + ); 167 167 + return ( 168 168 + <Popup 169 169 + onClose={opt(false)} 170 170 + containerProps={{ 171 171 + bottom: "$8", 172 172 + zIndex: 1000, 173 173 + }} 174 174 + bubbleProps={{ 175 175 + backgroundColor: "$accentBackground", 176 176 + gap: "$3", 177 177 + maxWidth: 400, 178 178 + }} 179 179 + > 180 180 + <H3 textAlign="center">Debug Recording</H3> 181 181 + <Text> 182 182 + Streamplace is beta software and it helps us to archive livestreams so 183 183 + we can later use them for debugging. Would you like to opt in to debug 184 184 + recording? 185 185 + </Text> 186 186 + <View flexDirection="row" gap="$2" f={1}> 187 187 + <Button f={3} backgroundColor="$accentColor" onPress={opt(true)}> 188 188 + Allow 189 189 + </Button> 190 190 + <Button f={3} onPress={opt(false)}> 191 191 + Don't Allow 192 192 + </Button> 193 193 + </View> 194 194 + </Popup> 195 195 + ); 196 196 + } 137 197 138 198 export function StreamSourcePicker({ 139 199 onPick,

+52

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

··· 1 1 + --- 2 2 + title: place.stream.server.settings 3 3 + description: Reference for the place.stream.server.settings lexicon 4 4 + --- 5 5 + 6 6 + **Lexicon Version:** 1 7 7 + 8 8 + ## Definitions 9 9 + 10 10 + <a name="main"></a> 11 11 + 12 12 + ### `main` 13 13 + 14 14 + **Type:** `record` 15 15 + 16 16 + Record containing user settings for a particular Streamplace node 17 17 + 18 18 + **Record Key:** `any` 19 19 + 20 20 + **Record Properties:** 21 21 + 22 22 + | Name | Type | Req'd | Description | Constraints | 23 23 + | ---------------- | --------- | ----- | ----------------------------------------------------------------------- | ----------- | 24 24 + | `debugRecording` | `boolean` | ❌ | Whether this node may archive your livestream for improving the service | | 25 25 + 26 26 + --- 27 27 + 28 28 + ## Lexicon Source 29 29 + 30 30 + ```json 31 31 + { 32 32 + "lexicon": 1, 33 33 + "id": "place.stream.server.settings", 34 34 + "defs": { 35 35 + "main": { 36 36 + "type": "record", 37 37 + "description": "Record containing user settings for a particular Streamplace node", 38 38 + "key": "any", 39 39 + "record": { 40 40 + "type": "object", 41 41 + "required": [], 42 42 + "properties": { 43 43 + "debugRecording": { 44 44 + "type": "boolean", 45 45 + "description": "Whether this node may archive your livestream for improving the service" 46 46 + } 47 47 + } 48 48 + } 49 49 + } 50 50 + } 51 51 + } 52 52 + ```

+21

lexicons/place/stream/server/settings.json

··· 1 1 + { 2 2 + "lexicon": 1, 3 3 + "id": "place.stream.server.settings", 4 4 + "defs": { 5 5 + "main": { 6 6 + "type": "record", 7 7 + "description": "Record containing user settings for a particular Streamplace node", 8 8 + "key": "any", 9 9 + "record": { 10 10 + "type": "object", 11 11 + "required": [], 12 12 + "properties": { 13 13 + "debugRecording": { 14 14 + "type": "boolean", 15 15 + "description": "Whether this node may archive your livestream for improving the service" 16 16 + } 17 17 + } 18 18 + } 19 19 + } 20 20 + } 21 21 + }

+30

pkg/api/api_internal.go

··· 20 20 "time" 21 21 22 22 "github.com/julienschmidt/httprouter" 23 23 + "github.com/pion/webrtc/v4" 23 24 "github.com/prometheus/client_golang/prometheus/promhttp" 24 25 sloghttp "github.com/samber/slog-http" 25 26 "golang.org/x/sync/errgroup" ··· 31 32 "stream.place/streamplace/pkg/model" 32 33 notificationpkg "stream.place/streamplace/pkg/notifications" 33 34 "stream.place/streamplace/pkg/renditions" 35 35 + "stream.place/streamplace/pkg/rtcrec" 34 36 v0 "stream.place/streamplace/pkg/schema/v0" 35 37 ) 36 38 ··· 599 601 return 600 602 } 601 603 if _, err := w.Write(bs); err != nil { 604 604 + log.Error(ctx, "error writing response", "error", err) 605 605 + } 606 606 + }) 607 607 + 608 608 + router.POST("/replay/:streamKey", func(w http.ResponseWriter, r *http.Request, p httprouter.Params) { 609 609 + key := p.ByName("streamKey") 610 610 + if key == "" { 611 611 + errors.WriteHTTPBadRequest(w, "streamKey required", nil) 612 612 + return 613 613 + } 614 614 + mediaSigner, err := a.MakeMediaSigner(ctx, key) 615 615 + if err != nil { 616 616 + errors.WriteHTTPUnauthorized(w, "invalid authorization key", err) 617 617 + return 618 618 + } 619 619 + pc, err := rtcrec.NewReplayPeerConnection(ctx, r.Body) 620 620 + if err != nil { 621 621 + errors.WriteHTTPInternalServerError(w, "unable to create replay peer connection", err) 622 622 + return 623 623 + } 624 624 + answer, err := a.MediaManager.WebRTCIngest(ctx, &webrtc.SessionDescription{SDP: "placeholder"}, mediaSigner, pc) 625 625 + if err != nil { 626 626 + errors.WriteHTTPInternalServerError(w, "unable to ingest web rtc", err) 627 627 + return 628 628 + } 629 629 + w.WriteHeader(200) 630 630 + if _, err := w.Write([]byte(answer.SDP)); err != nil { 631 631 + errors.WriteHTTPInternalServerError(w, "unable to write response", err) 602 632 log.Error(ctx, "error writing response", "error", err) 603 633 } 604 634 })

+6 -1

pkg/api/playback.go

··· 219 219 return 220 220 } 221 221 offer := webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(body)} 222 222 - answer, err := a.MediaManager.WebRTCIngest(ctx, &offer, mediaSigner) 222 222 + pc, err := a.MediaManager.NewPeerConnection(ctx, mediaSigner.Streamer()) 223 223 + if err != nil { 224 224 + errors.WriteHTTPInternalServerError(w, "unable to create peer connection", err) 225 225 + return 226 226 + } 227 227 + answer, err := a.MediaManager.WebRTCIngest(ctx, &offer, mediaSigner, pc) 223 228 if err != nil { 224 229 errors.WriteHTTPInternalServerError(w, "error playing back", err) 225 230 return

+1

pkg/atproto/firehose.go

··· 148 148 constants.APP_BSKY_GRAPH_FOLLOW, 149 149 constants.APP_BSKY_FEED_POST, 150 150 constants.APP_BSKY_GRAPH_BLOCK, 151 151 + constants.PLACE_STREAM_SERVER_SETTINGS, 151 152 } 152 153 153 154 func (atsync *ATProtoSynchronizer) handleCommitEventOps(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) {

+15

pkg/atproto/sync.go

··· 163 163 log.Error(ctx, "failed to create chat profile", "err", err) 164 164 } 165 165 166 166 + case *streamplace.ServerSettings: 167 167 + _, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) 168 168 + if err != nil { 169 169 + return fmt.Errorf("failed to sync bluesky repo: %w", err) 170 170 + } 171 171 + settings := &model.ServerSettings{ 172 172 + Server: rkey.String(), 173 173 + RepoDID: userDID, 174 174 + Record: recCBOR, 175 175 + } 176 176 + err = atsync.Model.UpdateServerSettings(ctx, settings) 177 177 + if err != nil { 178 178 + log.Error(ctx, "failed to create server settings", "err", err) 179 179 + } 180 180 + 166 181 case *bsky.FeedPost: 167 182 // jsonData, err := json.Marshal(d) 168 183 // if err != nil {

+9 -8

pkg/constants/constants.go

··· 1 1 package constants 2 2 3 3 - var PLACE_STREAM_KEY = "place.stream.key" //nolint:all 4 4 - var PLACE_STREAM_LIVESTREAM = "place.stream.livestream" //nolint:all 5 5 - var PLACE_STREAM_CHAT_MESSAGE = "place.stream.chat.message" //nolint:all 6 6 - var PLACE_STREAM_CHAT_PROFILE = "place.stream.chat.profile" //nolint:all 7 7 - var STREAMPLACE_SIGNING_KEY = "signingKey" //nolint:all 8 8 - var APP_BSKY_GRAPH_FOLLOW = "app.bsky.graph.follow" //nolint:all 9 9 - var APP_BSKY_FEED_POST = "app.bsky.feed.post" //nolint:all 10 10 - var APP_BSKY_GRAPH_BLOCK = "app.bsky.graph.block" //nolint:all 3 3 + var PLACE_STREAM_KEY = "place.stream.key" //nolint:all 4 4 + var PLACE_STREAM_LIVESTREAM = "place.stream.livestream" //nolint:all 5 5 + var PLACE_STREAM_CHAT_MESSAGE = "place.stream.chat.message" //nolint:all 6 6 + var PLACE_STREAM_CHAT_PROFILE = "place.stream.chat.profile" //nolint:all 7 7 + var PLACE_STREAM_SERVER_SETTINGS = "place.stream.server.settings" //nolint:all 8 8 + var STREAMPLACE_SIGNING_KEY = "signingKey" //nolint:all 9 9 + var APP_BSKY_GRAPH_FOLLOW = "app.bsky.graph.follow" //nolint:all 10 10 + var APP_BSKY_FEED_POST = "app.bsky.feed.post" //nolint:all 11 11 + var APP_BSKY_GRAPH_BLOCK = "app.bsky.graph.block" //nolint:all 11 12 12 13 const DID_KEY_PREFIX = "did:key" //nolint:all 13 14 const ADDRESS_KEY_PREFIX = "0x" //nolint:all

+1

pkg/gen/gen.go

··· 23 23 streamplace.ChatProfile{}, 24 24 streamplace.ChatProfile_Color{}, 25 25 streamplace.ChatMessage_ReplyRef{}, 26 26 + streamplace.ServerSettings{}, 26 27 ); err != nil { 27 28 panic(err) 28 29 }

+33

pkg/media/peer_connection.go

··· 1 1 + package media 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + 6 6 + "stream.place/streamplace/pkg/log" 7 7 + "stream.place/streamplace/pkg/rtcrec" 8 8 + ) 9 9 + 10 10 + func (mm *MediaManager) NewPeerConnection(ctx context.Context, user string) (rtcrec.PeerConnection, error) { 11 11 + shouldRecord := false 12 12 + settings, err := mm.model.GetServerSettings(ctx, mm.cli.PublicHost, user) 13 13 + if err != nil { 14 14 + return nil, err 15 15 + } 16 16 + if settings != nil { 17 17 + spsettings, err := settings.ToStreamplaceServerSettings() 18 18 + if err != nil { 19 19 + return nil, err 20 20 + } 21 21 + if spsettings.DebugRecording != nil { 22 22 + shouldRecord = *spsettings.DebugRecording 23 23 + } 24 24 + } 25 25 + if !shouldRecord { 26 26 + log.Warn(ctx, "no server settings found, will not record") 27 27 + } 28 28 + pionpc, err := mm.webrtcAPI.NewPeerConnection(mm.webrtcConfig) 29 29 + if err != nil { 30 30 + return nil, err 31 31 + } 32 32 + return rtcrec.NewRecordingPeerConnection(ctx, *mm.cli, user, pionpc, shouldRecord) 33 33 + }

+5 -10

pkg/media/webrtc_ingest.go

··· 12 12 "github.com/pion/rtcp" 13 13 "github.com/pion/webrtc/v4" 14 14 "stream.place/streamplace/pkg/log" 15 15 + "stream.place/streamplace/pkg/rtcrec" 15 16 ) 16 17 17 18 // This function remains in scope for the duration of a single users' playback 18 18 - func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionDescription, signer MediaSigner) (*webrtc.SessionDescription, error) { 19 19 + func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionDescription, signer MediaSigner, peerConnection rtcrec.PeerConnection) (*webrtc.SessionDescription, error) { 19 20 uu, err := uuid.NewV7() 20 21 if err != nil { 21 22 return nil, err 22 23 } 23 24 24 24 - ctx = log.WithLogValues(ctx, "webrtcID", uu.String(), "mediafunc", "WebRTCIngest") 25 25 - 26 26 - // Create a new RTCPeerConnection 27 27 - peerConnection, err := mm.webrtcAPI.NewPeerConnection(mm.webrtcConfig) 28 28 - if err != nil { 29 29 - return nil, fmt.Errorf("failed to create WebRTC peer connection: %w", err) 30 30 - } 25 25 + ctx = log.WithLogValues(ctx, "webrtcID", uu.String(), "mediafunc", "WebRTCIngest", "streamer", signer.Streamer()) 31 26 32 27 // Allow us to receive 1 audio track, and 1 video track 33 28 if _, err = peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio); err != nil { ··· 116 111 } 117 112 118 113 // Create channel that is blocked until ICE Gathering is complete 119 119 - gatherComplete := webrtc.GatheringCompletePromise(peerConnection) 114 114 + gatherComplete := rtcrec.GatheringCompletePromise(peerConnection) 120 115 121 116 go func() { 122 117 ticker := time.NewTicker(time.Second * 1) ··· 186 181 audioFirst := false 187 182 188 183 log.Warn(ctx, "setting OnTrack") 189 189 - peerConnection.OnTrack(func(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) { 184 184 + peerConnection.OnTrack(func(track rtcrec.TrackRemote, _ rtcrec.RTPReceiver) { 190 185 log.Warn(ctx, "OnTrack") 191 186 if track.Kind() == webrtc.RTPCodecTypeVideo { 192 187 // Send a PLI on an interval so that the publisher is pushing a keyframe every rtcpPLIInterval

+5

pkg/model/model.go

··· 92 92 UpdateOAuthSession(id string, session *oatproxy.OAuthSession) error 93 93 ListOAuthSessions() ([]oatproxy.OAuthSession, error) 94 94 GetSessionByDID(did string) (*oatproxy.OAuthSession, error) 95 95 + 96 96 + UpdateServerSettings(ctx context.Context, settings *ServerSettings) error 97 97 + GetServerSettings(ctx context.Context, server string, repoDID string) (*ServerSettings, error) 98 98 + DeleteServerSettings(ctx context.Context, server string, repoDID string) error 95 99 } 96 100 97 101 func MakeDB(dbURL string) (Model, error) { ··· 151 155 ChatMessage{}, 152 156 ChatProfile{}, 153 157 oatproxy.OAuthSession{}, 158 158 + ServerSettings{}, 154 159 } { 155 160 err = db.AutoMigrate(model) 156 161 if err != nil {

+70

pkg/model/server_settings.go

··· 1 1 + package model 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "errors" 6 6 + "fmt" 7 7 + "time" 8 8 + 9 9 + lexutil "github.com/bluesky-social/indigo/lex/util" 10 10 + "gorm.io/gorm" 11 11 + "stream.place/streamplace/pkg/streamplace" 12 12 + ) 13 13 + 14 14 + // ServerSettings represents a user's settings for a particular Streamplace node 15 15 + type ServerSettings struct { 16 16 + Server string `gorm:"primaryKey;column:server"` 17 17 + RepoDID string `gorm:"primaryKey;column:repo_did"` 18 18 + Record *[]byte `gorm:"column:record"` 19 19 + Created time.Time `gorm:"column:created;not null"` 20 20 + Updated time.Time `gorm:"column:updated;not null"` 21 21 + } 22 22 + 23 23 + // TableName specifies the table name for the ServerSettings model 24 24 + func (ServerSettings) TableName() string { 25 25 + return "server_settings" 26 26 + } 27 27 + 28 28 + // ToStreamplaceServerSettings converts the model to a streamplace ServerSettings 29 29 + func (m *ServerSettings) ToStreamplaceServerSettings() (*streamplace.ServerSettings, error) { 30 30 + if m.Record == nil { 31 31 + return nil, fmt.Errorf("no record data") 32 32 + } 33 33 + rec, err := lexutil.CborDecodeValue(*m.Record) 34 34 + if err != nil { 35 35 + return nil, fmt.Errorf("error decoding server settings: %w", err) 36 36 + } 37 37 + ss, ok := rec.(*streamplace.ServerSettings) 38 38 + if !ok { 39 39 + return nil, fmt.Errorf("invalid server settings") 40 40 + } 41 41 + return ss, nil 42 42 + } 43 43 + 44 44 + // UpdateServerSettings creates or updates a server settings record 45 45 + func (m *DBModel) UpdateServerSettings(ctx context.Context, settings *ServerSettings) error { 46 46 + now := time.Now() 47 47 + if settings.Created.IsZero() { 48 48 + settings.Created = now 49 49 + } 50 50 + settings.Updated = now 51 51 + return m.DB.Save(settings).Error 52 52 + } 53 53 + 54 54 + // GetServerSettings retrieves server settings for a given server and repoDID 55 55 + func (m *DBModel) GetServerSettings(ctx context.Context, server string, repoDID string) (*ServerSettings, error) { 56 56 + var settings ServerSettings 57 57 + err := m.DB.Where("server = ? AND repo_did = ?", server, repoDID).First(&settings).Error 58 58 + if errors.Is(err, gorm.ErrRecordNotFound) { 59 59 + return nil, nil 60 60 + } 61 61 + if err != nil { 62 62 + return nil, err 63 63 + } 64 64 + return &settings, nil 65 65 + } 66 66 + 67 67 + // DeleteServerSettings deletes server settings for a given server and repoDID 68 68 + func (m *DBModel) DeleteServerSettings(ctx context.Context, server string, repoDID string) error { 69 69 + return m.DB.Where("server = ? AND repo_did = ?", server, repoDID).Delete(&ServerSettings{}).Error 70 70 + }

+221

pkg/rtcrec/cbor.go

··· 1 1 + package rtcrec 2 2 + 3 3 + import ( 4 4 + "fmt" 5 5 + "io" 6 6 + "sort" 7 7 + "time" 8 8 + 9 9 + "github.com/fxamacker/cbor/v2" 10 10 + "github.com/pion/webrtc/v4" 11 11 + ) 12 12 + 13 13 + type WebRTCEventDecoder struct { 14 14 + dec *cbor.Decoder 15 15 + } 16 16 + 17 17 + func Opts() (cbor.EncMode, error) { 18 18 + opts := cbor.CoreDetEncOptions() 19 19 + opts.Time = cbor.TimeRFC3339Nano 20 20 + em, err := opts.EncMode() 21 21 + if err != nil { 22 22 + return nil, fmt.Errorf("failed to create encoder mode: %w", err) 23 23 + } 24 24 + return em, nil 25 25 + } 26 26 + 27 27 + func MakeWebRTCDecoder(r io.Reader) (*WebRTCEventDecoder, error) { 28 28 + dec := cbor.NewDecoder(r) 29 29 + return &WebRTCEventDecoder{dec: dec}, nil 30 30 + } 31 31 + 32 32 + func (d *WebRTCEventDecoder) Next() (*WebRTCEvent, error) { 33 33 + var ev WebRTCEvent 34 34 + err := d.dec.Decode(&ev) 35 35 + if err != nil { 36 36 + return nil, err 37 37 + } 38 38 + return &ev, err 39 39 + } 40 40 + 41 41 + type WebRTCEventGroup struct { 42 42 + Events map[string][]*WebRTCEvent 43 43 + Tracks map[webrtc.SSRC]map[string][]*WebRTCEvent 44 44 + FirstTime time.Time 45 45 + } 46 46 + 47 47 + const ( 48 48 + EventTypeOffer = "Offer" 49 49 + EventTypeCreateAnswer = "CreateAnswer" 50 50 + EventTypeSetRemoteDescription = "SetRemoteDescription" 51 51 + EventTypeSetLocalDescription = "SetLocalDescription" 52 52 + EventTypeLocalDescription = "LocalDescription" 53 53 + EventTypeICEConnectionState = "ICEConnectionStateChange" 54 54 + EventTypeConnectionState = "ConnectionStateChange" 55 55 + EventTypeTrack = "Track" 56 56 + EventTypeAddTransceiverFromKind = "AddTransceiverFromKind" 57 57 + EventTypeICEGatheringState = "ICEGatheringState" 58 58 + EventTypeDataChannel = "DataChannel" 59 59 + EventTypeNegotiationNeeded = "NegotiationNeeded" 60 60 + EventTypeTrackRead = "TrackRead" 61 61 + EventTypeTrackCodec = "TrackCodec" 62 62 + EventTypeTrackKind = "TrackKind" 63 63 + EventTypeTrackPayloadType = "TrackPayloadType" 64 64 + EventTypeTrackSSRC = "TrackSSRC" 65 65 + EventTypeUnknown = "Unknown" 66 66 + ) 67 67 + 68 68 + // ReadAllEvents reads all WebRTC events from a CBOR reader and organizes them by type. 69 69 + // Returns a map where keys are event types and values are slices of events of that type. 70 70 + func ReadAllEvents(r io.Reader) (*WebRTCEventGroup, error) { 71 71 + dec, err := MakeWebRTCDecoder(r) 72 72 + if err != nil { 73 73 + return nil, err 74 74 + } 75 75 + 76 76 + eventList := []*WebRTCEvent{} 77 77 + events := make(map[string][]*WebRTCEvent) 78 78 + tracks := make(map[webrtc.SSRC]map[string][]*WebRTCEvent) 79 79 + var firstTime time.Time 80 80 + for { 81 81 + ev, err := dec.Next() 82 82 + if err == io.EOF { 83 83 + break 84 84 + } 85 85 + if err != nil { 86 86 + return nil, err 87 87 + } 88 88 + eventList = append(eventList, ev) 89 89 + } 90 90 + 91 91 + sort.Slice(eventList, func(i, j int) bool { 92 92 + return eventList[i].Time.Before(eventList[j].Time) 93 93 + }) 94 94 + 95 95 + for _, ev := range eventList { 96 96 + if firstTime.IsZero() { 97 97 + firstTime = ev.Time 98 98 + } 99 99 + 100 100 + // Determine the event type based on which field is non-nil 101 101 + var eventType string 102 102 + var trackSSRC *webrtc.SSRC 103 103 + switch { 104 104 + case ev.Offer != nil: 105 105 + eventType = EventTypeOffer 106 106 + case ev.CreateAnswer != nil: 107 107 + eventType = EventTypeCreateAnswer 108 108 + case ev.SetRemoteDescription != nil: 109 109 + eventType = EventTypeSetRemoteDescription 110 110 + case ev.SetLocalDescription != nil: 111 111 + eventType = EventTypeSetLocalDescription 112 112 + case ev.LocalDescription != nil: 113 113 + eventType = EventTypeLocalDescription 114 114 + case ev.ICEConnectionStateChange != nil: 115 115 + eventType = EventTypeICEConnectionState 116 116 + case ev.ConnectionStateChange != nil: 117 117 + eventType = EventTypeConnectionState 118 118 + case ev.Track != nil: 119 119 + eventType = EventTypeTrack 120 120 + case ev.AddTransceiverFromKind != nil: 121 121 + eventType = EventTypeAddTransceiverFromKind 122 122 + case ev.ICEGatheringState != nil: 123 123 + eventType = EventTypeICEGatheringState 124 124 + case ev.DataChannel != nil: 125 125 + eventType = EventTypeDataChannel 126 126 + case ev.NegotiationNeeded != nil: 127 127 + eventType = EventTypeNegotiationNeeded 128 128 + 129 129 + case ev.TrackRead != nil: 130 130 + trackSSRC = &ev.TrackRead.SSRC 131 131 + eventType = EventTypeTrackRead 132 132 + case ev.TrackCodec != nil: 133 133 + trackSSRC = &ev.TrackCodec.SSRC 134 134 + eventType = EventTypeTrackCodec 135 135 + case ev.TrackKind != nil: 136 136 + trackSSRC = &ev.TrackKind.SSRC 137 137 + eventType = EventTypeTrackKind 138 138 + case ev.TrackPayloadType != nil: 139 139 + trackSSRC = &ev.TrackPayloadType.SSRC 140 140 + eventType = EventTypeTrackPayloadType 141 141 + case ev.TrackSSRC != nil: 142 142 + trackSSRC = &ev.TrackSSRC.SSRC 143 143 + eventType = EventTypeTrackSSRC 144 144 + default: 145 145 + eventType = EventTypeUnknown 146 146 + panic(fmt.Sprintf("unknown event type: %+v", ev)) 147 147 + } 148 148 + 149 149 + if trackSSRC != nil { 150 150 + if tracks[*trackSSRC] == nil { 151 151 + tracks[*trackSSRC] = make(map[string][]*WebRTCEvent) 152 152 + } 153 153 + if tracks[*trackSSRC][eventType] == nil { 154 154 + tracks[*trackSSRC][eventType] = []*WebRTCEvent{} 155 155 + } 156 156 + tracks[*trackSSRC][eventType] = append(tracks[*trackSSRC][eventType], ev) 157 157 + } 158 158 + 159 159 + if events[eventType] == nil { 160 160 + events[eventType] = []*WebRTCEvent{} 161 161 + } 162 162 + 163 163 + events[eventType] = append(events[eventType], ev) 164 164 + } 165 165 + 166 166 + return &WebRTCEventGroup{ 167 167 + Events: events, 168 168 + Tracks: tracks, 169 169 + FirstTime: firstTime, 170 170 + }, nil 171 171 + } 172 172 + 173 173 + func (g *WebRTCEventGroup) Peek(eventType string) *WebRTCEvent { 174 174 + if g.Events[eventType] == nil { 175 175 + panic(fmt.Sprintf("no events of type %s", eventType)) 176 176 + } 177 177 + if len(g.Events[eventType]) == 0 { 178 178 + return nil 179 179 + } 180 180 + return g.Events[eventType][0] 181 181 + } 182 182 + 183 183 + func (g *WebRTCEventGroup) PeekTrack(ssrc webrtc.SSRC, eventType string) *WebRTCEvent { 184 184 + if g.Tracks[ssrc] == nil { 185 185 + panic(fmt.Sprintf("no tracks for ssrc %d", ssrc)) 186 186 + } 187 187 + if g.Tracks[ssrc][eventType] == nil { 188 188 + panic(fmt.Sprintf("no events of type %s for ssrc %d", eventType, ssrc)) 189 189 + } 190 190 + if len(g.Tracks[ssrc][eventType]) == 0 { 191 191 + return nil 192 192 + } 193 193 + return g.Tracks[ssrc][eventType][0] 194 194 + } 195 195 + 196 196 + func (g *WebRTCEventGroup) Next(eventType string) *WebRTCEvent { 197 197 + if g.Events[eventType] == nil { 198 198 + panic(fmt.Sprintf("no events of type %s", eventType)) 199 199 + } 200 200 + if len(g.Events[eventType]) == 0 { 201 201 + return nil 202 202 + } 203 203 + ev := g.Events[eventType][0] 204 204 + g.Events[eventType] = g.Events[eventType][1:] 205 205 + return ev 206 206 + } 207 207 + 208 208 + func (g *WebRTCEventGroup) NextTrack(ssrc webrtc.SSRC, eventType string) *WebRTCEvent { 209 209 + if g.Tracks[ssrc] == nil { 210 210 + panic(fmt.Sprintf("no tracks for ssrc %d", ssrc)) 211 211 + } 212 212 + if g.Tracks[ssrc][eventType] == nil { 213 213 + panic(fmt.Sprintf("no events of type %s for ssrc %d", eventType, ssrc)) 214 214 + } 215 215 + if len(g.Tracks[ssrc][eventType]) == 0 { 216 216 + return nil 217 217 + } 218 218 + ev := g.Tracks[ssrc][eventType][0] 219 219 + g.Tracks[ssrc][eventType] = g.Tracks[ssrc][eventType][1:] 220 220 + return ev 221 221 + }

+77

pkg/rtcrec/cmd/rtcrec.go

··· 1 1 + package main 2 2 + 3 3 + import ( 4 4 + "encoding/json" 5 5 + "errors" 6 6 + "flag" 7 7 + "fmt" 8 8 + "io" 9 9 + "log" 10 10 + "os" 11 11 + 12 12 + "stream.place/streamplace/pkg/rtcrec" 13 13 + ) 14 14 + 15 15 + func main() { 16 16 + err := Start() 17 17 + if err != nil { 18 18 + log.Fatal(err) 19 19 + } 20 20 + } 21 21 + 22 22 + func Start() error { 23 23 + var path string 24 24 + flag.StringVar(&path, "path", "", "path to the file to decode") 25 25 + flag.Parse() 26 26 + if path == "" { 27 27 + return fmt.Errorf("path is required") 28 28 + } 29 29 + return DecodeFile(path) 30 30 + } 31 31 + 32 32 + func DecodeFile(path string) error { 33 33 + f, err := os.Open(path) 34 34 + if err != nil { 35 35 + return err 36 36 + } 37 37 + defer f.Close() 38 38 + dec, err := rtcrec.MakeWebRTCDecoder(f) 39 39 + if err != nil { 40 40 + return err 41 41 + } 42 42 + for { 43 43 + ev, err := dec.Next() 44 44 + if errors.Is(err, io.EOF) { 45 45 + return nil 46 46 + } 47 47 + if err != nil { 48 48 + return err 49 49 + } 50 50 + if ev.TrackRead != nil { 51 51 + // spitting out the data as base64 is pointless, replace with a label 52 52 + n := len(ev.TrackRead.Data) 53 53 + byteString := fmt.Sprintf("%d bytes", n) 54 54 + bs, err := json.Marshal(ev) 55 55 + if err != nil { 56 56 + return err 57 57 + } 58 58 + var m map[string]any 59 59 + err = json.Unmarshal(bs, &m) 60 60 + if err != nil { 61 61 + return err 62 62 + } 63 63 + m["trackRead"].(map[string]any)["data"] = byteString 64 64 + bs, err = json.Marshal(m) 65 65 + if err != nil { 66 66 + return err 67 67 + } 68 68 + fmt.Println(string(bs)) 69 69 + } else { 70 70 + bs, err := json.Marshal(ev) 71 71 + if err != nil { 72 72 + return err 73 73 + } 74 74 + fmt.Println(string(bs)) 75 75 + } 76 76 + } 77 77 + }

+236

pkg/rtcrec/recording_peerconnection.go

··· 1 1 + package rtcrec 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + "os" 7 7 + "time" 8 8 + 9 9 + "github.com/pion/rtcp" 10 10 + "github.com/pion/webrtc/v4" 11 11 + "stream.place/streamplace/pkg/aqtime" 12 12 + "stream.place/streamplace/pkg/config" 13 13 + "stream.place/streamplace/pkg/log" 14 14 + ) 15 15 + 16 16 + type RecordingPeerConnection struct { 17 17 + enabled bool 18 18 + pionpc *webrtc.PeerConnection 19 19 + file *os.File 20 20 + stream *RecorderStream 21 21 + } 22 22 + 23 23 + func NewRecordingPeerConnection(ctx context.Context, cli config.CLI, user string, pionpc *webrtc.PeerConnection, enabled bool) (PeerConnection, error) { 24 24 + if !enabled { 25 25 + return &RecordingPeerConnection{ 26 26 + pionpc: pionpc, 27 27 + enabled: enabled, 28 28 + }, nil 29 29 + } 30 30 + aqt := aqtime.FromTime(time.Now()) 31 31 + f, err := cli.DataFileCreate([]string{user, "rtcrec", fmt.Sprintf("%s.cbor", aqt.FileSafeString())}, true) 32 32 + if err != nil { 33 33 + return nil, fmt.Errorf("failed to create data file: %w", err) 34 34 + } 35 35 + log.Log(ctx, "logging webrtc session to file", "file", f.Name()) 36 36 + stream, err := NewRecorderStream(f) 37 37 + if err != nil { 38 38 + return nil, fmt.Errorf("failed to create recorder stream: %w", err) 39 39 + } 40 40 + return &RecordingPeerConnection{ 41 41 + pionpc: pionpc, 42 42 + file: f, 43 43 + stream: stream, 44 44 + enabled: enabled, 45 45 + }, nil 46 46 + } 47 47 + 48 48 + func (pc *RecordingPeerConnection) Do(f func()) { 49 49 + if pc.enabled { 50 50 + go f() 51 51 + } 52 52 + } 53 53 + 54 54 + func (pc *RecordingPeerConnection) Close() error { 55 55 + return pc.pionpc.Close() 56 56 + } 57 57 + 58 58 + func (pc *RecordingPeerConnection) CreateAnswer(options *webrtc.AnswerOptions) (webrtc.SessionDescription, error) { 59 59 + now := time.Now() 60 60 + ret, err := pc.pionpc.CreateAnswer(options) 61 61 + if err != nil { 62 62 + return ret, err 63 63 + } 64 64 + pc.Do(func() { 65 65 + pc.stream.Event(WebRTCEvent{ 66 66 + CreateAnswer: &CreateAnswer{ 67 67 + SDPAnswer: ret.SDP, 68 68 + }, 69 69 + Time: now, 70 70 + }) 71 71 + }) 72 72 + return ret, nil 73 73 + } 74 74 + 75 75 + func (pc *RecordingPeerConnection) SetLocalDescription(desc webrtc.SessionDescription) error { 76 76 + now := time.Now() 77 77 + pc.Do(func() { 78 78 + pc.stream.Event(WebRTCEvent{ 79 79 + SetRemoteDescription: &SetRemoteDescription{ 80 80 + SDPRemoteDescription: desc.SDP, 81 81 + }, 82 82 + Time: now, 83 83 + }) 84 84 + }) 85 85 + return pc.pionpc.SetLocalDescription(desc) 86 86 + } 87 87 + 88 88 + func (pc *RecordingPeerConnection) SetRemoteDescription(desc webrtc.SessionDescription) error { 89 89 + now := time.Now() 90 90 + pc.Do(func() { 91 91 + pc.stream.Event(WebRTCEvent{ 92 92 + SetRemoteDescription: &SetRemoteDescription{ 93 93 + SDPRemoteDescription: desc.SDP, 94 94 + }, 95 95 + Time: now, 96 96 + }) 97 97 + }) 98 98 + return pc.pionpc.SetRemoteDescription(desc) 99 99 + } 100 100 + 101 101 + func (pc *RecordingPeerConnection) LocalDescription() *webrtc.SessionDescription { 102 102 + now := time.Now() 103 103 + desc := pc.pionpc.LocalDescription() 104 104 + pc.Do(func() { 105 105 + pc.stream.Event(WebRTCEvent{ 106 106 + LocalDescription: &LocalDescription{ 107 107 + SDPLocalDescription: pc.pionpc.LocalDescription().SDP, 108 108 + }, 109 109 + Time: now, 110 110 + }) 111 111 + }) 112 112 + return desc 113 113 + } 114 114 + 115 115 + // func (pc *RecorderPeerConnection) RemoteDescription() *webrtc.SessionDescription { 116 116 + // return pc.pionpc.RemoteDescription() 117 117 + // } 118 118 + 119 119 + func (pc *RecordingPeerConnection) OnICEConnectionStateChange(f func(webrtc.ICEConnectionState)) { 120 120 + pc.pionpc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) { 121 121 + now := time.Now() 122 122 + pc.Do(func() { 123 123 + pc.stream.Event(WebRTCEvent{ 124 124 + ICEConnectionStateChange: &ICEConnectionStateChange{ 125 125 + ICEConnectionState: state, 126 126 + }, 127 127 + Time: now, 128 128 + }) 129 129 + }) 130 130 + f(state) 131 131 + }) 132 132 + } 133 133 + 134 134 + func (pc *RecordingPeerConnection) OnConnectionStateChange(f func(webrtc.PeerConnectionState)) { 135 135 + pc.pionpc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { 136 136 + now := time.Now() 137 137 + pc.Do(func() { 138 138 + pc.stream.Event(WebRTCEvent{ 139 139 + ConnectionStateChange: &ConnectionStateChange{ 140 140 + ConnectionState: state, 141 141 + }, 142 142 + Time: now, 143 143 + }) 144 144 + }) 145 145 + f(state) 146 146 + }) 147 147 + } 148 148 + 149 149 + func (pc *RecordingPeerConnection) OnTrack(f func(TrackRemote, RTPReceiver)) { 150 150 + pc.pionpc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) { 151 151 + now := time.Now() 152 152 + wrappedTrack := &RecordingTrackRemote{track: track, stream: pc.stream, pc: pc} 153 153 + id := track.ID() 154 154 + kind := track.Kind() 155 155 + ssrc := track.SSRC() 156 156 + payloadType := track.PayloadType() 157 157 + streamID := track.StreamID() 158 158 + msid := track.Msid() 159 159 + rid := track.RID() 160 160 + pc.Do(func() { 161 161 + pc.stream.Event(WebRTCEvent{ 162 162 + Track: &Track{ 163 163 + ID: id, 164 164 + Kind: kind, 165 165 + SSRC: ssrc, 166 166 + PayloadType: payloadType, 167 167 + StreamID: streamID, 168 168 + Msid: msid, 169 169 + RID: rid, 170 170 + }, 171 171 + Time: now, 172 172 + }) 173 173 + }) 174 174 + f(wrappedTrack, receiver) 175 175 + }) 176 176 + } 177 177 + 178 178 + func (pc *RecordingPeerConnection) WriteRTCP(pkts []rtcp.Packet) error { 179 179 + return pc.pionpc.WriteRTCP(pkts) 180 180 + } 181 181 + 182 182 + func (pc *RecordingPeerConnection) AddTransceiverFromKind(kind webrtc.RTPCodecType, init ...webrtc.RTPTransceiverInit) (RTPTransceiver, error) { 183 183 + now := time.Now() 184 184 + ret, err := pc.pionpc.AddTransceiverFromKind(kind, init...) 185 185 + pc.Do(func() { 186 186 + pc.stream.Event(WebRTCEvent{ 187 187 + AddTransceiverFromKind: &AddTransceiverFromKind{ 188 188 + Kind: kind, 189 189 + }, 190 190 + Time: now, 191 191 + }) 192 192 + }) 193 193 + return ret, err 194 194 + } 195 195 + 196 196 + func (pc *RecordingPeerConnection) ICEGatheringState() webrtc.ICEGatheringState { 197 197 + now := time.Now() 198 198 + state := pc.pionpc.ICEGatheringState() 199 199 + pc.Do(func() { 200 200 + pc.stream.Event(WebRTCEvent{ 201 201 + ICEGatheringState: &ICEGatheringState{ 202 202 + State: state, 203 203 + }, 204 204 + Time: now, 205 205 + }) 206 206 + }) 207 207 + return state 208 208 + } 209 209 + 210 210 + func (pc *RecordingPeerConnection) OnDataChannel(f func(*webrtc.DataChannel)) { 211 211 + pc.pionpc.OnDataChannel(func(dc *webrtc.DataChannel) { 212 212 + now := time.Now() 213 213 + pc.Do(func() { 214 214 + pc.stream.Event(WebRTCEvent{ 215 215 + DataChannel: &DataChannel{ 216 216 + Label: dc.Label(), 217 217 + }, 218 218 + Time: now, 219 219 + }) 220 220 + }) 221 221 + f(dc) 222 222 + }) 223 223 + } 224 224 + 225 225 + func (pc *RecordingPeerConnection) OnNegotiationNeeded(f func()) { 226 226 + pc.pionpc.OnNegotiationNeeded(func() { 227 227 + now := time.Now() 228 228 + pc.Do(func() { 229 229 + pc.stream.Event(WebRTCEvent{ 230 230 + NegotiationNeeded: &NegotiationNeeded{}, 231 231 + Time: now, 232 232 + }) 233 233 + }) 234 234 + f() 235 235 + }) 236 236 + }

+105

pkg/rtcrec/recording_track.go

··· 1 1 + package rtcrec 2 2 + 3 3 + import ( 4 4 + "time" 5 5 + 6 6 + "github.com/pion/interceptor" 7 7 + "github.com/pion/webrtc/v4" 8 8 + ) 9 9 + 10 10 + type RecordingTrackRemote struct { 11 11 + track *webrtc.TrackRemote 12 12 + stream *RecorderStream 13 13 + pc *RecordingPeerConnection 14 14 + } 15 15 + 16 16 + func (t *RecordingTrackRemote) do(f func()) { 17 17 + go f() 18 18 + } 19 19 + 20 20 + func (t *RecordingTrackRemote) Read(p []byte) (n int, attrs interceptor.Attributes, err error) { 21 21 + n, attrs, err = t.track.Read(p) 22 22 + now := time.Now() 23 23 + b2 := make([]byte, n) 24 24 + copy(b2, p) 25 25 + t.pc.Do(func() { 26 26 + errString := "" 27 27 + if err != nil { 28 28 + errString = err.Error() 29 29 + } 30 30 + t.stream.Event(WebRTCEvent{ 31 31 + TrackRead: &TrackRead{ 32 32 + Data: b2, 33 33 + SSRC: t.track.SSRC(), 34 34 + Count: n, 35 35 + // Attrs: attrs, 36 36 + Err: errString, 37 37 + }, 38 38 + Time: now, 39 39 + }) 40 40 + }) 41 41 + return n, attrs, err 42 42 + } 43 43 + 44 44 + func (t *RecordingTrackRemote) Codec() webrtc.RTPCodecParameters { 45 45 + now := time.Now() 46 46 + codec := t.track.Codec() 47 47 + t.pc.Do(func() { 48 48 + t.stream.Event(WebRTCEvent{ 49 49 + TrackCodec: &TrackCodec{ 50 50 + SSRC: t.track.SSRC(), 51 51 + Codec: codec, 52 52 + }, 53 53 + Time: now, 54 54 + }) 55 55 + }) 56 56 + return codec 57 57 + } 58 58 + 59 59 + func (t *RecordingTrackRemote) ID() string { 60 60 + return t.track.ID() 61 61 + } 62 62 + 63 63 + func (t *RecordingTrackRemote) Kind() webrtc.RTPCodecType { 64 64 + now := time.Now() 65 65 + kind := t.track.Kind() 66 66 + t.pc.Do(func() { 67 67 + t.stream.Event(WebRTCEvent{ 68 68 + TrackKind: &TrackKind{ 69 69 + SSRC: t.track.SSRC(), 70 70 + Kind: kind, 71 71 + }, 72 72 + Time: now, 73 73 + }) 74 74 + }) 75 75 + return kind 76 76 + } 77 77 + 78 78 + func (t *RecordingTrackRemote) PayloadType() webrtc.PayloadType { 79 79 + now := time.Now() 80 80 + payloadType := t.track.PayloadType() 81 81 + t.pc.Do(func() { 82 82 + t.stream.Event(WebRTCEvent{ 83 83 + TrackPayloadType: &TrackPayloadType{ 84 84 + SSRC: t.track.SSRC(), 85 85 + PayloadType: payloadType, 86 86 + }, 87 87 + Time: now, 88 88 + }) 89 89 + }) 90 90 + return payloadType 91 91 + } 92 92 + 93 93 + func (t *RecordingTrackRemote) SSRC() webrtc.SSRC { 94 94 + now := time.Now() 95 95 + ssrc := t.track.SSRC() 96 96 + t.pc.Do(func() { 97 97 + t.stream.Event(WebRTCEvent{ 98 98 + Time: now, 99 99 + TrackSSRC: &TrackSSRC{ 100 100 + SSRC: ssrc, 101 101 + }, 102 102 + }) 103 103 + }) 104 104 + return ssrc 105 105 + }

+148

pkg/rtcrec/replay_peerconnection.go

··· 1 1 + package rtcrec 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + "io" 7 7 + "time" 8 8 + 9 9 + "github.com/pion/rtcp" 10 10 + "github.com/pion/webrtc/v4" 11 11 + "stream.place/streamplace/pkg/log" 12 12 + ) 13 13 + 14 14 + type ReplayPeerConnection struct { 15 15 + startTime time.Time 16 16 + group *WebRTCEventGroup 17 17 + ctx context.Context 18 18 + } 19 19 + 20 20 + func NewReplayPeerConnection(ctx context.Context, r io.Reader) (PeerConnection, error) { 21 21 + group, err := ReadAllEvents(r) 22 22 + if err != nil { 23 23 + return nil, fmt.Errorf("failed to create web rtc decoder: %w", err) 24 24 + } 25 25 + 26 26 + return &ReplayPeerConnection{ 27 27 + startTime: time.Now(), 28 28 + group: group, 29 29 + ctx: context.Background(), 30 30 + }, nil 31 31 + } 32 32 + 33 33 + func (pc *ReplayPeerConnection) wait(label string, t time.Time) <-chan time.Time { 34 34 + now := time.Now() 35 35 + historicalDiff := t.Sub(pc.group.FirstTime) 36 36 + currentDiff := time.Since(pc.startTime) 37 37 + diff := historicalDiff - currentDiff 38 38 + log.Debug(pc.ctx, "waiting for event", "event", label, "diff", diff, "t", t, "first", pc.group.FirstTime, "startTime", pc.startTime, "now", now) 39 39 + return time.After(diff) 40 40 + } 41 41 + 42 42 + func (pc *ReplayPeerConnection) Close() error { 43 43 + // todo: implment stopping here 44 44 + return nil 45 45 + } 46 46 + 47 47 + func (pc *ReplayPeerConnection) CreateAnswer(options *webrtc.AnswerOptions) (webrtc.SessionDescription, error) { 48 48 + ev := pc.group.Peek(EventTypeCreateAnswer) 49 49 + if ev == nil { 50 50 + return webrtc.SessionDescription{}, fmt.Errorf("no create answer event found") 51 51 + } 52 52 + return webrtc.SessionDescription{SDP: ev.CreateAnswer.SDPAnswer}, nil 53 53 + } 54 54 + 55 55 + func (pc *ReplayPeerConnection) SetLocalDescription(desc webrtc.SessionDescription) error { 56 56 + return nil 57 57 + } 58 58 + 59 59 + func (pc *ReplayPeerConnection) SetRemoteDescription(desc webrtc.SessionDescription) error { 60 60 + return nil 61 61 + } 62 62 + 63 63 + func (pc *ReplayPeerConnection) LocalDescription() *webrtc.SessionDescription { 64 64 + ev := pc.group.Peek(EventTypeLocalDescription) 65 65 + if ev == nil { 66 66 + return nil 67 67 + } 68 68 + return &webrtc.SessionDescription{SDP: ev.LocalDescription.SDPLocalDescription} 69 69 + } 70 70 + 71 71 + // func (pc *ReplayPeerConnection) RemoteDescription() *webrtc.SessionDescription { 72 72 + // return pc.pionpc.RemoteDescription() 73 73 + // } 74 74 + 75 75 + func (pc *ReplayPeerConnection) OnICEConnectionStateChange(f func(webrtc.ICEConnectionState)) { 76 76 + go func() { 77 77 + for { 78 78 + ev := pc.group.Next(EventTypeICEConnectionState) 79 79 + if ev == nil { 80 80 + return 81 81 + } 82 82 + select { 83 83 + case <-pc.wait("OnICEConnectionStateChange", ev.Time): 84 84 + f(ev.ICEConnectionStateChange.ICEConnectionState) 85 85 + case <-pc.ctx.Done(): 86 86 + return 87 87 + } 88 88 + } 89 89 + }() 90 90 + } 91 91 + 92 92 + func (pc *ReplayPeerConnection) OnConnectionStateChange(f func(webrtc.PeerConnectionState)) { 93 93 + go func() { 94 94 + for { 95 95 + ev := pc.group.Next(EventTypeConnectionState) 96 96 + if ev == nil { 97 97 + return 98 98 + } 99 99 + select { 100 100 + case <-pc.wait("OnConnectionStateChange", ev.Time): 101 101 + f(ev.ConnectionStateChange.ConnectionState) 102 102 + case <-pc.ctx.Done(): 103 103 + return 104 104 + } 105 105 + } 106 106 + }() 107 107 + } 108 108 + 109 109 + func (pc *ReplayPeerConnection) OnTrack(f func(TrackRemote, RTPReceiver)) { 110 110 + go func() { 111 111 + for { 112 112 + ev := pc.group.Next(EventTypeTrack) 113 113 + if ev == nil { 114 114 + return 115 115 + } 116 116 + select { 117 117 + case <-pc.wait("OnTrack", ev.Time): 118 118 + track := &ReplayTrackRemote{ 119 119 + ssrc: ev.Track.SSRC, 120 120 + trackEvent: ev, 121 121 + events: pc.group.Tracks[ev.Track.SSRC], 122 122 + pc: pc, 123 123 + } 124 124 + go func() { 125 125 + f(track, nil) 126 126 + }() 127 127 + case <-pc.ctx.Done(): 128 128 + return 129 129 + } 130 130 + } 131 131 + }() 132 132 + } 133 133 + 134 134 + func (pc *ReplayPeerConnection) WriteRTCP(pkts []rtcp.Packet) error { 135 135 + return nil 136 136 + } 137 137 + 138 138 + func (pc *ReplayPeerConnection) AddTransceiverFromKind(kind webrtc.RTPCodecType, init ...webrtc.RTPTransceiverInit) (RTPTransceiver, error) { 139 139 + return nil, nil 140 140 + } 141 141 + 142 142 + func (pc *ReplayPeerConnection) ICEGatheringState() webrtc.ICEGatheringState { 143 143 + ev := pc.group.Peek(EventTypeICEGatheringState) 144 144 + if ev == nil { 145 145 + return webrtc.ICEGatheringStateNew 146 146 + } 147 147 + return ev.ICEGatheringState.State 148 148 + }

+64

pkg/rtcrec/replay_track.go

··· 1 1 + package rtcrec 2 2 + 3 3 + import ( 4 4 + "errors" 5 5 + "fmt" 6 6 + 7 7 + "github.com/pion/interceptor" 8 8 + "github.com/pion/webrtc/v4" 9 9 + ) 10 10 + 11 11 + type ReplayTrackRemote struct { 12 12 + ssrc webrtc.SSRC 13 13 + trackEvent *WebRTCEvent 14 14 + events map[string][]*WebRTCEvent 15 15 + pc *ReplayPeerConnection 16 16 + } 17 17 + 18 18 + func (t *ReplayTrackRemote) Read(p []byte) (n int, attrs interceptor.Attributes, err error) { 19 19 + ev := t.pc.group.NextTrack(t.ssrc, EventTypeTrackRead) 20 20 + if ev == nil { 21 21 + return 0, nil, nil 22 22 + } 23 23 + select { 24 24 + case <-t.pc.wait(fmt.Sprintf("TrackRead %s", t.trackEvent.Track.ID), ev.Time): 25 25 + var err error 26 26 + if ev.TrackRead.Err != "" { 27 27 + err = errors.New(ev.TrackRead.Err) 28 28 + } 29 29 + if ev.TrackRead.Data != nil { 30 30 + copied := copy(p, ev.TrackRead.Data) 31 31 + if copied != len(ev.TrackRead.Data) { 32 32 + panic("copied != len(ev.TrackRead.Data)") 33 33 + } 34 34 + } 35 35 + // log.Log(t.pc.ctx, "TrackRead", "trackId", t.trackEvent.Track.ID, "count", ev.TrackRead.Count, "err", err) 36 36 + return ev.TrackRead.Count, nil, err 37 37 + case <-t.pc.ctx.Done(): 38 38 + return 0, nil, t.pc.ctx.Err() 39 39 + } 40 40 + } 41 41 + 42 42 + func (t *ReplayTrackRemote) Codec() webrtc.RTPCodecParameters { 43 43 + ev := t.pc.group.PeekTrack(t.ssrc, EventTypeTrackCodec) 44 44 + if ev == nil { 45 45 + panic("no codec found") 46 46 + } 47 47 + return ev.TrackCodec.Codec 48 48 + } 49 49 + 50 50 + func (t *ReplayTrackRemote) ID() string { 51 51 + return t.trackEvent.Track.ID 52 52 + } 53 53 + 54 54 + func (t *ReplayTrackRemote) Kind() webrtc.RTPCodecType { 55 55 + return t.trackEvent.Track.Kind 56 56 + } 57 57 + 58 58 + func (t *ReplayTrackRemote) PayloadType() webrtc.PayloadType { 59 59 + return t.trackEvent.Track.PayloadType 60 60 + } 61 61 + 62 62 + func (t *ReplayTrackRemote) SSRC() webrtc.SSRC { 63 63 + return t.ssrc 64 64 + }

+56

pkg/rtcrec/rtcrec.go

··· 1 1 + package rtcrec 2 2 + 3 3 + import ( 4 4 + "time" 5 5 + 6 6 + "github.com/pion/interceptor" 7 7 + "github.com/pion/rtcp" 8 8 + "github.com/pion/webrtc/v4" 9 9 + ) 10 10 + 11 11 + type PeerConnection interface { 12 12 + AddTransceiverFromKind(kind webrtc.RTPCodecType, init ...webrtc.RTPTransceiverInit) (RTPTransceiver, error) 13 13 + Close() error 14 14 + SetRemoteDescription(description webrtc.SessionDescription) error 15 15 + CreateAnswer(options *webrtc.AnswerOptions) (webrtc.SessionDescription, error) 16 16 + SetLocalDescription(description webrtc.SessionDescription) error 17 17 + OnICEConnectionStateChange(func(webrtc.ICEConnectionState)) 18 18 + OnConnectionStateChange(func(webrtc.PeerConnectionState)) 19 19 + OnTrack(func(TrackRemote, RTPReceiver)) 20 20 + // OnDataChannel(func(*webrtc.DataChannel)) 21 21 + // OnNegotiationNeeded(func()) 22 22 + WriteRTCP(pkts []rtcp.Packet) error 23 23 + ICEGatheringState() webrtc.ICEGatheringState 24 24 + LocalDescription() *webrtc.SessionDescription 25 25 + } 26 26 + 27 27 + type RTPTransceiver interface { 28 28 + } 29 29 + 30 30 + type TrackRemote interface { 31 31 + Read(p []byte) (n int, attrs interceptor.Attributes, err error) 32 32 + Kind() webrtc.RTPCodecType 33 33 + PayloadType() webrtc.PayloadType 34 34 + Codec() webrtc.RTPCodecParameters 35 35 + SSRC() webrtc.SSRC 36 36 + } 37 37 + 38 38 + type RTPReceiver interface { 39 39 + } 40 40 + 41 41 + func GatheringCompletePromise(pc PeerConnection) <-chan struct{} { 42 42 + recorder, ok := pc.(*RecordingPeerConnection) 43 43 + if ok { 44 44 + return webrtc.GatheringCompletePromise(recorder.pionpc) 45 45 + } 46 46 + _, ok = pc.(*ReplayPeerConnection) 47 47 + if ok { 48 48 + ch := make(chan struct{}) 49 49 + go func() { 50 50 + <-time.After(100 * time.Millisecond) 51 51 + ch <- struct{}{} 52 52 + }() 53 53 + return ch 54 54 + } 55 55 + panic("unknown peer connection type") 56 56 + }

+152

pkg/rtcrec/webrtc_recording.go

··· 1 1 + package rtcrec 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + "io" 7 7 + "time" 8 8 + 9 9 + "github.com/fxamacker/cbor/v2" 10 10 + "github.com/pion/webrtc/v4" 11 11 + "stream.place/streamplace/pkg/log" 12 12 + ) 13 13 + 14 14 + type WebRTCRecording struct { 15 15 + Events []WebRTCEvent `json:"events,omitempty"` 16 16 + } 17 17 + 18 18 + type WebRTCEvent struct { 19 19 + Offer *Offer `json:"offer,omitempty"` 20 20 + CreateAnswer *CreateAnswer `json:"answer,omitempty"` 21 21 + SetRemoteDescription *SetRemoteDescription `json:"setRemoteDescription,omitempty"` 22 22 + SetLocalDescription *SetLocalDescription `json:"setLocalDescription,omitempty"` 23 23 + LocalDescription *LocalDescription `json:"localDescription,omitempty"` 24 24 + ICEConnectionStateChange *ICEConnectionStateChange `json:"iceConnectionStateChange,omitempty"` 25 25 + ConnectionStateChange *ConnectionStateChange `json:"connectionStateChange,omitempty"` 26 26 + Track *Track `json:"track,omitempty"` 27 27 + TrackRead *TrackRead `json:"trackRead,omitempty"` 28 28 + TrackCodec *TrackCodec `json:"trackCodec,omitempty"` 29 29 + TrackKind *TrackKind `json:"trackKind,omitempty"` 30 30 + TrackPayloadType *TrackPayloadType `json:"trackPayloadType,omitempty"` 31 31 + TrackSSRC *TrackSSRC `json:"trackSSRC,omitempty"` 32 32 + AddTransceiverFromKind *AddTransceiverFromKind `json:"addTransceiverFromKind,omitempty"` 33 33 + ICEGatheringState *ICEGatheringState `json:"iceGatheringState,omitempty"` 34 34 + DataChannel *DataChannel `json:"dataChannel,omitempty"` 35 35 + NegotiationNeeded *NegotiationNeeded `json:"negotiationNeeded,omitempty"` 36 36 + Time time.Time `json:"time,omitempty"` 37 37 + } 38 38 + 39 39 + func (e *WebRTCEvent) Detail() WebRTCEventDetail { 40 40 + if e.Offer != nil { 41 41 + return e.Offer 42 42 + } 43 43 + if e.CreateAnswer != nil { 44 44 + return e.CreateAnswer 45 45 + } 46 46 + return nil 47 47 + } 48 48 + 49 49 + type WebRTCEventDetail interface{} 50 50 + 51 51 + type Offer struct { 52 52 + SDPOffer string `json:"sdpOffer,omitempty"` 53 53 + } 54 54 + 55 55 + type CreateAnswer struct { 56 56 + SDPAnswer string `json:"sdpAnswer,omitempty"` 57 57 + } 58 58 + 59 59 + type SetRemoteDescription struct { 60 60 + SDPRemoteDescription string `json:"sdpRemoteDescription,omitempty"` 61 61 + } 62 62 + 63 63 + type SetLocalDescription struct { 64 64 + SDPLocalDescription string `json:"sdpRemoteDescription,omitempty"` 65 65 + } 66 66 + 67 67 + type LocalDescription struct { 68 68 + SDPLocalDescription string `json:"sdpLocalDescription,omitempty"` 69 69 + } 70 70 + 71 71 + type ICEConnectionStateChange struct { 72 72 + ICEConnectionState webrtc.ICEConnectionState `json:"iceConnectionState,omitempty"` 73 73 + } 74 74 + 75 75 + type ConnectionStateChange struct { 76 76 + ConnectionState webrtc.PeerConnectionState `json:"connectionState,omitempty"` 77 77 + } 78 78 + 79 79 + type Track struct { 80 80 + ID string `json:"id,omitempty"` 81 81 + Kind webrtc.RTPCodecType `json:"kind,omitempty"` 82 82 + SSRC webrtc.SSRC `json:"ssrc,omitempty"` 83 83 + PayloadType webrtc.PayloadType `json:"payloadType,omitempty"` 84 84 + StreamID string `json:"streamId,omitempty"` 85 85 + Msid string `json:"msid,omitempty"` 86 86 + RID string `json:"rid,omitempty"` 87 87 + } 88 88 + 89 89 + type TrackRead struct { 90 90 + SSRC webrtc.SSRC `json:"ssrc,omitempty"` 91 91 + Data []byte `json:"data,omitempty"` 92 92 + Count int `json:"count,omitempty"` 93 93 + Err string `json:"err,omitempty"` 94 94 + } 95 95 + 96 96 + type TrackCodec struct { 97 97 + SSRC webrtc.SSRC `json:"ssrc,omitempty"` 98 98 + Codec webrtc.RTPCodecParameters `json:"codec,omitempty"` 99 99 + } 100 100 + 101 101 + type TrackKind struct { 102 102 + SSRC webrtc.SSRC `json:"ssrc,omitempty"` 103 103 + Kind webrtc.RTPCodecType `json:"kind,omitempty"` 104 104 + } 105 105 + 106 106 + type TrackPayloadType struct { 107 107 + SSRC webrtc.SSRC `json:"ssrc,omitempty"` 108 108 + PayloadType webrtc.PayloadType `json:"payloadType,omitempty"` 109 109 + } 110 110 + 111 111 + type TrackSSRC struct { 112 112 + SSRC webrtc.SSRC `json:"ssrc,omitempty"` 113 113 + } 114 114 + 115 115 + type RecorderStream struct { 116 116 + encoder *cbor.Encoder 117 117 + } 118 118 + 119 119 + func NewRecorderStream(w io.Writer) (*RecorderStream, error) { 120 120 + opts := cbor.CoreDetEncOptions() 121 121 + opts.Time = cbor.TimeRFC3339Nano 122 122 + em, err := opts.EncMode() 123 123 + if err != nil { 124 124 + return nil, fmt.Errorf("failed to create encoder mode: %w", err) 125 125 + } 126 126 + encoder := em.NewEncoder(w) 127 127 + 128 128 + return &RecorderStream{ 129 129 + encoder: encoder, 130 130 + }, nil 131 131 + } 132 132 + 133 133 + func (s *RecorderStream) Event(event WebRTCEvent) { 134 134 + err := s.encoder.Encode(event) 135 135 + if err != nil { 136 136 + log.Log(context.Background(), "error encoding event", "error", err) 137 137 + } 138 138 + } 139 139 + 140 140 + type AddTransceiverFromKind struct { 141 141 + Kind webrtc.RTPCodecType `json:"kind,omitempty"` 142 142 + } 143 143 + 144 144 + type ICEGatheringState struct { 145 145 + State webrtc.ICEGatheringState `json:"state,omitempty"` 146 146 + } 147 147 + 148 148 + type DataChannel struct { 149 149 + Label string `json:"label,omitempty"` 150 150 + } 151 151 + 152 152 + type NegotiationNeeded struct{}

+72

pkg/rtcrec/webrtc_recording_test.go

··· 1 1 + package rtcrec 2 2 + 3 3 + import ( 4 4 + "io" 5 5 + "os" 6 6 + "testing" 7 7 + "time" 8 8 + 9 9 + "github.com/fxamacker/cbor/v2" 10 10 + "github.com/stretchr/testify/require" 11 11 + ) 12 12 + 13 13 + func TestWebRTCRecording(t *testing.T) { 14 14 + // Create a temporary file for testing 15 15 + tmpfile, err := os.CreateTemp("", "webrtc-recording-test-*") 16 16 + require.NoError(t, err) 17 17 + 18 18 + // Create recorder stream writing to temp file 19 19 + recorder, err := NewRecorderStream(tmpfile) 20 20 + require.NoError(t, err) 21 21 + 22 22 + // Test recording an offer event 23 23 + offerEvent := WebRTCEvent{ 24 24 + Offer: &Offer{ 25 25 + SDPOffer: "test-offer", 26 26 + }, 27 27 + Time: time.Now().UTC(), 28 28 + } 29 29 + recorder.Event(offerEvent) 30 30 + 31 31 + // Test recording an answer event 32 32 + answerEvent := WebRTCEvent{ 33 33 + CreateAnswer: &CreateAnswer{ 34 34 + SDPAnswer: "test-answer", 35 35 + }, 36 36 + Time: time.Now().UTC(), 37 37 + } 38 38 + recorder.Event(answerEvent) 39 39 + 40 40 + // err = recorder.Close() 41 41 + // require.NoError(t, err) 42 42 + err = tmpfile.Close() 43 43 + require.NoError(t, err) 44 44 + 45 45 + tmpfile, err = os.Open(tmpfile.Name()) 46 46 + require.NoError(t, err) 47 47 + defer tmpfile.Close() 48 48 + 49 49 + dec := cbor.NewDecoder(tmpfile) 50 50 + 51 51 + evs := []WebRTCEvent{} 52 52 + err = nil 53 53 + for err == nil { 54 54 + var ev WebRTCEvent 55 55 + err = dec.Decode(&ev) 56 56 + if err == nil { 57 57 + evs = append(evs, ev) 58 58 + } 59 59 + } 60 60 + 61 61 + require.ErrorIs(t, err, io.EOF) 62 62 + 63 63 + off, ok := evs[0].Detail().(*Offer) 64 64 + require.True(t, ok) 65 65 + ans, ok := evs[1].Detail().(*CreateAnswer) 66 66 + require.True(t, ok) 67 67 + 68 68 + require.Equal(t, 2, len(evs)) 69 69 + require.Equal(t, off.SDPOffer, offerEvent.Offer.SDPOffer) 70 70 + require.Equal(t, ans.SDPAnswer, answerEvent.CreateAnswer.SDPAnswer) 71 71 + 72 72 + }

+1

pkg/rtcrec/wrapped_peerconnection.go

··· 1 1 + package rtcrec

+159

pkg/streamplace/cbor_gen.go

··· 2610 2610 2611 2611 return nil 2612 2612 } 2613 2613 + func (t *ServerSettings) MarshalCBOR(w io.Writer) error { 2614 2614 + if t == nil { 2615 2615 + _, err := w.Write(cbg.CborNull) 2616 2616 + return err 2617 2617 + } 2618 2618 + 2619 2619 + cw := cbg.NewCborWriter(w) 2620 2620 + fieldCount := 2 2621 2621 + 2622 2622 + if t.DebugRecording == nil { 2623 2623 + fieldCount-- 2624 2624 + } 2625 2625 + 2626 2626 + if _, err := cw.Write(cbg.CborEncodeMajorType(cbg.MajMap, uint64(fieldCount))); err != nil { 2627 2627 + return err 2628 2628 + } 2629 2629 + 2630 2630 + // t.LexiconTypeID (string) (string) 2631 2631 + if len("$type") > 1000000 { 2632 2632 + return xerrors.Errorf("Value in field \"$type\" was too long") 2633 2633 + } 2634 2634 + 2635 2635 + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("$type"))); err != nil { 2636 2636 + return err 2637 2637 + } 2638 2638 + if _, err := cw.WriteString(string("$type")); err != nil { 2639 2639 + return err 2640 2640 + } 2641 2641 + 2642 2642 + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("place.stream.server.settings"))); err != nil { 2643 2643 + return err 2644 2644 + } 2645 2645 + if _, err := cw.WriteString(string("place.stream.server.settings")); err != nil { 2646 2646 + return err 2647 2647 + } 2648 2648 + 2649 2649 + // t.DebugRecording (bool) (bool) 2650 2650 + if t.DebugRecording != nil { 2651 2651 + 2652 2652 + if len("debugRecording") > 1000000 { 2653 2653 + return xerrors.Errorf("Value in field \"debugRecording\" was too long") 2654 2654 + } 2655 2655 + 2656 2656 + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("debugRecording"))); err != nil { 2657 2657 + return err 2658 2658 + } 2659 2659 + if _, err := cw.WriteString(string("debugRecording")); err != nil { 2660 2660 + return err 2661 2661 + } 2662 2662 + 2663 2663 + if t.DebugRecording == nil { 2664 2664 + if _, err := cw.Write(cbg.CborNull); err != nil { 2665 2665 + return err 2666 2666 + } 2667 2667 + } else { 2668 2668 + if err := cbg.WriteBool(w, *t.DebugRecording); err != nil { 2669 2669 + return err 2670 2670 + } 2671 2671 + } 2672 2672 + } 2673 2673 + return nil 2674 2674 + } 2675 2675 + 2676 2676 + func (t *ServerSettings) UnmarshalCBOR(r io.Reader) (err error) { 2677 2677 + *t = ServerSettings{} 2678 2678 + 2679 2679 + cr := cbg.NewCborReader(r) 2680 2680 + 2681 2681 + maj, extra, err := cr.ReadHeader() 2682 2682 + if err != nil { 2683 2683 + return err 2684 2684 + } 2685 2685 + defer func() { 2686 2686 + if err == io.EOF { 2687 2687 + err = io.ErrUnexpectedEOF 2688 2688 + } 2689 2689 + }() 2690 2690 + 2691 2691 + if maj != cbg.MajMap { 2692 2692 + return fmt.Errorf("cbor input should be of type map") 2693 2693 + } 2694 2694 + 2695 2695 + if extra > cbg.MaxLength { 2696 2696 + return fmt.Errorf("ServerSettings: map struct too large (%d)", extra) 2697 2697 + } 2698 2698 + 2699 2699 + n := extra 2700 2700 + 2701 2701 + nameBuf := make([]byte, 14) 2702 2702 + for i := uint64(0); i < n; i++ { 2703 2703 + nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 1000000) 2704 2704 + if err != nil { 2705 2705 + return err 2706 2706 + } 2707 2707 + 2708 2708 + if !ok { 2709 2709 + // Field doesn't exist on this type, so ignore it 2710 2710 + if err := cbg.ScanForLinks(cr, func(cid.Cid) {}); err != nil { 2711 2711 + return err 2712 2712 + } 2713 2713 + continue 2714 2714 + } 2715 2715 + 2716 2716 + switch string(nameBuf[:nameLen]) { 2717 2717 + // t.LexiconTypeID (string) (string) 2718 2718 + case "$type": 2719 2719 + 2720 2720 + { 2721 2721 + sval, err := cbg.ReadStringWithMax(cr, 1000000) 2722 2722 + if err != nil { 2723 2723 + return err 2724 2724 + } 2725 2725 + 2726 2726 + t.LexiconTypeID = string(sval) 2727 2727 + } 2728 2728 + // t.DebugRecording (bool) (bool) 2729 2729 + case "debugRecording": 2730 2730 + 2731 2731 + { 2732 2732 + b, err := cr.ReadByte() 2733 2733 + if err != nil { 2734 2734 + return err 2735 2735 + } 2736 2736 + if b != cbg.CborNull[0] { 2737 2737 + if err := cr.UnreadByte(); err != nil { 2738 2738 + return err 2739 2739 + } 2740 2740 + 2741 2741 + maj, extra, err = cr.ReadHeader() 2742 2742 + if err != nil { 2743 2743 + return err 2744 2744 + } 2745 2745 + if maj != cbg.MajOther { 2746 2746 + return fmt.Errorf("booleans must be major type 7") 2747 2747 + } 2748 2748 + 2749 2749 + var val bool 2750 2750 + switch extra { 2751 2751 + case 20: 2752 2752 + val = false 2753 2753 + case 21: 2754 2754 + val = true 2755 2755 + default: 2756 2756 + return fmt.Errorf("booleans are either major type 7, value 20 or 21 (got %d)", extra) 2757 2757 + } 2758 2758 + t.DebugRecording = &val 2759 2759 + } 2760 2760 + } 2761 2761 + 2762 2762 + default: 2763 2763 + // Field doesn't exist on this type, so ignore it 2764 2764 + if err := cbg.ScanForLinks(r, func(cid.Cid) {}); err != nil { 2765 2765 + return err 2766 2766 + } 2767 2767 + } 2768 2768 + } 2769 2769 + 2770 2770 + return nil 2771 2771 + }

+19

pkg/streamplace/serversettings.go

··· 1 1 + // Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. 2 2 + 3 3 + package streamplace 4 4 + 5 5 + // schema: place.stream.server.settings 6 6 + 7 7 + import ( 8 8 + "github.com/bluesky-social/indigo/lex/util" 9 9 + ) 10 10 + 11 11 + func init() { 12 12 + util.RegisterType("place.stream.server.settings", &ServerSettings{}) 13 13 + } // 14 14 + // RECORDTYPE: ServerSettings 15 15 + type ServerSettings struct { 16 16 + LexiconTypeID string `json:"$type,const=place.stream.server.settings" cborgen:"$type,const=place.stream.server.settings"` 17 17 + // debugRecording: Whether this node may archive your livestream for improving the service 18 18 + DebugRecording *bool `json:"debugRecording,omitempty" cborgen:"debugRecording,omitempty"` 19 19 + }