api: Allow patching a stream recordingSpec (#2220)

* api: Fix the profiles fields

- Add documentation to stream's and asset's profiles
- Reference doc from payloads
- Use right type for recordingSpec.profiles
- Propagate profiles on upload APIs as well (didn't know they already were configurable there)
- Avoid some repetition by cross-referencing payloads

* api: Repeat SDK workaround for recordingSpec.profiles

* api: Support recordingSpec on stream patch as well

* api: Add a couple tests

packages/api/src/controllers/asset.ts

@@ -1,42 +1,20 @@
-import { authorizer } from "../middleware";
-import { validatePost } from "../middleware";
-import { Request, RequestHandler, Router, Response } from "express";
-import jwt, { JwtPayload } from "jsonwebtoken";
-import { v4 as uuid } from "uuid";
+import { FileStore as TusFileStore } from "@tus/file-store";
+import { S3Store as TusS3Store } from "@tus/s3-store";
 import {
-  Server as TusServer,
   EVENTS as TUS_EVENTS,
+  Server as TusServer,
   Upload as TusUpload,
 } from "@tus/server";
-import { S3Store as TusS3Store } from "@tus/s3-store";
-import { FileStore as TusFileStore } from "@tus/file-store";
+import { Request, Response, Router } from "express";
+import mung from "express-mung";
+import httpProxy from "http-proxy";
+import jwt, { JwtPayload } from "jsonwebtoken";
 import _ from "lodash";
-import {
-  makeNextHREF,
-  parseFilters,
-  parseOrder,
-  getS3PresignedUrl,
-  toStringValues,
-  pathJoin,
-  getObjectStoreS3Config,
-  reqUseReplica,
-  isValidBase64,
-  mapInputCreatorId,
-  addDefaultProjectId,
-} from "./helpers";
-import { db } from "../store";
+import os from "os";
 import sql from "sql-template-strings";
-import {
-  ForbiddenError,
-  UnprocessableEntityError,
-  NotFoundError,
-  BadRequestError,
-  InternalServerError,
-  UnauthorizedError,
-  NotImplementedError,
-} from "../store/errors";
-import httpProxy from "http-proxy";
-import { generateUniquePlaybackId } from "./generate-keys";
+import { v4 as uuid } from "uuid";
+import { authorizer, validatePost } from "../middleware";
+import { CliArgs } from "../parse-cli";
 import {
   Asset,
   AssetPatchPayload,
@@ -46,20 +24,40 @@ import {
   NewAssetPayload,
   ObjectStore,
   PlaybackPolicy,
-  Project,
   Task,
 } from "../schema/types";
-import { WithID } from "../store/types";
-import Queue from "../store/queue";
-import { taskScheduler, ensureQueueCapacity } from "../task/scheduler";
-import os from "os";
+import { db } from "../store";
+import {
+  BadRequestError,
+  ForbiddenError,
+  InternalServerError,
+  NotFoundError,
+  NotImplementedError,
+  UnauthorizedError,
+  UnprocessableEntityError,
+} from "../store/errors";
 import {
   ensureExperimentSubject,
   isExperimentSubject,
 } from "../store/experiment-table";
-import { CliArgs } from "../parse-cli";
-import mung from "express-mung";
+import Queue from "../store/queue";
+import { WithID } from "../store/types";
+import { ensureQueueCapacity, taskScheduler } from "../task/scheduler";
 import { getClips } from "./clip";
+import { generateUniquePlaybackId } from "./generate-keys";
+import {
+  addDefaultProjectId,
+  getObjectStoreS3Config,
+  getS3PresignedUrl,
+  isValidBase64,
+  makeNextHREF,
+  mapInputCreatorId,
+  parseFilters,
+  parseOrder,
+  pathJoin,
+  reqUseReplica,
+  toStringValues,
+} from "./helpers";
 
 // 7 Days
 const DELETE_ASSET_DELAY = 7 * 24 * 60 * 60 * 1000;
@@ -197,15 +195,22 @@ export async function validateAssetPayload(
   const userId = req.user.id;
   const payload = req.body as NewAssetPayload;
 
-  if (payload.objectStoreId) {
-    const os = await getActiveObjectStore(payload.objectStoreId);
+  const { name, profiles, staticMp4, creatorId, objectStoreId, storage } =
+    payload;
+
+  if (objectStoreId) {
+    const os = await getActiveObjectStore(objectStoreId);
     if (os.userId !== userId) {
       throw new ForbiddenError(
         `the provided object store is not owned by user`,
       );
     }
   }
 
+  if (profiles && !profiles.length) {
+    throw new BadRequestError("assets must have at least one profile");
+  }
+
   // Validate playbackPolicy on creation to generate resourceId & check if unifiedAccessControlConditions is present when using lit_signing_condition
   const playbackPolicy = await validateAssetPlaybackPolicy(
     payload,
@@ -237,14 +242,15 @@ export async function validateAssetPayload(
       phase: source.type === "directUpload" ? "uploading" : "waiting",
       updatedAt: createdAt,
     },
-    name: payload.name,
+    name,
     source,
-    staticMp4: payload.staticMp4,
+    ...(profiles ? { profiles } : null), // avoid serializing null profiles on the asset,
+    staticMp4,
     projectId: req.project?.id,
-    creatorId: mapInputCreatorId(payload.creatorId),
+    creatorId: mapInputCreatorId(creatorId),
     playbackPolicy,
-    objectStoreId: payload.objectStoreId || (await defaultObjectStoreId(req)),
-    storage: storageInputToState(payload.storage),
+    objectStoreId: objectStoreId || (await defaultObjectStoreId(req)),
+    storage: storageInputToState(storage),
   };
 }
 

packages/api/src/controllers/stream.test.ts

@@ -17,11 +17,10 @@ import {
   AuxTestServer,
   TestClient,
   clearDatabase,
+  createApiToken,
+  createProject,
   setupUsers,
   startAuxTestServer,
-  createProject,
-  createApiToken,
-  useApiTokenWithProject,
 } from "../test-helpers";
 import serverPromise, { TestServer } from "../test-server";
 import { semaphore, sleep } from "../util";
@@ -1266,6 +1265,35 @@ describe("controllers/stream", () => {
         expect(json.errors[0]).toContain("additionalProperties");
       });
 
+      it("should disallow adding recordingSpec without record=true", async () => {
+        const res = await client.patch(patchPath, {
+          recordingSpec: { profiles: [{ bitrate: 1600000 }] },
+        });
+        expect(res.status).toBe(400);
+        const json = await res.json();
+        expect(json.errors[0]).toContain(
+          "recordingSpec is only supported with record=true",
+        );
+      });
+
+      it("should allow patching recordingSpec with record=true", async () => {
+        const res = await client.patch(patchPath, {
+          record: true,
+          recordingSpec: { profiles: [{ bitrate: 3000000 }] },
+        });
+        expect(res.status).toBe(204);
+      });
+
+      it("should remove null recordingSpec.profiles", async () => {
+        const res = await client.patch(patchPath, {
+          record: true,
+          recordingSpec: { profiles: null },
+        });
+        expect(res.status).toBe(204);
+        const updated = await db.stream.get(stream.id);
+        expect(updated.recordingSpec).toEqual({});
+      });
+
       it("should validate field types", async () => {
         const testTypeErr = async (payload: any) => {
           let res = await client.patch(patchPath, payload);
@@ -1425,13 +1453,37 @@ describe("controllers/stream", () => {
     });
 
     it("should not accept additional properties for creating a stream", async () => {
-      const postMockLivepeerStream = JSON.parse(JSON.stringify(postMockStream));
-      postMockLivepeerStream.livepeer = "livepeer";
-      const res = await client.post("/stream", { ...postMockLivepeerStream });
+      const res = await client.post("/stream", {
+        ...postMockStream,
+        livepeer: "livepeer",
+      });
       expect(res.status).toBe(422);
       const stream = await res.json();
       expect(stream.id).toBeUndefined();
     });
+
+    it("should not accept recordingSpec without record", async () => {
+      const res = await client.post("/stream", {
+        ...postMockStream,
+        recordingSpec: { profiles: [{ bitrate: 1600000 }] },
+      });
+      expect(res.status).toBe(400);
+      const json = await res.json();
+      expect(json.errors[0]).toContain(
+        "recordingSpec is only supported with record=true",
+      );
+    });
+
+    it("should remove null profiles from recordingSpec", async () => {
+      const res = await client.post("/stream", {
+        ...postMockStream,
+        record: true,
+        recordingSpec: { profiles: null },
+      });
+      expect(res.status).toBe(201);
+      const stream = await res.json();
+      expect(stream.recordingSpec).toEqual({});
+    });
   });
 
   describe("stream endpoint with api key", () => {

packages/api/src/controllers/stream.ts

@@ -1405,7 +1405,7 @@ async function handleCreateStream(req: Request, payload: NewStreamPayload) {
     playbackId += "-test";
   }
 
-  const { objectStoreId } = payload;
+  let { objectStoreId, recordingSpec } = payload;
   if (objectStoreId) {
     const store = await db.objectStore.get(objectStoreId);
     if (!store || store.deleted || store.disabled) {
@@ -1415,10 +1415,25 @@ async function handleCreateStream(req: Request, payload: NewStreamPayload) {
     }
   }
 
+  if (recordingSpec) {
+    if (!payload.record) {
+      throw new BadRequestError(
+        `recordingSpec is only supported with record=true`,
+      );
+    }
+    if (!recordingSpec.profiles) {
+      // remove null profiles from the recordingSpec. it's only supported on the
+      // input as an SDK workaround but we want to avoid serializing them as null.
+      const { profiles, ...rest } = recordingSpec;
+      recordingSpec = rest;
+    }
+  }
+
   let doc: DBStream = {
     ...payload,
-    profiles: payload.profiles || req.config.defaultStreamProfiles,
     kind: "stream",
+    profiles: payload.profiles || req.config.defaultStreamProfiles,
+    recordingSpec,
     userId: req.user.id,
     creatorId: mapInputCreatorId(payload.creatorId),
     renditions: {},
@@ -1932,6 +1947,7 @@ app.patch(
       userTags,
       creatorId,
       profiles,
+      recordingSpec,
     } = payload;
     if (record != undefined && stream.isActive && stream.record != record) {
       res.status(400);
@@ -1953,6 +1969,22 @@ app.patch(
       creatorId: mapInputCreatorId(creatorId),
     };
 
+    if (recordingSpec) {
+      const { record } = typeof payload.record === "boolean" ? payload : stream;
+      if (!record) {
+        throw new BadRequestError(
+          `recordingSpec is only supported with record=true`,
+        );
+      }
+      if (!recordingSpec.profiles) {
+        // remove null profiles from the recordingSpec. it's only supported on the
+        // input as an SDK workaround but we want to avoid serializing them as null.
+        const { profiles, ...rest } = recordingSpec;
+        recordingSpec = rest;
+      }
+      patch = { ...patch, recordingSpec };
+    }
+
     if (multistream) {
       multistream = await validateMultistreamOpts(
         req.user.id,