From f630c4c6cebe5730d70bbd91028ceecad86aeb3e Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Sun, 18 Oct 2026 15:15:14 -0700 Subject: [PATCH] Keep stream rendition config on the node; OBS encodes it Desktop's configuration is atproto records that stay on the node: place.stream.local.{put,list,delete}Record, shaped like com.atproto.repo.*, keep a signed-in author's records of the collections the node knows (localdb's local_records), ready to move to spaces later. The first two are place.stream.ingest.videoRendition and audioRendition. The OBS plugin's configuration request now gets encoder_configurations from the streamer's video renditions (one video track for now) and audio_configurations from their audio codecs, for the full mix and any stems. Entries name a codec instead of an OBS encoder type: the plugin uses OBS's own streaming encoder when it is that codec, under the node's bitrate, 1s keyframes and no B-frames, else the first encoder of the codec. Desktop no longer passes an Ethereum address as the node's only allowed streamer, which turned away the signed-in user's own stream keys. --- js/desktop/src/index.ts | 21 +- js/desktop/src/preload.ts | 5 +- .../place-stream-ingest-audiorendition.md | 63 ++++++ .../ingest/place-stream-ingest-defs.md | 32 +++ .../place-stream-ingest-videorendition.md | 85 ++++++++ .../local/place-stream-local-deleterecord.md | 86 ++++++++ .../local/place-stream-local-listrecords.md | 114 +++++++++++ .../local/place-stream-local-putrecord.md | 115 +++++++++++ .../content/docs/lex-reference/openapi.json | 192 ++++++++++++++++++ .../place/stream/ingest/audioRendition.json | 27 +++ lexicons/place/stream/ingest/defs.json | 9 + .../place/stream/ingest/videoRendition.json | 40 ++++ lexicons/place/stream/local/deleteRecord.json | 30 +++ lexicons/place/stream/local/listRecords.json | 40 ++++ lexicons/place/stream/local/putRecord.json | 52 +++++ obs-plugin/README.md | 8 +- obs-plugin/obs.def | 2 + obs-plugin/src/streamplace-output.c | 62 ++++-- pkg/api/fmp4_ingest.go | 138 ++++++++++--- pkg/api/fmp4_ingest_test.go | 31 ++- pkg/localdb/local_record.go | 39 ++++ pkg/localdb/localdb.go | 4 + pkg/placestream/ingestaudiorendition.go | 53 +++++ pkg/placestream/ingestdefs.go | 27 +++ pkg/placestream/ingestvideorendition.go | 56 +++++ pkg/placestream/localdeleterecord.go | 72 +++++++ pkg/placestream/locallistrecords.go | 80 ++++++++ pkg/placestream/localputrecord.go | 77 +++++++ pkg/spxrpc/local_record.go | 118 +++++++++++ pkg/spxrpc/local_record_test.go | 51 +++++ pkg/spxrpc/stubs.go | 52 +++++ 31 files changed, 1705 insertions(+), 76 deletions(-) create mode 100644 js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-audiorendition.md create mode 100644 js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-videorendition.md create mode 100644 js/docs/src/content/docs/lex-reference/local/place-stream-local-deleterecord.md create mode 100644 js/docs/src/content/docs/lex-reference/local/place-stream-local-listrecords.md create mode 100644 js/docs/src/content/docs/lex-reference/local/place-stream-local-putrecord.md create mode 100644 lexicons/place/stream/ingest/audioRendition.json create mode 100644 lexicons/place/stream/ingest/videoRendition.json create mode 100644 lexicons/place/stream/local/deleteRecord.json create mode 100644 lexicons/place/stream/local/listRecords.json create mode 100644 lexicons/place/stream/local/putRecord.json create mode 100644 pkg/localdb/local_record.go create mode 100644 pkg/placestream/ingestaudiorendition.go create mode 100644 pkg/placestream/ingestvideorendition.go create mode 100644 pkg/placestream/localdeleterecord.go create mode 100644 pkg/placestream/locallistrecords.go create mode 100644 pkg/placestream/localputrecord.go create mode 100644 pkg/spxrpc/local_record.go create mode 100644 pkg/spxrpc/local_record_test.go diff --git a/js/desktop/src/index.ts b/js/desktop/src/index.ts index b0164e07c..2609b4739 100644 --- a/js/desktop/src/index.ts +++ b/js/desktop/src/index.ts @@ -1,7 +1,7 @@ -import { app, dialog, ipcMain } from "electron"; +import { app, dialog } from "electron"; import { parseArgs } from "node:util"; import "source-map-support/register"; -import { generatePrivateKey, privateKeyToAccount } from "viem/accounts"; +import { generatePrivateKey } from "viem/accounts"; import getEnv from "./env"; import makeNode from "./node"; import runTests, { allTestNames } from "./tests/test-runner"; @@ -38,19 +38,10 @@ if (require("electron-squirrel-startup")) { ); app.on("ready", async () => { - let privateKey: `0x${string}`; - if (process.env.AQD_ADMIN_ACCOUNT_KEY) { - privateKey = process.env.AQD_ADMIN_ACCOUNT_KEY as `0x${string}`; - } else { - privateKey = generatePrivateKey(); - } - ipcMain.handle("getPrivateKey", () => privateKey); - const account = privateKeyToAccount(privateKey); - const env = { - SP_ADMIN_ACCOUNT: account.address.toLowerCase(), - SP_ALLOWED_STREAMS: account.address.toLowerCase(), - }; if (args["self-test"]) { + // The self-tests stream with a key of their own. + const privateKey = (process.env.AQD_ADMIN_ACCOUNT_KEY ?? + generatePrivateKey()) as `0x${string}`; app.on("window-all-closed", () => { // need to override this to prevent the app from quitting }); @@ -67,7 +58,7 @@ if (require("electron-squirrel-startup")) { } } else { try { - await start(env); + await start({}); } catch (e) { console.error(e); const dialogOpts: Electron.MessageBoxOptions = { diff --git a/js/desktop/src/preload.ts b/js/desktop/src/preload.ts index 88c342170..a02a09595 100644 --- a/js/desktop/src/preload.ts +++ b/js/desktop/src/preload.ts @@ -1,8 +1,5 @@ -import { contextBridge, ipcRenderer } from "electron"; +import { contextBridge } from "electron"; contextBridge.exposeInMainWorld("SP_ELECTRON", { isElectron: true, }); -contextBridge.exposeInMainWorld("getPrivateKey", () => { - return ipcRenderer.invoke("getPrivateKey"); -}); diff --git a/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-audiorendition.md b/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-audiorendition.md new file mode 100644 index 000000000..f4e2da746 --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-audiorendition.md @@ -0,0 +1,63 @@ +--- +title: place.stream.ingest.audioRendition +description: Reference for the place.stream.ingest.audioRendition lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `record` + +An audio codec the streamer's encoder sends each of its audio tracks in (the full mix and any stems). A node keeps these locally (place.stream.local.putRecord). + +**Record Key:** `any` + +**Record Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ----------- | --------- | ----- | ------------------------ | --------------------------- | +| `codec` | `string` | ✅ | | Known Values: `aac`, `opus` | +| `bitrate` | `integer` | ✅ | Target bitrate, in kbps. | Min: 1 | +| `createdAt` | `string` | ✅ | | Format: `datetime` | + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.ingest.audioRendition", + "defs": { + "main": { + "type": "record", + "key": "any", + "description": "An audio codec the streamer's encoder sends each of its audio tracks in (the full mix and any stems). A node keeps these locally (place.stream.local.putRecord).", + "record": { + "type": "object", + "required": ["codec", "bitrate", "createdAt"], + "properties": { + "codec": { + "type": "string", + "knownValues": ["aac", "opus"] + }, + "bitrate": { + "type": "integer", + "minimum": 1, + "description": "Target bitrate, in kbps." + }, + "createdAt": { + "type": "string", + "format": "datetime" + } + } + } + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-defs.md b/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-defs.md index 209b898aa..1dd0782b5 100644 --- a/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-defs.md +++ b/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-defs.md @@ -7,6 +7,23 @@ description: Reference for the place.stream.ingest.defs lexicon ## Definitions + + +### `framerate` + +**Type:** `object` + +A frame rate as a fraction, like 60/1 or 30000/1001. + +**Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ------------- | --------- | ----- | ----------- | ----------- | +| `numerator` | `integer` | ✅ | | Min: 1 | +| `denominator` | `integer` | ✅ | | Min: 1 | + +--- + ### `ingest` @@ -66,6 +83,21 @@ Something the streamer should fix about how they're connected. "lexicon": 1, "id": "place.stream.ingest.defs", "defs": { + "framerate": { + "type": "object", + "description": "A frame rate as a fraction, like 60/1 or 30000/1001.", + "required": ["numerator", "denominator"], + "properties": { + "numerator": { + "type": "integer", + "minimum": 1 + }, + "denominator": { + "type": "integer", + "minimum": 1 + } + } + }, "ingest": { "type": "object", "description": "An ingest URL for a Streamplace station.", diff --git a/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-videorendition.md b/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-videorendition.md new file mode 100644 index 000000000..d3b1ef38f --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-videorendition.md @@ -0,0 +1,85 @@ +--- +title: place.stream.ingest.videoRendition +description: Reference for the place.stream.ingest.videoRendition lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `record` + +A video track the streamer's encoder sends to the node they stream to, which tells the encoder about it when it connects (the Streamplace OBS plugin asks). A node keeps these locally (place.stream.local.putRecord). + +**Record Key:** `any` + +**Record Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ----------- | ----------------------------------------------------------------------------------------- | ----- | ------------------------ | -------------------- | +| `codec` | `string` | ✅ | | Known Values: `h264` | +| `width` | `integer` | ✅ | | Min: 1 | +| `height` | `integer` | ✅ | | Min: 1 | +| `framerate` | [`place.stream.ingest.defs#framerate`](/lex-reference/place-stream-ingest-defs#framerate) | ✅ | | | +| `bitrate` | `integer` | ✅ | Target bitrate, in kbps. | Min: 1 | +| `createdAt` | `string` | ✅ | | Format: `datetime` | + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.ingest.videoRendition", + "defs": { + "main": { + "type": "record", + "key": "any", + "description": "A video track the streamer's encoder sends to the node they stream to, which tells the encoder about it when it connects (the Streamplace OBS plugin asks). A node keeps these locally (place.stream.local.putRecord).", + "record": { + "type": "object", + "required": [ + "codec", + "width", + "height", + "framerate", + "bitrate", + "createdAt" + ], + "properties": { + "codec": { + "type": "string", + "knownValues": ["h264"] + }, + "width": { + "type": "integer", + "minimum": 1 + }, + "height": { + "type": "integer", + "minimum": 1 + }, + "framerate": { + "type": "ref", + "ref": "place.stream.ingest.defs#framerate" + }, + "bitrate": { + "type": "integer", + "minimum": 1, + "description": "Target bitrate, in kbps." + }, + "createdAt": { + "type": "string", + "format": "datetime" + } + } + } + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/local/place-stream-local-deleterecord.md b/js/docs/src/content/docs/lex-reference/local/place-stream-local-deleterecord.md new file mode 100644 index 000000000..f1506fdfb --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/local/place-stream-local-deleterecord.md @@ -0,0 +1,86 @@ +--- +title: place.stream.local.deleteRecord +description: Reference for the place.stream.local.deleteRecord lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `procedure` + +Delete a record the caller keeps on this node. Shaped like com.atproto.repo.deleteRecord; deleting one that doesn't exist succeeds. + +**Parameters:** _(None defined)_ + +**Input:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| ------------ | -------- | ----- | ----------- | ---------------------------------------- | +| `repo` | `string` | ✅ | | Format: `did` | +| `collection` | `string` | ✅ | | Format: `nsid` | +| `rkey` | `string` | ✅ | | Format: `record-key`
Max Length: 512 | + +**Output:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +_(No properties defined)_ + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.local.deleteRecord", + "defs": { + "main": { + "type": "procedure", + "description": "Delete a record the caller keeps on this node. Shaped like com.atproto.repo.deleteRecord; deleting one that doesn't exist succeeds.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repo", "collection", "rkey"], + "properties": { + "repo": { + "type": "string", + "format": "did" + }, + "collection": { + "type": "string", + "format": "nsid" + }, + "rkey": { + "type": "string", + "format": "record-key", + "maxLength": 512 + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "properties": {} + } + } + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/local/place-stream-local-listrecords.md b/js/docs/src/content/docs/lex-reference/local/place-stream-local-listrecords.md new file mode 100644 index 000000000..a5d5aa16f --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/local/place-stream-local-listrecords.md @@ -0,0 +1,114 @@ +--- +title: place.stream.local.listRecords +description: Reference for the place.stream.local.listRecords lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `query` + +List the caller's records of a collection kept on this node, in record-key order. Shaped like com.atproto.repo.listRecords. + +**Parameters:** + +| Name | Type | Req'd | Description | Constraints | +| ------------ | -------- | ----- | ----------- | -------------- | +| `repo` | `string` | ✅ | | Format: `did` | +| `collection` | `string` | ✅ | | Format: `nsid` | + +**Output:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| --------- | ----------------------------- | ----- | ----------- | ----------- | +| `records` | Array of [`#record`](#record) | ✅ | | | + +--- + + + +### `record` + +**Type:** `object` + +**Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ------- | --------- | ----- | ----------- | ---------------- | +| `uri` | `string` | ✅ | | Format: `at-uri` | +| `cid` | `string` | ✅ | | Format: `cid` | +| `value` | `unknown` | ✅ | | | + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.local.listRecords", + "defs": { + "main": { + "type": "query", + "description": "List the caller's records of a collection kept on this node, in record-key order. Shaped like com.atproto.repo.listRecords.", + "parameters": { + "type": "params", + "required": ["repo", "collection"], + "properties": { + "repo": { + "type": "string", + "format": "did" + }, + "collection": { + "type": "string", + "format": "nsid" + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["records"], + "properties": { + "records": { + "type": "array", + "items": { + "type": "ref", + "ref": "#record" + } + } + } + } + } + }, + "record": { + "type": "object", + "required": ["uri", "cid", "value"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri" + }, + "cid": { + "type": "string", + "format": "cid" + }, + "value": { + "type": "unknown" + } + } + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/local/place-stream-local-putrecord.md b/js/docs/src/content/docs/lex-reference/local/place-stream-local-putrecord.md new file mode 100644 index 000000000..6e401576f --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/local/place-stream-local-putrecord.md @@ -0,0 +1,115 @@ +--- +title: place.stream.local.putRecord +description: Reference for the place.stream.local.putRecord lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `procedure` + +Write a record that stays on this node instead of the caller's repo, such as its configuration for this node. Shaped like com.atproto.repo.putRecord. + +**Parameters:** _(None defined)_ + +**Input:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| ------------ | --------- | ----- | ----------------------------------------------------------------------- | ---------------------------------------- | +| `repo` | `string` | ✅ | The caller's DID; a node keeps local records for their own author only. | Format: `did` | +| `collection` | `string` | ✅ | One of the collections this node keeps locally. | Format: `nsid` | +| `rkey` | `string` | ✅ | | Format: `record-key`
Max Length: 512 | +| `record` | `unknown` | ✅ | | | + +**Output:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| ----- | -------- | ----- | ----------- | ---------------- | +| `uri` | `string` | ✅ | | Format: `at-uri` | +| `cid` | `string` | ✅ | | Format: `cid` | + +**Possible Errors:** + +- `InvalidCollection`: This node does not keep records of the collection locally. + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.local.putRecord", + "defs": { + "main": { + "type": "procedure", + "description": "Write a record that stays on this node instead of the caller's repo, such as its configuration for this node. Shaped like com.atproto.repo.putRecord.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repo", "collection", "rkey", "record"], + "properties": { + "repo": { + "type": "string", + "format": "did", + "description": "The caller's DID; a node keeps local records for their own author only." + }, + "collection": { + "type": "string", + "format": "nsid", + "description": "One of the collections this node keeps locally." + }, + "rkey": { + "type": "string", + "format": "record-key", + "maxLength": 512 + }, + "record": { + "type": "unknown" + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["uri", "cid"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri" + }, + "cid": { + "type": "string", + "format": "cid" + } + } + } + }, + "errors": [ + { + "name": "InvalidCollection", + "description": "This node does not keep records of the collection locally." + } + ] + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index c8c512894..638a780e6 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -3440,6 +3440,183 @@ } } }, + "/xrpc/place.stream.local.deleteRecord": { + "post": { + "summary": "Delete a record the caller keeps on this node. Shaped like com.atproto.repo.deleteRecord; deleting one that doesn't exist succeeds.", + "operationId": "place.stream.local.deleteRecord", + "tags": ["place.stream.local"], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": {} + } + } + } + } + }, + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "repo": { + "type": "string", + "format": "did" + }, + "collection": { + "type": "string", + "format": "nsid" + }, + "rkey": { + "type": "string", + "maxLength": 512, + "format": "record-key" + } + }, + "required": ["repo", "collection", "rkey"] + } + } + } + } + } + }, + "/xrpc/place.stream.local.listRecords": { + "get": { + "summary": "List the caller's records of a collection kept on this node, in record-key order. Shaped like com.atproto.repo.listRecords.", + "operationId": "place.stream.local.listRecords", + "tags": ["place.stream.local"], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "records": { + "type": "array", + "items": { + "$ref": "#/components/schemas/place.stream.local.listRecords_record" + } + } + }, + "required": ["records"] + } + } + } + } + }, + "parameters": [ + { + "name": "repo", + "in": "query", + "required": true, + "schema": { + "type": "string", + "format": "did" + } + }, + { + "name": "collection", + "in": "query", + "required": true, + "schema": { + "type": "string", + "format": "nsid" + } + } + ] + } + }, + "/xrpc/place.stream.local.putRecord": { + "post": { + "summary": "Write a record that stays on this node instead of the caller's repo, such as its configuration for this node. Shaped like com.atproto.repo.putRecord.", + "operationId": "place.stream.local.putRecord", + "tags": ["place.stream.local"], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "uri": { + "type": "string", + "format": "uri" + }, + "cid": { + "type": "string", + "format": "cid" + } + }, + "required": ["uri", "cid"] + } + } + } + }, + "400": { + "description": "Bad Request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["error", "message"], + "properties": { + "error": { + "type": "string", + "oneOf": [ + { + "const": "InvalidCollection" + } + ] + }, + "message": { + "type": "string" + } + } + } + } + } + } + }, + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "repo": { + "type": "string", + "description": "The caller's DID; a node keeps local records for their own author only.", + "format": "did" + }, + "collection": { + "type": "string", + "description": "One of the collections this node keeps locally.", + "format": "nsid" + }, + "rkey": { + "type": "string", + "maxLength": 512, + "format": "record-key" + }, + "record": {} + }, + "required": ["repo", "collection", "rkey", "record"] + } + } + } + } + } + }, "/xrpc/place.stream.live.denyTeleport": { "post": { "summary": "Deny an incoming teleport request.", @@ -7625,6 +7802,21 @@ "place.stream.video": { "description": "Unknown type" }, + "place.stream.local.listRecords_record": { + "type": "object", + "properties": { + "uri": { + "type": "string", + "format": "uri" + }, + "cid": { + "type": "string", + "format": "cid" + }, + "value": {} + }, + "required": ["uri", "cid", "value"] + }, "place.stream.livestream_livestreamView": { "type": "object", "properties": { diff --git a/lexicons/place/stream/ingest/audioRendition.json b/lexicons/place/stream/ingest/audioRendition.json new file mode 100644 index 000000000..84e21faa6 --- /dev/null +++ b/lexicons/place/stream/ingest/audioRendition.json @@ -0,0 +1,27 @@ +{ + "lexicon": 1, + "id": "place.stream.ingest.audioRendition", + "defs": { + "main": { + "type": "record", + "key": "any", + "description": "An audio codec the streamer's encoder sends each of its audio tracks in (the full mix and any stems). A node keeps these locally (place.stream.local.putRecord).", + "record": { + "type": "object", + "required": ["codec", "bitrate", "createdAt"], + "properties": { + "codec": { + "type": "string", + "knownValues": ["aac", "opus"] + }, + "bitrate": { + "type": "integer", + "minimum": 1, + "description": "Target bitrate, in kbps." + }, + "createdAt": { "type": "string", "format": "datetime" } + } + } + } + } +} diff --git a/lexicons/place/stream/ingest/defs.json b/lexicons/place/stream/ingest/defs.json index 9107716cb..b97e9464d 100644 --- a/lexicons/place/stream/ingest/defs.json +++ b/lexicons/place/stream/ingest/defs.json @@ -2,6 +2,15 @@ "lexicon": 1, "id": "place.stream.ingest.defs", "defs": { + "framerate": { + "type": "object", + "description": "A frame rate as a fraction, like 60/1 or 30000/1001.", + "required": ["numerator", "denominator"], + "properties": { + "numerator": { "type": "integer", "minimum": 1 }, + "denominator": { "type": "integer", "minimum": 1 } + } + }, "ingest": { "type": "object", "description": "An ingest URL for a Streamplace station.", diff --git a/lexicons/place/stream/ingest/videoRendition.json b/lexicons/place/stream/ingest/videoRendition.json new file mode 100644 index 000000000..3064d3254 --- /dev/null +++ b/lexicons/place/stream/ingest/videoRendition.json @@ -0,0 +1,40 @@ +{ + "lexicon": 1, + "id": "place.stream.ingest.videoRendition", + "defs": { + "main": { + "type": "record", + "key": "any", + "description": "A video track the streamer's encoder sends to the node they stream to, which tells the encoder about it when it connects (the Streamplace OBS plugin asks). A node keeps these locally (place.stream.local.putRecord).", + "record": { + "type": "object", + "required": [ + "codec", + "width", + "height", + "framerate", + "bitrate", + "createdAt" + ], + "properties": { + "codec": { + "type": "string", + "knownValues": ["h264"] + }, + "width": { "type": "integer", "minimum": 1 }, + "height": { "type": "integer", "minimum": 1 }, + "framerate": { + "type": "ref", + "ref": "place.stream.ingest.defs#framerate" + }, + "bitrate": { + "type": "integer", + "minimum": 1, + "description": "Target bitrate, in kbps." + }, + "createdAt": { "type": "string", "format": "datetime" } + } + } + } + } +} diff --git a/lexicons/place/stream/local/deleteRecord.json b/lexicons/place/stream/local/deleteRecord.json new file mode 100644 index 000000000..95176554e --- /dev/null +++ b/lexicons/place/stream/local/deleteRecord.json @@ -0,0 +1,30 @@ +{ + "lexicon": 1, + "id": "place.stream.local.deleteRecord", + "defs": { + "main": { + "type": "procedure", + "description": "Delete a record the caller keeps on this node. Shaped like com.atproto.repo.deleteRecord; deleting one that doesn't exist succeeds.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repo", "collection", "rkey"], + "properties": { + "repo": { "type": "string", "format": "did" }, + "collection": { "type": "string", "format": "nsid" }, + "rkey": { + "type": "string", + "format": "record-key", + "maxLength": 512 + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { "type": "object", "properties": {} } + } + } + } +} diff --git a/lexicons/place/stream/local/listRecords.json b/lexicons/place/stream/local/listRecords.json new file mode 100644 index 000000000..b6317baa9 --- /dev/null +++ b/lexicons/place/stream/local/listRecords.json @@ -0,0 +1,40 @@ +{ + "lexicon": 1, + "id": "place.stream.local.listRecords", + "defs": { + "main": { + "type": "query", + "description": "List the caller's records of a collection kept on this node, in record-key order. Shaped like com.atproto.repo.listRecords.", + "parameters": { + "type": "params", + "required": ["repo", "collection"], + "properties": { + "repo": { "type": "string", "format": "did" }, + "collection": { "type": "string", "format": "nsid" } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["records"], + "properties": { + "records": { + "type": "array", + "items": { "type": "ref", "ref": "#record" } + } + } + } + } + }, + "record": { + "type": "object", + "required": ["uri", "cid", "value"], + "properties": { + "uri": { "type": "string", "format": "at-uri" }, + "cid": { "type": "string", "format": "cid" }, + "value": { "type": "unknown" } + } + } + } +} diff --git a/lexicons/place/stream/local/putRecord.json b/lexicons/place/stream/local/putRecord.json new file mode 100644 index 000000000..b45280150 --- /dev/null +++ b/lexicons/place/stream/local/putRecord.json @@ -0,0 +1,52 @@ +{ + "lexicon": 1, + "id": "place.stream.local.putRecord", + "defs": { + "main": { + "type": "procedure", + "description": "Write a record that stays on this node instead of the caller's repo, such as its configuration for this node. Shaped like com.atproto.repo.putRecord.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repo", "collection", "rkey", "record"], + "properties": { + "repo": { + "type": "string", + "format": "did", + "description": "The caller's DID; a node keeps local records for their own author only." + }, + "collection": { + "type": "string", + "format": "nsid", + "description": "One of the collections this node keeps locally." + }, + "rkey": { + "type": "string", + "format": "record-key", + "maxLength": 512 + }, + "record": { "type": "unknown" } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["uri", "cid"], + "properties": { + "uri": { "type": "string", "format": "at-uri" }, + "cid": { "type": "string", "format": "cid" } + } + } + }, + "errors": [ + { + "name": "InvalidCollection", + "description": "This node does not keep records of the collection locally." + } + ] + } + } +} diff --git a/obs-plugin/README.md b/obs-plugin/README.md index c1fd7aff6..f2c88cb19 100644 --- a/obs-plugin/README.md +++ b/obs-plugin/README.md @@ -60,8 +60,12 @@ When you start streaming, the plugin does three things in order. - `ingest_endpoints`: the plugin uses the first entry whose `protocol` is `FMP4`, replacing `{stream_key}` in its `url_template`. - `encoder_configurations` replaces the encoders from OBS's output settings - with one video encoder per entry (H.264 encoder types only). Each entry - becomes one video track. + with one video encoder per entry, each becoming one video track. As a + Streamplace extension an entry names a `codec` (`h264`) instead of an OBS + encoder `type`: the plugin uses the encoder from OBS's output settings + when it is that codec, with its settings under the entry's `settings`, + and otherwise the first available encoder of the codec. `width`, + `height` and `framerate` scale it from the canvas. - `audio_configurations.live` creates one audio encoder per entry. `codec` is `aac` or `opus`, and `track_id` is the zero-based OBS audio mixer track to encode. Each such track is written with `tkhd.alternate_group` set to diff --git a/obs-plugin/obs.def b/obs-plugin/obs.def index 07143a06d..0676b1575 100644 --- a/obs-plugin/obs.def +++ b/obs-plugin/obs.def @@ -27,6 +27,7 @@ EXPORTS obs_data_array_item obs_data_array_push_back obs_data_array_release + obs_data_apply obs_data_create obs_data_create_from_json obs_data_create_from_json_file @@ -44,6 +45,7 @@ EXPORTS obs_data_set_string obs_encoder_audio obs_encoder_get_codec + obs_encoder_get_id obs_encoder_get_extra_data obs_encoder_get_frame_rate_divisor obs_encoder_get_height diff --git a/obs-plugin/src/streamplace-output.c b/obs-plugin/src/streamplace-output.c index 90f8d0989..ef3964245 100644 --- a/obs-plugin/src/streamplace-output.c +++ b/obs-plugin/src/streamplace-output.c @@ -212,16 +212,6 @@ static bool find_ingest_url(obs_data_t *config, const char *key, struct dstr *ur return !dstr_is_empty(url); } -static bool encoder_type_available(const char *type) -{ - const char *id; - for (size_t i = 0; obs_enum_encoder_types(i, &id); i++) { - if (strcmp(id, type) == 0) - return true; - } - return false; -} - static const char *audio_encoder_for_codec(const char *codec) { const char *id; @@ -260,22 +250,52 @@ static int lookup_enum(const struct enum_name *names, const char *name, int fall return fallback; } -static obs_encoder_t *create_video_encoder(obs_data_t *config, size_t idx, const struct obs_video_info *ovi, - struct dstr *err) +/* The server names a codec rather than an OBS encoder type (a Streamplace + * extension to GoLiveApi): which encoders exist is this machine's business. + * The streamer's own streaming encoder wins when it is that codec, keeping + * their choice of hardware and settings; otherwise the first usable encoder of + * the codec does. *settings gets a copy of the settings to start from. */ +static const char *video_encoder_for_codec(obs_encoder_t *own, const char *codec, obs_data_t **settings) +{ + *settings = obs_data_create(); + if (own && strcmp(obs_encoder_get_codec(own), codec) == 0) { + obs_data_t *own_settings = obs_encoder_get_settings(own); + obs_data_apply(*settings, own_settings); + obs_data_release(own_settings); + return obs_encoder_get_id(own); + } + const char *id; + for (size_t i = 0; obs_enum_encoder_types(i, &id); i++) { + if (obs_get_encoder_type(id) == OBS_ENCODER_VIDEO && strcmp(obs_get_encoder_codec(id), codec) == 0 && + !(obs_get_encoder_caps(id) & (OBS_ENCODER_CAP_DEPRECATED | OBS_ENCODER_CAP_INTERNAL))) + return id; + } + return NULL; +} + +static obs_encoder_t *create_video_encoder(struct sp_output *sp, obs_data_t *config, size_t idx, + const struct obs_video_info *ovi, struct dstr *err) { - const char *type = obs_data_get_string(config, "type"); - if (!encoder_type_available(type)) { - dstr_printf(err, "Video encoder '%s' is not available in this OBS", type); + const char *codec = obs_data_get_string(config, "codec"); + if (strcmp(codec, "h264") != 0) { + dstr_printf(err, "The server asked for '%s' video; this plugin only sends H.264", codec); return NULL; } - if (strcmp(obs_get_encoder_codec(type), "h264") != 0) { - dstr_printf(err, "Video encoder '%s' is not H.264, the only video codec Streamplace accepts", type); + obs_data_t *settings; + const char *type = video_encoder_for_codec(sp->obs_video, codec, &settings); + if (!type) { + obs_data_release(settings); + dstr_printf(err, "No %s video encoder is available in this OBS", codec); return NULL; } - obs_data_t *settings = obs_data_get_obj(config, "settings"); - if (!settings) - settings = obs_data_create(); + /* The server's settings (bitrate, keyframe interval, ...) go over the + * encoder's own. */ + obs_data_t *server_settings = obs_data_get_obj(config, "settings"); + if (server_settings) { + obs_data_apply(settings, server_settings); + obs_data_release(server_settings); + } /* Keeps keyframes aligned across renditions. */ obs_data_set_bool(settings, "disable_scenecut", true); @@ -331,7 +351,7 @@ static bool create_video_encoders(struct sp_output *sp, obs_data_array_t *config bool ok = true; for (size_t i = 0; ok && i < count; i++) { obs_data_t *config = obs_data_array_item(configs, i); - encoders[i] = create_video_encoder(config, i, &ovi, err); + encoders[i] = create_video_encoder(sp, config, i, &ovi, err); ok = encoders[i] && obs_encoder_set_group(encoders[i], group); obs_data_release(config); } diff --git a/pkg/api/fmp4_ingest.go b/pkg/api/fmp4_ingest.go index ed061c4c2..9c71ea87e 100644 --- a/pkg/api/fmp4_ingest.go +++ b/pkg/api/fmp4_ingest.go @@ -10,9 +10,11 @@ import ( "github.com/google/uuid" "github.com/julienschmidt/httprouter" + glex "github.com/streamplace/glex/runtime" "stream.place/streamplace/pkg/errors" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/media" + "stream.place/streamplace/pkg/placestream" ) // The Streamplace OBS plugin asks for its stream configuration with the same @@ -38,18 +40,41 @@ type clientConfiguration struct { Meta clientConfigurationMeta `json:"meta"` Status clientConfigurationStatus `json:"status"` IngestEndpoints []ingestEndpoint `json:"ingest_endpoints,omitempty"` + // EncoderConfigurations lists the video encoders the plugin creates, one + // track each; absent, the plugin keeps the encoder from OBS's own output + // settings. + EncoderConfigurations []videoConfiguration `json:"encoder_configurations,omitempty"` // AudioConfigurations lists the audio encoders the plugin creates, keyed // by the output they feed ("live"); absent, the plugin keeps the encoder // from OBS's own output settings. AudioConfigurations map[string][]audioConfiguration `json:"audio_configurations,omitempty"` } +// videoConfiguration is a GoLiveApi encoder configuration, except that it +// names a codec rather than an OBS encoder type, a Streamplace extension: +// which encoders exist is the streamer's machine's business, so the plugin +// picks one (the streamer's own streaming encoder when it is that codec). +type videoConfiguration struct { + Codec string `json:"codec"` + Width int64 `json:"width"` + Height int64 `json:"height"` + Framerate placestream.IngestDefs_Framerate `json:"framerate"` + // Settings are the encoder settings the plugin applies over the + // encoder's own; OBS's H.264 encoders share these names. + Settings struct { + Bitrate int64 `json:"bitrate"` + RateControl string `json:"rate_control"` + KeyintSec int `json:"keyint_sec"` + BFrames int `json:"bf"` + } `json:"settings"` +} + type audioConfiguration struct { Codec string `json:"codec"` TrackID int `json:"track_id"` Channels int `json:"channels"` Settings struct { - Bitrate int `json:"bitrate"` + Bitrate int64 `json:"bitrate"` } `json:"settings"` } @@ -73,42 +98,93 @@ type ingestEndpoint struct { // OBS allows six audio encoders per output, and each track takes two. const maxStems = 2 -// stemAudioConfigurations lists the encoders the plugin should create when -// the streamer named mixer tracks beyond the first: an AAC and an Opus -// encode of track 0 (the full mix) and of each named track in track order, -// up to maxStems of them. With no named track it returns nil, and the plugin -// keeps the encoder from OBS's output settings: one audio track, as today. -func stemAudioConfigurations(tracks []audioTrackPreference) []audioConfiguration { +// defaultAudioRenditions are the codecs each audio track is encoded in when +// the streamer hasn't chosen (place.stream.ingest.audioRendition). +var defaultAudioRenditions = []*placestream.IngestAudioRendition{ + {Codec: "aac", Bitrate: 160}, + {Codec: "opus", Bitrate: 128}, +} + +// audioConfigurations lists the audio encoders the plugin should create: one +// per rendition (or the defaults) for track 0 (the full mix) and for each +// mixer track the streamer named beyond the first, in track order, up to +// maxStems of them. With no renditions and no named tracks it returns nil, +// and the plugin keeps the encoder from OBS's output settings. +func audioConfigurations(tracks []audioTrackPreference, renditions []*placestream.IngestAudioRendition) []audioConfiguration { var stems []int for _, t := range tracks { if t.TrackID >= 1 && t.TrackID <= 5 && t.Name != "" { stems = append(stems, t.TrackID) } } - if len(stems) == 0 { + if len(stems) == 0 && len(renditions) == 0 { return nil } + if len(renditions) == 0 { + renditions = defaultAudioRenditions + } slices.Sort(stems) ids := append([]int{0}, stems[:min(len(stems), maxStems)]...) var out []audioConfiguration for _, id := range ids { - for _, c := range []struct { - codec string - bitrate int - }{{"aac", 160}, {"opus", 128}} { - cfg := audioConfiguration{Codec: c.codec, TrackID: id, Channels: 2} - cfg.Settings.Bitrate = c.bitrate + for _, r := range renditions { + cfg := audioConfiguration{Codec: r.Codec, TrackID: id, Channels: 2} + cfg.Settings.Bitrate = r.Bitrate out = append(out, cfg) } } return out } +// videoConfigurations lists the video encoders the plugin should create from +// the streamer's renditions: none, and the plugin keeps OBS's own. A stream +// has one video track for now, so only the first rendition is encoded. +func videoConfigurations(renditions []*placestream.IngestVideoRendition) []videoConfiguration { + var out []videoConfiguration + for _, r := range renditions[:min(len(renditions), 1)] { + cfg := videoConfiguration{Codec: r.Codec, Width: r.Width, Height: r.Height, Framerate: r.Framerate} + cfg.Settings.Bitrate = r.Bitrate + cfg.Settings.RateControl = "CBR" + // Segments are a GoP each, and MUXL takes no B-frames. + cfg.Settings.KeyintSec = 1 + cfg.Settings.BFrames = 0 + out = append(out, cfg) + } + return out +} + +// localRecords returns repoDID's records of collection kept on this node, +// as T. +func localRecords[T any, PT interface { + *T + glex.Record +}](a *StreamplaceAPI, repoDID, collection string) ([]*T, error) { + rows, err := a.LocalDB.ListLocalRecords(repoDID, collection) + if err != nil { + return nil, err + } + var out []*T + for _, row := range rows { + value := &glex.LexiconTypeDecoder{} + if err := value.UnmarshalCBOR(row.Record); err != nil { + return nil, fmt.Errorf("decoding %s: %w", row.URI(), err) + } + rec, err := glex.RecordAs[T, PT](value.Val) + if err != nil { + return nil, fmt.Errorf("decoding %s: %w", row.URI(), err) + } + out = append(out, rec) + } + return out, nil +} + // HandleClientConfiguration answers the plugin's configuration request. A bad // stream key is reported in status, as GoLiveApi clients expect, so OBS shows -// the message. Unless the streamer named mixer tracks (see -// stemAudioConfigurations) the response names no encoders, so the plugin keeps -// the encoders from OBS's output settings: one video and one audio track. +// the message. The encoders come from the streamer's renditions kept on this +// node (place.stream.ingest.*Rendition) and their named mixer tracks (see +// audioConfigurations); with none of either the response names no +// encoders, so the plugin keeps the encoders from OBS's output settings: one +// video and one audio track. func (a *StreamplaceAPI) HandleClientConfiguration(ctx context.Context) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { var req clientConfigurationRequest @@ -124,7 +200,8 @@ func (a *StreamplaceAPI) HandleClientConfiguration(ctx context.Context) http.Han }, Status: clientConfigurationStatus{Result: "success"}, } - if problem := a.streamKeyProblem(r.Context(), req.Authentication); problem != "" { + streamer, problem := a.streamKeyStreamer(r.Context(), req.Authentication) + if problem != "" { config.Status = clientConfigurationStatus{Result: "error", HTMLEnUS: problem} } else { scheme := "http" @@ -135,7 +212,18 @@ func (a *StreamplaceAPI) HandleClientConfiguration(ctx context.Context) http.Han Protocol: "FMP4", URLTemplate: fmt.Sprintf("%s://%s/api/ingest/fmp4/{stream_key}", scheme, r.Host), }} - if live := stemAudioConfigurations(req.Preferences.AudioTracks); live != nil { + video, err := localRecords[placestream.IngestVideoRendition](a, streamer, "place.stream.ingest.videoRendition") + if err != nil { + errors.WriteHTTPInternalServerError(w, "could not read video renditions", err) + return + } + audio, err := localRecords[placestream.IngestAudioRendition](a, streamer, "place.stream.ingest.audioRendition") + if err != nil { + errors.WriteHTTPInternalServerError(w, "could not read audio renditions", err) + return + } + config.EncoderConfigurations = videoConfigurations(video) + if live := audioConfigurations(req.Preferences.AudioTracks, audio); live != nil { config.AudioConfigurations = map[string][]audioConfiguration{"live": live} } } @@ -146,17 +234,17 @@ func (a *StreamplaceAPI) HandleClientConfiguration(ctx context.Context) http.Han } } -// streamKeyProblem explains to the streamer why key may not stream, or returns -// "" if it may. -func (a *StreamplaceAPI) streamKeyProblem(ctx context.Context, key string) string { +// streamKeyStreamer returns whose key it is, or explains to the streamer why +// key may not stream. +func (a *StreamplaceAPI) streamKeyStreamer(ctx context.Context, key string) (streamer, problem string) { ms, err := a.MakeMediaSigner(ctx, key) if err != nil { - return "This stream key isn't valid. Copy a fresh one from your Streamplace dashboard." + return "", "This stream key isn't valid. Copy a fresh one from your Streamplace dashboard." } if err := a.checkBanned(ctx, ms.Streamer()); err != nil { - return "This account can't stream right now." + return "", "This account can't stream right now." } - return "" + return ms.Streamer(), "" } // HandleFMP4Ingest accepts the OBS plugin's push: a chunked POST of fragmented diff --git a/pkg/api/fmp4_ingest_test.go b/pkg/api/fmp4_ingest_test.go index 6b3818139..22d174ae1 100644 --- a/pkg/api/fmp4_ingest_test.go +++ b/pkg/api/fmp4_ingest_test.go @@ -5,14 +5,16 @@ import ( "testing" "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/placestream" ) // The plugin's configuration request names the OBS mixer tracks; named tracks -// beyond the first become stems, each encoded as AAC and Opus beside the full -// mix, in track order and at most two of them. With nothing named the -// response carries no audio configuration at all, so the plugin keeps the -// single encoder from OBS's own settings. -func TestStemAudioConfigurations(t *testing.T) { +// beyond the first become stems, each encoded like the full mix (AAC and Opus +// unless the streamer chose codecs), in track order and at most two of them. +// With nothing named and nothing chosen the response carries no audio +// configuration at all, so the plugin keeps the single encoder from OBS's own +// settings. +func TestAudioConfigurations(t *testing.T) { decode := func(body string) clientConfigurationRequest { var req clientConfigurationRequest require.NoError(t, json.Unmarshal([]byte(body), &req)) @@ -21,18 +23,18 @@ func TestStemAudioConfigurations(t *testing.T) { unnamed := decode(`{"authentication":"k","preferences":{"audio_tracks":[ {"track_id":0,"name":""},{"track_id":1,"name":""},{"track_id":2,"name":""}]}}`) - require.Nil(t, stemAudioConfigurations(unnamed.Preferences.AudioTracks), "unnamed tracks keep OBS's own encoder") - require.Nil(t, stemAudioConfigurations(nil), "a request without preferences keeps OBS's own encoder") + require.Nil(t, audioConfigurations(unnamed.Preferences.AudioTracks, nil), "unnamed tracks keep OBS's own encoder") + require.Nil(t, audioConfigurations(nil, nil), "a request without preferences keeps OBS's own encoder") onlyFirstNamed := decode(`{"preferences":{"audio_tracks":[{"track_id":0,"name":"Mix"}]}}`) - require.Nil(t, stemAudioConfigurations(onlyFirstNamed.Preferences.AudioTracks), "naming only the full mix makes no stems") + require.Nil(t, audioConfigurations(onlyFirstNamed.Preferences.AudioTracks, nil), "naming only the full mix makes no stems") // Three named stems arrive out of order: the two lowest mixer tracks win, // and the full mix (track 0) always comes first. named := decode(`{"preferences":{"audio_tracks":[ {"track_id":0,"name":""},{"track_id":4,"name":"Game"},{"track_id":1,"name":"Music"}, {"track_id":2,"name":""},{"track_id":3,"name":"Mic"},{"track_id":5,"name":"Alerts"}]}}`) - got, err := json.Marshal(stemAudioConfigurations(named.Preferences.AudioTracks)) + got, err := json.Marshal(audioConfigurations(named.Preferences.AudioTracks, nil)) require.NoError(t, err) require.JSONEq(t, `[ {"codec":"aac","track_id":0,"channels":2,"settings":{"bitrate":160}}, @@ -41,4 +43,15 @@ func TestStemAudioConfigurations(t *testing.T) { {"codec":"opus","track_id":1,"channels":2,"settings":{"bitrate":128}}, {"codec":"aac","track_id":3,"channels":2,"settings":{"bitrate":160}}, {"codec":"opus","track_id":3,"channels":2,"settings":{"bitrate":128}}]`, string(got)) + + // Chosen codecs replace the defaults, for the full mix alone too. + opus := []*placestream.IngestAudioRendition{{Codec: "opus", Bitrate: 96}} + got, err = json.Marshal(audioConfigurations(unnamed.Preferences.AudioTracks, opus)) + require.NoError(t, err) + require.JSONEq(t, `[{"codec":"opus","track_id":0,"channels":2,"settings":{"bitrate":96}}]`, string(got)) + got, err = json.Marshal(audioConfigurations(named.Preferences.AudioTracks[:2], opus)) + require.NoError(t, err) + require.JSONEq(t, `[ + {"codec":"opus","track_id":0,"channels":2,"settings":{"bitrate":96}}, + {"codec":"opus","track_id":4,"channels":2,"settings":{"bitrate":96}}]`, string(got)) } diff --git a/pkg/localdb/local_record.go b/pkg/localdb/local_record.go new file mode 100644 index 000000000..1e6365dd7 --- /dev/null +++ b/pkg/localdb/local_record.go @@ -0,0 +1,39 @@ +package localdb + +import "fmt" + +// LocalRecord is an atproto record that stays on this node instead of its +// author's repo, such as their configuration for this node (see +// place.stream.local.putRecord). Record is its CBOR encoding. +type LocalRecord struct { + RepoDID string `gorm:"column:repo_did;primaryKey"` + Collection string `gorm:"column:collection;primaryKey"` + Rkey string `gorm:"column:rkey;primaryKey"` + CID string `gorm:"column:cid;not null"` + Record []byte `gorm:"column:record;not null"` +} + +// URI is the record's at:// URI, as it would be in its author's repo. +func (r *LocalRecord) URI() string { + return fmt.Sprintf("at://%s/%s/%s", r.RepoDID, r.Collection, r.Rkey) +} + +// PutLocalRecord creates the record, or replaces the one at its key. +func (m *LocalDatabase) PutLocalRecord(record *LocalRecord) error { + return m.DB.Save(record).Error +} + +// ListLocalRecords returns repoDID's records of collection, in rkey order. +func (m *LocalDatabase) ListLocalRecords(repoDID, collection string) ([]LocalRecord, error) { + var records []LocalRecord + err := m.DB.Where("repo_did = ? AND collection = ?", repoDID, collection).Order("rkey").Find(&records).Error + if err != nil { + return nil, fmt.Errorf("list local records: %w", err) + } + return records, nil +} + +// DeleteLocalRecord deletes the record at the key; deleting none is no error. +func (m *LocalDatabase) DeleteLocalRecord(repoDID, collection, rkey string) error { + return m.DB.Where("repo_did = ? AND collection = ? AND rkey = ?", repoDID, collection, rkey).Delete(&LocalRecord{}).Error +} diff --git a/pkg/localdb/localdb.go b/pkg/localdb/localdb.go index 1e180b3ca..daed019d8 100644 --- a/pkg/localdb/localdb.go +++ b/pkg/localdb/localdb.go @@ -27,6 +27,9 @@ type LocalDB interface { GetViewLogSalt(date string) ([]byte, error) PutViewLogSalt(date string, salt []byte) error DeleteViewLogSaltsBefore(date string) error + PutLocalRecord(record *LocalRecord) error + ListLocalRecords(repoDID, collection string) ([]LocalRecord, error) + DeleteLocalRecord(repoDID, collection, rkey string) error } type LocalDatabase struct { @@ -72,6 +75,7 @@ func MakeDB(dbURL string) (LocalDB, error) { for _, model := range []any{ Segment{}, ViewLogSalt{}, + LocalRecord{}, } { err = db.AutoMigrate(model) if err != nil { diff --git a/pkg/placestream/ingestaudiorendition.go b/pkg/placestream/ingestaudiorendition.go new file mode 100644 index 000000000..198765eec --- /dev/null +++ b/pkg/placestream/ingestaudiorendition.go @@ -0,0 +1,53 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.ingest.audioRendition + +package placestream + +import ( + "encoding/json" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +func init() { + glex.RegisterType("place.stream.ingest.audioRendition", &IngestAudioRendition{}) +} + +// An audio codec the streamer's encoder sends each of its audio tracks in (the full mix and any stems). A node keeps these locally (place.stream.local.putRecord). +type IngestAudioRendition struct { + LexiconTypeID string `json:"$type,omitempty"` + // bitrate: Target bitrate, in kbps. + Bitrate int64 `json:"bitrate"` + Codec string `json:"codec"` + CreatedAt string `json:"createdAt"` +} + +// RecordTypeID implements glex.Record. +func (t *IngestAudioRendition) RecordTypeID() string { return "place.stream.ingest.audioRendition" } + +// MarshalJSON stamps the $type field, like MarshalCBOR does. The value +// receiver operates on a copy, so the record is never mutated and both +// IngestAudioRendition and *IngestAudioRendition marshal with $type. +func (t IngestAudioRendition) MarshalJSON() ([]byte, error) { + t.LexiconTypeID = "place.stream.ingest.audioRendition" + type alias IngestAudioRendition + return json.Marshal((alias)(t)) +} + +func (t *IngestAudioRendition) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.ingest.audioRendition" + return glex.MarshalCBOR(w, &cp) +} + +func (t *IngestAudioRendition) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} diff --git a/pkg/placestream/ingestdefs.go b/pkg/placestream/ingestdefs.go index 8e6059833..24e55a616 100644 --- a/pkg/placestream/ingestdefs.go +++ b/pkg/placestream/ingestdefs.go @@ -11,6 +11,33 @@ import ( cbg "github.com/whyrusleeping/cbor-gen" ) +// IngestDefs_Framerate is a "framerate" in the place.stream.ingest.defs schema. +// +// A frame rate as a fraction, like 60/1 or 30000/1001. +type IngestDefs_Framerate struct { + LexiconTypeID string `json:"$type,omitempty"` + Denominator int64 `json:"denominator"` + Numerator int64 `json:"numerator"` +} + +// RecordTypeID implements glex.Record. +func (t *IngestDefs_Framerate) RecordTypeID() string { return "place.stream.ingest.defs#framerate" } + +func (t *IngestDefs_Framerate) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.ingest.defs#framerate" + return glex.MarshalCBOR(w, &cp) +} + +func (t *IngestDefs_Framerate) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + // IngestDefs_Ingest is a "ingest" in the place.stream.ingest.defs schema. // // An ingest URL for a Streamplace station. diff --git a/pkg/placestream/ingestvideorendition.go b/pkg/placestream/ingestvideorendition.go new file mode 100644 index 000000000..421e7ec69 --- /dev/null +++ b/pkg/placestream/ingestvideorendition.go @@ -0,0 +1,56 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.ingest.videoRendition + +package placestream + +import ( + "encoding/json" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +func init() { + glex.RegisterType("place.stream.ingest.videoRendition", &IngestVideoRendition{}) +} + +// A video track the streamer's encoder sends to the node they stream to, which tells the encoder about it when it connects (the Streamplace OBS plugin asks). A node keeps these locally (place.stream.local.putRecord). +type IngestVideoRendition struct { + LexiconTypeID string `json:"$type,omitempty"` + // bitrate: Target bitrate, in kbps. + Bitrate int64 `json:"bitrate"` + Codec string `json:"codec"` + CreatedAt string `json:"createdAt"` + Framerate IngestDefs_Framerate `json:"framerate"` + Height int64 `json:"height"` + Width int64 `json:"width"` +} + +// RecordTypeID implements glex.Record. +func (t *IngestVideoRendition) RecordTypeID() string { return "place.stream.ingest.videoRendition" } + +// MarshalJSON stamps the $type field, like MarshalCBOR does. The value +// receiver operates on a copy, so the record is never mutated and both +// IngestVideoRendition and *IngestVideoRendition marshal with $type. +func (t IngestVideoRendition) MarshalJSON() ([]byte, error) { + t.LexiconTypeID = "place.stream.ingest.videoRendition" + type alias IngestVideoRendition + return json.Marshal((alias)(t)) +} + +func (t *IngestVideoRendition) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.ingest.videoRendition" + return glex.MarshalCBOR(w, &cp) +} + +func (t *IngestVideoRendition) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} diff --git a/pkg/placestream/localdeleterecord.go b/pkg/placestream/localdeleterecord.go new file mode 100644 index 000000000..fb817fd93 --- /dev/null +++ b/pkg/placestream/localdeleterecord.go @@ -0,0 +1,72 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.local.deleteRecord + +package placestream + +import ( + "context" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +type LocalDeleteRecord_Input struct { + LexiconTypeID string `json:"$type,omitempty"` + Collection string `json:"collection"` + Repo string `json:"repo"` + Rkey string `json:"rkey"` +} + +// RecordTypeID implements glex.Record. +func (t *LocalDeleteRecord_Input) RecordTypeID() string { return "place.stream.local.deleteRecord" } + +func (t *LocalDeleteRecord_Input) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.local.deleteRecord" + return glex.MarshalCBOR(w, &cp) +} + +func (t *LocalDeleteRecord_Input) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +type LocalDeleteRecord_Output struct { + LexiconTypeID string `json:"$type,omitempty"` +} + +// RecordTypeID implements glex.Record. +func (t *LocalDeleteRecord_Output) RecordTypeID() string { return "place.stream.local.deleteRecord" } + +func (t *LocalDeleteRecord_Output) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.local.deleteRecord" + return glex.MarshalCBOR(w, &cp) +} + +func (t *LocalDeleteRecord_Output) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +// LocalDeleteRecord calls the XRPC method "place.stream.local.deleteRecord". +// +// Delete a record the caller keeps on this node. Shaped like com.atproto.repo.deleteRecord; deleting one that doesn't exist succeeds. +func LocalDeleteRecord(ctx context.Context, c glex.LexClient, input *LocalDeleteRecord_Input) (*LocalDeleteRecord_Output, error) { + var out LocalDeleteRecord_Output + + if err := c.LexDo(ctx, glex.Procedure, "application/json", "place.stream.local.deleteRecord", nil, input, &out); err != nil { + return nil, err + } + return &out, nil +} diff --git a/pkg/placestream/locallistrecords.go b/pkg/placestream/locallistrecords.go new file mode 100644 index 000000000..616a668ad --- /dev/null +++ b/pkg/placestream/locallistrecords.go @@ -0,0 +1,80 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.local.listRecords + +package placestream + +import ( + "context" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +type LocalListRecords_Output struct { + LexiconTypeID string `json:"$type,omitempty"` + Records []LocalListRecords_Record `json:"records"` +} + +// RecordTypeID implements glex.Record. +func (t *LocalListRecords_Output) RecordTypeID() string { return "place.stream.local.listRecords" } + +func (t *LocalListRecords_Output) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.local.listRecords" + return glex.MarshalCBOR(w, &cp) +} + +func (t *LocalListRecords_Output) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +// LocalListRecords calls the XRPC method "place.stream.local.listRecords". +// +// List the caller's records of a collection kept on this node, in record-key order. Shaped like com.atproto.repo.listRecords. +func LocalListRecords(ctx context.Context, c glex.LexClient, collection string, repo string) (*LocalListRecords_Output, error) { + var out LocalListRecords_Output + + params := map[string]interface{}{} + params["collection"] = collection + params["repo"] = repo + + if err := c.LexDo(ctx, glex.Query, "", "place.stream.local.listRecords", params, nil, &out); err != nil { + return nil, err + } + return &out, nil +} + +// LocalListRecords_Record is a "record" in the place.stream.local.listRecords schema. +type LocalListRecords_Record struct { + LexiconTypeID string `json:"$type,omitempty"` + Cid string `json:"cid"` + Uri string `json:"uri"` + Value *glex.LexiconTypeDecoder `json:"value"` +} + +// RecordTypeID implements glex.Record. +func (t *LocalListRecords_Record) RecordTypeID() string { + return "place.stream.local.listRecords#record" +} + +func (t *LocalListRecords_Record) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.local.listRecords#record" + return glex.MarshalCBOR(w, &cp) +} + +func (t *LocalListRecords_Record) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} diff --git a/pkg/placestream/localputrecord.go b/pkg/placestream/localputrecord.go new file mode 100644 index 000000000..5786bf04d --- /dev/null +++ b/pkg/placestream/localputrecord.go @@ -0,0 +1,77 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.local.putRecord + +package placestream + +import ( + "context" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +type LocalPutRecord_Input struct { + LexiconTypeID string `json:"$type,omitempty"` + // collection: One of the collections this node keeps locally. + Collection string `json:"collection"` + Record *glex.LexiconTypeDecoder `json:"record"` + // repo: The caller's DID; a node keeps local records for their own author only. + Repo string `json:"repo"` + Rkey string `json:"rkey"` +} + +// RecordTypeID implements glex.Record. +func (t *LocalPutRecord_Input) RecordTypeID() string { return "place.stream.local.putRecord" } + +func (t *LocalPutRecord_Input) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.local.putRecord" + return glex.MarshalCBOR(w, &cp) +} + +func (t *LocalPutRecord_Input) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +type LocalPutRecord_Output struct { + LexiconTypeID string `json:"$type,omitempty"` + Cid string `json:"cid"` + Uri string `json:"uri"` +} + +// RecordTypeID implements glex.Record. +func (t *LocalPutRecord_Output) RecordTypeID() string { return "place.stream.local.putRecord" } + +func (t *LocalPutRecord_Output) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.local.putRecord" + return glex.MarshalCBOR(w, &cp) +} + +func (t *LocalPutRecord_Output) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +// LocalPutRecord calls the XRPC method "place.stream.local.putRecord". +// +// Write a record that stays on this node instead of the caller's repo, such as its configuration for this node. Shaped like com.atproto.repo.putRecord. +func LocalPutRecord(ctx context.Context, c glex.LexClient, input *LocalPutRecord_Input) (*LocalPutRecord_Output, error) { + var out LocalPutRecord_Output + + if err := c.LexDo(ctx, glex.Procedure, "application/json", "place.stream.local.putRecord", nil, input, &out); err != nil { + return nil, err + } + return &out, nil +} diff --git a/pkg/spxrpc/local_record.go b/pkg/spxrpc/local_record.go new file mode 100644 index 000000000..992b20081 --- /dev/null +++ b/pkg/spxrpc/local_record.go @@ -0,0 +1,118 @@ +package spxrpc + +import ( + "context" + "fmt" + "net/http" + "slices" + + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/bluesky-social/indigo/xrpc" + "github.com/labstack/echo/v4" + glex "github.com/streamplace/glex/runtime" + "stream.place/streamplace/pkg/localdb" + "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/spid" +) + +// localCollections are the collections a node keeps for their authors +// instead of their repos, each with a check that the node can act on a +// record: an encoder told to use a codec it doesn't have fails to stream. +var localCollections = map[string]func(glex.Record) error{ + "place.stream.ingest.videoRendition": func(rec glex.Record) error { + r := rec.(*placestream.IngestVideoRendition) + if r.Codec != "h264" { + return fmt.Errorf("unsupported video codec %q", r.Codec) + } + if r.Width < 1 || r.Height < 1 || r.Bitrate < 1 || r.Framerate.Numerator < 1 || r.Framerate.Denominator < 1 { + return fmt.Errorf("width, height, bitrate and framerate must be positive") + } + return nil + }, + "place.stream.ingest.audioRendition": func(rec glex.Record) error { + r := rec.(*placestream.IngestAudioRendition) + if !slices.Contains([]string{"aac", "opus"}, r.Codec) { + return fmt.Errorf("unsupported audio codec %q", r.Codec) + } + if r.Bitrate < 1 { + return fmt.Errorf("bitrate must be positive") + } + return nil + }, +} + +// localRecordCaller checks that the caller is signed in as repo and that +// collection is one this node keeps locally. +func localRecordCaller(ctx context.Context, repo, collection string) error { + session := GetCaller(ctx) + if session == nil { + return echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") + } + if session.DID != repo { + return echo.NewHTTPError(http.StatusForbidden, "local records can only be kept for your own repo") + } + if _, ok := localCollections[collection]; !ok { + return &xrpc.Error{ + StatusCode: http.StatusBadRequest, + Wrapped: &xrpc.XRPCError{ErrStr: "InvalidCollection", Message: fmt.Sprintf("this node does not keep %s records locally", collection)}, + } + } + return nil +} + +func (s *Server) handlePlaceStreamLocalPutRecord(ctx context.Context, input *placestream.LocalPutRecord_Input) (*placestream.LocalPutRecord_Output, error) { + if err := localRecordCaller(ctx, input.Repo, input.Collection); err != nil { + return nil, err + } + if _, err := syntax.ParseRecordKey(input.Rkey); err != nil { + return nil, echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("invalid rkey: %s", err)) + } + if input.Record == nil || input.Record.Val == nil || input.Record.Val.RecordTypeID() != input.Collection { + return nil, echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("record must be a %s", input.Collection)) + } + if err := localCollections[input.Collection](input.Record.Val); err != nil { + return nil, echo.NewHTTPError(http.StatusBadRequest, err.Error()) + } + record, err := input.Record.MarshalCBOR() + if err != nil { + return nil, fmt.Errorf("encoding record: %w", err) + } + cid, err := spid.GetCIDFromBytes(record) + if err != nil { + return nil, fmt.Errorf("computing record cid: %w", err) + } + row := &localdb.LocalRecord{RepoDID: input.Repo, Collection: input.Collection, Rkey: input.Rkey, CID: cid.String(), Record: record} + if err := s.localDB.PutLocalRecord(row); err != nil { + return nil, err + } + return &placestream.LocalPutRecord_Output{Uri: row.URI(), Cid: row.CID}, nil +} + +func (s *Server) handlePlaceStreamLocalListRecords(ctx context.Context, collection string, repo string) (*placestream.LocalListRecords_Output, error) { + if err := localRecordCaller(ctx, repo, collection); err != nil { + return nil, err + } + rows, err := s.localDB.ListLocalRecords(repo, collection) + if err != nil { + return nil, err + } + out := &placestream.LocalListRecords_Output{Records: []placestream.LocalListRecords_Record{}} + for _, row := range rows { + value := &glex.LexiconTypeDecoder{} + if err := value.UnmarshalCBOR(row.Record); err != nil { + return nil, fmt.Errorf("decoding %s: %w", row.URI(), err) + } + out.Records = append(out.Records, placestream.LocalListRecords_Record{Uri: row.URI(), Cid: row.CID, Value: value}) + } + return out, nil +} + +func (s *Server) handlePlaceStreamLocalDeleteRecord(ctx context.Context, input *placestream.LocalDeleteRecord_Input) (*placestream.LocalDeleteRecord_Output, error) { + if err := localRecordCaller(ctx, input.Repo, input.Collection); err != nil { + return nil, err + } + if err := s.localDB.DeleteLocalRecord(input.Repo, input.Collection, input.Rkey); err != nil { + return nil, err + } + return &placestream.LocalDeleteRecord_Output{}, nil +} diff --git a/pkg/spxrpc/local_record_test.go b/pkg/spxrpc/local_record_test.go new file mode 100644 index 000000000..897eace32 --- /dev/null +++ b/pkg/spxrpc/local_record_test.go @@ -0,0 +1,51 @@ +package spxrpc + +import ( + "context" + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/localdb" + "stream.place/streamplace/pkg/placestream" +) + +// A node keeps its local collections for the signed-in author only, refuses +// records it couldn't act on, and hands back what it kept with its $type. +func TestLocalRecords(t *testing.T) { + ldb, err := localdb.MakeDB(":memory:") + require.NoError(t, err) + s := &Server{localDB: ldb} + const me = "did:plc:me" + ctx := context.WithValue(context.Background(), pdsAuthContextKey, me) + const video = "place.stream.ingest.videoRendition" + put := func(body string) error { + var input placestream.LocalPutRecord_Input + require.NoError(t, json.Unmarshal([]byte(body), &input)) + _, err := s.handlePlaceStreamLocalPutRecord(ctx, &input) + return err + } + record := func(codec string) string { + return `{"$type":"` + video + `","codec":"` + codec + `","width":1920,"height":1080,"framerate":{"numerator":60,"denominator":1},"bitrate":6000,"createdAt":"2026-10-18T00:00:00Z"}` + } + + require.NoError(t, put(`{"repo":"`+me+`","collection":"`+video+`","rkey":"main","record":`+record("h264")+`}`)) + require.ErrorContains(t, put(`{"repo":"did:plc:other","collection":"`+video+`","rkey":"main","record":`+record("h264")+`}`), "your own repo") + require.ErrorContains(t, put(`{"repo":"`+me+`","collection":"place.stream.livestream","rkey":"main","record":`+record("h264")+`}`), "InvalidCollection") + require.ErrorContains(t, put(`{"repo":"`+me+`","collection":"place.stream.ingest.audioRendition","rkey":"main","record":`+record("h264")+`}`), "must be a place.stream.ingest.audioRendition") + require.ErrorContains(t, put(`{"repo":"`+me+`","collection":"`+video+`","rkey":"main","record":`+record("vp8")+`}`), "unsupported video codec") + + list, err := s.handlePlaceStreamLocalListRecords(ctx, video, me) + require.NoError(t, err) + require.Len(t, list.Records, 1) + require.Equal(t, "at://"+me+"/"+video+"/main", list.Records[0].Uri) + got, err := json.Marshal(list.Records[0].Value) + require.NoError(t, err) + require.JSONEq(t, record("h264"), string(got), "the record as written, h264 kept over the refused vp8") + + _, err = s.handlePlaceStreamLocalDeleteRecord(ctx, &placestream.LocalDeleteRecord_Input{Repo: me, Collection: video, Rkey: "main"}) + require.NoError(t, err) + list, err = s.handlePlaceStreamLocalListRecords(ctx, video, me) + require.NoError(t, err) + require.Empty(t, list.Records) +} diff --git a/pkg/spxrpc/stubs.go b/pkg/spxrpc/stubs.go index 318cbc37b..d1c639e6d 100644 --- a/pkg/spxrpc/stubs.go +++ b/pkg/spxrpc/stubs.go @@ -444,6 +444,9 @@ func (s *Server) RegisterHandlersPlacestream(e *echo.Echo) error { e.GET("/xrpc/place.stream.live.searchActorsTypeahead", s.HandlePlaceStreamLiveSearchActorsTypeahead) e.POST("/xrpc/place.stream.live.startLivestream", s.HandlePlaceStreamLiveStartLivestream) e.POST("/xrpc/place.stream.live.stopLivestream", s.HandlePlaceStreamLiveStopLivestream) + e.POST("/xrpc/place.stream.local.deleteRecord", s.HandlePlaceStreamLocalDeleteRecord) + e.GET("/xrpc/place.stream.local.listRecords", s.HandlePlaceStreamLocalListRecords) + e.POST("/xrpc/place.stream.local.putRecord", s.HandlePlaceStreamLocalPutRecord) e.POST("/xrpc/place.stream.media.createUpload", s.HandlePlaceStreamMediaCreateUpload) e.POST("/xrpc/place.stream.media.deleteVideo", s.HandlePlaceStreamMediaDeleteVideo) e.POST("/xrpc/place.stream.media.finalizeLivestream", s.HandlePlaceStreamMediaFinalizeLivestream) @@ -920,6 +923,55 @@ func (s *Server) HandlePlaceStreamLiveStopLivestream(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamLocalDeleteRecord(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamLocalDeleteRecord") + defer span.End() + var body placestream.LocalDeleteRecord_Input + if err := c.Bind(&body); err != nil { + return err + } + var out *placestream.LocalDeleteRecord_Output + var handleErr error + // func (s *Server) handlePlaceStreamLocalDeleteRecord(ctx context.Context,body *placestream.LocalDeleteRecord_Input) (*placestream.LocalDeleteRecord_Output, error) + out, handleErr = s.handlePlaceStreamLocalDeleteRecord(ctx, &body) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + +func (s *Server) HandlePlaceStreamLocalListRecords(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamLocalListRecords") + defer span.End() + collection := c.QueryParam("collection") + repo := c.QueryParam("repo") + var out *placestream.LocalListRecords_Output + var handleErr error + // func (s *Server) handlePlaceStreamLocalListRecords(ctx context.Context,collection string,repo string) (*placestream.LocalListRecords_Output, error) + out, handleErr = s.handlePlaceStreamLocalListRecords(ctx, collection, repo) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + +func (s *Server) HandlePlaceStreamLocalPutRecord(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamLocalPutRecord") + defer span.End() + var body placestream.LocalPutRecord_Input + if err := c.Bind(&body); err != nil { + return err + } + var out *placestream.LocalPutRecord_Output + var handleErr error + // func (s *Server) handlePlaceStreamLocalPutRecord(ctx context.Context,body *placestream.LocalPutRecord_Input) (*placestream.LocalPutRecord_Output, error) + out, handleErr = s.handlePlaceStreamLocalPutRecord(ctx, &body) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + func (s *Server) HandlePlaceStreamMediaCreateUpload(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamMediaCreateUpload") defer span.End() -- 2.54.0