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()