서버, 클라이언트, 서비스 워커 공용 코드를 lib/shared로 분리하고 브라우저 전용 모듈을 lib/client로 재구성

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/lib/client/transfer/chunkedUpload.ts

@@ -0,0 +1,177 @@
+import { sha256 } from "@noble/hashes/sha2.js";
+import axios from "axios";
+import pLimit from "p-limit";
+import { encodeToBase64, encryptChunk } from "#lib/shared/crypto/index.js";
+import { CHUNK_SIZE, getChunkCount, getEncryptedSize } from "#lib/shared/format/index.js";
+import { BoundedQueue } from "#lib/utils/concurrency/index.js";
+
+interface UploadStats {
+  progress: number;
+  rate: number;
+}
+
+interface EncryptedChunk {
+  index: number;
+  data: ArrayBuffer;
+}
+
+const createSpeedMeter = (timeWindow = 3000, minInterval = 200, warmupPeriod = 500) => {
+  const samples: { t: number; b: number }[] = [];
+  let lastSpeed = 0;
+  let startTime: number | null = null;
+
+  return (bytesNow?: number) => {
+    if (bytesNow === undefined) return lastSpeed;
+
+    const now = performance.now();
+
+    // Initialize start time on first call
+    if (startTime === null) {
+      startTime = now;
+    }
+
+    // Check if enough time has passed since the last sample
+    const lastSample = samples[samples.length - 1];
+    if (lastSample && now - lastSample.t < minInterval) {
+      return lastSpeed;
+    }
+
+    samples.push({ t: now, b: bytesNow });
+
+    // Remove old samples outside the time window
+    const cutoff = now - timeWindow;
+    while (samples.length > 2 && samples[0]!.t < cutoff) samples.shift();
+
+    // Need at least 2 samples to calculate speed
+    if (samples.length < 2) {
+      return lastSpeed;
+    }
+
+    const first = samples[0]!;
+    const dt = now - first.t;
+    const db = bytesNow - first.b;
+
+    if (dt >= minInterval) {
+      const instantSpeed = (db / dt) * 1000;
+      // Apply EMA for smoother speed transitions
+      const alpha = 0.3;
+      const rawSpeed =
+        lastSpeed === 0 ? instantSpeed : alpha * instantSpeed + (1 - alpha) * lastSpeed;
+
+      // Apply warmup ramp to prevent initial overestimation
+      const elapsed = now - startTime;
+      const warmupWeight = Math.min(1, elapsed / warmupPeriod);
+      lastSpeed = rawSpeed * warmupWeight;
+    }
+
+    return lastSpeed;
+  };
+};
+
+const encryptChunkData = async (chunk: Blob, dataKey: CryptoKey): Promise<ArrayBuffer> => {
+  return await encryptChunk(await chunk.arrayBuffer(), dataKey);
+};
+
+const uploadEncryptedChunk = async (
+  uploadId: string,
+  chunkIndex: number,
+  encrypted: ArrayBuffer,
+  onChunkProgress: (chunkIndex: number, loaded: number) => void,
+) => {
+  await axios.post(`/api/upload/${uploadId}/chunks/${chunkIndex + 1}`, encrypted, {
+    headers: { "Content-Type": "application/octet-stream" },
+    onUploadProgress(e) {
+      onChunkProgress(chunkIndex, e.loaded ?? 0);
+    },
+  });
+
+  onChunkProgress(chunkIndex, encrypted.byteLength);
+};
+
+export const uploadBlob = async (
+  uploadId: string,
+  blob: Blob,
+  dataKey: CryptoKey,
+  options?: { concurrency?: number; onProgress?: (s: UploadStats) => void },
+) => {
+  const onProgress = options?.onProgress;
+  const networkConcurrency = options?.concurrency ?? 4;
+  const maxQueueSize = 8;
+
+  const totalChunks = getChunkCount(blob.size);
+  const totalBytes = getEncryptedSize(blob.size);
+
+  const uploadedByChunk = new Array<number>(totalChunks).fill(0);
+  const speedMeter = createSpeedMeter(3000, 200);
+  const hash = sha256.create();
+
+  const emit = () => {
+    if (!onProgress) return;
+
+    const uploadedBytes = uploadedByChunk.reduce((a, b) => a + b, 0);
+    const rate = speedMeter(uploadedBytes);
+    const progress = Math.min(1, uploadedBytes / totalBytes);
+
+    onProgress({ progress, rate });
+  };
+
+  const onChunkProgress = (idx: number, loaded: number) => {
+    uploadedByChunk[idx] = loaded;
+    emit();
+  };
+
+  const queue = new BoundedQueue<EncryptedChunk>(maxQueueSize);
+  let encryptionError: Error | null = null;
+
+  // Producer: encrypt chunks and push to queue
+  const encryptionProducer = async () => {
+    try {
+      for (let i = 0; i < totalChunks; i++) {
+        const chunk = blob.slice(i * CHUNK_SIZE, (i + 1) * CHUNK_SIZE);
+        const data = await encryptChunkData(chunk, dataKey);
+        hash.update(new Uint8Array(data));
+        await queue.push({ index: i, data });
+      }
+    } catch (e) {
+      encryptionError = e instanceof Error ? e : new Error(String(e));
+    } finally {
+      queue.close();
+    }
+  };
+
+  // Consumer: upload chunks from queue with concurrency limit
+  const uploadConsumer = async () => {
+    const limit = pLimit(networkConcurrency);
+    const activeTasks = new Set<Promise<void>>();
+
+    while (true) {
+      const item = await queue.pop();
+      if (item === null) break;
+      if (encryptionError) throw encryptionError;
+
+      const task = limit(async () => {
+        try {
+          await uploadEncryptedChunk(uploadId, item.index, item.data, onChunkProgress);
+        } finally {
+          // @ts-expect-error
+          item.data = null;
+        }
+      });
+
+      activeTasks.add(task);
+      task.finally(() => activeTasks.delete(task));
+
+      if (activeTasks.size >= networkConcurrency) {
+        await Promise.race(activeTasks);
+      }
+    }
+
+    await Promise.all(activeTasks);
+  };
+
+  // Run producer and consumer concurrently
+  await Promise.all([encryptionProducer(), uploadConsumer()]);
+
+  onProgress?.({ progress: 1, rate: speedMeter() });
+  return { encContentHash: encodeToBase64(hash.digest()) };
+};