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

@@ -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;
+  });
+});

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<Buffer> {
+  return new Promise<Buffer>((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)");
+}

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<VideoUploadResult> {
+    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<VideoUploadResult> {
+    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(

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<VideoUploadResult> {
+    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<string, unknown>,
   ): Promise<void> {