diff --git a/apps/api/src/group-files/ocis-webdav.ts b/apps/api/src/group-files/ocis-webdav.ts index 5d4707f..8db8678 100644 --- a/apps/api/src/group-files/ocis-webdav.ts +++ b/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 { 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; - const queue: Buffer[] = []; - let sourceEnded = false; - let sourceError: Error | undefined; - let wakeTransport: (() => void) | undefined; - let wakePump: (() => void) | undefined; - - const requestStream = new ReadableStream({ - async pull(controller): Promise { - 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((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 => { - 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((resolve) => { - wakePump = resolve; - }); - wakePump = undefined; - } - } - sourceEnded = true; - wakeTransport?.(); - })().catch((err: unknown) => { - sourceError = err instanceof Error ? err : new Error(String(err)); - wakeTransport?.(); - wakePump?.(); - }); - + for await (const chunk of body) { + const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array | string); + hash.update(buffer); + sizeBytes += buffer.byteLength; + chunks.push(buffer); + } + const payload = Buffer.concat(chunks); const response = await this.fetcher(this.uploadUrl(groupId, safeName), { method: "PUT", - headers: { ...this.authHeaders(), "content-type": contentType }, - body: requestStream as unknown as ReadableStream, - duplex: "half", - } satisfies FetchInitWithDuplex); - if (!response.ok) { - requestStream.cancel().catch(() => undefined); - await pumped; - throw new Error(`oCIS upload failed with ${response.status}`); - } - await pumped; + headers: { ...this.authHeaders(), "content-type": contentType, "content-length": String(payload.byteLength) }, + body: payload, + }); + if (!response.ok) throw new Error(`oCIS upload failed with ${response.status}`); const etag = response.headers.get("oc-etag") ?? response.headers.get("etag"); const resourceId = response.headers.get("oc-fileid"); return {