export const PRIVATE_UPLOAD_PART_BYTES = 8 * 1024 * 1024;

export async function* readUploadParts(
  body: ReadableStream<Uint8Array>,
  expectedBytes: number,
  partBytes = PRIVATE_UPLOAD_PART_BYTES,
): AsyncGenerator<Buffer> {
  if (!Number.isSafeInteger(expectedBytes) || expectedBytes < 1 ||
      !Number.isSafeInteger(partBytes) || partBytes < 1) {
    throw new Error("upload-size-invalid");
  }
  const reader = body.getReader();
  let pending: Buffer[] = [];
  let pendingBytes = 0;
  let receivedBytes = 0;
  let completed = false;
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      if (!value?.byteLength) continue;
      receivedBytes += value.byteLength;
      if (receivedBytes > expectedBytes) throw new Error("upload-size-mismatch");
      const bytes = Buffer.from(value.buffer, value.byteOffset, value.byteLength);
      for (let offset = 0; offset < bytes.length;) {
        const count = Math.min(bytes.length - offset, partBytes - pendingBytes);
        pending.push(bytes.subarray(offset, offset + count));
        pendingBytes += count;
        offset += count;
        if (pendingBytes === partBytes) {
          yield Buffer.concat(pending, pendingBytes);
          pending = [];
          pendingBytes = 0;
        }
      }
    }
    if (receivedBytes !== expectedBytes) throw new Error("upload-size-mismatch");
    if (pendingBytes > 0) yield Buffer.concat(pending, pendingBytes);
    completed = true;
  } finally {
    if (!completed) {
      await reader.cancel().catch(() => undefined);
    }
    reader.releaseLock();
  }
}