Repository Public
1// Upload manager. A module-level store: transfers keep running while the user moves between folders and screens, as2// long as the page stays open. Three files travel at once; each one is a resumable session on the server.3import type { QueryClient } from '@tanstack/react-query';4import { useSyncExternalStore } from 'react';5import { ApiError, del, get, patch, post } from '../api.ts';6import { AUTH_KEY } from '../session.ts';7import type { FileItem, UploadState } from '../types.ts';8import { transfer, wholeFileSha } from '../upload-engine.ts';9import { type Picked, XKEY } from './lib.ts';1011export type Decision = 'version' | 'rename' | 'skip';12export type UpState = 'queued' | 'uploading' | 'paused' | 'finishing' | 'conflict' | 'done' | 'skipped' | 'error';13export interface UploadItem {14 key: number; batch: number; file: File; name: string; size: number; folderId: string; relativePath?: string;15 /** Set when the upload is a new version of a known file. */16 fileId?: string;17 uploadId?: string; sent: number; state: UpState; error?: unknown; speed: number;18 /** Bytes the server already held when an earlier session was continued (0 = started from scratch). */19 resumedFrom: number;20 result?: FileItem; created?: boolean; decided?: Decision;21}22/** A session started before a page reload. It continues when the user picks the same file again. */23export interface Pending { uploadId: string; name: string; size: number; lastModified: number; folderId: string; relativePath?: string }24export interface Snapshot { items: UploadItem[]; pending: Pending[]; batchSizes: Record<number, number> }2526const PARALLEL = 3;27const STORE = 'spd.uploads.pending';28const ACTIVE: UpState[] = ['queued', 'uploading', 'finishing', 'conflict'];2930let items: UploadItem[] = [];31let seq = 0, batchSeq = 0, running = 0;32const ctrls = new Map<number, AbortController>();33const started = new Set<number>(); // keys with a run() in flight34const decisions = new Map<number, Decision>(); // batch → answer given with "apply to all"35const hashes = new WeakMap<File, Promise<string | undefined>>();36const listeners = new Set<() => void>();37let client: QueryClient | null = null;38let snapshot: Snapshot = { items, pending: [], batchSizes: {} };3940const loadPending = (): Pending[] => { try { const v = JSON.parse(localStorage.getItem(STORE) ?? '[]'); return Array.isArray(v) ? v : []; } catch { return []; } };41const savePending = (p: Pending[]) => { try { localStorage.setItem(STORE, JSON.stringify(p.slice(-500))); } catch { /* storage full or disabled: resuming after a reload is lost, nothing else */ } };42const dropPending = (uploadId?: string) => { if (uploadId) savePending(loadPending().filter((p) => p.uploadId !== uploadId)); };4344const warn = (e: BeforeUnloadEvent) => { e.preventDefault(); e.returnValue = ''; };45function emit(): void {46 const mine = new Set(items.map((i) => i.uploadId));47 const batchSizes: Record<number, number> = {};48 for (const i of items) batchSizes[i.batch] = (batchSizes[i.batch] ?? 0) + 1;49 snapshot = { items, pending: loadPending().filter((p) => !mine.has(p.uploadId)), batchSizes };50 if (items.some((i) => ACTIVE.includes(i.state))) window.addEventListener('beforeunload', warn); else window.removeEventListener('beforeunload', warn);51 listeners.forEach((f) => f());52}53const find = (key: number) => items.find((i) => i.key === key);54function set(key: number, p: Partial<UploadItem>): void { items = items.map((i) => (i.key === key ? { ...i, ...p } : i)); emit(); }5556let refreshTimer: ReturnType<typeof setTimeout> | undefined;57function refresh(): void { // several files finishing together refresh the listing once58 clearTimeout(refreshTimer);59 refreshTimer = setTimeout(() => { void client?.invalidateQueries({ queryKey: [XKEY] }); void client?.invalidateQueries({ queryKey: AUTH_KEY }); }, 250);60}61const isAbort = (e: unknown) => (e as Error)?.name === 'AbortError';62const refused = (e: unknown) => e instanceof ApiError && e.status >= 400 && e.status < 500;6364function pump(): void {65 for (const it of items) {66 if (running >= PARALLEL) return;67 if (it.state === 'queued' && !started.has(it.key)) void run(it.key);68 }69}70async function run(key: number): Promise<void> {71 running++; started.add(key);72 const ctrl = new AbortController();73 ctrls.set(key, ctrl);74 const signal = ctrl.signal;75 try {76 const it = find(key)!;77 set(key, { state: 'uploading', error: undefined, speed: 0 });78 let uploadId = it.uploadId, startAt = 0;79 if (!uploadId && !it.fileId) { // the same file was being sent before a reload: continue that session80 const p = loadPending().find((x) => x.name === it.name && x.size === it.size && x.lastModified === it.file.lastModified && x.folderId === it.folderId && (x.relativePath ?? '') === (it.relativePath ?? '') && !items.some((o) => o.uploadId === x.uploadId));81 if (p) uploadId = p.uploadId;82 }83 if (uploadId) {84 try {85 const s = await get<UploadState>(`/uploads/${uploadId}`, { signal });86 if (s.status === 'done') { set(key, { uploadId, state: 'done', sent: it.size }); dropPending(uploadId); refresh(); return; }87 if (s.status !== 'active' || s.size !== it.size) throw new ApiError(410, 'upload_closed', 'closed');88 startAt = s.received;89 set(key, { uploadId, sent: startAt, resumedFrom: it.uploadId ? it.resumedFrom : startAt });90 } catch (e) {91 if (!refused(e)) throw e;92 dropPending(uploadId); uploadId = undefined; set(key, { uploadId: undefined, sent: 0, resumedFrom: 0 });93 }94 }95 if (!uploadId) {96 const s = await post<UploadState>('/uploads', { folderId: it.folderId, name: it.name, size: it.size, relativePath: it.relativePath, onConflict: 'fail', fileId: it.fileId }, { signal });97 uploadId = s.id;98 set(key, { uploadId });99 if (!it.fileId) savePending([...loadPending(), { uploadId, name: it.name, size: it.size, lastModified: it.file.lastModified, folderId: it.folderId, relativePath: it.relativePath }]);100 }101 const t0 = performance.now(), from = startAt;102 await transfer({103 file: it.file, startAt, signal, chunkPath: (o) => `/uploads/${uploadId}?offset=${o}`, statusPath: `/uploads/${uploadId}`,104 onProgress: (sent) => { const dt = (performance.now() - t0) / 1000; set(key, { sent, speed: dt > 0.2 && sent > from ? (sent - from) / dt : 0 }); },105 });106 await finish(key);107 } catch (e) {108 if (isAbort(e)) return; // paused or cancelled: the state was set by whoever aborted109 if (refused(e)) dropPending(find(key)?.uploadId); // the server will not take this session any more110 if (find(key)) set(key, { state: 'error', error: e });111 } finally {112 ctrls.delete(key); started.delete(key); running--;113 pump();114 }115}116/** Ask the server to seal the file. A name collision stops here until the user (or the batch rule) decides. */117async function finish(key: number): Promise<void> {118 const it = find(key);119 if (!it?.uploadId) return;120 set(key, { state: 'finishing', sent: it.size });121 if (!hashes.has(it.file)) hashes.set(it.file, wholeFileSha(it.file));122 const sha256 = await hashes.get(it.file);123 try {124 const r = await post<UploadState & { file: FileItem; created?: boolean }>(`/uploads/${it.uploadId}/complete`, sha256 ? { sha256 } : {});125 dropPending(it.uploadId);126 if (find(key)) set(key, { state: 'done', result: r.file, created: r.created });127 refresh();128 } catch (e) {129 if (!(e instanceof ApiError) || e.code !== 'name_conflict' || it.fileId || it.decided) throw e;130 const rule = decisions.get(it.batch);131 if (rule) return decide(key, rule);132 set(key, { state: 'conflict' });133 }134}135async function decide(key: number, d: Decision): Promise<void> {136 const it = find(key);137 if (!it?.uploadId) return;138 try {139 if (d === 'skip') {140 await del(`/uploads/${it.uploadId}`).catch((e) => { if (!refused(e)) throw e; });141 dropPending(it.uploadId);142 set(key, { state: 'skipped', decided: d });143 return;144 }145 set(key, { state: 'finishing', decided: d });146 await patch(`/uploads/${it.uploadId}/options`, { onConflict: d });147 await finish(key);148 } catch (e) { if (find(key)) set(key, { state: 'error', error: e }); }149}150151export const uploads = {152 /** Called once by the explorer so finished files refresh the listings. */153 bind(qc: QueryClient): void { client = qc; },154 enqueue(picked: Picked[], folderId: string, o: { fileId?: string } = {}): void {155 if (!picked.length) return;156 const batch = ++batchSeq;157 items = [...items, ...picked.map((p): UploadItem => ({ key: ++seq, batch, file: p.file, name: p.file.name, size: p.file.size, folderId, relativePath: p.relativePath, fileId: o.fileId, sent: 0, state: 'queued', speed: 0, resumedFrom: 0 }))];158 emit(); pump();159 },160 pause(key: number): void {161 const it = find(key);162 if (!it || (it.state !== 'uploading' && it.state !== 'queued')) return;163 set(key, { state: 'paused', speed: 0 });164 ctrls.get(key)?.abort();165 },166 /** Resume a paused file or retry a failed one: the server is asked where it stopped. */167 resume(key: number): void {168 const it = find(key);169 if (!it || (it.state !== 'paused' && it.state !== 'error')) return;170 set(key, { state: 'queued', error: undefined, decided: undefined });171 pump();172 },173 cancel(key: number): void {174 const it = find(key);175 if (!it) return;176 ctrls.get(key)?.abort();177 items = items.filter((i) => i.key !== key);178 if (it.uploadId && it.state !== 'done' && it.state !== 'skipped') { dropPending(it.uploadId); void del(`/uploads/${it.uploadId}`).catch(() => {}); }179 emit(); pump();180 },181 /** Answer a name collision; `all` applies the same answer to every collision of the batch, present and to come. */182 resolve(key: number, d: Decision, all: boolean): void {183 const it = find(key);184 if (!it) return;185 if (all) decisions.set(it.batch, d);186 for (const x of all ? items.filter((i) => i.batch === it.batch && i.state === 'conflict') : [it]) void decide(x.key, d);187 },188 clearFinished(): void { items = items.filter((i) => i.state !== 'done' && i.state !== 'skipped'); emit(); },189 /** Give up an interrupted session from before a reload: its bytes are discarded on the server. */190 forget(uploadId: string): void { dropPending(uploadId); void del(`/uploads/${uploadId}`).catch(() => {}); emit(); },191 isActive: (i: UploadItem) => ACTIVE.includes(i.state),192};193snapshot = { items, pending: loadPending(), batchSizes: {} };194195export function useUploads(): Snapshot {196 return useSyncExternalStore((cb) => { listeners.add(cb); return () => listeners.delete(cb); }, () => snapshot);197}198