spb/drive Public

1/**2 * ─────────────────────────────────────────────3 *  SPB Drive — Personal Cloud Drive4 * ─────────────────────────────────────────────5 *  Author  : Simon-Pierre Boucher6 *  Contact : contact@spboucher.ai7 *  File    : src/storage/upload.mjs8 *  Purpose : Chunked + resumable uploads (init / chunk / complete / status)9 *  License : MIT © Simon-Pierre Boucher10 * ─────────────────────────────────────────────11 */1213import { randomBytes } from 'node:crypto';14import { createReadStream, createWriteStream, existsSync } from 'node:fs';15import { mkdir, readdir, rm } from 'node:fs/promises';16import path from 'node:path';17import { pipeline } from 'node:stream/promises';18import { config } from '../config.mjs';19import { getDb } from '../db/db.mjs';20import { putBlob } from './blobs.mjs';21import { createFileNode, mustGetNode, validateName } from './nodes.mjs';2223const uploadDir = (id) => path.join(config.uploadsDir, id);2425/**26 * Start (or describe) an upload session.27 * @returns {{uploadId: string, chunkSize: number, nChunks: number, have: number[]}}28 */29export async function initUpload({ parentId, name, size }) {30  validateName(name);31  mustGetNode(parentId);32  const sz = Number(size);33  if (!Number.isFinite(sz) || sz < 0) throw Object.assign(new Error('Bad size'), { statusCode: 400 });3435  const id = randomBytes(16).toString('hex');36  const nChunks = Math.max(1, Math.ceil(sz / config.chunkSize));37  getDb()38    .prepare(39      `INSERT INTO uploads (id, parent_id, name, size, chunk_size, n_chunks, created)40       VALUES (?, ?, ?, ?, ?, ?, ?)`,41    )42    .run(id, parentId, name, sz, config.chunkSize, nChunks, Date.now());43  await mkdir(uploadDir(id), { recursive: true, mode: 0o700 });44  return { uploadId: id, chunkSize: config.chunkSize, nChunks, have: [] };45}4647/**48 * Tag an upload session with its origin (e.g. 'req:<fileRequestId>') so49 * public file-request endpoints can only touch their own sessions.50 */51export function markUploadSource(id, source) {52  getDb().prepare('UPDATE uploads SET source = ? WHERE id = ?').run(source, String(id));53}5455/** Look up an upload session row or throw 404. */56export function getUpload(id) {57  const row = getDb().prepare('SELECT * FROM uploads WHERE id = ?').get(String(id));58  if (!row) throw Object.assign(new Error('Unknown upload'), { statusCode: 404 });59  return row;60}6162/** Which chunk indexes are already on disk (for resume). */63export async function chunksPresent(id) {64  getUpload(id);65  try {66    const entries = await readdir(uploadDir(id));67    return entries68      .filter((e) => /^\d+$/.test(e))69      .map(Number)70      .sort((a, b) => a - b);71  } catch {72    return [];73  }74}7576/** Persist one chunk from a request stream. */77export async function writeChunk(id, n, stream) {78  const up = getUpload(id);79  const idx = Number(n);80  if (!Number.isInteger(idx) || idx < 0 || idx >= up.n_chunks) {81    throw Object.assign(new Error('Chunk index out of range'), { statusCode: 400 });82  }83  await mkdir(uploadDir(id), { recursive: true, mode: 0o700 });84  const tmp = path.join(uploadDir(id), `${idx}.part`);85  await pipeline(stream, createWriteStream(tmp, { mode: 0o600 }));86  const { rename } = await import('node:fs/promises');87  await rename(tmp, path.join(uploadDir(id), String(idx)));88}8990/**91 * Assemble chunks → CAS blob → file node. Cleans up the session.92 * @param {'keep-both'|'replace'|'skip'} conflict93 */94export async function completeUpload(id, { conflict = 'keep-both', mime = null, ip = '' } = {}) {95  const up = getUpload(id);96  const have = await chunksPresent(id);97  if (have.length !== up.n_chunks) {98    throw Object.assign(99      new Error(`Missing chunks: have ${have.length}/${up.n_chunks}`),100      { statusCode: 409 },101    );102  }103104  async function* concat() {105    for (let i = 0; i < up.n_chunks; i += 1) {106      const chunkFile = path.join(uploadDir(id), String(i));107      if (!existsSync(chunkFile)) throw new Error(`Chunk ${i} vanished`);108      yield* createReadStream(chunkFile);109    }110  }111112  const { sha, size } = await putBlob(concat());113  if (size !== up.size) {114    throw Object.assign(115      new Error(`Size mismatch: expected ${up.size}, got ${size}`),116      { statusCode: 409 },117    );118  }119120  const node = createFileNode(up.parent_id, up.name, { sha, size, mime, conflict, ip });121  await abortUpload(id);122  return node;123}124125/** Drop an upload session and its chunk files. */126export async function abortUpload(id) {127  getDb().prepare('DELETE FROM uploads WHERE id = ?').run(String(id));128  await rm(uploadDir(String(id)), { recursive: true, force: true });129}130131/** Remove upload sessions older than 48 h (daily job). */132export async function pruneUploads() {133  const stale = getDb()134    .prepare('SELECT id FROM uploads WHERE created < ?')135    .all(Date.now() - 48 * 3_600_000);136  for (const { id } of stale) await abortUpload(id);137  return stale.length;138}139