diff --git a/js/docs/src/content/docs/lex-reference/media/place-stream-media-deletevideo.md b/js/docs/src/content/docs/lex-reference/media/place-stream-media-deletevideo.md new file mode 100644 index 000000000..f5789d7bd --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/media/place-stream-media-deletevideo.md @@ -0,0 +1,108 @@ +--- +title: place.stream.media.deleteVideo +description: Reference for the place.stream.media.deleteVideo lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `procedure` + +Delete a place.stream.video record and, on request, the track records its source points at. The content blob and the upload stay, so the record can be published again from the same upload. The caller must be the record's owner or hold livestream.manage from them; the records are deleted with the owner's session. + +**Parameters:** _(None defined)_ + +**Input:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| -------- | --------- | ----- | -------------------------------------- | ---------------- | +| `uri` | `string` | ✅ | | Format: `at-uri` | +| `tracks` | `boolean` | ❌ | Also delete the video's track records. | Default: `false` | + +**Output:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| --------- | ----------------- | ----- | -------------------------------------- | ----------- | +| `deleted` | Array of `string` | ✅ | Every record deleted, the video first. | | + +**Possible Errors:** + +- `NotPermitted`: The caller is neither the owner nor a moderator with livestream.manage from them. +- `VideoNotFound`: No such video record. + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.media.deleteVideo", + "defs": { + "main": { + "type": "procedure", + "description": "Delete a place.stream.video record and, on request, the track records its source points at. The content blob and the upload stay, so the record can be published again from the same upload. The caller must be the record's owner or hold livestream.manage from them; the records are deleted with the owner's session.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["uri"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri" + }, + "tracks": { + "type": "boolean", + "default": false, + "description": "Also delete the video's track records." + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["deleted"], + "properties": { + "deleted": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "Every record deleted, the video first." + } + } + } + }, + "errors": [ + { + "name": "NotPermitted", + "description": "The caller is neither the owner nor a moderator with livestream.manage from them." + }, + { + "name": "VideoNotFound", + "description": "No such video record." + } + ] + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/media/place-stream-media-finalizelivestream.md b/js/docs/src/content/docs/lex-reference/media/place-stream-media-finalizelivestream.md index fa896206f..5c8918e5a 100644 --- a/js/docs/src/content/docs/lex-reference/media/place-stream-media-finalizelivestream.md +++ b/js/docs/src/content/docs/lex-reference/media/place-stream-media-finalizelivestream.md @@ -13,7 +13,7 @@ description: Reference for the place.stream.media.finalizeLivestream lexicon **Type:** `procedure` -Turn a finished livestream into a VOD. The server concatenates the MUXL segments it recorded for the livestream into a single content blob, derives the playback sidecars, and publishes the place.stream.media.track records. Processing completes server-side and creates a draft VOD; the client does not need to poll or publish — the user publishes the draft later via place.stream.vod.publishDraft. Returns the draft's ats:// URI so the client can navigate to it, alongside an uploadId for backwards compatibility. +Turn a finished livestream into a VOD. The server concatenates the MUXL segments it recorded for the livestream(s) into a single content blob, derives the playback sidecars, and publishes the place.stream.media.track records. By default it leaves a draft VOD the streamer publishes later via place.stream.vod.publishDraft; with publish set it publishes the place.stream.video record as soon as the VOD is finalized. The caller must be the streamer, or a moderator they have granted livestream.manage; every record is written with the streamer's session. **Parameters:** _(None defined)_ @@ -24,9 +24,14 @@ Turn a finished livestream into a VOD. The server concatenates the MUXL segments **Schema Type:** `object` -| Name | Type | Req'd | Description | Constraints | -| ------------ | -------- | ----- | ----------------------------------------------------------------------------------------------------------- | ---------------- | -| `livestream` | `string` | ✅ | AT-URI of the place.stream.livestream record to finalize into a VOD. Must belong to the authenticated user. | Format: `at-uri` | +| Name | Type | Req'd | Description | Constraints | +| --------------- | ----------------- | ----- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------- | +| `livestream` | `string` | ❌ | AT-URI of the place.stream.livestream record to finalize into a VOD. Either this or livestreams is required. | Format: `at-uri` | +| `livestreams` | Array of `string` | ❌ | Several livestream records whose recordings make up one VOD: a streamer who started a new record mid-stream splits the recording across them. Recordings are concatenated in the records' creation order, whatever order they are listed in. All must belong to one streamer. | | +| `title` | `string` | ❌ | Title of the video record; the first livestream's title when empty. | Max Length: 1000 | +| `description` | `string` | ❌ | Description of the video record. | Max Length: 10000 | +| `publish` | `boolean` | ❌ | Publish the place.stream.video record as soon as the VOD is finalized, instead of leaving a draft. | Default: `false` | +| `endLivestream` | `boolean` | ❌ | Also end any of the livestream records that were never stopped, so the streamer's page stops reading as live. | Default: `false` | **Output:** @@ -35,15 +40,19 @@ Turn a finished livestream into a VOD. The server concatenates the MUXL segments **Schema Type:** `object` -| Name | Type | Req'd | Description | Constraints | -| ---------- | -------- | ----- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------- | -| `uploadId` | `string` | ✅ | Identifier for the finalize job. Retained for backwards compatibility; the draft flow no longer requires the client to poll getUploadStatus. | | -| `draftUri` | `string` | ✅ | The ats:// URI of the draft VOD created for this finalize. The draft reaches status 'ready' when processing completes; the user publishes it from the Drafts tab via place.stream.vod.publishDraft. | | +| Name | Type | Req'd | Description | Constraints | +| ------------- | ----------------- | ----- | ----------------------------------------------------------------------------------------------------------------------------------------------- | ----------- | +| `uploadId` | `string` | ✅ | Identifier for the finalize job. Retained for backwards compatibility; the draft flow no longer requires the client to poll getUploadStatus. | | +| `draftUri` | `string` | ❌ | The ats:// URI of the draft VOD created for this finalize, when publish is not set. The draft reaches status 'ready' when processing completes. | | +| `livestreams` | Array of `string` | ❌ | The livestream records the VOD is made of, in recording order. | | +| `objects` | `integer` | ❌ | How many recorded objects will be concatenated. | | +| `ended` | Array of `string` | ❌ | Livestream records that are now ended (already, or by endLivestream). | | **Possible Errors:** -- `LivestreamNotFound`: No livestream with the given URI is known, or it does not belong to the authenticated user. +- `LivestreamNotFound`: No livestream with the given URI is known to this node, or the livestreams belong to different streamers. - `NoRecording`: The livestream has no recorded MUXL segments to finalize (recording was not enabled, or none completed). +- `NotPermitted`: The caller is neither the streamer nor a moderator with livestream.manage from them. --- @@ -56,17 +65,45 @@ Turn a finished livestream into a VOD. The server concatenates the MUXL segments "defs": { "main": { "type": "procedure", - "description": "Turn a finished livestream into a VOD. The server concatenates the MUXL segments it recorded for the livestream into a single content blob, derives the playback sidecars, and publishes the place.stream.media.track records. Processing completes server-side and creates a draft VOD; the client does not need to poll or publish — the user publishes the draft later via place.stream.vod.publishDraft. Returns the draft's ats:// URI so the client can navigate to it, alongside an uploadId for backwards compatibility.", + "description": "Turn a finished livestream into a VOD. The server concatenates the MUXL segments it recorded for the livestream(s) into a single content blob, derives the playback sidecars, and publishes the place.stream.media.track records. By default it leaves a draft VOD the streamer publishes later via place.stream.vod.publishDraft; with publish set it publishes the place.stream.video record as soon as the VOD is finalized. The caller must be the streamer, or a moderator they have granted livestream.manage; every record is written with the streamer's session.", "input": { "encoding": "application/json", "schema": { "type": "object", - "required": ["livestream"], + "required": [], "properties": { "livestream": { "type": "string", "format": "at-uri", - "description": "AT-URI of the place.stream.livestream record to finalize into a VOD. Must belong to the authenticated user." + "description": "AT-URI of the place.stream.livestream record to finalize into a VOD. Either this or livestreams is required." + }, + "livestreams": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "Several livestream records whose recordings make up one VOD: a streamer who started a new record mid-stream splits the recording across them. Recordings are concatenated in the records' creation order, whatever order they are listed in. All must belong to one streamer." + }, + "title": { + "type": "string", + "maxLength": 1000, + "description": "Title of the video record; the first livestream's title when empty." + }, + "description": { + "type": "string", + "maxLength": 10000, + "description": "Description of the video record." + }, + "publish": { + "type": "boolean", + "default": false, + "description": "Publish the place.stream.video record as soon as the VOD is finalized, instead of leaving a draft." + }, + "endLivestream": { + "type": "boolean", + "default": false, + "description": "Also end any of the livestream records that were never stopped, so the streamer's page stops reading as live." } } } @@ -75,7 +112,7 @@ Turn a finished livestream into a VOD. The server concatenates the MUXL segments "encoding": "application/json", "schema": { "type": "object", - "required": ["uploadId", "draftUri"], + "required": ["uploadId"], "properties": { "uploadId": { "type": "string", @@ -83,7 +120,27 @@ Turn a finished livestream into a VOD. The server concatenates the MUXL segments }, "draftUri": { "type": "string", - "description": "The ats:// URI of the draft VOD created for this finalize. The draft reaches status 'ready' when processing completes; the user publishes it from the Drafts tab via place.stream.vod.publishDraft." + "description": "The ats:// URI of the draft VOD created for this finalize, when publish is not set. The draft reaches status 'ready' when processing completes." + }, + "livestreams": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "The livestream records the VOD is made of, in recording order." + }, + "objects": { + "type": "integer", + "description": "How many recorded objects will be concatenated." + }, + "ended": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "Livestream records that are now ended (already, or by endLivestream)." } } } @@ -91,11 +148,15 @@ Turn a finished livestream into a VOD. The server concatenates the MUXL segments "errors": [ { "name": "LivestreamNotFound", - "description": "No livestream with the given URI is known, or it does not belong to the authenticated user." + "description": "No livestream with the given URI is known to this node, or the livestreams belong to different streamers." }, { "name": "NoRecording", "description": "The livestream has no recorded MUXL segments to finalize (recording was not enabled, or none completed)." + }, + { + "name": "NotPermitted", + "description": "The caller is neither the streamer nor a moderator with livestream.manage from them." } ] } diff --git a/js/docs/src/content/docs/lex-reference/media/place-stream-media-listvideos.md b/js/docs/src/content/docs/lex-reference/media/place-stream-media-listvideos.md new file mode 100644 index 000000000..1229bbb60 --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/media/place-stream-media-listvideos.md @@ -0,0 +1,227 @@ +--- +title: place.stream.media.listVideos +description: Reference for the place.stream.media.listVideos lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `query` + +A streamer's place.stream.video records (read from their PDS) and this node's uploads for them: everything updateVideo, deleteVideo and publishVideo take. The caller must be the streamer or hold livestream.manage from them. + +**Parameters:** + +| Name | Type | Req'd | Description | Constraints | +| ------ | -------- | ----- | ------------------- | ------------- | +| `repo` | `string` | ✅ | The streamer's DID. | Format: `did` | + +**Output:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| --------- | ----------------------------- | ----- | ----------- | ------------- | +| `repoDID` | `string` | ✅ | | Format: `did` | +| `videos` | Array of [`#video`](#video) | ✅ | | | +| `uploads` | Array of [`#upload`](#upload) | ✅ | | | + +**Possible Errors:** + +- `NotPermitted`: The caller is neither the streamer nor a moderator with livestream.manage from them. + +--- + + + +### `video` + +**Type:** `object` + +**Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ------------- | ----------------- | ----- | ------------------------------------------------- | ---------------- | +| `uri` | `string` | ✅ | | Format: `at-uri` | +| `cid` | `string` | ✅ | | Format: `cid` | +| `title` | `string` | ✅ | | | +| `durationMs` | `integer` | ❌ | | | +| `createdAt` | `string` | ❌ | | | +| `tracks` | Array of `string` | ✅ | The track records the video's source points at. | | +| `connections` | Array of `string` | ❌ | The livestream records the video is connected to. | | +| `thumb` | `boolean` | ❌ | Whether the record carries a thumbnail. | | + +--- + + + +### `upload` + +**Type:** `object` + +**Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ------------ | --------- | ----- | -------------------------------------------------------------- | ------------------ | +| `uploadId` | `string` | ✅ | | | +| `status` | `string` | ✅ | Processing status: pending, processing, done or error. | | +| `error` | `string` | ❌ | | | +| `backend` | `string` | ✅ | | | +| `location` | `string` | ❌ | For a finalized recording, the livestream record it came from. | | +| `contentCid` | `string` | ❌ | | | +| `durationMs` | `integer` | ❌ | | | +| `blobSize` | `integer` | ❌ | | | +| `tracks` | `integer` | ❌ | How many track records the upload has published. | | +| `createdAt` | `string` | ❌ | | Format: `datetime` | + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.media.listVideos", + "defs": { + "main": { + "type": "query", + "description": "A streamer's place.stream.video records (read from their PDS) and this node's uploads for them: everything updateVideo, deleteVideo and publishVideo take. The caller must be the streamer or hold livestream.manage from them.", + "parameters": { + "type": "params", + "required": ["repo"], + "properties": { + "repo": { + "type": "string", + "format": "did", + "description": "The streamer's DID." + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repoDID", "videos", "uploads"], + "properties": { + "repoDID": { + "type": "string", + "format": "did" + }, + "videos": { + "type": "array", + "items": { + "type": "ref", + "ref": "#video" + } + }, + "uploads": { + "type": "array", + "items": { + "type": "ref", + "ref": "#upload" + } + } + } + } + }, + "errors": [ + { + "name": "NotPermitted", + "description": "The caller is neither the streamer nor a moderator with livestream.manage from them." + } + ] + }, + "video": { + "type": "object", + "required": ["uri", "cid", "title", "tracks"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri" + }, + "cid": { + "type": "string", + "format": "cid" + }, + "title": { + "type": "string" + }, + "durationMs": { + "type": "integer" + }, + "createdAt": { + "type": "string" + }, + "tracks": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "The track records the video's source points at." + }, + "connections": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "The livestream records the video is connected to." + }, + "thumb": { + "type": "boolean", + "description": "Whether the record carries a thumbnail." + } + } + }, + "upload": { + "type": "object", + "required": ["uploadId", "status", "backend"], + "properties": { + "uploadId": { + "type": "string" + }, + "status": { + "type": "string", + "description": "Processing status: pending, processing, done or error." + }, + "error": { + "type": "string" + }, + "backend": { + "type": "string" + }, + "location": { + "type": "string", + "description": "For a finalized recording, the livestream record it came from." + }, + "contentCid": { + "type": "string" + }, + "durationMs": { + "type": "integer" + }, + "blobSize": { + "type": "integer" + }, + "tracks": { + "type": "integer", + "description": "How many track records the upload has published." + }, + "createdAt": { + "type": "string", + "format": "datetime" + } + } + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/media/place-stream-media-publishvideo.md b/js/docs/src/content/docs/lex-reference/media/place-stream-media-publishvideo.md index 371f4bd70..dfffcf7fc 100644 --- a/js/docs/src/content/docs/lex-reference/media/place-stream-media-publishvideo.md +++ b/js/docs/src/content/docs/lex-reference/media/place-stream-media-publishvideo.md @@ -13,7 +13,7 @@ description: Reference for the place.stream.media.publishVideo lexicon **Type:** `procedure` -Publish a place.stream.video record for a finished upload, server-side. The caller supplies the record it would otherwise putRecord itself; the server overrides the fields it is authoritative about (source tracks and durationMs, taken from the processed upload) and, if the record carries no thumb, generates one from the video and attaches it. The record is written to the authenticated user's repo. +Publish a place.stream.video record for a finished upload, server-side. The caller supplies the record it would otherwise putRecord itself; the server overrides the fields it is authoritative about (source tracks and durationMs, taken from the processed upload) and, if the record carries no thumb, generates one from the video and attaches it. The record is written to the authenticated user's repo. The caller must own the upload, or hold livestream.manage from its owner; the record is written with the owner's session. Without a record, one is built from the upload's livestream(s) (title, tags, activity, connections) and the given title and description. **Parameters:** _(None defined)_ @@ -24,10 +24,13 @@ Publish a place.stream.video record for a finished upload, server-side. The call **Schema Type:** `object` -| Name | Type | Req'd | Description | Constraints | -| ---------- | --------------------------------------------------------- | ----- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------- | -| `uploadId` | `string` | ✅ | The upload ID returned by place.stream.media.createUpload. Its processing must be complete (status 'done'). | | -| `record` | [`place.stream.video`](/lex-reference/place-stream-video) | ✅ | A place.stream.video record. The server overrides `source` and `durationMs` from the processed upload, and fills in `thumb` with a generated thumbnail when the supplied record has none. | | +| Name | Type | Req'd | Description | Constraints | +| ------------- | --------------------------------------------------------- | ----- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------- | +| `uploadId` | `string` | ✅ | The upload ID returned by place.stream.media.createUpload. Its processing must be complete (status 'done'). | | +| `record` | [`place.stream.video`](/lex-reference/place-stream-video) | ❌ | A place.stream.video record. The server overrides `source` and `durationMs` from the processed upload, and fills in `thumb` with a generated thumbnail when the supplied record has none. | | +| `livestreams` | Array of `string` | ❌ | Livestream records the video came from, used to build the record when none is given; defaults to the upload's own livestream for a finalized recording. | | +| `title` | `string` | ❌ | Title for a record built by the server; the first livestream's title when empty. | Max Length: 1000 | +| `description` | `string` | ❌ | Description for a record built by the server. | Max Length: 10000 | **Output:** @@ -57,12 +60,12 @@ Publish a place.stream.video record for a finished upload, server-side. The call "defs": { "main": { "type": "procedure", - "description": "Publish a place.stream.video record for a finished upload, server-side. The caller supplies the record it would otherwise putRecord itself; the server overrides the fields it is authoritative about (source tracks and durationMs, taken from the processed upload) and, if the record carries no thumb, generates one from the video and attaches it. The record is written to the authenticated user's repo.", + "description": "Publish a place.stream.video record for a finished upload, server-side. The caller supplies the record it would otherwise putRecord itself; the server overrides the fields it is authoritative about (source tracks and durationMs, taken from the processed upload) and, if the record carries no thumb, generates one from the video and attaches it. The record is written to the authenticated user's repo. The caller must own the upload, or hold livestream.manage from its owner; the record is written with the owner's session. Without a record, one is built from the upload's livestream(s) (title, tags, activity, connections) and the given title and description.", "input": { "encoding": "application/json", "schema": { "type": "object", - "required": ["uploadId", "record"], + "required": ["uploadId"], "properties": { "uploadId": { "type": "string", @@ -72,6 +75,24 @@ Publish a place.stream.video record for a finished upload, server-side. The call "type": "ref", "ref": "place.stream.video", "description": "A place.stream.video record. The server overrides `source` and `durationMs` from the processed upload, and fills in `thumb` with a generated thumbnail when the supplied record has none." + }, + "livestreams": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "Livestream records the video came from, used to build the record when none is given; defaults to the upload's own livestream for a finalized recording." + }, + "title": { + "type": "string", + "maxLength": 1000, + "description": "Title for a record built by the server; the first livestream's title when empty." + }, + "description": { + "type": "string", + "maxLength": 10000, + "description": "Description for a record built by the server." } } } diff --git a/js/docs/src/content/docs/lex-reference/media/place-stream-media-updatevideo.md b/js/docs/src/content/docs/lex-reference/media/place-stream-media-updatevideo.md new file mode 100644 index 000000000..ffb784b8b --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/media/place-stream-media-updatevideo.md @@ -0,0 +1,127 @@ +--- +title: place.stream.media.updateVideo +description: Reference for the place.stream.media.updateVideo lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `procedure` + +Rewrite a place.stream.video record in place: retitle, redescribe or retag it, or point it at a finished upload's tracks (publishing them if the upload has none yet: the repair for a record published without tracks, or the way to point a record at a re-finalized VOD). Fields left out are kept; an empty description or tag list clears it. The caller must be the record's owner or hold livestream.manage from them; the record is written with the owner's session. + +**Parameters:** _(None defined)_ + +**Input:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| ------------- | ----------------- | ----- | ------------------------------------------------------------------------------------------ | ----------------- | +| `uri` | `string` | ✅ | The place.stream.video record. | Format: `at-uri` | +| `title` | `string` | ❌ | | Max Length: 1000 | +| `description` | `string` | ❌ | | Max Length: 10000 | +| `tags` | Array of `string` | ❌ | | | +| `uploadId` | `string` | ❌ | A finished upload whose tracks become the record's source and whose duration the record's. | | + +**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:** + +- `NotPermitted`: The caller is neither the owner nor a moderator with livestream.manage from them. +- `VideoNotFound`: No such video record. + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.media.updateVideo", + "defs": { + "main": { + "type": "procedure", + "description": "Rewrite a place.stream.video record in place: retitle, redescribe or retag it, or point it at a finished upload's tracks (publishing them if the upload has none yet: the repair for a record published without tracks, or the way to point a record at a re-finalized VOD). Fields left out are kept; an empty description or tag list clears it. The caller must be the record's owner or hold livestream.manage from them; the record is written with the owner's session.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["uri"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri", + "description": "The place.stream.video record." + }, + "title": { + "type": "string", + "maxLength": 1000 + }, + "description": { + "type": "string", + "maxLength": 10000 + }, + "tags": { + "type": "array", + "items": { + "type": "string", + "maxLength": 64 + } + }, + "uploadId": { + "type": "string", + "description": "A finished upload whose tracks become the record's source and whose duration the record's." + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["uri", "cid"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri" + }, + "cid": { + "type": "string", + "format": "cid" + } + } + } + }, + "errors": [ + { + "name": "NotPermitted", + "description": "The caller is neither the owner nor a moderator with livestream.manage from them." + }, + { + "name": "VideoNotFound", + "description": "No such video record." + } + ] + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index 7d56e79e6..0aacc932f 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -2759,9 +2759,88 @@ } } }, + "/xrpc/place.stream.media.deleteVideo": { + "post": { + "summary": "Delete a place.stream.video record and, on request, the track records its source points at. The content blob and the upload stay, so the record can be published again from the same upload. The caller must be the record's owner or hold livestream.manage from them; the records are deleted with the owner's session.", + "operationId": "place.stream.media.deleteVideo", + "tags": ["place.stream.media"], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "deleted": { + "type": "array", + "description": "Every record deleted, the video first.", + "items": { + "type": "string", + "format": "uri" + } + } + }, + "required": ["deleted"] + } + } + } + }, + "400": { + "description": "Bad Request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["error", "message"], + "properties": { + "error": { + "type": "string", + "oneOf": [ + { + "const": "NotPermitted" + }, + { + "const": "VideoNotFound" + } + ] + }, + "message": { + "type": "string" + } + } + } + } + } + } + }, + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "uri": { + "type": "string", + "format": "uri" + }, + "tracks": { + "type": "boolean", + "description": "Also delete the video's track records.", + "default": false + } + }, + "required": ["uri"] + } + } + } + } + } + }, "/xrpc/place.stream.media.finalizeLivestream": { "post": { - "summary": "Turn a finished livestream into a VOD. The server concatenates the MUXL segments it recorded for the livestream into a single content blob, derives the playback sidecars, and publishes the place.stream.media.track records. Processing completes server-side and creates a draft VOD; the client does not need to poll or publish — the user publishes the draft later via place.stream.vod.publishDraft. Returns the draft's ats:// URI so the client can navigate to it, alongside an uploadId for backwards compatibility.", + "summary": "Turn a finished livestream into a VOD. The server concatenates the MUXL segments it recorded for the livestream(s) into a single content blob, derives the playback sidecars, and publishes the place.stream.media.track records. By default it leaves a draft VOD the streamer publishes later via place.stream.vod.publishDraft; with publish set it publishes the place.stream.video record as soon as the VOD is finalized. The caller must be the streamer, or a moderator they have granted livestream.manage; every record is written with the streamer's session.", "operationId": "place.stream.media.finalizeLivestream", "tags": ["place.stream.media"], "responses": { @@ -2778,10 +2857,30 @@ }, "draftUri": { "type": "string", - "description": "The ats:// URI of the draft VOD created for this finalize. The draft reaches status 'ready' when processing completes; the user publishes it from the Drafts tab via place.stream.vod.publishDraft." + "description": "The ats:// URI of the draft VOD created for this finalize, when publish is not set. The draft reaches status 'ready' when processing completes." + }, + "livestreams": { + "type": "array", + "description": "The livestream records the VOD is made of, in recording order.", + "items": { + "type": "string", + "format": "uri" + } + }, + "objects": { + "type": "integer", + "description": "How many recorded objects will be concatenated." + }, + "ended": { + "type": "array", + "description": "Livestream records that are now ended (already, or by endLivestream).", + "items": { + "type": "string", + "format": "uri" + } } }, - "required": ["uploadId", "draftUri"] + "required": ["uploadId"] } } } @@ -2802,6 +2901,9 @@ }, { "const": "NoRecording" + }, + { + "const": "NotPermitted" } ] }, @@ -2823,11 +2925,38 @@ "properties": { "livestream": { "type": "string", - "description": "AT-URI of the place.stream.livestream record to finalize into a VOD. Must belong to the authenticated user.", + "description": "AT-URI of the place.stream.livestream record to finalize into a VOD. Either this or livestreams is required.", "format": "uri" + }, + "livestreams": { + "type": "array", + "description": "Several livestream records whose recordings make up one VOD: a streamer who started a new record mid-stream splits the recording across them. Recordings are concatenated in the records' creation order, whatever order they are listed in. All must belong to one streamer.", + "items": { + "type": "string", + "format": "uri" + } + }, + "title": { + "type": "string", + "description": "Title of the video record; the first livestream's title when empty.", + "maxLength": 1000 + }, + "description": { + "type": "string", + "description": "Description of the video record.", + "maxLength": 10000 + }, + "publish": { + "type": "boolean", + "description": "Publish the place.stream.video record as soon as the VOD is finalized, instead of leaving a draft.", + "default": false + }, + "endLivestream": { + "type": "boolean", + "description": "Also end any of the livestream records that were never stopped, so the streamer's page stops reading as live.", + "default": false } - }, - "required": ["livestream"] + } } } } @@ -3064,9 +3193,84 @@ ] } }, + "/xrpc/place.stream.media.listVideos": { + "get": { + "summary": "A streamer's place.stream.video records (read from their PDS) and this node's uploads for them: everything updateVideo, deleteVideo and publishVideo take. The caller must be the streamer or hold livestream.manage from them.", + "operationId": "place.stream.media.listVideos", + "tags": ["place.stream.media"], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "repoDID": { + "type": "string", + "format": "did" + }, + "videos": { + "type": "array", + "items": { + "$ref": "#/components/schemas/place.stream.media.listVideos_video" + } + }, + "uploads": { + "type": "array", + "items": { + "$ref": "#/components/schemas/place.stream.media.listVideos_upload" + } + } + }, + "required": ["repoDID", "videos", "uploads"] + } + } + } + }, + "400": { + "description": "Bad Request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["error", "message"], + "properties": { + "error": { + "type": "string", + "oneOf": [ + { + "const": "NotPermitted" + } + ] + }, + "message": { + "type": "string" + } + } + } + } + } + } + }, + "parameters": [ + { + "name": "repo", + "in": "query", + "required": true, + "description": "The streamer's DID.", + "schema": { + "type": "string", + "description": "The streamer's DID.", + "format": "did" + } + } + ] + } + }, "/xrpc/place.stream.media.publishVideo": { "post": { - "summary": "Publish a place.stream.video record for a finished upload, server-side. The caller supplies the record it would otherwise putRecord itself; the server overrides the fields it is authoritative about (source tracks and durationMs, taken from the processed upload) and, if the record carries no thumb, generates one from the video and attaches it. The record is written to the authenticated user's repo.", + "summary": "Publish a place.stream.video record for a finished upload, server-side. The caller supplies the record it would otherwise putRecord itself; the server overrides the fields it is authoritative about (source tracks and durationMs, taken from the processed upload) and, if the record carries no thumb, generates one from the video and attaches it. The record is written to the authenticated user's repo. The caller must own the upload, or hold livestream.manage from its owner; the record is written with the owner's session. Without a record, one is built from the upload's livestream(s) (title, tags, activity, connections) and the given title and description.", "operationId": "place.stream.media.publishVideo", "tags": ["place.stream.media"], "responses": { @@ -3135,9 +3339,121 @@ "record": { "$ref": "#/components/schemas/place.stream.video", "description": "A place.stream.video record. The server overrides `source` and `durationMs` from the processed upload, and fills in `thumb` with a generated thumbnail when the supplied record has none." + }, + "livestreams": { + "type": "array", + "description": "Livestream records the video came from, used to build the record when none is given; defaults to the upload's own livestream for a finalized recording.", + "items": { + "type": "string", + "format": "uri" + } + }, + "title": { + "type": "string", + "description": "Title for a record built by the server; the first livestream's title when empty.", + "maxLength": 1000 + }, + "description": { + "type": "string", + "description": "Description for a record built by the server.", + "maxLength": 10000 } }, - "required": ["uploadId", "record"] + "required": ["uploadId"] + } + } + } + } + } + }, + "/xrpc/place.stream.media.updateVideo": { + "post": { + "summary": "Rewrite a place.stream.video record in place: retitle, redescribe or retag it, or point it at a finished upload's tracks (publishing them if the upload has none yet: the repair for a record published without tracks, or the way to point a record at a re-finalized VOD). Fields left out are kept; an empty description or tag list clears it. The caller must be the record's owner or hold livestream.manage from them; the record is written with the owner's session.", + "operationId": "place.stream.media.updateVideo", + "tags": ["place.stream.media"], + "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": "NotPermitted" + }, + { + "const": "VideoNotFound" + } + ] + }, + "message": { + "type": "string" + } + } + } + } + } + } + }, + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "uri": { + "type": "string", + "description": "The place.stream.video record.", + "format": "uri" + }, + "title": { + "type": "string", + "maxLength": 1000 + }, + "description": { + "type": "string", + "maxLength": 10000 + }, + "tags": { + "type": "array", + "items": { + "type": "string", + "maxLength": 64 + } + }, + "uploadId": { + "type": "string", + "description": "A finished upload whose tracks become the record's source and whose duration the record's." + } + }, + "required": ["uri"] } } } @@ -6998,6 +7314,89 @@ }, "required": ["uri", "cid", "author", "record"] }, + "place.stream.media.listVideos_video": { + "type": "object", + "properties": { + "uri": { + "type": "string", + "format": "uri" + }, + "cid": { + "type": "string", + "format": "cid" + }, + "title": { + "type": "string" + }, + "durationMs": { + "type": "integer" + }, + "createdAt": { + "type": "string" + }, + "tracks": { + "type": "array", + "description": "The track records the video's source points at.", + "items": { + "type": "string", + "format": "uri" + } + }, + "connections": { + "type": "array", + "description": "The livestream records the video is connected to.", + "items": { + "type": "string", + "format": "uri" + } + }, + "thumb": { + "type": "boolean", + "description": "Whether the record carries a thumbnail." + } + }, + "required": ["uri", "cid", "title", "tracks"] + }, + "place.stream.media.listVideos_upload": { + "type": "object", + "properties": { + "uploadId": { + "type": "string" + }, + "status": { + "type": "string", + "description": "Processing status: pending, processing, done or error." + }, + "error": { + "type": "string" + }, + "backend": { + "type": "string" + }, + "location": { + "type": "string", + "description": "For a finalized recording, the livestream record it came from." + }, + "contentCid": { + "type": "string" + }, + "durationMs": { + "type": "integer" + }, + "blobSize": { + "type": "integer" + }, + "tracks": { + "type": "integer", + "description": "How many track records the upload has published." + }, + "createdAt": { + "type": "string", + "format": "date-time" + } + }, + "required": ["uploadId", "status", "backend"] + }, "place.stream.video": { "description": "Unknown type" }, diff --git a/lexicons/place/stream/media/deleteVideo.json b/lexicons/place/stream/media/deleteVideo.json new file mode 100644 index 000000000..42defa9d0 --- /dev/null +++ b/lexicons/place/stream/media/deleteVideo.json @@ -0,0 +1,55 @@ +{ + "lexicon": 1, + "id": "place.stream.media.deleteVideo", + "defs": { + "main": { + "type": "procedure", + "description": "Delete a place.stream.video record and, on request, the track records its source points at. The content blob and the upload stay, so the record can be published again from the same upload. The caller must be the record's owner or hold livestream.manage from them; the records are deleted with the owner's session.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["uri"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri" + }, + "tracks": { + "type": "boolean", + "default": false, + "description": "Also delete the video's track records." + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["deleted"], + "properties": { + "deleted": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "Every record deleted, the video first." + } + } + } + }, + "errors": [ + { + "name": "NotPermitted", + "description": "The caller is neither the owner nor a moderator with livestream.manage from them." + }, + { + "name": "VideoNotFound", + "description": "No such video record." + } + ] + } + } +} diff --git a/lexicons/place/stream/media/finalizeLivestream.json b/lexicons/place/stream/media/finalizeLivestream.json index 615234b86..aa4981edb 100644 --- a/lexicons/place/stream/media/finalizeLivestream.json +++ b/lexicons/place/stream/media/finalizeLivestream.json @@ -4,17 +4,45 @@ "defs": { "main": { "type": "procedure", - "description": "Turn a finished livestream into a VOD. The server concatenates the MUXL segments it recorded for the livestream into a single content blob, derives the playback sidecars, and publishes the place.stream.media.track records. Processing completes server-side and creates a draft VOD; the client does not need to poll or publish — the user publishes the draft later via place.stream.vod.publishDraft. Returns the draft's ats:// URI so the client can navigate to it, alongside an uploadId for backwards compatibility.", + "description": "Turn a finished livestream into a VOD. The server concatenates the MUXL segments it recorded for the livestream(s) into a single content blob, derives the playback sidecars, and publishes the place.stream.media.track records. By default it leaves a draft VOD the streamer publishes later via place.stream.vod.publishDraft; with publish set it publishes the place.stream.video record as soon as the VOD is finalized. The caller must be the streamer, or a moderator they have granted livestream.manage; every record is written with the streamer's session.", "input": { "encoding": "application/json", "schema": { "type": "object", - "required": ["livestream"], + "required": [], "properties": { "livestream": { "type": "string", "format": "at-uri", - "description": "AT-URI of the place.stream.livestream record to finalize into a VOD. Must belong to the authenticated user." + "description": "AT-URI of the place.stream.livestream record to finalize into a VOD. Either this or livestreams is required." + }, + "livestreams": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "Several livestream records whose recordings make up one VOD: a streamer who started a new record mid-stream splits the recording across them. Recordings are concatenated in the records' creation order, whatever order they are listed in. All must belong to one streamer." + }, + "title": { + "type": "string", + "maxLength": 1000, + "description": "Title of the video record; the first livestream's title when empty." + }, + "description": { + "type": "string", + "maxLength": 10000, + "description": "Description of the video record." + }, + "publish": { + "type": "boolean", + "default": false, + "description": "Publish the place.stream.video record as soon as the VOD is finalized, instead of leaving a draft." + }, + "endLivestream": { + "type": "boolean", + "default": false, + "description": "Also end any of the livestream records that were never stopped, so the streamer's page stops reading as live." } } } @@ -23,7 +51,7 @@ "encoding": "application/json", "schema": { "type": "object", - "required": ["uploadId", "draftUri"], + "required": ["uploadId"], "properties": { "uploadId": { "type": "string", @@ -31,7 +59,27 @@ }, "draftUri": { "type": "string", - "description": "The ats:// URI of the draft VOD created for this finalize. The draft reaches status 'ready' when processing completes; the user publishes it from the Drafts tab via place.stream.vod.publishDraft." + "description": "The ats:// URI of the draft VOD created for this finalize, when publish is not set. The draft reaches status 'ready' when processing completes." + }, + "livestreams": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "The livestream records the VOD is made of, in recording order." + }, + "objects": { + "type": "integer", + "description": "How many recorded objects will be concatenated." + }, + "ended": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "Livestream records that are now ended (already, or by endLivestream)." } } } @@ -39,11 +87,15 @@ "errors": [ { "name": "LivestreamNotFound", - "description": "No livestream with the given URI is known, or it does not belong to the authenticated user." + "description": "No livestream with the given URI is known to this node, or the livestreams belong to different streamers." }, { "name": "NoRecording", "description": "The livestream has no recorded MUXL segments to finalize (recording was not enabled, or none completed)." + }, + { + "name": "NotPermitted", + "description": "The caller is neither the streamer nor a moderator with livestream.manage from them." } ] } diff --git a/lexicons/place/stream/media/listVideos.json b/lexicons/place/stream/media/listVideos.json new file mode 100644 index 000000000..d65c75b54 --- /dev/null +++ b/lexicons/place/stream/media/listVideos.json @@ -0,0 +1,137 @@ +{ + "lexicon": 1, + "id": "place.stream.media.listVideos", + "defs": { + "main": { + "type": "query", + "description": "A streamer's place.stream.video records (read from their PDS) and this node's uploads for them: everything updateVideo, deleteVideo and publishVideo take. The caller must be the streamer or hold livestream.manage from them.", + "parameters": { + "type": "params", + "required": ["repo"], + "properties": { + "repo": { + "type": "string", + "format": "did", + "description": "The streamer's DID." + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["repoDID", "videos", "uploads"], + "properties": { + "repoDID": { + "type": "string", + "format": "did" + }, + "videos": { + "type": "array", + "items": { + "type": "ref", + "ref": "#video" + } + }, + "uploads": { + "type": "array", + "items": { + "type": "ref", + "ref": "#upload" + } + } + } + } + }, + "errors": [ + { + "name": "NotPermitted", + "description": "The caller is neither the streamer nor a moderator with livestream.manage from them." + } + ] + }, + "video": { + "type": "object", + "required": ["uri", "cid", "title", "tracks"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri" + }, + "cid": { + "type": "string", + "format": "cid" + }, + "title": { + "type": "string" + }, + "durationMs": { + "type": "integer" + }, + "createdAt": { + "type": "string" + }, + "tracks": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "The track records the video's source points at." + }, + "connections": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "The livestream records the video is connected to." + }, + "thumb": { + "type": "boolean", + "description": "Whether the record carries a thumbnail." + } + } + }, + "upload": { + "type": "object", + "required": ["uploadId", "status", "backend"], + "properties": { + "uploadId": { + "type": "string" + }, + "status": { + "type": "string", + "description": "Processing status: pending, processing, done or error." + }, + "error": { + "type": "string" + }, + "backend": { + "type": "string" + }, + "location": { + "type": "string", + "description": "For a finalized recording, the livestream record it came from." + }, + "contentCid": { + "type": "string" + }, + "durationMs": { + "type": "integer" + }, + "blobSize": { + "type": "integer" + }, + "tracks": { + "type": "integer", + "description": "How many track records the upload has published." + }, + "createdAt": { + "type": "string", + "format": "datetime" + } + } + } + } +} diff --git a/lexicons/place/stream/media/publishVideo.json b/lexicons/place/stream/media/publishVideo.json index 4f7a7f4f7..2cf5e04bb 100644 --- a/lexicons/place/stream/media/publishVideo.json +++ b/lexicons/place/stream/media/publishVideo.json @@ -4,12 +4,12 @@ "defs": { "main": { "type": "procedure", - "description": "Publish a place.stream.video record for a finished upload, server-side. The caller supplies the record it would otherwise putRecord itself; the server overrides the fields it is authoritative about (source tracks and durationMs, taken from the processed upload) and, if the record carries no thumb, generates one from the video and attaches it. The record is written to the authenticated user's repo.", + "description": "Publish a place.stream.video record for a finished upload, server-side. The caller supplies the record it would otherwise putRecord itself; the server overrides the fields it is authoritative about (source tracks and durationMs, taken from the processed upload) and, if the record carries no thumb, generates one from the video and attaches it. The record is written to the authenticated user's repo. The caller must own the upload, or hold livestream.manage from its owner; the record is written with the owner's session. Without a record, one is built from the upload's livestream(s) (title, tags, activity, connections) and the given title and description.", "input": { "encoding": "application/json", "schema": { "type": "object", - "required": ["uploadId", "record"], + "required": ["uploadId"], "properties": { "uploadId": { "type": "string", @@ -19,6 +19,24 @@ "type": "ref", "ref": "place.stream.video", "description": "A place.stream.video record. The server overrides `source` and `durationMs` from the processed upload, and fills in `thumb` with a generated thumbnail when the supplied record has none." + }, + "livestreams": { + "type": "array", + "items": { + "type": "string", + "format": "at-uri" + }, + "description": "Livestream records the video came from, used to build the record when none is given; defaults to the upload's own livestream for a finalized recording." + }, + "title": { + "type": "string", + "maxLength": 1000, + "description": "Title for a record built by the server; the first livestream's title when empty." + }, + "description": { + "type": "string", + "maxLength": 10000, + "description": "Description for a record built by the server." } } } diff --git a/lexicons/place/stream/media/updateVideo.json b/lexicons/place/stream/media/updateVideo.json new file mode 100644 index 000000000..763cd384c --- /dev/null +++ b/lexicons/place/stream/media/updateVideo.json @@ -0,0 +1,70 @@ +{ + "lexicon": 1, + "id": "place.stream.media.updateVideo", + "defs": { + "main": { + "type": "procedure", + "description": "Rewrite a place.stream.video record in place: retitle, redescribe or retag it, or point it at a finished upload's tracks (publishing them if the upload has none yet: the repair for a record published without tracks, or the way to point a record at a re-finalized VOD). Fields left out are kept; an empty description or tag list clears it. The caller must be the record's owner or hold livestream.manage from them; the record is written with the owner's session.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["uri"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri", + "description": "The place.stream.video record." + }, + "title": { + "type": "string", + "maxLength": 1000 + }, + "description": { + "type": "string", + "maxLength": 10000 + }, + "tags": { + "type": "array", + "items": { + "type": "string", + "maxLength": 64 + } + }, + "uploadId": { + "type": "string", + "description": "A finished upload whose tracks become the record's source and whose duration the record's." + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["uri", "cid"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri" + }, + "cid": { + "type": "string", + "format": "cid" + } + } + } + }, + "errors": [ + { + "name": "NotPermitted", + "description": "The caller is neither the owner nor a moderator with livestream.manage from them." + }, + { + "name": "VideoNotFound", + "description": "No such video record." + } + ] + } + } +} diff --git a/pkg/api/admin_livestream.go b/pkg/api/admin_livestream.go deleted file mode 100644 index f469c394c..000000000 --- a/pkg/api/admin_livestream.go +++ /dev/null @@ -1,268 +0,0 @@ -package api - -import ( - "context" - "encoding/json" - "net/http" - "sort" - "strings" - - "github.com/google/uuid" - "github.com/julienschmidt/httprouter" - "stream.place/streamplace/pkg/comatproto" - "stream.place/streamplace/pkg/errors" - "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/model" - "stream.place/streamplace/pkg/placestream" - "stream.place/streamplace/pkg/statedb" -) - -// finalizeLivestreamRequest is the body of POST /finalize-livestream on the -// internal API. -type finalizeLivestreamRequest struct { - // Livestream is the at:// URI of the place.stream.livestream record whose - // recording becomes the VOD (every recorded object under it, in order). - Livestream string `json:"livestream"` - // Livestreams names several records whose recordings make up one VOD: - // a streamer who started a new record mid-stream (some clients do that - // on a title change) split the recording across them. Objects are - // concatenated in recording order across all of them. Either field or - // both may be given. - Livestreams []string `json:"livestreams"` - // Title and Description of the video record; the first livestream's - // title when empty. - Title string `json:"title"` - Description string `json:"description"` - // Publish (default true) publishes the video record as soon as the VOD - // is finalized; false leaves an upload the streamer publishes from the - // app's Livestreams tab. - Publish *bool `json:"publish"` - // EndLivestream also sets endedAt on any of the records the streamer - // never stopped, so their page stops reading as live. - EndLivestream bool `json:"endLivestream"` -} - -type finalizeLivestreamResponse struct { - UploadID string `json:"uploadId"` - RepoDID string `json:"repoDID"` - Livestreams []string `json:"livestreams"` - Objects int `json:"objects"` - Bytes int64 `json:"bytes"` - Publish bool `json:"publish"` - Title string `json:"title"` - Ended []string `json:"ended,omitempty"` - EndErrors []string `json:"endErrors,omitempty"` - TaskQueue string `json:"taskQueue"` -} - -// livestreamItem is one livestream record: the indexed row and its decoded -// record. -type livestreamItem struct { - ls *model.Livestream - rec *placestream.Livestream -} - -// HandleFinalizeLivestream (internal API, POST /finalize-livestream) turns a -// recorded livestream into a VOD and, by default, publishes it on the -// streamer's channel: the operator's counterpart of the app's Livestreams -// tab, for a stream whose streamer can't or won't click through (a client's -// event, a stream that was never stopped), and the only path that can join -// a recording split across livestream records. The records are written with -// the streamer's stored OAuth session, as the app path does; nothing here -// can act for a streamer who never signed in to this node. Loopback only, -// like the rest of the internal API. -func (a *StreamplaceAPI) HandleFinalizeLivestream(ctx context.Context) httprouter.Handle { - return func(w http.ResponseWriter, r *http.Request, _ httprouter.Params) { - var req finalizeLivestreamRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - errors.WriteHTTPBadRequest(w, "invalid request body", err) - return - } - var uris []string - if req.Livestream != "" { - uris = append(uris, req.Livestream) - } - for _, u := range req.Livestreams { - if u != "" && !contains(uris, u) { - uris = append(uris, u) - } - } - if len(uris) == 0 { - errors.WriteHTTPBadRequest(w, "livestream (at:// URI) or livestreams is required", nil) - return - } - items, herr := a.livestreamItems(uris) - if herr != nil { - herr.write(w) - return - } - repoDID := items[0].ls.RepoDID - ordered := make([]string, len(items)) - for i, it := range items { - ordered[i] = it.ls.URI - } - segs, err := a.StatefulDB.ListS3SegmentsForLivestreams(ctx, ordered) - if err != nil { - errors.WriteHTTPInternalServerError(w, "list recorded objects", err) - return - } - if len(segs) == 0 { - errors.WriteHTTPNotFound(w, "no completed recording objects for these livestreams", nil) - return - } - var total int64 - for _, s := range segs { - total += s.Size - } - if session, err := a.StatefulDB.GetSessionByDID(repoDID); err != nil || session == nil { - errors.WriteHTTPBadRequest(w, "the streamer has no stored session on this node; they must sign in once before their records can be written", err) - return - } - publish := req.Publish == nil || *req.Publish - uu, err := uuid.NewV7() - if err != nil { - errors.WriteHTTPInternalServerError(w, "uuid", err) - return - } - uploadID := uu.String() - if err := a.StatefulDB.CreateUpload(ctx, &statedb.Upload{ - ID: uploadID, - RepoDID: repoDID, - MimeType: "video/mp4", - Backend: "live", - Location: ordered[0], - }); err != nil { - errors.WriteHTTPInternalServerError(w, "create upload", err) - return - } - task := statedb.FinalizeLivestreamVODTask{UploadID: uploadID, RepoDID: repoDID, LivestreamURI: ordered[0], LivestreamURIs: ordered} - video := videoDraftForLivestreams(items, req.Title, req.Description) - if publish { - task.Publish = video - } - if _, err := a.StatefulDB.EnqueueTask(ctx, statedb.TaskFinalizeLivestreamVOD, task, statedb.WithTaskKey("finalize-vod:"+uploadID)); err != nil { - errors.WriteHTTPInternalServerError(w, "enqueue finalize task", err) - return - } - log.Log(ctx, "operator finalize: livestream VOD queued", "livestreams", ordered, "repoDID", repoDID, "uploadId", uploadID, "objects", len(segs), "bytes", total, "publish", publish) - resp := finalizeLivestreamResponse{UploadID: uploadID, RepoDID: repoDID, Livestreams: ordered, Objects: len(segs), Bytes: total, Publish: publish, Title: video.Title, TaskQueue: statedb.TaskFinalizeLivestreamVOD} - for _, it := range items { - if it.rec.EndedAt != nil { - resp.Ended = append(resp.Ended, it.ls.URI) - continue - } - if !req.EndLivestream { - continue - } - if err := a.StatefulDB.EndLivestreamRecord(ctx, it.ls, it.rec); err != nil { - log.Error(ctx, "operator finalize: could not end livestream record", "livestream", it.ls.URI, "error", err) - resp.EndErrors = append(resp.EndErrors, it.ls.URI+": "+err.Error()) - } else { - resp.Ended = append(resp.Ended, it.ls.URI) - } - } - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(resp); err != nil { - log.Error(ctx, "error writing response", "error", err) - } - } -} - -// httpError is a handler failure with the status it should be reported as. -type httpError struct { - status int - msg string - err error -} - -func (e *httpError) write(w http.ResponseWriter) { - switch e.status { - case http.StatusBadRequest: - errors.WriteHTTPBadRequest(w, e.msg, e.err) - case http.StatusNotFound: - errors.WriteHTTPNotFound(w, e.msg, e.err) - default: - errors.WriteHTTPInternalServerError(w, e.msg, e.err) - } -} - -// livestreamItems resolves livestream URIs to their indexed rows and decoded -// records, checks they belong to one streamer, and orders them by creation: -// recording order is record order, whatever order the caller listed them in. -func (a *StreamplaceAPI) livestreamItems(uris []string) ([]livestreamItem, *httpError) { - items := make([]livestreamItem, 0, len(uris)) - for _, u := range uris { - ls, err := a.Model.GetLivestream(u) - if err != nil { - return nil, &httpError{http.StatusInternalServerError, "get livestream " + u, err} - } - if ls == nil { - return nil, &httpError{http.StatusNotFound, "livestream not indexed on this node: " + u, nil} - } - view, err := ls.ToLivestreamView() - if err != nil { - return nil, &httpError{http.StatusInternalServerError, "decode livestream " + u, err} - } - rec, ok := view.Record.Val.(*placestream.Livestream) - if !ok { - return nil, &httpError{http.StatusInternalServerError, "record is not a place.stream.livestream: " + u, nil} - } - items = append(items, livestreamItem{ls: ls, rec: rec}) - } - for _, it := range items[1:] { - if it.ls.RepoDID != items[0].ls.RepoDID { - return nil, &httpError{http.StatusBadRequest, "the livestreams belong to different streamers", nil} - } - } - sort.SliceStable(items, func(i, j int) bool { return items[i].rec.CreatedAt < items[j].rec.CreatedAt }) - return items, nil -} - -func contains(list []string, v string) bool { - for _, x := range list { - if x == v { - return true - } - } - return false -} - -// videoDraftForLivestreams describes the place.stream.video record for the -// VOD of one or more livestream records: the first one's title (or the given -// one), description, tags and activity, connected to every livestream record -// the way the app's draft is connected to its one. Duration, source tracks -// and thumbnail are filled in at publish time from the finalized upload. -func videoDraftForLivestreams(items []livestreamItem, title, description string) *statedb.VideoDraft { - first := items[0].rec - title = strings.TrimSpace(title) - if title == "" { - title = strings.TrimSpace(first.Title) - } - if title == "" { - title = "Livestream" - } - v := &statedb.VideoDraft{ - Title: title, - Tags: first.Tags, - } - for _, it := range items { - v.Connections = append(v.Connections, placestream.Video_Connections_Elem{ - Video_Connection: &placestream.Video_Connection{ - LexiconTypeID: "place.stream.video#connection", - Ref: &comatproto.RepoStrongRef{Uri: it.ls.URI, Cid: it.ls.CID}, - }, - }) - } - if d := strings.TrimSpace(description); d != "" { - v.Description = &d - } - if first.Activity != nil { - switch { - case first.Activity.Defs_ActivityGame != nil: - v.Activity = &placestream.Video_Activity{Defs_ActivityGame: first.Activity.Defs_ActivityGame} - case first.Activity.Defs_ActivityLabel != nil: - v.Activity = &placestream.Video_Activity{Defs_ActivityLabel: first.Activity.Defs_ActivityLabel} - } - } - return v -} diff --git a/pkg/api/admin_video.go b/pkg/api/admin_video.go deleted file mode 100644 index 4a92cf0e5..000000000 --- a/pkg/api/admin_video.go +++ /dev/null @@ -1,277 +0,0 @@ -package api - -import ( - "context" - "encoding/json" - "errors" - "net/http" - "strings" - "time" - - "github.com/bluesky-social/indigo/atproto/syntax" - "github.com/julienschmidt/httprouter" - sperrors "stream.place/streamplace/pkg/errors" - "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/statedb" - "stream.place/streamplace/pkg/vod" -) - -// The operator's video CRUD on the internal API (loopback only): the -// counterpart of /finalize-livestream for everything after the finalize. -// -// GET /videos?repo= the repo's video records and its uploads -// POST /videos publish a video record for a finished upload -// PUT /videos retitle / redescribe / retag a record, or point -// it at an upload's tracks (repair or re-finalize) -// DELETE /videos delete a record (and, on request, its tracks) -// -// Every write is made with the streamer's stored session, as the finalize -// is; the content blob and the upload row are never touched, so a deleted -// record can be published again from the same upload. - -type videoListItem struct { - URI string `json:"uri"` - CID string `json:"cid"` - Title string `json:"title"` - DurationMs int64 `json:"durationMs"` - CreatedAt string `json:"createdAt"` - Tracks []string `json:"tracks"` - Connections []string `json:"connections,omitempty"` - Thumb bool `json:"thumb"` -} - -type uploadListItem struct { - UploadID string `json:"uploadId"` - Status string `json:"status"` - Error string `json:"error,omitempty"` - Backend string `json:"backend"` - Location string `json:"location"` - ContentCID string `json:"contentCid,omitempty"` - DurationMs int64 `json:"durationMs"` - BlobSize int64 `json:"blobSize"` - Tracks int `json:"tracks"` - CreatedAt time.Time `json:"createdAt"` -} - -type videoListResponse struct { - RepoDID string `json:"repoDID"` - Videos []videoListItem `json:"videos"` - Uploads []uploadListItem `json:"uploads"` -} - -// HandleListVideos (GET /videos?repo=) lists a streamer's video records -// (from their PDS) and this node's uploads for them: the ids the other -// routes take. -func (a *StreamplaceAPI) HandleListVideos(ctx context.Context) httprouter.Handle { - return func(w http.ResponseWriter, r *http.Request, _ httprouter.Params) { - did := strings.TrimSpace(r.URL.Query().Get("repo")) - if _, err := syntax.ParseDID(did); err != nil { - sperrors.WriteHTTPBadRequest(w, "repo must be a DID", err) - return - } - recs, err := vod.ListVideoRecords(ctx, a.StatefulDB, did) - if err != nil { - sperrors.WriteHTTPInternalServerError(w, "list video records", err) - return - } - uploads, err := a.StatefulDB.ListUploadsForRepo(ctx, did) - if err != nil { - sperrors.WriteHTTPInternalServerError(w, "list uploads", err) - return - } - resp := videoListResponse{RepoDID: did, Videos: []videoListItem{}, Uploads: []uploadListItem{}} - for _, rec := range recs { - item := videoListItem{URI: rec.URI, CID: rec.CID, Tracks: rec.TrackURIs()} - if item.Tracks == nil { - item.Tracks = []string{} - } - if rec.Video != nil { - item.Title = rec.Video.Title - item.DurationMs = rec.Video.DurationMs - item.CreatedAt = rec.Video.CreatedAt - item.Thumb = rec.Video.Thumb != nil - for _, c := range rec.Video.Connections { - if c.Video_Connection != nil && c.Video_Connection.Ref != nil { - item.Connections = append(item.Connections, c.Video_Connection.Ref.Uri) - } - } - } - resp.Videos = append(resp.Videos, item) - } - for _, u := range uploads { - item := uploadListItem{UploadID: u.ID, Status: u.ProcessingStatus, Error: u.ProcessingError, Backend: u.Backend, - Location: u.Location, ContentCID: u.ContentCID, DurationMs: u.DurationMS, BlobSize: u.BlobSize, CreatedAt: u.CreatedAt} - if u.TrackURIs != "" { - var refs []struct{ URI string } - if json.Unmarshal([]byte(u.TrackURIs), &refs) == nil { - item.Tracks = len(refs) - } - } - resp.Uploads = append(resp.Uploads, item) - } - writeJSON(w, http.StatusOK, resp) - } -} - -type publishVideoRequest struct { - // UploadID is a finished upload of the streamer's (a finalized - // livestream, or any processed upload). - UploadID string `json:"uploadId"` - // Title and Description of the record. Title falls back to the first - // connected livestream's title, then "Livestream". - Title string `json:"title"` - Description string `json:"description"` - // Livestreams are the livestream records to connect the video to (the - // records whose recording it is). When empty and the upload came from - // a livestream, that livestream. - Livestreams []string `json:"livestreams"` -} - -// HandlePublishVideo (POST /videos) publishes a place.stream.video record on -// the streamer's channel for a finished upload: what /finalize-livestream -// does after the finalize, on its own, for an upload finalized with -// publish=false, a publish that failed, or a record that was deleted. -func (a *StreamplaceAPI) HandlePublishVideo(ctx context.Context) httprouter.Handle { - return func(w http.ResponseWriter, r *http.Request, _ httprouter.Params) { - var req publishVideoRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - sperrors.WriteHTTPBadRequest(w, "invalid request body", err) - return - } - if req.UploadID == "" { - sperrors.WriteHTTPBadRequest(w, "uploadId is required", nil) - return - } - upload, err := a.StatefulDB.GetUpload(ctx, req.UploadID) - if err != nil { - sperrors.WriteHTTPInternalServerError(w, "get upload", err) - return - } - if upload == nil { - sperrors.WriteHTTPNotFound(w, "no such upload", nil) - return - } - if upload.ProcessingStatus != "done" { - sperrors.WriteHTTPBadRequest(w, "upload is not finished processing (status "+upload.ProcessingStatus+")", nil) - return - } - uris := req.Livestreams - if len(uris) == 0 && strings.HasPrefix(upload.Location, "at://") { - uris = []string{upload.Location} - } - var items []livestreamItem - if len(uris) > 0 { - var herr *httpError - items, herr = a.livestreamItems(uris) - if herr != nil { - herr.write(w) - return - } - if items[0].ls.RepoDID != upload.RepoDID { - sperrors.WriteHTTPBadRequest(w, "the livestreams belong to a different streamer than the upload", nil) - return - } - } - var draft *statedb.VideoDraft - if len(items) > 0 { - draft = videoDraftForLivestreams(items, req.Title, req.Description) - } else { - title := strings.TrimSpace(req.Title) - if title == "" { - title = "Video" - } - draft = &statedb.VideoDraft{Title: title} - if d := strings.TrimSpace(req.Description); d != "" { - draft.Description = &d - } - } - uri, cid, err := vod.PublishVideo(ctx, a.StatefulDB, a.PlaybackStore, upload.RepoDID, upload.ID, draft.Record()) - if err != nil { - sperrors.WriteHTTPInternalServerError(w, "publish video", err) - return - } - log.Log(ctx, "operator publish: video record published", "uri", uri, "cid", cid, "uploadId", upload.ID) - writeJSON(w, http.StatusOK, map[string]any{"uri": uri, "cid": cid, "uploadId": upload.ID, "title": draft.Title}) - } -} - -type updateVideoRequest struct { - URI string `json:"uri"` - // Absent fields are left alone; "" clears description, [] clears tags. - Title *string `json:"title"` - Description *string `json:"description"` - Tags []string `json:"tags"` - // UploadID makes the upload's tracks the record's source (publishing - // them if the upload has none yet) and its duration the record's. - UploadID string `json:"uploadId"` -} - -// HandleUpdateVideo (PUT /videos) rewrites a video record in place. -func (a *StreamplaceAPI) HandleUpdateVideo(ctx context.Context) httprouter.Handle { - return func(w http.ResponseWriter, r *http.Request, _ httprouter.Params) { - var req updateVideoRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - sperrors.WriteHTTPBadRequest(w, "invalid request body", err) - return - } - if req.URI == "" { - sperrors.WriteHTTPBadRequest(w, "uri is required", nil) - return - } - cid, err := vod.UpdateVideo(ctx, a.StatefulDB, a.PlaybackStore, req.URI, vod.VideoUpdate{ - Title: req.Title, Description: req.Description, Tags: req.Tags, UploadID: req.UploadID, - }) - if err != nil { - writeVideoError(w, "update video", err) - return - } - writeJSON(w, http.StatusOK, map[string]any{"uri": req.URI, "cid": cid}) - } -} - -type deleteVideoRequest struct { - URI string `json:"uri"` - // Tracks also deletes the track records the video's source points at. - Tracks bool `json:"tracks"` -} - -// HandleDeleteVideo (DELETE /videos) deletes a video record and, on request, -// its track records. The content blob and the upload stay. -func (a *StreamplaceAPI) HandleDeleteVideo(ctx context.Context) httprouter.Handle { - return func(w http.ResponseWriter, r *http.Request, _ httprouter.Params) { - var req deleteVideoRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - sperrors.WriteHTTPBadRequest(w, "invalid request body", err) - return - } - if req.URI == "" { - sperrors.WriteHTTPBadRequest(w, "uri is required", nil) - return - } - deleted, err := vod.DeleteVideo(ctx, a.StatefulDB, req.URI, req.Tracks) - if err != nil { - writeVideoError(w, "delete video", err) - return - } - writeJSON(w, http.StatusOK, map[string]any{"deleted": deleted}) - } -} - -func writeVideoError(w http.ResponseWriter, msg string, err error) { - switch { - case errors.Is(err, vod.ErrNotAVideo): - sperrors.WriteHTTPBadRequest(w, msg, err) - case errors.Is(err, vod.ErrUploadNotFound): - sperrors.WriteHTTPNotFound(w, msg+": no such upload for this streamer", err) - case errors.Is(err, vod.ErrUploadNotReady): - sperrors.WriteHTTPBadRequest(w, msg+": upload is not finished processing", err) - default: - sperrors.WriteHTTPInternalServerError(w, msg, err) - } -} - -func writeJSON(w http.ResponseWriter, status int, v any) { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(status) - _ = json.NewEncoder(w).Encode(v) -} diff --git a/pkg/api/api_internal.go b/pkg/api/api_internal.go index 965396755..75e958c1f 100644 --- a/pkg/api/api_internal.go +++ b/pkg/api/api_internal.go @@ -114,18 +114,6 @@ func (a *StreamplaceAPI) InternalHandler(ctx context.Context) (http.Handler, err // Pull a VOD blob from another Streamplace node into our playback // store and attest to it via place.stream.media.origin. router.POST("/vod-transfer", a.HandleVODTransfer(ctx)) - // Turn a recorded livestream into a VOD on the streamer's channel, on - // the operator's say-so (the streamer's own path is the app's - // Livestreams tab); see HandleFinalizeLivestream. - router.POST("/finalize-livestream", a.HandleFinalizeLivestream(ctx)) - // The operator's hands on a streamer's video records after the - // finalize: list them (with the node's uploads for the repo), publish - // one for a finished upload, retitle or repair one, delete one. - router.GET("/videos", a.HandleListVideos(ctx)) - router.POST("/videos", a.HandlePublishVideo(ctx)) - router.PUT("/videos", a.HandleUpdateVideo(ctx)) - router.DELETE("/videos", a.HandleDeleteVideo(ctx)) - // Rebuild the local media.origin index from our own server repo, for // blobs we host but never indexed (a dropped firehose event, or a // --secure node whose self-subscription never connected). diff --git a/pkg/moderation/permissions.go b/pkg/moderation/permissions.go index d78f8e873..9bde0f132 100644 --- a/pkg/moderation/permissions.go +++ b/pkg/moderation/permissions.go @@ -30,8 +30,15 @@ var ActionPermissions = map[string]string{ "createVodGate": PermissionVodCommentHide, "deleteVodGate": PermissionVodCommentHide, "updateLivestream": PermissionLivestreamManage, - "createPin": PermissionMessagePin, - "deletePin": PermissionMessagePin, + // A streamer's recordings and video records: what the app's Livestreams + // tab does, for a moderator the streamer trusts with their channel. + "finalizeLivestream": PermissionLivestreamManage, + "listVideos": PermissionLivestreamManage, + "publishVideo": PermissionLivestreamManage, + "updateVideo": PermissionLivestreamManage, + "deleteVideo": PermissionLivestreamManage, + "createPin": PermissionMessagePin, + "deletePin": PermissionMessagePin, } // PermissionChecker validates moderation permissions diff --git a/pkg/moderation/permissions_test.go b/pkg/moderation/permissions_test.go index e30907949..1509c1100 100644 --- a/pkg/moderation/permissions_test.go +++ b/pkg/moderation/permissions_test.go @@ -112,6 +112,9 @@ func TestActionPermissions_Mapping(t *testing.T) { require.Equal(t, PermissionBan, ActionPermissions["deleteBlock"]) require.Equal(t, PermissionHide, ActionPermissions["createGate"]) require.Equal(t, PermissionHide, ActionPermissions["deleteGate"]) + for _, action := range []string{"finalizeLivestream", "listVideos", "publishVideo", "updateVideo", "deleteVideo"} { + require.Equal(t, PermissionLivestreamManage, ActionPermissions[action], action) + } require.Equal(t, PermissionLivestreamManage, ActionPermissions["updateLivestream"]) } diff --git a/pkg/placestream/mediadeletevideo.go b/pkg/placestream/mediadeletevideo.go new file mode 100644 index 000000000..93052905d --- /dev/null +++ b/pkg/placestream/mediadeletevideo.go @@ -0,0 +1,74 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.media.deleteVideo + +package placestream + +import ( + "context" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +type MediaDeleteVideo_Input struct { + LexiconTypeID string `json:"$type,omitempty"` + // tracks: Also delete the video's track records. + Tracks *bool `json:"tracks,omitempty"` + Uri string `json:"uri"` +} + +// RecordTypeID implements glex.Record. +func (t *MediaDeleteVideo_Input) RecordTypeID() string { return "place.stream.media.deleteVideo" } + +func (t *MediaDeleteVideo_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.media.deleteVideo" + return glex.MarshalCBOR(w, &cp) +} + +func (t *MediaDeleteVideo_Input) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +type MediaDeleteVideo_Output struct { + LexiconTypeID string `json:"$type,omitempty"` + // deleted: Every record deleted, the video first. + Deleted []string `json:"deleted"` +} + +// RecordTypeID implements glex.Record. +func (t *MediaDeleteVideo_Output) RecordTypeID() string { return "place.stream.media.deleteVideo" } + +func (t *MediaDeleteVideo_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.media.deleteVideo" + return glex.MarshalCBOR(w, &cp) +} + +func (t *MediaDeleteVideo_Output) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +// MediaDeleteVideo calls the XRPC method "place.stream.media.deleteVideo". +// +// Delete a place.stream.video record and, on request, the track records its source points at. The content blob and the upload stay, so the record can be published again from the same upload. The caller must be the record's owner or hold livestream.manage from them; the records are deleted with the owner's session. +func MediaDeleteVideo(ctx context.Context, c glex.LexClient, input *MediaDeleteVideo_Input) (*MediaDeleteVideo_Output, error) { + var out MediaDeleteVideo_Output + + if err := c.LexDo(ctx, glex.Procedure, "application/json", "place.stream.media.deleteVideo", nil, input, &out); err != nil { + return nil, err + } + return &out, nil +} diff --git a/pkg/placestream/mediafinalizelivestream.go b/pkg/placestream/mediafinalizelivestream.go index 37d088876..ffa25468e 100644 --- a/pkg/placestream/mediafinalizelivestream.go +++ b/pkg/placestream/mediafinalizelivestream.go @@ -14,8 +14,18 @@ import ( type MediaFinalizeLivestream_Input struct { LexiconTypeID string `json:"$type,omitempty"` - // livestream: AT-URI of the place.stream.livestream record to finalize into a VOD. Must belong to the authenticated user. - Livestream string `json:"livestream"` + // description: Description of the video record. + Description *string `json:"description,omitempty"` + // endLivestream: Also end any of the livestream records that were never stopped, so the streamer's page stops reading as live. + EndLivestream *bool `json:"endLivestream,omitempty"` + // livestream: AT-URI of the place.stream.livestream record to finalize into a VOD. Either this or livestreams is required. + Livestream *string `json:"livestream,omitempty"` + // livestreams: Several livestream records whose recordings make up one VOD: a streamer who started a new record mid-stream splits the recording across them. Recordings are concatenated in the records' creation order, whatever order they are listed in. All must belong to one streamer. + Livestreams []string `json:"livestreams,omitempty"` + // publish: Publish the place.stream.video record as soon as the VOD is finalized, instead of leaving a draft. + Publish *bool `json:"publish,omitempty"` + // title: Title of the video record; the first livestream's title when empty. + Title *string `json:"title,omitempty"` } // RecordTypeID implements glex.Record. @@ -40,8 +50,14 @@ func (t *MediaFinalizeLivestream_Input) UnmarshalCBOR(r io.Reader) error { type MediaFinalizeLivestream_Output struct { LexiconTypeID string `json:"$type,omitempty"` - // draftUri: The ats:// URI of the draft VOD created for this finalize. The draft reaches status 'ready' when processing completes; the user publishes it from the Drafts tab via place.stream.vod.publishDraft. - DraftUri string `json:"draftUri"` + // draftUri: The ats:// URI of the draft VOD created for this finalize, when publish is not set. The draft reaches status 'ready' when processing completes. + DraftUri *string `json:"draftUri,omitempty"` + // ended: Livestream records that are now ended (already, or by endLivestream). + Ended []string `json:"ended,omitempty"` + // livestreams: The livestream records the VOD is made of, in recording order. + Livestreams []string `json:"livestreams,omitempty"` + // objects: How many recorded objects will be concatenated. + Objects *int64 `json:"objects,omitempty"` // uploadId: Identifier for the finalize job. Retained for backwards compatibility; the draft flow no longer requires the client to poll getUploadStatus. UploadId string `json:"uploadId"` } @@ -68,7 +84,7 @@ func (t *MediaFinalizeLivestream_Output) UnmarshalCBOR(r io.Reader) error { // MediaFinalizeLivestream calls the XRPC method "place.stream.media.finalizeLivestream". // -// Turn a finished livestream into a VOD. The server concatenates the MUXL segments it recorded for the livestream into a single content blob, derives the playback sidecars, and publishes the place.stream.media.track records. Processing completes server-side and creates a draft VOD; the client does not need to poll or publish — the user publishes the draft later via place.stream.vod.publishDraft. Returns the draft's ats:// URI so the client can navigate to it, alongside an uploadId for backwards compatibility. +// Turn a finished livestream into a VOD. The server concatenates the MUXL segments it recorded for the livestream(s) into a single content blob, derives the playback sidecars, and publishes the place.stream.media.track records. By default it leaves a draft VOD the streamer publishes later via place.stream.vod.publishDraft; with publish set it publishes the place.stream.video record as soon as the VOD is finalized. The caller must be the streamer, or a moderator they have granted livestream.manage; every record is written with the streamer's session. func MediaFinalizeLivestream(ctx context.Context, c glex.LexClient, input *MediaFinalizeLivestream_Input) (*MediaFinalizeLivestream_Output, error) { var out MediaFinalizeLivestream_Output diff --git a/pkg/placestream/medialistvideos.go b/pkg/placestream/medialistvideos.go new file mode 100644 index 000000000..20c8f3ee5 --- /dev/null +++ b/pkg/placestream/medialistvideos.go @@ -0,0 +1,125 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.media.listVideos + +package placestream + +import ( + "context" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +type MediaListVideos_Output struct { + LexiconTypeID string `json:"$type,omitempty"` + RepoDID string `json:"repoDID"` + Uploads []MediaListVideos_Upload `json:"uploads"` + Videos []MediaListVideos_Video `json:"videos"` +} + +// RecordTypeID implements glex.Record. +func (t *MediaListVideos_Output) RecordTypeID() string { return "place.stream.media.listVideos" } + +func (t *MediaListVideos_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.media.listVideos" + return glex.MarshalCBOR(w, &cp) +} + +func (t *MediaListVideos_Output) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +// MediaListVideos calls the XRPC method "place.stream.media.listVideos". +// +// A streamer's place.stream.video records (read from their PDS) and this node's uploads for them: everything updateVideo, deleteVideo and publishVideo take. The caller must be the streamer or hold livestream.manage from them. +// +// repo: The streamer's DID. +func MediaListVideos(ctx context.Context, c glex.LexClient, repo string) (*MediaListVideos_Output, error) { + var out MediaListVideos_Output + + params := map[string]interface{}{} + params["repo"] = repo + + if err := c.LexDo(ctx, glex.Query, "", "place.stream.media.listVideos", params, nil, &out); err != nil { + return nil, err + } + return &out, nil +} + +// MediaListVideos_Upload is a "upload" in the place.stream.media.listVideos schema. +type MediaListVideos_Upload struct { + LexiconTypeID string `json:"$type,omitempty"` + Backend string `json:"backend"` + BlobSize *int64 `json:"blobSize,omitempty"` + ContentCid *string `json:"contentCid,omitempty"` + CreatedAt *string `json:"createdAt,omitempty"` + DurationMs *int64 `json:"durationMs,omitempty"` + Error *string `json:"error,omitempty"` + // location: For a finalized recording, the livestream record it came from. + Location *string `json:"location,omitempty"` + // status: Processing status: pending, processing, done or error. + Status string `json:"status"` + // tracks: How many track records the upload has published. + Tracks *int64 `json:"tracks,omitempty"` + UploadId string `json:"uploadId"` +} + +// RecordTypeID implements glex.Record. +func (t *MediaListVideos_Upload) RecordTypeID() string { return "place.stream.media.listVideos#upload" } + +func (t *MediaListVideos_Upload) 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.media.listVideos#upload" + return glex.MarshalCBOR(w, &cp) +} + +func (t *MediaListVideos_Upload) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +// MediaListVideos_Video is a "video" in the place.stream.media.listVideos schema. +type MediaListVideos_Video struct { + LexiconTypeID string `json:"$type,omitempty"` + Cid string `json:"cid"` + // connections: The livestream records the video is connected to. + Connections []string `json:"connections,omitempty"` + CreatedAt *string `json:"createdAt,omitempty"` + DurationMs *int64 `json:"durationMs,omitempty"` + // thumb: Whether the record carries a thumbnail. + Thumb *bool `json:"thumb,omitempty"` + Title string `json:"title"` + // tracks: The track records the video's source points at. + Tracks []string `json:"tracks"` + Uri string `json:"uri"` +} + +// RecordTypeID implements glex.Record. +func (t *MediaListVideos_Video) RecordTypeID() string { return "place.stream.media.listVideos#video" } + +func (t *MediaListVideos_Video) 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.media.listVideos#video" + return glex.MarshalCBOR(w, &cp) +} + +func (t *MediaListVideos_Video) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} diff --git a/pkg/placestream/mediapublishvideo.go b/pkg/placestream/mediapublishvideo.go index bd953f17f..c646333d0 100644 --- a/pkg/placestream/mediapublishvideo.go +++ b/pkg/placestream/mediapublishvideo.go @@ -14,8 +14,14 @@ import ( type MediaPublishVideo_Input struct { LexiconTypeID string `json:"$type,omitempty"` + // description: Description for a record built by the server. + Description *string `json:"description,omitempty"` + // livestreams: Livestream records the video came from, used to build the record when none is given; defaults to the upload's own livestream for a finalized recording. + Livestreams []string `json:"livestreams,omitempty"` // record: A place.stream.video record. The server overrides `source` and `durationMs` from the processed upload, and fills in `thumb` with a generated thumbnail when the supplied record has none. - Record Video `json:"record"` + Record *Video `json:"record,omitempty"` + // title: Title for a record built by the server; the first livestream's title when empty. + Title *string `json:"title,omitempty"` // uploadId: The upload ID returned by place.stream.media.createUpload. Its processing must be complete (status 'done'). UploadId string `json:"uploadId"` } @@ -66,7 +72,7 @@ func (t *MediaPublishVideo_Output) UnmarshalCBOR(r io.Reader) error { // MediaPublishVideo calls the XRPC method "place.stream.media.publishVideo". // -// Publish a place.stream.video record for a finished upload, server-side. The caller supplies the record it would otherwise putRecord itself; the server overrides the fields it is authoritative about (source tracks and durationMs, taken from the processed upload) and, if the record carries no thumb, generates one from the video and attaches it. The record is written to the authenticated user's repo. +// Publish a place.stream.video record for a finished upload, server-side. The caller supplies the record it would otherwise putRecord itself; the server overrides the fields it is authoritative about (source tracks and durationMs, taken from the processed upload) and, if the record carries no thumb, generates one from the video and attaches it. The record is written to the authenticated user's repo. The caller must own the upload, or hold livestream.manage from its owner; the record is written with the owner's session. Without a record, one is built from the upload's livestream(s) (title, tags, activity, connections) and the given title and description. func MediaPublishVideo(ctx context.Context, c glex.LexClient, input *MediaPublishVideo_Input) (*MediaPublishVideo_Output, error) { var out MediaPublishVideo_Output diff --git a/pkg/placestream/mediaupdatevideo.go b/pkg/placestream/mediaupdatevideo.go new file mode 100644 index 000000000..283ebac3d --- /dev/null +++ b/pkg/placestream/mediaupdatevideo.go @@ -0,0 +1,78 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.media.updateVideo + +package placestream + +import ( + "context" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +type MediaUpdateVideo_Input struct { + LexiconTypeID string `json:"$type,omitempty"` + Description *string `json:"description,omitempty"` + Tags []string `json:"tags,omitempty"` + Title *string `json:"title,omitempty"` + // uploadId: A finished upload whose tracks become the record's source and whose duration the record's. + UploadId *string `json:"uploadId,omitempty"` + // uri: The place.stream.video record. + Uri string `json:"uri"` +} + +// RecordTypeID implements glex.Record. +func (t *MediaUpdateVideo_Input) RecordTypeID() string { return "place.stream.media.updateVideo" } + +func (t *MediaUpdateVideo_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.media.updateVideo" + return glex.MarshalCBOR(w, &cp) +} + +func (t *MediaUpdateVideo_Input) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +type MediaUpdateVideo_Output struct { + LexiconTypeID string `json:"$type,omitempty"` + Cid string `json:"cid"` + Uri string `json:"uri"` +} + +// RecordTypeID implements glex.Record. +func (t *MediaUpdateVideo_Output) RecordTypeID() string { return "place.stream.media.updateVideo" } + +func (t *MediaUpdateVideo_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.media.updateVideo" + return glex.MarshalCBOR(w, &cp) +} + +func (t *MediaUpdateVideo_Output) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +// MediaUpdateVideo calls the XRPC method "place.stream.media.updateVideo". +// +// Rewrite a place.stream.video record in place: retitle, redescribe or retag it, or point it at a finished upload's tracks (publishing them if the upload has none yet: the repair for a record published without tracks, or the way to point a record at a re-finalized VOD). Fields left out are kept; an empty description or tag list clears it. The caller must be the record's owner or hold livestream.manage from them; the record is written with the owner's session. +func MediaUpdateVideo(ctx context.Context, c glex.LexClient, input *MediaUpdateVideo_Input) (*MediaUpdateVideo_Output, error) { + var out MediaUpdateVideo_Output + + if err := c.LexDo(ctx, glex.Procedure, "application/json", "place.stream.media.updateVideo", nil, input, &out); err != nil { + return nil, err + } + return &out, nil +} diff --git a/pkg/spxrpc/media_manage.go b/pkg/spxrpc/media_manage.go new file mode 100644 index 000000000..4dac84ac8 --- /dev/null +++ b/pkg/spxrpc/media_manage.go @@ -0,0 +1,286 @@ +package spxrpc + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "sort" + "strings" + + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/labstack/echo/v4" + "stream.place/streamplace/pkg/comatproto" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/model" + "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/statedb" + "stream.place/streamplace/pkg/vod" +) + +// A streamer's recordings and video records, through XRPC, for the streamer +// and for the moderators they have granted livestream.manage: finalize a +// recording into a VOD (place.stream.media.finalizeLivestream), publish a +// record for a finished upload (publishVideo), list the records and this +// node's uploads (listVideos), retitle or repair a record (updateVideo), +// delete one (deleteVideo). Every write goes through the streamer's stored +// session, the way delegated moderation does, so nothing here can act for a +// streamer who never signed in to this node. + +// requireLivestreamManage is the authorization every handler here shares: +// the caller must be streamer, or hold livestream.manage from them; the +// streamer must have a stored session for the writes. +func (s *Server) requireLivestreamManage(ctx context.Context, streamer, action string) (*DelegatedModerationContext, error) { + modCtx, err := s.GetDelegatedModerationContext(ctx, streamer, action) + if err != nil { + var he *echo.HTTPError + if errors.As(err, &he) && he.Code == http.StatusForbidden { + return nil, echo.NewHTTPError(http.StatusForbidden, "NotPermitted: the caller is neither the streamer nor a moderator with livestream.manage from them") + } + return nil, err + } + return modCtx, nil +} + +// livestreamItem is one livestream record: the indexed row and its decoded +// record. +type livestreamItem struct { + ls *model.Livestream + rec *placestream.Livestream +} + +// livestreamItems resolves livestream URIs to their indexed rows and decoded +// records, checks they belong to one streamer, and orders them by creation: +// recording order is record order, whatever order the caller listed them in. +func (s *Server) livestreamItems(uris []string) ([]livestreamItem, error) { + items := make([]livestreamItem, 0, len(uris)) + for _, u := range uris { + ls, err := s.model.GetLivestream(u) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "get livestream "+u+": "+err.Error()) + } + if ls == nil { + return nil, echo.NewHTTPError(http.StatusNotFound, "LivestreamNotFound: not indexed on this node: "+u) + } + view, err := ls.ToLivestreamView() + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "decode livestream "+u+": "+err.Error()) + } + rec, ok := view.Record.Val.(*placestream.Livestream) + if !ok { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "record is not a place.stream.livestream: "+u) + } + items = append(items, livestreamItem{ls: ls, rec: rec}) + } + for _, it := range items[1:] { + if it.ls.RepoDID != items[0].ls.RepoDID { + return nil, echo.NewHTTPError(http.StatusNotFound, "LivestreamNotFound: the livestreams belong to different streamers") + } + } + sort.SliceStable(items, func(i, j int) bool { return items[i].rec.CreatedAt < items[j].rec.CreatedAt }) + return items, nil +} + +// uniqueURIs merges the single and plural forms of a livestream argument. +func uniqueURIs(one *string, many []string) []string { + var out []string + seen := map[string]bool{} + add := func(u string) { + u = strings.TrimSpace(u) + if u != "" && !seen[u] { + seen[u] = true + out = append(out, u) + } + } + if one != nil { + add(*one) + } + for _, u := range many { + add(u) + } + return out +} + +func deref(p *string) string { + if p == nil { + return "" + } + return *p +} + +// videoDraftForLivestreams describes the place.stream.video record for the +// VOD of one or more livestream records: the first one's title (or the given +// one), description, tags and activity, connected to every livestream record +// the way the app's draft is connected to its one. Duration, source tracks +// and thumbnail are filled in at publish time from the finalized upload. +func videoDraftForLivestreams(items []livestreamItem, title, description string) *statedb.VideoDraft { + first := items[0].rec + title = strings.TrimSpace(title) + if title == "" { + title = strings.TrimSpace(first.Title) + } + if title == "" { + title = "Livestream" + } + v := &statedb.VideoDraft{ + Title: title, + Tags: first.Tags, + } + for _, it := range items { + v.Connections = append(v.Connections, placestream.Video_Connections_Elem{ + Video_Connection: &placestream.Video_Connection{ + LexiconTypeID: "place.stream.video#connection", + Ref: &comatproto.RepoStrongRef{Uri: it.ls.URI, Cid: it.ls.CID}, + }, + }) + } + if d := strings.TrimSpace(description); d != "" { + v.Description = &d + } + if first.Activity != nil { + switch { + case first.Activity.Defs_ActivityGame != nil: + v.Activity = &placestream.Video_Activity{Defs_ActivityGame: first.Activity.Defs_ActivityGame} + case first.Activity.Defs_ActivityLabel != nil: + v.Activity = &placestream.Video_Activity{Defs_ActivityLabel: first.Activity.Defs_ActivityLabel} + } + } + return v +} + +// videoOwner is the repo a place.stream.video URI lives in. +func videoOwner(uri string) (string, error) { + aturi, err := syntax.ParseATURI(uri) + if err != nil || aturi.Collection().String() != "place.stream.video" { + return "", echo.NewHTTPError(http.StatusBadRequest, "uri must be a place.stream.video record") + } + return aturi.Authority().String(), nil +} + +// videoError maps the vod package's failures onto XRPC errors. +func videoError(what string, err error) error { + switch { + case errors.Is(err, vod.ErrNotAVideo): + return echo.NewHTTPError(http.StatusBadRequest, what+": "+err.Error()) + case errors.Is(err, vod.ErrUploadNotFound): + return echo.NewHTTPError(http.StatusNotFound, what+": no such upload for this streamer") + case errors.Is(err, vod.ErrUploadNotReady): + return echo.NewHTTPError(http.StatusConflict, what+": upload is not finished processing") + case strings.Contains(strings.ToLower(err.Error()), "could not locate record") || strings.Contains(strings.ToLower(err.Error()), "record not found"): + return echo.NewHTTPError(http.StatusNotFound, "VideoNotFound: "+err.Error()) + } + return echo.NewHTTPError(http.StatusInternalServerError, what+": "+err.Error()) +} + +// handlePlaceStreamMediaListVideos lists a streamer's video records and this +// node's uploads for them. +func (s *Server) handlePlaceStreamMediaListVideos(ctx context.Context, repo string) (*placestream.MediaListVideos_Output, error) { + did := strings.TrimSpace(repo) + if _, err := syntax.ParseDID(did); err != nil { + return nil, echo.NewHTTPError(http.StatusBadRequest, "repo must be a DID") + } + if _, err := s.requireLivestreamManage(ctx, did, "listVideos"); err != nil { + return nil, err + } + recs, err := vod.ListVideoRecords(ctx, s.statefulDB, did) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "list video records: "+err.Error()) + } + uploads, err := s.statefulDB.ListUploadsForRepo(ctx, did) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "list uploads: "+err.Error()) + } + out := &placestream.MediaListVideos_Output{RepoDID: did, Videos: []placestream.MediaListVideos_Video{}, Uploads: []placestream.MediaListVideos_Upload{}} + for _, rec := range recs { + item := placestream.MediaListVideos_Video{Uri: rec.URI, Cid: rec.CID, Tracks: rec.TrackURIs()} + if item.Tracks == nil { + item.Tracks = []string{} + } + if rec.Video != nil { + item.Title = rec.Video.Title + d := rec.Video.DurationMs + item.DurationMs = &d + created := rec.Video.CreatedAt + item.CreatedAt = &created + thumb := rec.Video.Thumb != nil + item.Thumb = &thumb + for _, c := range rec.Video.Connections { + if c.Video_Connection != nil && c.Video_Connection.Ref != nil { + item.Connections = append(item.Connections, c.Video_Connection.Ref.Uri) + } + } + } + out.Videos = append(out.Videos, item) + } + for _, u := range uploads { + item := placestream.MediaListVideos_Upload{UploadId: u.ID, Status: u.ProcessingStatus, Backend: u.Backend} + if u.ProcessingError != "" { + e := u.ProcessingError + item.Error = &e + } + if u.Location != "" { + l := u.Location + item.Location = &l + } + if u.ContentCID != "" { + c := u.ContentCID + item.ContentCid = &c + } + dur, size := u.DurationMS, u.BlobSize + item.DurationMs, item.BlobSize = &dur, &size + created := u.CreatedAt.UTC().Format("2006-01-02T15:04:05.000Z") + item.CreatedAt = &created + if u.TrackURIs != "" { + var refs []struct{ URI string } + if json.Unmarshal([]byte(u.TrackURIs), &refs) == nil { + n := int64(len(refs)) + item.Tracks = &n + } + } + out.Uploads = append(out.Uploads, item) + } + return out, nil +} + +// handlePlaceStreamMediaUpdateVideo rewrites a video record in place. +func (s *Server) handlePlaceStreamMediaUpdateVideo(ctx context.Context, body *placestream.MediaUpdateVideo_Input) (*placestream.MediaUpdateVideo_Output, error) { + owner, err := videoOwner(body.Uri) + if err != nil { + return nil, err + } + if _, err := s.requireLivestreamManage(ctx, owner, "updateVideo"); err != nil { + return nil, err + } + if s.playbackStore == nil { + return nil, echo.NewHTTPError(http.StatusServiceUnavailable, "playback store not configured") + } + cid, err := vod.UpdateVideo(ctx, s.statefulDB, s.playbackStore, body.Uri, vod.VideoUpdate{ + Title: body.Title, Description: body.Description, Tags: body.Tags, UploadID: deref(body.UploadId), + }) + if err != nil { + return nil, videoError("update video", err) + } + log.Log(ctx, "video record updated", "uri", body.Uri, "cid", cid) + return &placestream.MediaUpdateVideo_Output{Uri: body.Uri, Cid: cid}, nil +} + +// handlePlaceStreamMediaDeleteVideo deletes a video record and, on request, +// its track records. +func (s *Server) handlePlaceStreamMediaDeleteVideo(ctx context.Context, body *placestream.MediaDeleteVideo_Input) (*placestream.MediaDeleteVideo_Output, error) { + owner, err := videoOwner(body.Uri) + if err != nil { + return nil, err + } + if _, err := s.requireLivestreamManage(ctx, owner, "deleteVideo"); err != nil { + return nil, err + } + deleted, err := vod.DeleteVideo(ctx, s.statefulDB, body.Uri, body.Tracks != nil && *body.Tracks) + if err != nil { + return nil, videoError("delete video", err) + } + log.Log(ctx, "video record deleted", "uri", body.Uri, "deleted", deleted) + if deleted == nil { + deleted = []string{} + } + return &placestream.MediaDeleteVideo_Output{Deleted: deleted}, nil +} diff --git a/pkg/api/admin_livestream_test.go b/pkg/spxrpc/media_manage_test.go similarity index 85% rename from pkg/api/admin_livestream_test.go rename to pkg/spxrpc/media_manage_test.go index d7b8e61ba..e8efde263 100644 --- a/pkg/api/admin_livestream_test.go +++ b/pkg/spxrpc/media_manage_test.go @@ -1,4 +1,4 @@ -package api +package spxrpc import ( "encoding/json" @@ -66,3 +66,19 @@ func TestVideoDraftPayloadRoundTrip(t *testing.T) { require.Equal(t, a.ls.CID, rec.Connections[0].Video_Connection.Ref.Cid) require.Equal(t, "place.stream.video", rec.RecordTypeID()) } + +func TestUniqueURIs(t *testing.T) { + one := " at://a " + require.Equal(t, []string{"at://a", "at://b"}, uniqueURIs(&one, []string{"at://b", "", "at://a"})) + require.Nil(t, uniqueURIs(nil, nil)) +} + +func TestVideoOwner(t *testing.T) { + did, err := videoOwner("at://did:plc:abc/place.stream.video/3k") + require.NoError(t, err) + require.Equal(t, "did:plc:abc", did) + _, err = videoOwner("at://did:plc:abc/place.stream.livestream/3k") + require.Error(t, err) + _, err = videoOwner("nope") + require.Error(t, err) +} diff --git a/pkg/spxrpc/place_stream_media_finalizelivestream.go b/pkg/spxrpc/place_stream_media_finalizelivestream.go index d6d4d9c1b..e58e2a7e1 100644 --- a/pkg/spxrpc/place_stream_media_finalizelivestream.go +++ b/pkg/spxrpc/place_stream_media_finalizelivestream.go @@ -3,47 +3,58 @@ package spxrpc import ( "context" "net/http" + "strings" "github.com/google/uuid" "github.com/labstack/echo/v4" - "github.com/streamplace/oatproxy/pkg/oatproxy" "stream.place/streamplace/pkg/log" placestream "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/statedb" ) -// handlePlaceStreamMediaFinalizeLivestream turns a finished livestream into a -// VOD. It creates a synthetic Upload row and enqueues a background finalize -// task that concatenates the livestream's recorded MUXL objects into a content -// blob and publishes the track records. It also creates a draft VOD in the -// 'processing' state (inheriting the livestream's metadata) that the user -// publishes later from the Drafts tab via place.stream.vod.publishDraft — the -// client no longer polls getUploadStatus or calls publishVideo. Returns the -// draft's ats:// URI so the client can navigate to it. +// handlePlaceStreamMediaFinalizeLivestream turns a recorded livestream (or +// several records of one recording) into a VOD. It creates a synthetic +// Upload row and enqueues a background finalize task that concatenates the +// recorded MUXL objects into a content blob and publishes the track +// records. By default it also creates a draft VOD in the 'processing' state +// that the streamer publishes later from the Drafts tab; with publish set, +// the task publishes the place.stream.video record itself as soon as the +// VOD is finalized, with the streamer's stored session. The caller is the +// streamer or a moderator with livestream.manage from them. func (s *Server) handlePlaceStreamMediaFinalizeLivestream(ctx context.Context, body *placestream.MediaFinalizeLivestream_Input) (*placestream.MediaFinalizeLivestream_Output, error) { - session, _ := oatproxy.GetOAuthSession(ctx) - if session == nil { - return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session required") + uris := uniqueURIs(body.Livestream, body.Livestreams) + if len(uris) == 0 { + return nil, echo.NewHTTPError(http.StatusBadRequest, "livestream or livestreams is required") } - ctx = log.WithLogValues(ctx, "func", "finalizeLivestream", "did", session.DID, "livestream", body.Livestream) - if body.Livestream == "" { - return nil, echo.NewHTTPError(http.StatusBadRequest, "livestream is required") + items, err := s.livestreamItems(uris) + if err != nil { + return nil, err + } + streamer := items[0].ls.RepoDID + ordered := make([]string, len(items)) + for i, it := range items { + ordered[i] = it.ls.URI + } + modCtx, err := s.requireLivestreamManage(ctx, streamer, "finalizeLivestream") + if err != nil { + return nil, err } + ctx = log.WithLogValues(ctx, "func", "finalizeLivestream", "did", streamer, "by", modCtx.ModeratorDID, "livestreams", strings.Join(ordered, ",")) + // A banned account can't mint new VODs (the playback gates would hide them // regardless, but skip the work and the dead records). - if banned, err := s.accountBanned(session.DID); err != nil { + if banned, err := s.accountBanned(streamer); err != nil { return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) } else if banned { return nil, echo.NewHTTPError(http.StatusForbidden, "account is not permitted to publish videos") } - ls, err := s.model.GetLivestream(body.Livestream) + segs, err := s.statefulDB.ListS3SegmentsForLivestreams(ctx, ordered) if err != nil { - return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) + return nil, echo.NewHTTPError(http.StatusInternalServerError, "list recorded objects: "+err.Error()) } - if ls == nil || ls.RepoDID != session.DID { - // Don't distinguish "not found" from "not yours" — same response. - return nil, echo.NewHTTPError(http.StatusNotFound, "livestream not found") + if len(segs) == 0 { + return nil, echo.NewHTTPError(http.StatusNotFound, "NoRecording: no completed recording objects for these livestreams") } // Synthetic Upload row so the client reuses the getUploadStatus / @@ -55,34 +66,52 @@ func (s *Server) handlePlaceStreamMediaFinalizeLivestream(ctx context.Context, b uploadID := uu.String() if err := s.statefulDB.CreateUpload(ctx, &statedb.Upload{ ID: uploadID, - RepoDID: session.DID, + RepoDID: streamer, MimeType: "video/mp4", Backend: "live", - Location: body.Livestream, + Location: ordered[0], }); err != nil { return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) } - // Create a draft VOD in the 'processing' state, inheriting the livestream's - // metadata and linking back to it via connections. The client no longer polls - // or publishes — the draft reaches 'ready' server-side and the user - // publishes it later from the Drafts tab. - draft, err := s.createLivestreamDraft(ctx, session.DID, uploadID, ls) - if err != nil { - return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) - } - - if _, err := s.statefulDB.EnqueueTask(ctx, statedb.TaskFinalizeLivestreamVOD, statedb.FinalizeLivestreamVODTask{ - UploadID: uploadID, - RepoDID: session.DID, - LivestreamURI: body.Livestream, - }, statedb.WithTaskKey("finalize-vod:"+uploadID)); err != nil { - _, _ = s.statefulDB.DeleteDraft(ctx, draft.URI) - return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) + publish := body.Publish != nil && *body.Publish + task := statedb.FinalizeLivestreamVODTask{UploadID: uploadID, RepoDID: streamer, LivestreamURI: ordered[0], LivestreamURIs: ordered} + out := &placestream.MediaFinalizeLivestream_Output{UploadId: uploadID, Livestreams: ordered} + objects := int64(len(segs)) + out.Objects = &objects + if publish { + task.Publish = videoDraftForLivestreams(items, deref(body.Title), deref(body.Description)) + } else { + // A draft VOD in the 'processing' state, inheriting the first + // livestream's metadata and linking back to it; it reaches 'ready' + // server-side and the streamer publishes it from the Drafts tab. + draft, err := s.createLivestreamDraft(ctx, streamer, uploadID, items[0].ls) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) + } + out.DraftUri = &draft.URI } - if err != nil { + if _, err := s.statefulDB.EnqueueTask(ctx, statedb.TaskFinalizeLivestreamVOD, task, statedb.WithTaskKey("finalize-vod:"+uploadID)); err != nil { + if out.DraftUri != nil { + _, _ = s.statefulDB.DeleteDraft(ctx, *out.DraftUri) + } return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) } + log.Log(ctx, "livestream VOD queued", "uploadId", uploadID, "objects", len(segs), "publish", publish) - return &placestream.MediaFinalizeLivestream_Output{UploadId: uploadID, DraftUri: draft.URI}, nil + for _, it := range items { + if it.rec.EndedAt != nil { + out.Ended = append(out.Ended, it.ls.URI) + continue + } + if body.EndLivestream == nil || !*body.EndLivestream { + continue + } + if err := s.statefulDB.EndLivestreamRecord(ctx, it.ls, it.rec); err != nil { + log.Error(ctx, "could not end livestream record", "livestream", it.ls.URI, "error", err) + } else { + out.Ended = append(out.Ended, it.ls.URI) + } + } + return out, nil } diff --git a/pkg/spxrpc/place_stream_media_publishvideo.go b/pkg/spxrpc/place_stream_media_publishvideo.go index fbc01fd29..55ccd63bb 100644 --- a/pkg/spxrpc/place_stream_media_publishvideo.go +++ b/pkg/spxrpc/place_stream_media_publishvideo.go @@ -2,44 +2,83 @@ package spxrpc import ( "context" - "errors" "net/http" + "strings" "github.com/labstack/echo/v4" - "github.com/streamplace/oatproxy/pkg/oatproxy" "stream.place/streamplace/pkg/log" placestream "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/statedb" "stream.place/streamplace/pkg/vod" ) // handlePlaceStreamMediaPublishVideo creates a place.stream.video record in // the authenticated user's repo for a finished upload. The client hands us // the record it would otherwise putRecord itself; the server overrides the -// authoritative fields (source tracks + durationMs) from the processed -// upload and backfills a generated thumbnail when the record carries none. + +// handlePlaceStreamMediaPublishVideo publishes a place.stream.video record +// for a finished upload of the streamer's: the record the caller supplies, +// or one built from the upload's livestream(s) and the given title and +// description (what the finalize does after the finalize, on its own, for an +// upload finalized without publish, a publish that failed, or a record that +// was deleted). The caller is the upload's owner or a moderator with +// livestream.manage from them; the record is written with the owner's +// session. func (s *Server) handlePlaceStreamMediaPublishVideo(ctx context.Context, body *placestream.MediaPublishVideo_Input) (*placestream.MediaPublishVideo_Output, error) { - session, _ := oatproxy.GetOAuthSession(ctx) - if session == nil { - return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session required") - } - ctx = log.WithLogValues(ctx, "func", "publishVideo", "did", session.DID, "uploadId", body.UploadId) if body.UploadId == "" { return nil, echo.NewHTTPError(http.StatusBadRequest, "uploadId is required") } if s.playbackStore == nil { return nil, echo.NewHTTPError(http.StatusServiceUnavailable, "playback store not configured") } - - uri, cid, err := vod.PublishVideo(ctx, s.statefulDB, s.playbackStore, session.DID, body.UploadId, &body.Record) + upload, err := s.statefulDB.GetUpload(ctx, body.UploadId) if err != nil { - switch { - case errors.Is(err, vod.ErrUploadNotFound): - return nil, echo.NewHTTPError(http.StatusNotFound, "upload not found") - case errors.Is(err, vod.ErrUploadNotReady): - return nil, echo.NewHTTPError(http.StatusConflict, "upload is still processing") - default: - return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) + return nil, echo.NewHTTPError(http.StatusInternalServerError, "get upload: "+err.Error()) + } + if upload == nil { + return nil, echo.NewHTTPError(http.StatusNotFound, "upload not found") + } + modCtx, err := s.requireLivestreamManage(ctx, upload.RepoDID, "publishVideo") + if err != nil { + return nil, err + } + ctx = log.WithLogValues(ctx, "func", "publishVideo", "did", upload.RepoDID, "by", modCtx.ModeratorDID, "uploadId", body.UploadId) + + record := body.Record + if record == nil { + uris := body.Livestreams + if len(uris) == 0 && strings.HasPrefix(upload.Location, "at://") { + uris = []string{upload.Location} + } + var items []livestreamItem + if len(uris) > 0 { + if items, err = s.livestreamItems(uris); err != nil { + return nil, err + } + if items[0].ls.RepoDID != upload.RepoDID { + return nil, echo.NewHTTPError(http.StatusBadRequest, "the livestreams belong to a different streamer than the upload") + } + } + var draft *statedb.VideoDraft + if len(items) > 0 { + draft = videoDraftForLivestreams(items, deref(body.Title), deref(body.Description)) + } else { + title := strings.TrimSpace(deref(body.Title)) + if title == "" { + title = "Video" + } + draft = &statedb.VideoDraft{Title: title} + if d := strings.TrimSpace(deref(body.Description)); d != "" { + draft.Description = &d + } } + record = draft.Record() + } + + uri, cid, err := vod.PublishVideo(ctx, s.statefulDB, s.playbackStore, upload.RepoDID, upload.ID, record) + if err != nil { + return nil, videoError("publish video", err) } + log.Log(ctx, "video record published", "uri", uri, "cid", cid) return &placestream.MediaPublishVideo_Output{Uri: uri, Cid: cid}, nil } diff --git a/pkg/spxrpc/stubs.go b/pkg/spxrpc/stubs.go index 63b8ce69f..a0c5c2301 100644 --- a/pkg/spxrpc/stubs.go +++ b/pkg/spxrpc/stubs.go @@ -442,11 +442,14 @@ func (s *Server) RegisterHandlersPlacestream(e *echo.Echo) error { e.POST("/xrpc/place.stream.live.startLivestream", s.HandlePlaceStreamLiveStartLivestream) e.POST("/xrpc/place.stream.live.stopLivestream", s.HandlePlaceStreamLiveStopLivestream) 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) e.GET("/xrpc/place.stream.media.getUploadStatus", s.HandlePlaceStreamMediaGetUploadStatus) e.GET("/xrpc/place.stream.media.getVideo", s.HandlePlaceStreamMediaGetVideo) e.GET("/xrpc/place.stream.media.getVideoList", s.HandlePlaceStreamMediaGetVideoList) + e.GET("/xrpc/place.stream.media.listVideos", s.HandlePlaceStreamMediaListVideos) e.POST("/xrpc/place.stream.media.publishVideo", s.HandlePlaceStreamMediaPublishVideo) + e.POST("/xrpc/place.stream.media.updateVideo", s.HandlePlaceStreamMediaUpdateVideo) e.POST("/xrpc/place.stream.moderation.createBlock", s.HandlePlaceStreamModerationCreateBlock) e.POST("/xrpc/place.stream.moderation.createGate", s.HandlePlaceStreamModerationCreateGate) e.POST("/xrpc/place.stream.moderation.createPin", s.HandlePlaceStreamModerationCreatePin) @@ -869,6 +872,23 @@ func (s *Server) HandlePlaceStreamMediaCreateUpload(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamMediaDeleteVideo(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamMediaDeleteVideo") + defer span.End() + var body placestream.MediaDeleteVideo_Input + if err := c.Bind(&body); err != nil { + return err + } + var out *placestream.MediaDeleteVideo_Output + var handleErr error + // func (s *Server) handlePlaceStreamMediaDeleteVideo(ctx context.Context,body *placestream.MediaDeleteVideo_Input) (*placestream.MediaDeleteVideo_Output, error) + out, handleErr = s.handlePlaceStreamMediaDeleteVideo(ctx, &body) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + func (s *Server) HandlePlaceStreamMediaFinalizeLivestream(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamMediaFinalizeLivestream") defer span.End() @@ -937,6 +957,20 @@ func (s *Server) HandlePlaceStreamMediaGetVideoList(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamMediaListVideos(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamMediaListVideos") + defer span.End() + repo := c.QueryParam("repo") + var out *placestream.MediaListVideos_Output + var handleErr error + // func (s *Server) handlePlaceStreamMediaListVideos(ctx context.Context,repo string) (*placestream.MediaListVideos_Output, error) + out, handleErr = s.handlePlaceStreamMediaListVideos(ctx, repo) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + func (s *Server) HandlePlaceStreamMediaPublishVideo(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamMediaPublishVideo") defer span.End() @@ -954,6 +988,23 @@ func (s *Server) HandlePlaceStreamMediaPublishVideo(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamMediaUpdateVideo(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamMediaUpdateVideo") + defer span.End() + var body placestream.MediaUpdateVideo_Input + if err := c.Bind(&body); err != nil { + return err + } + var out *placestream.MediaUpdateVideo_Output + var handleErr error + // func (s *Server) handlePlaceStreamMediaUpdateVideo(ctx context.Context,body *placestream.MediaUpdateVideo_Input) (*placestream.MediaUpdateVideo_Output, error) + out, handleErr = s.handlePlaceStreamMediaUpdateVideo(ctx, &body) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + func (s *Server) HandlePlaceStreamModerationCreateBlock(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamModerationCreateBlock") defer span.End()