Fortsetzbare Datei-Uploads in Angular mit RxJS

(
  node --input-type=module <<'NODE' &&
import { existsSync, realpathSync } from 'node:fs'
import { dirname, join } from 'node:path'
let directory = realpathSync('.')
while (true) {
  if (existsSync(join(directory, '.pnp.cjs')) || existsSync(join(directory, '.pnp.js'))) {
    console.error('Create this demo outside the enclosing Yarn Plug’n’Play project.')
    process.exit(1)
  }
  const parent = dirname(directory)
  if (parent === directory) break
  directory = parent
}
NODE
  mkdir chunked-upload-app &&
  cd chunked-upload-app &&
  printf 'nodeLinker: node-modules\nenableGlobalCache: false\n' > .yarnrc.yml &&
  touch yarn.lock
)

{
  "name": "chunked-upload-app",
  "private": true,
  "type": "module",
  "packageManager": "yarn@4.12.0",
  "scripts": {
    "build": "ng build",
    "start": "ng serve --host 127.0.0.1 --port 4200"
  },
  "browserslist": ["Chrome 145"],
  "dependencies": {
    "@angular/build": "22.2.0",
    "@angular/cli": "22.2.0",
    "@angular/common": "22.2.0",
    "@angular/compiler": "22.2.0",
    "@angular/compiler-cli": "22.2.0",
    "@angular/core": "22.2.0",
    "@angular/platform-browser": "22.2.0",
    "rxjs": "7.8.2",
    "tslib": "2.8.1",
    "typescript": "6.0.3"
  }
}

{
  "version": 1,
  "cli": { "analytics": false, "cache": { "enabled": false } },
  "projects": {
    "demo": {
      "projectType": "application",
      "root": ".",
      "architect": {
        "build": {
          "builder": "@angular/build:application",
          "options": {
            "browser": "src/main.ts",
            "tsConfig": "tsconfig.json",
            "index": "src/index.html",
            "outputPath": "dist",
            "styles": [],
            "assets": []
          }
        },
        "serve": {
          "builder": "@angular/build:dev-server",
          "options": {
            "buildTarget": "demo:build",
            "proxyConfig": "proxy.conf.json"
          }
        }
      }
    }
  }
}

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "preserve",
    "moduleResolution": "bundler",
    "experimentalDecorators": true,
    "strict": true,
    "lib": ["ES2022", "DOM"],
    "types": [],
    "rewriteRelativeImportExtensions": true
  },
  "angularCompilerOptions": { "strictTemplates": true },
  "files": ["src/main.ts"]
}

<!doctype html>
<html lang="en">
  <head>
    <meta charset="utf-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <base href="/" />
    <title>Chunked Angular upload</title>
  </head>
  <body><app-root></app-root></body>
</html>

{
  "/api/**": { "target": "http://127.0.0.1:4317", "secure": false }
}

import { provideHttpClient, withInterceptors, withXhr } from '@angular/common/http'
import type { ApplicationConfig } from '@angular/core'

export const appConfig: ApplicationConfig = {
  providers: [
    provideHttpClient(
      withXhr(),
      withInterceptors([
        (request, next) => next(request.clone({ setHeaders: { 'X-Demo-Upload': '1' } })),
      ]),
    ),
  ],
}

import {
  closeSync,
  fsyncSync,
  ftruncateSync,
  linkSync,
  mkdirSync,
  openSync,
  rmSync,
  statSync,
  writeSync,
} from 'node:fs'
import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'
import { resolve } from 'node:path'

type Session = { size: number; offset: number; complete: boolean; canceled: boolean }
class RequestError extends Error {
  readonly status: number
  constructor(status: number, message: string) {
    super(message)
    this.status = status
  }
}

const maxFile = 100 * 1024 * 1024
const maxChunk = 1024 * 1024
const sessions = new Map<string, Session>()
const directory = resolve('uploads')
const port = Number(process.env.PORT ?? '4317')
const appOrigin = process.env.APP_ORIGIN ?? 'http://127.0.0.1:4200'
if (!Number.isInteger(port) || port < 1 || port > 65535) {
  throw new Error('PORT must be between 1 and 65535.')
}
if (new URL(appOrigin).origin !== appOrigin || new URL(appOrigin).hostname !== '127.0.0.1') {
  throw new Error('APP_ORIGIN must be a loopback URL such as http://127.0.0.1:4200.')
}
mkdirSync(directory, { recursive: true, mode: 0o700 })

function reply(response: ServerResponse, status: number, value?: unknown): void {
  response.writeHead(status, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' })
  response.end(value === undefined ? undefined : JSON.stringify(value))
}

async function body(request: IncomingMessage, limit: number): Promise<Buffer> {
  const parts: Buffer[] = []
  let length = 0
  // Keep the socket available for a 413 response when stopping the body reader early.
  for await (const part of request.iterator({ destroyOnReturn: false })) {
    if (!(part instanceof Buffer)) throw new RequestError(400, 'Expected binary request data.')
    length += part.length
    if (length > limit) {
      request.resume()
      throw new RequestError(413, 'Request body is too large.')
    }
    parts.push(part)
  }
  return Buffer.concat(parts)
}

function reserve(id: string, size: number): Session {
  const existing = sessions.get(id)
  if (existing) {
    if (existing.canceled) throw new RequestError(410, 'Upload was canceled.')
    if (existing.size !== size) throw new RequestError(409, 'Upload size conflicts.')
    return existing
  }
  if (sessions.size >= 32) throw new RequestError(429, 'Restart the demo receiver after 32 IDs.')
  const file = openSync(resolve(directory, `${id}.part`), 'wx', 0o600)
  closeSync(file)
  const session = { size, offset: 0, complete: false, canceled: false }
  sessions.set(id, session)
  return session
}

async function handle(request: IncomingMessage, response: ServerResponse): Promise<void> {
  if (
    request.headers['x-demo-upload'] !== '1' ||
    (request.headers.origin !== undefined && request.headers.origin !== appOrigin)
  ) {
    throw new RequestError(403, 'Use the local demo application.')
  }
  const match = /^\/api\/uploads\/([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})(\/bytes|\/complete)?$/u.exec(request.url ?? '')
  if (!match) throw new RequestError(404, 'Unknown upload route.')
  const id = match[1]
  if (!id) throw new RequestError(400, 'Missing upload ID.')
  const operation = match[2] ?? ''
  const part = resolve(directory, `${id}.part`)
  const output = resolve(directory, `${id}.bin`)

  if (request.method === 'PUT' && operation === '') {
    let input: unknown
    try {
      input = JSON.parse((await body(request, 512)).toString('utf8'))
    } catch (error) {
      if (error instanceof RequestError) throw error
      throw new RequestError(400, 'Expected JSON containing size.')
    }
    if (
      typeof input !== 'object' || input === null || !('size' in input) ||
      typeof input.size !== 'number' || !Number.isSafeInteger(input.size) ||
      input.size < 1 || input.size > maxFile
    ) {
      throw new RequestError(400, 'Choose a file between 1 byte and 100 MiB.')
    }
    const session = reserve(id, input.size)
    reply(response, 200, { offset: session.offset, complete: session.complete })
    return
  }

  // Read first, then recheck state: cancellation can arrive while a body is pending.
  const data = await body(request, operation === '/bytes' ? maxChunk : 512)
  let session = sessions.get(id)
  if (request.method === 'DELETE' && operation === '') {
    if (session?.complete) throw new RequestError(409, 'Upload was already published.')
    if (!session) {
      if (sessions.size >= 32) throw new RequestError(429, 'Restart the demo receiver after 32 IDs.')
      session = { size: 0, offset: 0, complete: false, canceled: true }
      sessions.set(id, session)
    }
    session.canceled = true
    rmSync(part, { force: true })
    reply(response, 204)
    return
  }
  if (!session || session.canceled) throw new RequestError(410, 'Upload is unavailable.')

  if (request.method === 'PUT' && operation === '/bytes') {
    const range = /^bytes (\d+)-(\d+)\/(\d+)$/u.exec(request.headers['content-range'] ?? '')
    if (!range) throw new RequestError(400, 'Expected Content-Range.')
    const start = Number(range[1])
    const end = Number(range[2]) + 1
    const total = Number(range[3])
    if (
      !Number.isSafeInteger(start) || !Number.isSafeInteger(end) ||
      !Number.isSafeInteger(total) || session.complete || start !== session.offset ||
      total !== session.size || end <= start || end > total || data.length !== end - start
    ) {
      throw new RequestError(409, 'Range or body does not match the committed offset.')
    }
    // These synchronous mutations cannot interleave in this single Node process.
    const file = openSync(part, 'r+')
    try {
      let written = 0
      while (written < data.length) {
        const count = writeSync(file, data, written, data.length - written, start + written)
        if (count === 0) throw new Error('Disk write made no progress.')
        written += count
      }
      ftruncateSync(file, end)
      fsyncSync(file)
      session.offset = end
    } finally {
      closeSync(file)
    }
    reply(response, 204)
    return
  }

  if (request.method === 'POST' && operation === '/complete') {
    if (!session.complete) {
      if (session.offset !== session.size || statSync(part).size !== session.size) {
        throw new RequestError(409, 'Upload still has missing bytes.')
      }
      // A hard link publishes on this filesystem without replacing an existing output.
      linkSync(part, output)
      session.complete = true
      rmSync(part, { force: true })
      console.log(`Published uploads/${id}.bin`)
    }
    reply(response, 204)
    return
  }
  throw new RequestError(405, 'Unsupported upload operation.')
}

const server = createServer((request, response) => {
  handle(request, response).catch((error: unknown) => {
    if (response.destroyed) return
    if (error instanceof RequestError) {
      reply(response, error.status, { message: error.message })
      return
    }
    console.error('Receiver could not process the upload.')
    reply(response, 500, { message: 'Receiver could not process the upload.' })
  })
})
server.requestTimeout = 30_000
server.on('error', () => {
  console.error('Receiver failed to listen. Check whether its port is already in use.')
  process.exitCode = 1
})
server.listen(port, '127.0.0.1', () => console.log(`Receiver ready on http://127.0.0.1:${port}`))

import { HttpClient, HttpEventType } from '@angular/common/http'
import { Injectable, inject } from '@angular/core'
import {
  BehaviorSubject,
  EMPTY,
  Subscription,
  catchError,
  concatMap,
  concatWith,
  defer,
  finalize,
  range,
  tap,
} from 'rxjs'

interface UploadState {
  progress: number
  running: boolean
  canceling: boolean
  complete: boolean
  message: string
}

@Injectable({ providedIn: 'root' })
export class FileUploadService {
  readonly state = new BehaviorSubject<UploadState>({
    progress: 0,
    running: false,
    canceling: false,
    complete: false,
    message: 'Select a file.',
  })
  readonly #http = inject(HttpClient)
  #request = new Subscription()
  #file: File | null = null
  #id = ''
  #offset = 0
  readonly #chunkSize = 1024 * 1024

  select(file: File): void {
    if (this.state.value.canceling) return
    this.pause()
    this.#file = null
    if (file.size === 0 || file.size > 100 * 1024 * 1024) {
      this.state.next({
        progress: 0,
        running: false,
        canceling: false,
        complete: false,
        message: 'Choose a file between 1 byte and 100 MiB.',
      })
      return
    }
    this.#file = file
    this.#id = crypto.randomUUID()
    this.#offset = 0
    this.state.next({
      progress: 0,
      running: false,
      canceling: false,
      complete: false,
      message: 'Ready.',
    })
    this.resume()
  }

  resume(): void {
    const file = this.#file
    if (!file || this.state.value.running || this.state.value.canceling || this.state.value.complete) {
      return
    }
    const url = `/api/uploads/${this.#id}`
    this.state.next({ ...this.state.value, running: true, message: 'Uploading…' })
    this.#request = this.#http
      .put<unknown>(url, { size: file.size }, { timeout: 30_000 })
      .pipe(
        concatMap((status) => {
          if (
            typeof status !== 'object' ||
            status === null ||
            !('offset' in status) ||
            typeof status.offset !== 'number' ||
            !Number.isSafeInteger(status.offset) ||
            status.offset < 0 ||
            status.offset > file.size ||
            !('complete' in status) ||
            typeof status.complete !== 'boolean' ||
            (status.complete && status.offset !== file.size)
          ) {
            throw new Error('Invalid upload state')
          }
          this.#offset = status.offset
          this.progress(file)
          if (status.complete) {
            this.finished()
            return EMPTY
          }
          const offset = status.offset
          return range(0, Math.ceil((file.size - offset) / this.#chunkSize)).pipe(
            concatMap((index) =>
              defer(() => {
                const start = offset + index * this.#chunkSize
                const end = Math.min(start + this.#chunkSize, file.size)
                return this.#http
                  .put(`${url}/bytes`, file.slice(start, end), {
                    headers: { 'Content-Range': `bytes ${start}-${end - 1}/${file.size}` },
                    observe: 'events',
                    reportProgress: true,
                    timeout: 30_000,
                  })
                  .pipe(
                    tap((event) => {
                      if (event.type === HttpEventType.UploadProgress) {
                        this.progress(file, Math.min(event.loaded, end - start))
                      }
                      if (event.type === HttpEventType.Response) {
                        if (event.status !== 204) throw new Error('Chunk was not committed')
                        this.#offset = end
                        this.progress(file)
                      }
                    }),
                  )
              }),
            ),
            // Finalization runs once, including when the server already has every byte.
            concatWith(
              defer(() =>
                this.#http
                  .post(`${url}/complete`, {}, { observe: 'response', timeout: 30_000 })
                  .pipe(
                    tap((response) => {
                      if (response.status !== 204) throw new Error('Publication was not confirmed')
                      this.finished()
                    }),
                  ),
              ),
            ),
          )
        }),
        finalize(() => this.state.next({ ...this.state.value, running: false })),
      )
      .subscribe({
        error: () =>
          this.state.next({ ...this.state.value, message: 'Upload interrupted. Resume to retry.' }),
      })
  }

  pause(): void {
    if (this.state.value.canceling) return
    this.#request.unsubscribe()
    if (this.#file && !this.state.value.complete) {
      this.progress(this.#file)
      this.state.next({ ...this.state.value, running: false, message: 'Paused.' })
    }
  }

  cancel(): void {
    if (this.state.value.canceling) return
    this.pause()
    if (!this.#file || this.state.value.complete) return
    const url = `/api/uploads/${this.#id}`
    this.#file = null
    let confirmed = false
    this.state.next({
      progress: 0,
      running: true,
      canceling: true,
      complete: false,
      message: 'Canceling…',
    })
    this.#request = this.#http
      .delete(url, { observe: 'response', timeout: 30_000 })
      .pipe(
        tap((response) => {
          if (response.status !== 204) throw new Error('Cancellation was not confirmed')
          confirmed = true
        }),
        // Errors and interrupted requests both leave cleanup unconfirmed.
        catchError(() => EMPTY),
        // Publish one settled state so observers can safely select another file.
        finalize(() =>
          this.state.next({
            ...this.state.value,
            running: false,
            canceling: false,
            message: confirmed ? 'Canceled.' : 'Cancellation unconfirmed. Check your uploads.',
          }),
        ),
      )
      .subscribe()
  }

  destroy(): void {
    this.pause()
    // Destruction also interrupts cleanup; its finalizer reports the unconfirmed result.
    this.#request.unsubscribe()
  }

  private progress(file: File, loaded = 0): void {
    this.state.next({
      ...this.state.value,
      progress: Math.min(99, Math.floor((100 * (this.#offset + loaded)) / file.size)),
    })
  }

  private finished(): void {
    this.state.next({
      progress: 100,
      running: false,
      canceling: false,
      complete: true,
      message: 'Upload complete.',
    })
  }
}

import { AsyncPipe } from '@angular/common'
import { Component, inject, OnDestroy } from '@angular/core'
import { FileUploadService } from './file-upload.service.ts'

@Component({
  selector: 'app-root',
  standalone: true,
  imports: [AsyncPipe],
  template: `
    @if (upload.state | async; as state) {
      <label>File <input type="file" (change)="select($event)" [disabled]="state.running" /></label>
      <progress aria-label="Upload progress" max="100" [value]="state.progress"></progress>
      <p role="status">{{ state.message }}</p>
      <button (click)="upload.pause()" [disabled]="!state.running || state.canceling">Pause</button>
      <button (click)="upload.resume()" [disabled]="state.running || state.canceling || state.complete">Resume</button>
      <button (click)="upload.cancel()" [disabled]="state.canceling || state.complete">Cancel</button>
    }
  `,
})
export class App implements OnDestroy {
  readonly upload = inject(FileUploadService)

  select(event: Event): void {
    if (!(event.target instanceof HTMLInputElement)) return
    const file = event.target.files?.[0]
    event.target.value = ''
    if (file) this.upload.select(file)
  }

  ngOnDestroy(): void {
    this.upload.destroy()
  }
}

import { bootstrapApplication } from '@angular/platform-browser'
import { App } from './app/app.ts'
import { appConfig } from './app/app.config.ts'

bootstrapApplication(App, appConfig).catch(() => console.error('Application startup failed.'))

(
  cd chunked-upload-app &&
  corepack yarn install &&
  corepack yarn build
)

(
  cd chunked-upload-app &&
  node receiver.mts
)

(
  cd chunked-upload-app &&
  corepack yarn start
)