spb/drive Public

1/**2 * ─────────────────────────────────────────────3 *  SPB Drive — Personal Cloud Drive4 * ─────────────────────────────────────────────5 *  Author  : Simon-Pierre Boucher6 *  Contact : contact@spboucher.ai7 *  File    : src/storage/blobs.mjs8 *  Purpose : Content-addressed blob store — sha256 CAS, refcounts, GC9 *  License : MIT © Simon-Pierre Boucher10 * ─────────────────────────────────────────────11 */1213import { createHash, randomBytes } from 'node:crypto';14import { createReadStream, createWriteStream, existsSync } from 'node:fs';15import { mkdir, rename, rm, stat } 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';2021/** Absolute on-disk path for a blob sha. */22export function blobPath(sha) {23  return path.join(config.filesDir, sha.slice(0, 2), sha.slice(2, 4), sha);24}2526/**27 * Store a readable stream in the CAS. Streams to a temp file while hashing,28 * then moves into place — identical content dedupes naturally.29 * @param {NodeJS.ReadableStream} stream30 * @returns {Promise<{sha: string, size: number, deduped: boolean}>}31 */32export async function putBlob(stream) {33  const tmp = path.join(config.cacheDir, 'tmp', `put-${randomBytes(8).toString('hex')}`);34  const hash = createHash('sha256');35  let size = 0;3637  await pipeline(38    stream,39    async function* (source) {40      for await (const chunk of source) {41        hash.update(chunk);42        size += chunk.length;43        yield chunk;44      }45    },46    createWriteStream(tmp, { mode: 0o600 }),47  );4849  const sha = hash.digest('hex');50  const dest = blobPath(sha);51  const deduped = existsSync(dest);52  if (deduped) {53    await rm(tmp, { force: true });54  } else {55    await mkdir(path.dirname(dest), { recursive: true, mode: 0o700 });56    await rename(tmp, dest);57  }5859  getDb()60    .prepare(61      `INSERT INTO blobs (sha, size, refcount, created) VALUES (?, ?, 0, ?)62       ON CONFLICT(sha) DO NOTHING`,63    )64    .run(sha, size, Date.now());65  return { sha, size, deduped };66}6768/** Increment a blob's refcount (one node now points at it). */69export function refBlob(sha) {70  getDb().prepare('UPDATE blobs SET refcount = refcount + 1 WHERE sha = ?').run(sha);71}7273/** Decrement a blob's refcount. Physical deletion happens in gcBlobs(). */74export function unrefBlob(sha) {75  getDb().prepare('UPDATE blobs SET refcount = MAX(refcount - 1, 0) WHERE sha = ?').run(sha);76}7778/** Readable stream over a blob (optionally a byte range). */79export function blobStream(sha, { start, end } = {}) {80  return createReadStream(blobPath(sha), start !== undefined ? { start, end } : {});81}8283/** Blob size from disk (throws if missing). */84export async function blobSize(sha) {85  return (await stat(blobPath(sha))).size;86}8788/**89 * Delete blobs with refcount 0 (and their cache artifacts).90 * @returns {Promise<number>} number of blobs removed91 */92export async function gcBlobs() {93  const db = getDb();94  const orphans = db.prepare('SELECT sha FROM blobs WHERE refcount <= 0').all();95  for (const { sha } of orphans) {96    await rm(blobPath(sha), { force: true });97    for (const sub of ['thumbs', 'transcode', 'office', 'peaks', 'text']) {98      const dir = path.join(config.cacheDir, sub);99      for (const suffix of ['', '.256.webp', '.512.webp', '.pdf', '.mp4', '.json', '.txt']) {100        await rm(path.join(dir, `${sha}${suffix}`), { force: true });101      }102    }103    db.prepare('DELETE FROM blobs WHERE sha = ?').run(sha);104  }105  return orphans.length;106}107108/** Aggregate storage stats for /healthz and the storage meter. */109export function storageStats() {110  const db = getDb();111  const blobs = db.prepare('SELECT COUNT(*) AS n, COALESCE(SUM(size), 0) AS bytes FROM blobs WHERE refcount > 0').get();112  const nodes = db.prepare("SELECT COUNT(*) AS n FROM nodes WHERE trashed_at IS NULL AND id != 1").get();113  const byType = db114    .prepare(115      `SELECT CASE116         WHEN mime LIKE 'image/%' THEN 'images'117         WHEN mime LIKE 'video/%' THEN 'video'118         WHEN mime LIKE 'audio/%' THEN 'audio'119         WHEN mime LIKE 'application/pdf' OR mime LIKE '%word%' OR mime LIKE '%sheet%'120           OR mime LIKE '%presentation%' OR mime LIKE 'text/%' THEN 'docs'121         WHEN mime LIKE '%zip%' OR mime LIKE '%tar%' OR mime LIKE '%7z%' OR mime LIKE '%rar%' THEN 'archives'122         ELSE 'other' END AS bucket,123         COALESCE(SUM(size), 0) AS bytes, COUNT(*) AS n124       FROM nodes WHERE type = 'file' AND trashed_at IS NULL GROUP BY bucket`,125    )126    .all();127  return { blobCount: blobs.n, usedBytes: blobs.bytes, nodeCount: nodes.n, byType };128}129