From 177ca6eb2b294cfc50ad015d0b156c408281fe61 Mon Sep 17 00:00:00 2001 From: glass beadian Date: Wed, 14 Oct 2026 14:16:19 +0000 Subject: [PATCH] feat: video library imports from Drive and chat media (WS2 backend) POST /video/import-drive {resourceId} pulls through the oCIS WebDAV service-account client; POST /video/import-chat {mxcUri} pulls from the Matrix content repository (MATRIX_MEDIA_TOKEN, homeserver falls back to ARCHIVE_PROVISIONER_HOMESERVER_URL). Both cap at 256 MiB (buffered v1, streaming follow-up), store through the same encrypted content- addressed path as uploads, probe, and audit video.imported with provenance. --- apps/api/src/video/import-sources.test.ts | 43 ++++++++++++ apps/api/src/video/import-sources.ts | 85 +++++++++++++++++++++++ apps/api/src/video/video.controller.ts | 21 ++++++ apps/api/src/video/video.service.ts | 19 ++++- 4 files changed, 167 insertions(+), 1 deletion(-) create mode 100644 apps/api/src/video/import-sources.test.ts create mode 100644 apps/api/src/video/import-sources.ts diff --git a/apps/api/src/video/import-sources.test.ts b/apps/api/src/video/import-sources.test.ts new file mode 100644 index 0000000..b303d9a --- /dev/null +++ b/apps/api/src/video/import-sources.test.ts @@ -0,0 +1,43 @@ +import { Readable } from "node:stream"; +import { describe, expect, it } from "vitest"; +import { BadRequestException } from "@nestjs/common"; +import { IMPORT_MAX_BYTES, bufferWithCap, matrixMediaConfig, parseMxcUri } from "./import-sources.js"; + +describe("parseMxcUri", () => { + it("accepts well-formed mxc uris and rejects everything else", () => { + expect(parseMxcUri("mxc://chat.occupymedia.cc/AbC123_-")).toEqual({ server: "chat.occupymedia.cc", mediaId: "AbC123_-" }); + expect(parseMxcUri(" mxc://server/abc ")).toEqual({ server: "server", mediaId: "abc" }); + for (const bad of ["https://server/abc", "mxc:///abc", "mxc://server/", "mxc://server/a/b", "mxc://server/abc?x=1", ""]) { + expect(() => parseMxcUri(bad)).toThrow(BadRequestException); + } + }); +}); + +describe("bufferWithCap", () => { + const streamOf = (chunks: Buffer[]) => new Readable({ read() { for (const c of chunks) this.push(c); this.push(null); } }); + + it("concatenates a stream under the cap", async () => { + const bytes = await bufferWithCap(streamOf([Buffer.alloc(10, 1), Buffer.alloc(10, 2)]), 100); + expect(bytes).toHaveLength(20); + }); + + it("refuses streams beyond the cap and destroys them", async () => { + await expect(bufferWithCap(streamOf([Buffer.alloc(60), Buffer.alloc(60)]), 64)).rejects.toBeInstanceOf(BadRequestException); + expect(IMPORT_MAX_BYTES).toBe(256 * 1024 * 1024); + }); +}); + +describe("matrixMediaConfig", () => { + it("is null without a token and reads the fallback homeserver", () => { + const previous = { ...process.env }; + delete process.env.MATRIX_MEDIA_TOKEN; + expect(matrixMediaConfig()).toBeNull(); + process.env.MATRIX_MEDIA_TOKEN = "tok"; + delete process.env.MATRIX_MEDIA_HOMESERVER_URL; + process.env.ARCHIVE_PROVISIONER_HOMESERVER_URL = "https://chat.example/"; + expect(matrixMediaConfig()).toEqual({ homeserver: "https://chat.example", token: "tok" }); + process.env.MATRIX_MEDIA_HOMESERVER_URL = "https://media.example"; + expect(matrixMediaConfig()?.homeserver).toBe("https://media.example"); + process.env = previous; + }); +}); diff --git a/apps/api/src/video/import-sources.ts b/apps/api/src/video/import-sources.ts new file mode 100644 index 0000000..1522409 --- /dev/null +++ b/apps/api/src/video/import-sources.ts @@ -0,0 +1,85 @@ +import { Readable } from "node:stream"; +import { BadRequestException, ServiceUnavailableException } from "@nestjs/common"; +import { OcisNotFoundError, OcisWebdavClient, ocisConfigFromEnv } from "../group-files/ocis-webdav.js"; +/** Imports buffer the whole file in memory; 256 MiB is the v1 cap (streaming is a follow-up). */ +export const IMPORT_MAX_BYTES = 256 * 1024 * 1024; + +export interface MxcUri { + server: string; + mediaId: string; +} + +/** Strict mxc://server/mediaId (v3 media ids are opaque non-slash segments). */ +export function parseMxcUri(raw: string): MxcUri { + const match = /^mxc:\/\/([^/\s]+)\/([A-Za-z0-9_-]+)$/.exec(raw.trim()); + if (match === null) throw new BadRequestException("mxcUri must look like mxc://server/mediaId"); + return { server: match[1]!, mediaId: match[2]! }; +} + +export async function bufferWithCap(stream: Readable, capBytes: number): Promise { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + let total = 0; + stream.on("data", (chunk: Buffer) => { + total += chunk.length; + if (total > capBytes) { + stream.destroy(); + reject(new BadRequestException(`file exceeds the ${Math.round(capBytes / (1024 * 1024))} MiB import cap`)); + return; + } + chunks.push(chunk); + }); + stream.on("end", () => resolve(Buffer.concat(chunks))); + stream.on("error", reject); + }); +} + +let ocisClient: OcisWebdavClient | null = null; + +/** Drive import: pull one oCIS resource through the service-account WebDAV client. */ +export async function loadDriveBytes(resourceId: string): Promise<{ bytes: Buffer; mime: string }> { + ocisClient ??= new OcisWebdavClient(ocisConfigFromEnv()); + const download = await ocisClient.download(resourceId).catch((error: unknown) => { + if (error instanceof OcisNotFoundError) throw new BadRequestException("no such Drive resource"); + throw new ServiceUnavailableException("Drive is unreachable — try again later"); + }); + const bytes = await bufferWithCap(download.stream, IMPORT_MAX_BYTES); + return { bytes, mime: download.contentType ?? "application/octet-stream" }; +} + +export function matrixMediaConfig(): { homeserver: string; token: string } | null { + const token = process.env.MATRIX_MEDIA_TOKEN?.trim() ?? ""; + if (token === "") return null; + const homeserver = (process.env.MATRIX_MEDIA_HOMESERVER_URL ?? process.env.ARCHIVE_PROVISIONER_HOMESERVER_URL ?? "").replace(/\/+$/, ""); + if (homeserver === "") return null; + return { homeserver, token }; +} + +/** Chat import: pull one MXC from the homeserver's authenticated media repository. */ +export async function loadMatrixBytes(mxc: MxcUri): Promise<{ bytes: Buffer; mime: string }> { + const config = matrixMediaConfig(); + if (config === null) throw new ServiceUnavailableException("chat media import is not configured: set MATRIX_MEDIA_TOKEN"); + const encode = (value: string) => encodeURIComponent(value); + const paths = [ + `/_matrix/client/v1/media/download/${encode(mxc.server)}/${encode(mxc.mediaId)}`, + `/_matrix/client/v3/media/download/${encode(mxc.server)}/${encode(mxc.mediaId)}`, + ]; + let lastStatus = 0; + for (const path of paths) { + const res = await fetch(`${config.homeserver}${path}`, { + headers: { authorization: `Bearer ${config.token}`, accept: "application/octet-stream" }, + redirect: "error", + }).catch(() => null); + if (res === null) throw new ServiceUnavailableException("chat homeserver unreachable — try again later"); + if (res.status === 404 && path.startsWith("/_matrix/client/v1/")) { lastStatus = 404; continue; } + if (!res.ok) { + lastStatus = res.status; + if (res.status === 404) throw new BadRequestException("no such chat media (or it has expired)"); + throw new ServiceUnavailableException(`chat media download failed (HTTP ${String(lastStatus)})`); + } + if (res.body === null) throw new ServiceUnavailableException("chat media download returned an empty body"); + const bytes = await bufferWithCap(Readable.fromWeb(res.body as import("node:stream/web").ReadableStream), IMPORT_MAX_BYTES); + return { bytes, mime: res.headers.get("content-type")?.split(";")[0]?.trim() || "application/octet-stream" }; + } + throw new BadRequestException("no such chat media (or it has expired)"); +} diff --git a/apps/api/src/video/video.controller.ts b/apps/api/src/video/video.controller.ts index 1a6b2f0..f4612b6 100644 --- a/apps/api/src/video/video.controller.ts +++ b/apps/api/src/video/video.controller.ts @@ -24,6 +24,7 @@ import { type VideoUploadResult, } from "./video.service.js"; import { MAX_VIDEO_UPLOAD_BYTES } from "./upload-limits.js"; +import { loadDriveBytes, loadMatrixBytes, parseMxcUri } from "./import-sources.js"; interface VideoFile { buffer: Buffer; @@ -60,6 +61,26 @@ export class VideoController { return this.video.uploadVideo(session.workspaceId, dto.storageKey, dto.mimeType, dto.sizeBytes); } + @Post("import-drive") + @RequirePermission("media.manage") + async importDrive(@Req() request: Request & AuthenticatedRequest, @Body() body: unknown): Promise { + const session = requireSession(request); + const input = requireObject(body, "body"); + const resourceId = requireString(input.resourceId, "resourceId"); + const { bytes, mime } = await loadDriveBytes(resourceId); + return this.video.importVideoBytes(session.workspaceId, bytes, mime, "drive", resourceId); + } + + @Post("import-chat") + @RequirePermission("media.manage") + async importChat(@Req() request: Request & AuthenticatedRequest, @Body() body: unknown): Promise { + const session = requireSession(request); + const input = requireObject(body, "body"); + const mxc = parseMxcUri(requireString(input.mxcUri, "mxcUri")); + const { bytes, mime } = await loadMatrixBytes(mxc); + return this.video.importVideoBytes(session.workspaceId, bytes, mime, "chat", `${mxc.server}/${mxc.mediaId}`); + } + @Get() @RequirePermission("media.view") async list( diff --git a/apps/api/src/video/video.service.ts b/apps/api/src/video/video.service.ts index 6cdc28e..4147579 100644 --- a/apps/api/src/video/video.service.ts +++ b/apps/api/src/video/video.service.ts @@ -103,6 +103,23 @@ export class VideoService { return this.registerAndProbe(workspaceId, stored.key, stored.contentType, stored.sizeBytes); } + async importVideoBytes( + workspaceId: string, + bytes: Buffer, + mimeType: string, + source: "drive" | "chat", + sourceRef: string, + ): Promise { + validateUpload("pending", mimeType, bytes.length); + const storage = this.requireStorage(); + const stored = await storage.putContent(bytes, mimeType).catch(() => { + throw new ServiceUnavailableException("Video storage unavailable: could not store imported video."); + }); + const result = await this.registerAndProbe(workspaceId, stored.key, stored.contentType, stored.sizeBytes); + await this.audit(workspaceId, "video.imported", result.videoAsset.id, { source, sourceRef, sizeBytes: bytes.length }); + return result; + } + async uploadVideo( workspaceId: string, storageKey: string, @@ -356,7 +373,7 @@ export class VideoService { private async audit( workspaceId: string, - action: "video.uploaded" | "video.probed" | "video.deleted", + action: "video.uploaded" | "video.probed" | "video.deleted" | "video.imported", targetId: string, diff: Record, ): Promise {