oCIS upload: drain to buffer + explicit content-length (reva rejects chunked PUTs)

apps/api/src/group-files/ocis-webdav.ts

@@ -183,84 +183,24 @@ export class OcisWebdavClient {
    */
   async upload(groupId: string, safeName: string, body: Readable, contentType: string): Promise<OcisUploadResult> {
     await this.ensureGroupFolder(groupId);
+    // reva rejects chunked PUTs, so the source drains into a bounded buffer
+    // (the upload cap is 20 MiB) and goes out with an explicit content-length.
     const hash = createHash("sha256");
+    const chunks: Buffer[] = [];
     let sizeBytes = 0;
-    let queuedBytes = 0;
+    for await (const chunk of body) {
-    const queue: Buffer[] = [];
+      const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array | string);
-    let sourceEnded = false;
+      hash.update(buffer);
-    let sourceError: Error | undefined;
+      sizeBytes += buffer.byteLength;
-    let wakeTransport: (() => void) | undefined;
+      chunks.push(buffer);
-    let wakePump: (() => void) | undefined;
+    }
-
+    const payload = Buffer.concat(chunks);
-    const requestStream = new ReadableStream<Buffer>({
-      async pull(controller): Promise<void> {
-        if (queue.length > 0) {
-          const next = queue.shift()!;
-          queuedBytes -= next.byteLength;
-          controller.enqueue(next);
-          wakePump?.();
-          return;
-        }
-        if (sourceEnded) {
-          if (sourceError !== undefined) controller.error(sourceError);
-          else controller.close();
-          return;
-        }
-        await new Promise<void>((resolve) => {
-          wakeTransport = resolve;
-        });
-        wakeTransport = undefined;
-        if (queue.length > 0) {
-          controller.enqueue(queue.shift()!);
-          wakePump?.();
-        } else if (sourceError !== undefined) {
-          controller.error(sourceError);
-        } else if (sourceEnded) {
-          controller.close();
-        }
-      },
-      cancel(): void {
-        // The transport hung up (e.g. it rejected early): keep draining the
-        // source so the digest/size accounting still completes.
-        wakeTransport?.();
-      },
-    });
-
-    const pumped = (async (): Promise<void> => {
-      for await (const chunk of body) {
-        const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array | string);
-        hash.update(buffer);
-        sizeBytes += buffer.byteLength;
-        queuedBytes += buffer.byteLength;
-        queue.push(buffer);
-        wakeTransport?.();
-        while (queuedBytes > UPLOAD_BUFFER_CAP) {
-          await new Promise<void>((resolve) => {
-            wakePump = resolve;
-          });
-          wakePump = undefined;
-        }
-      }
-      sourceEnded = true;
-      wakeTransport?.();
-    })().catch((err: unknown) => {
-      sourceError = err instanceof Error ? err : new Error(String(err));
-      wakeTransport?.();
-      wakePump?.();
-    });
-
     const response = await this.fetcher(this.uploadUrl(groupId, safeName), {
       method: "PUT",
-      headers: { ...this.authHeaders(), "content-type": contentType },
+      headers: { ...this.authHeaders(), "content-type": contentType, "content-length": String(payload.byteLength) },
-      body: requestStream as unknown as ReadableStream,
+      body: payload,
-      duplex: "half",
+    });
-    } satisfies FetchInitWithDuplex);
+    if (!response.ok) throw new Error(`oCIS upload failed with ${response.status}`);
-    if (!response.ok) {
-      requestStream.cancel().catch(() => undefined);
-      await pumped;
-      throw new Error(`oCIS upload failed with ${response.status}`);
-    }
-    await pumped;
     const etag = response.headers.get("oc-etag") ?? response.headers.get("etag");
     const resourceId = response.headers.get("oc-fileid");
     return {