Function createMultipartUpload

apps/sim/lib/uploads/core/storage-service.ts:261–361

(options: {
  key: string
  context: StorageContext
  contentType: string
})

 * buffered and written via a single {@link uploadFile} on `complete`.
 */
export async function createMultipartUpload(options: {
  key: string
  context: StorageContext
  contentType: string
}): Promise<MultipartUploadHandle> {
  const { key, context, contentType } = options
  const config = getStorageConfig(context)
  const cloud = hasCloudStorage()

  let backend: MultipartBackend | null = null
  // Accumulate writes as references, not a growing buffer — concatenating only when a part fills
  // (or on complete) keeps total copying ~O(bytes) instead of O(bytes × writes).
  let pendingChunks: Buffer[] = []
  let pendingBytes = 0
  let totalBytes = 0
  let partNumber = 0
  let aborted = false
  let firstError: unknown
  const inflight = new Set<Promise<void>>()

  /** Merge the accumulated chunks into one ArrayBuffer-backed buffer (which `uploadFile` expects). */
  const drainPending = (): Buffer<ArrayBuffer> => Buffer.concat(pendingChunks, pendingBytes)

  const ensureBackend = async (): Promise<MultipartBackend> => {
    if (!backend) {
      backend = USE_BLOB_STORAGE
        ? await createBlobBackend(key, createBlobConfig(config), contentType)
        : await createS3Backend(key, createS3Config(config), contentType, context)
    }
    return backend
  }

  const dispatchPart = async (body: Buffer): Promise<void> => {
    // Bound concurrency: wait for a free slot before starting another part.
    while (inflight.size >= MULTIPART_MAX_INFLIGHT) await Promise.race(inflight)
    if (firstError) throw firstError
    const be = await ensureBackend()
    const partNo = ++partNumber
    const p = be
      .uploadPart(partNo, body)
      .catch((err) => {
        firstError ??= err
      })
      .finally(() => {
        inflight.delete(p)
      })
    inflight.add(p)
  }

  const abort = async (): Promise<void> => {
    aborted = true
    await Promise.allSettled(inflight)
    if (backend) await backend.abort().catch(() => {})
  }

  return {
    async write(chunk) {
      if (aborted) throw new Error('Multipart upload already aborted')