"""Content-addressed blob store (SPEC §4 "Blob store", §6.4). Layout: ``${BLOB_ROOT}/{user_id}/{sha256[:2]}/{sha256}`` (filesystem default). Content-addressed, so identical payloads dedupe. Postgres holds metadata only; deleting a user removes their blob subtree. """ from __future__ import annotations import hashlib import uuid from pathlib import Path from app.config import get_settings def _root() -> Path: return Path(get_settings().blob_root) def sha256_hex(data: bytes) -> str: """Return the hex sha256 digest of ``data``.""" return hashlib.sha256(data).hexdigest() def blob_path(user_id: uuid.UUID, sha: str) -> Path: """Return the on-disk path for a user's blob (does not create it).""" return _root() / str(user_id) / sha[:2] / sha def store(user_id: uuid.UUID, data: bytes) -> str: """Store ``data`` content-addressed; return its sha256. Idempotent (dedupes).""" sha = sha256_hex(data) path = blob_path(user_id, sha) if not path.exists(): path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(".tmp") tmp.write_bytes(data) tmp.replace(path) # atomic within the same directory return sha def load(user_id: uuid.UUID, sha: str) -> bytes: """Load a blob by sha256. Raises FileNotFoundError if absent.""" return blob_path(user_id, sha).read_bytes() def exists(user_id: uuid.UUID, sha: str) -> bool: """Return True if the blob is present.""" return blob_path(user_id, sha).exists() def remove(user_id: uuid.UUID, sha: str) -> None: """Delete a blob if present (used by account/user deletion). Idempotent.""" blob_path(user_id, sha).unlink(missing_ok=True)