soulcraft/brainy
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions

feat(8.0)!: delete fork/branch/commit/history/versions — superseded by the Db API

Browse source 
The COW version-control surface (fork, branches, checkout, commit,
getHistory/streamHistory, asOfCommit, brain.versions) is gone, along with
its subsystems: src/versioning/, the COW object store (CommitLog,
CommitObject, RefManager, TreeObject), HistoricalStorageAdapter, and the
TypeAwareHNSWIndex path it kept alive. The Db API (now/transact/asOf/
with/persist/restore) is the one versioning model in 8.0.

Survivors and replacements:
- BlobStorage survives (the VFS stores file content through it), relocated
  to src/storage/blobStorage.ts with binaryDataCodec.ts; its adapter
  interface is now BlobStoreAdapter, slimmed to the consumed surface
  (write/read/has/delete/getMetadata + MIME-aware compression policy).
- brain.migrate() backup branches are replaced by persist-before-migrate:
  MigrateOptions.backupTo persists a hard-link snapshot of the current
  generation before any transform runs; MigrationResult.backupPath reports
  it, and brain.restore(path) brings it back wholesale.
- CLI: cow.ts (fork/branch/checkout/history/migrate) is replaced by
  snapshot.ts — snapshot <path>, restore <path>, history (tx-log),
  generation.
- New public read API: brain.transactionLog({limit}) exposes the reified
  tx-log (generation/timestamp/meta, newest first) that backs the CLI
  history command; TxLogEntry is exported.

Tests: superseded suites deleted; fork/commit blocks excised from shared
suites; BlobStorage tests relocated + reworked against the slimmed store;
migration tests now prove the backupTo snapshot/restore round trip; new
transactionLog coverage in db-mvcc.
This commit is contained in:
David Snelling 2026-06-10 15:22:47 -07:00
parent 431cd64406
commit 8f93add705
64 changed files with 1444 additions and 16313 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/storage/adapters/baseStorageAdapter.ts
View file

@ -152,9 +152,8 @@ export abstract class BaseStorageAdapter implements StorageAdapter {
// "graph-lsm/source/sstable-123" "<root>/_blobs/graph-lsm/source/sstable-123.bin"
//
// i.e. the key's "/"-separated segments are nested under a `_blobs/` prefix and
// suffixed with `.bin`. Blobs are deliberately NOT branch-scoped (COW): they
// are immutable, content-addressed segments managed by their producer,
// mirroring how COW metadata under `_cow/` is also kept global.
// suffixed with `.bin`. Blobs are immutable, content-addressed segments
// managed by their producer.
/**
* Persist a raw binary blob under `key`, writing the bytes verbatim (no JSON

46
src/storage/adapters/fileSystemStorage.ts
View file

@ -912,7 +912,7 @@ export class FileSystemStorage extends BaseStorage {
if (entry.isFile()) {
// Handle multiple compression formats for broad compatibility
// - .json.gz: Standard entity/metadata files (JSON compressed)
// - .gz: COW files (refs, blobs, commits - raw compressed)
// - .gz: Raw compressed payloads (e.g. blob-store binary values)
// - .json: Uncompressed JSON files
if (entry.name.endsWith('.json.gz')) {
// Strip .gz extension and add the .json path
@ -923,7 +923,7 @@ export class FileSystemStorage extends BaseStorage {
seen.add(normalizedPath)
}
} else if (entry.name.endsWith('.gz')) {
// COW files stored as .gz (not .json.gz)
// Raw payloads stored as .gz (not .json.gz)
// Strip .gz extension and return path
const normalizedName = entry.name.slice(0, -3) // Remove .gz
const normalizedPath = path.join(prefix, normalizedName)
@ -1387,12 +1387,10 @@ export class FileSystemStorage extends BaseStorage {
}
}
// Clear the entire branches/ directory (branch-based storage)
// Bug fix: Data is stored in branches/main/entities/, not just entities/
// The branch-based structure was introduced for COW support
const branchesDir = path.join(this.rootDir, 'branches')
if (await this.directoryExists(branchesDir)) {
await removeDirectoryContents(branchesDir)
// Clear the canonical entity/verb data area
const entitiesDir = path.join(this.rootDir, 'entities')
if (await this.directoryExists(entitiesDir)) {
await removeDirectoryContents(entitiesDir)
}
// Remove all files in both system directories
@ -1401,19 +1399,16 @@ export class FileSystemStorage extends BaseStorage {
await removeDirectoryContents(this.indexDir)
}
// Remove COW (copy-on-write) version control data
// This directory stores all git-like versioning data (commits, trees, blobs, refs)
// Must be deleted to fully clear all data including version history
const cowDir = path.join(this.rootDir, '_cow')
if (await this.directoryExists(cowDir)) {
// Delete the entire _cow/ directory (not just contents)
await fs.promises.rm(cowDir, { recursive: true, force: true })
// Remove the content-addressed blob store (VFS file content)
const casDir = path.join(this.rootDir, '_cas')
if (await this.directoryExists(casDir)) {
// Delete the entire _cas/ directory (not just contents)
await fs.promises.rm(casDir, { recursive: true, force: true })
// Reset COW managers (but don't disable COW - it's always enabled)
// COW will re-initialize automatically on next use
this.refManager = undefined
// Drop the BlobStorage instance — its LRU cache holds deleted blobs.
// initializeBlobStorage() re-creates it (brain.clear() does this before
// the VFS is rebuilt).
this.blobStorage = undefined
this.commitLog = undefined
}
// Clear the statistics cache
@ -1426,7 +1421,7 @@ export class FileSystemStorage extends BaseStorage {
;(this as any).totalVerbCount = 0
// Clear write-through cache (inherited from BaseStorage)
// Without this, readWithInheritance() would return stale cached data
// Without this, readCanonicalObject() would return stale cached data
// after clear(), causing "ghost" entities to appear
this.clearWriteCache()
}
@ -1457,17 +1452,6 @@ export class FileSystemStorage extends BaseStorage {
return result
}
/**
* Check if COW has been explicitly disabled via clear()
* Fixes bug where clear() doesn't persist across instance restarts
* @returns true if marker file exists, false otherwise
* @protected
*/
/**
* Removed checkClearMarker() and createClearMarker() methods
* COW is now always enabled - marker files are no longer used
*/
/**
* Get information about storage usage and capacity
*/

571
src/storage/adapters/historicalStorageAdapter.ts
View file

@ -1,571 +0,0 @@
/**
* Historical Storage Adapter
*
* Provides lazy-loading read-only access to a historical commit state.
* Uses LRU cache to bound memory usage and prevent eager-loading of entire history.
*
* Architecture:
* - Extends BaseStorage to inherit all storage infrastructure
* - Wraps an underlying storage adapter to access commit state
* - Implements lazy-loading with LRU cache (bounded memory)
* - All writes throw read-only errors
* - All reads load from historical commit state on-demand
*
* Usage:
* const historical = new HistoricalStorageAdapter({
* underlyingStorage: brain.storage as BaseStorage,
* commitId: 'abc123...',
* cacheSize: 10000 // LRU cache size
* })
* await historical.init()
*
* Performance:
* - O(1) cache lookups for frequently accessed entities
* - Bounded memory: max cacheSize entities in memory
* - Lazy loading: only loads entities when accessed
* - No eager-loading of entire commit state
*
* Production-ready, billion-scale historical queries
*/
import { BaseStorage } from '../baseStorage.js'
import { CommitLog } from '../cow/CommitLog.js'
import { BlobStorage } from '../cow/BlobStorage.js'
import {
HNSWNoun,
HNSWVerb,
NounMetadata,
VerbMetadata,
StatisticsData
} from '../../coreTypes.js'
/**
* Simple LRU Cache implementation
* Bounds memory usage by evicting least-recently-used items
*/
class LRUCache<T> {
private cache = new Map<string, T>()
private accessOrder: string[] = []
private maxSize: number
constructor(maxSize: number = 10000) {
this.maxSize = maxSize
}
get(key: string): T | undefined {
const value = this.cache.get(key)
if (value !== undefined) {
// Move to end (most recently used)
this.accessOrder = this.accessOrder.filter(k => k !== key)
this.accessOrder.push(key)
}
return value
}
set(key: string, value: T): void {
// Remove if already exists
if (this.cache.has(key)) {
this.accessOrder = this.accessOrder.filter(k => k !== key)
}
// Add to cache
this.cache.set(key, value)
this.accessOrder.push(key)
// Evict oldest if over capacity
if (this.cache.size > this.maxSize) {
const oldest = this.accessOrder.shift()
if (oldest) {
this.cache.delete(oldest)
}
}
}
has(key: string): boolean {
return this.cache.has(key)
}
clear(): void {
this.cache.clear()
this.accessOrder = []
}
get size(): number {
return this.cache.size
}
}
export interface HistoricalStorageAdapterOptions {
/** Underlying storage to access commit state from */
underlyingStorage: BaseStorage
/** Commit ID to load historical state from */
commitId: string
/** Max number of entities to cache (default: 10000) */
cacheSize?: number
/** Branch containing the commit (default: 'main') */
branch?: string
}
/**
* Historical Storage Adapter
*
* Lazy-loading, read-only storage adapter for historical commit state.
* Implements billion-scale time-travel queries with bounded memory.
*/
export class HistoricalStorageAdapter extends BaseStorage {
private underlyingStorage: BaseStorage
private commitId: string
private branch: string
private cacheSize: number
// LRU caches for lazy-loaded entities
private cache: LRUCache<any>
// Historical commit state (loaded lazily) - must match BaseStorage visibility
public commitLog?: CommitLog
public blobStorage?: BlobStorage
constructor(options: HistoricalStorageAdapterOptions) {
super()
this.underlyingStorage = options.underlyingStorage
this.commitId = options.commitId
this.branch = options.branch || 'main'
this.cacheSize = options.cacheSize || 10000
this.cache = new LRUCache(this.cacheSize)
}
/**
* Initialize historical storage adapter
* Loads commit metadata but NOT entity data (lazy loading)
*/
public async init(): Promise<void> {
// Get COW components from underlying storage
// Fixed property names - use public properties without underscore prefix
this.commitLog = this.underlyingStorage.commitLog
this.blobStorage = this.underlyingStorage.blobStorage
if (!this.commitLog || !this.blobStorage) {
throw new Error(
'Historical storage requires underlying storage to have COW enabled. ' +
'Call brain.init() first to initialize COW.'
)
}
// Verify commit exists
const commit = await this.commitLog.getCommit(this.commitId)
if (!commit) {
throw new Error(`Commit not found: ${this.commitId}`)
}
// Initialize GraphAdjacencyIndex and type statistics
await super.init()
}
// ============= Abstract Method Implementations =============
/**
* Read object from historical commit state
* Uses LRU cache to avoid repeated blob reads
*/
protected async readObjectFromPath(path: string): Promise<any | null> {
// Check cache first
if (this.cache.has(path)) {
return this.cache.get(path) || null
}
try {
// Import COW classes
const { CommitObject } = await import('../cow/CommitObject.js')
const { TreeObject } = await import('../cow/TreeObject.js')
const { isNullHash } = await import('../cow/constants.js')
// Read commit
const commit = await CommitObject.read(this.blobStorage!, this.commitId)
if (isNullHash(commit.tree)) {
return null
}
// Read tree
const tree = await TreeObject.read(this.blobStorage!, commit.tree)
// Walk tree to find matching path
for await (const entry of TreeObject.walk(this.blobStorage!, tree)) {
if (entry.type === 'blob' && entry.name === path) {
// Read blob data
const blobData = await this.blobStorage!.read(entry.hash)
const data = JSON.parse(blobData.toString())
// Cache the result
this.cache.set(path, data)
return data
}
}
return null
} catch (error) {
// Path doesn't exist in historical state
return null
}
}
/**
* List objects under path in historical commit state
*/
protected async listObjectsUnderPath(prefix: string): Promise<string[]> {
try {
// Import COW classes
const { CommitObject } = await import('../cow/CommitObject.js')
const { TreeObject } = await import('../cow/TreeObject.js')
const { isNullHash } = await import('../cow/constants.js')
// Read commit
const commit = await CommitObject.read(this.blobStorage!, this.commitId)
if (isNullHash(commit.tree)) {
return []
}
// Read tree
const tree = await TreeObject.read(this.blobStorage!, commit.tree)
// Walk tree to find all paths matching prefix
const paths: string[] = []
for await (const entry of TreeObject.walk(this.blobStorage!, tree)) {
if (entry.name.startsWith(prefix)) {
paths.push(entry.name)
}
}
return paths
} catch (error) {
return []
}
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
protected async writeObjectToPath(path: string, data: any): Promise<void> {
throw new Error(
`Historical storage is read-only. Cannot write to path: ${path}`
)
}
/**
* DELETE BLOCKED: Historical storage is read-only
*/
protected async deleteObjectFromPath(path: string): Promise<void> {
throw new Error(
`Historical storage is read-only. Cannot delete path: ${path}`
)
}
// ===========================================================================
// Raw binary-blob primitive (read-only)
// ===========================================================================
/**
* WRITE BLOCKED: Historical storage is read-only.
*
* @param key - The blob key (unused; included for the error message).
* @throws Always — historical state is immutable.
*/
public async saveBinaryBlob(key: string, _data: Buffer): Promise<void> {
throw new Error(
`Historical storage is read-only. Cannot save binary blob: ${key}`
)
}
/**
* Load the raw bytes of a binary blob from the historical commit state by
* walking the commit tree for the blob entry at `_blobs/<key>.bin` and reading
* its content-addressed bytes verbatim (no JSON decode). Returns `null` if the
* blob was not present in this commit.
*
* @param key - The blob key (same convention as the live adapters).
* @returns The blob bytes as stored at this commit, or `null` if absent.
*/
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
try {
const { CommitObject } = await import('../cow/CommitObject.js')
const { TreeObject } = await import('../cow/TreeObject.js')
const { isNullHash } = await import('../cow/constants.js')
const commit = await CommitObject.read(this.blobStorage!, this.commitId)
if (isNullHash(commit.tree)) {
return null
}
const tree = await TreeObject.read(this.blobStorage!, commit.tree)
const blobName = `_blobs/${key}.bin`
for await (const entry of TreeObject.walk(this.blobStorage!, tree)) {
if (entry.type === 'blob' && entry.name === blobName) {
return await this.blobStorage!.read(entry.hash)
}
}
return null
} catch (error) {
// Blob not present in historical state
return null
}
}
/**
* DELETE BLOCKED: Historical storage is read-only.
*
* @param key - The blob key (unused; included for the error message).
* @throws Always — historical state is immutable.
*/
public async deleteBinaryBlob(key: string): Promise<void> {
throw new Error(
`Historical storage is read-only. Cannot delete binary blob: ${key}`
)
}
/**
* Historical state lives in the content-addressed commit store, not on the
* local filesystem, so there is no mmap-able path. Always returns `null`.
*
* @param _key - The blob key (unused).
* @returns Always `null`.
*/
public getBinaryBlobPath(_key: string): string | null {
return null
}
/**
* Get storage statistics from historical commit
*/
protected async getStatisticsData(): Promise<StatisticsData | null> {
return await this.readObjectFromPath('_system/statistics.json')
}
/**
* WRITE BLOCKED: Cannot save statistics to historical storage
*/
protected async saveStatisticsData(data: StatisticsData): Promise<void> {
throw new Error('Historical storage is read-only. Cannot save statistics.')
}
/**
* Clear cache (does not affect historical data)
*/
public async clear(): Promise<void> {
this.cache.clear()
}
/**
* Get storage status
*/
public async getStorageStatus(): Promise<{
type: string
used: number
quota: number | null
details?: Record<string, any>
}> {
return {
type: 'historical',
used: this.cache.size,
quota: this.cacheSize,
details: {
commitId: this.commitId,
branch: this.branch,
cached: this.cache.size,
maxCache: this.cacheSize,
readOnly: true
}
}
}
/**
* Check if COW has been explicitly disabled via clear()
* No-op for HistoricalStorageAdapter (read-only, doesn't manage COW)
* @returns Always false (read-only adapter doesn't manage COW state)
* @protected
*/
/**
* Removed checkClearMarker() and createClearMarker() methods
* COW is now always enabled - marker files are no longer used
*/
// ============= Override Write Methods (Read-Only) =============
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async saveNoun(noun: HNSWNoun): Promise<void> {
throw new Error('Historical storage is read-only. Cannot save noun.')
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async saveNounMetadata(id: string, metadata: NounMetadata): Promise<void> {
throw new Error('Historical storage is read-only. Cannot save noun metadata.')
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async deleteNoun(id: string): Promise<void> {
throw new Error('Historical storage is read-only. Cannot delete noun.')
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async deleteNounMetadata(id: string): Promise<void> {
throw new Error('Historical storage is read-only. Cannot delete noun metadata.')
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async saveVerb(verb: HNSWVerb): Promise<void> {
throw new Error('Historical storage is read-only. Cannot save verb.')
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async saveVerbMetadata(id: string, metadata: VerbMetadata): Promise<void> {
throw new Error('Historical storage is read-only. Cannot save verb metadata.')
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async deleteVerb(id: string): Promise<void> {
throw new Error('Historical storage is read-only. Cannot delete verb.')
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async deleteVerbMetadata(id: string): Promise<void> {
throw new Error('Historical storage is read-only. Cannot delete verb metadata.')
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async saveMetadata(id: string, metadata: any): Promise<void> {
throw new Error('Historical storage is read-only. Cannot save metadata.')
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async saveVectorIndexData(nounId: string, hnswData: {
level: number
connections: Record<string, string[]>
}): Promise<void> {
throw new Error('Historical storage is read-only. Cannot save HNSW data.')
}
/**
* WRITE BLOCKED: Historical storage is read-only
*/
public async saveHNSWSystem(systemData: {
entryPointId: string | null
maxLevel: number
}): Promise<void> {
throw new Error('Historical storage is read-only. Cannot save HNSW system data.')
}
// ============= Additional Abstract Methods =============
/**
* Get noun vector from historical state
*/
public async getNounVector(id: string): Promise<number[] | null> {
const noun = await this.getNoun(id)
return noun?.vector || null
}
/**
* Get HNSW data from historical state
*/
public async getVectorIndexData(nounId: string): Promise<{
level: number
connections: Record<string, string[]>
} | null> {
const path = `_system/hnsw/nodes/${nounId}.json`
return await this.readObjectFromPath(path)
}
/**
* Get HNSW system data from historical state
*/
public async getHNSWSystem(): Promise<{
entryPointId: string | null
maxLevel: number
} | null> {
return await this.readObjectFromPath('_system/hnsw/system.json')
}
/**
* Initialize counts (no-op for historical storage)
* Counts are loaded from historical state metadata
*/
protected async initializeCounts(): Promise<void> {
// No-op: Historical storage doesn't need to initialize counts
// They're read from commit state metadata
}
/**
* WRITE BLOCKED: Cannot persist counts to historical storage
*/
protected async persistCounts(): Promise<void> {
// No-op: Historical storage is read-only
}
// ===========================================================================
// Generational record layer (8.0 MVCC) — not applicable to this adapter
//
// HistoricalStorageAdapter is the pre-8.0 commit-based time-travel view and
// is read-only by construction. The generation store never registers its
// write hooks on reader-mode instances, so these methods are unreachable in
// normal operation; they throw explicitly rather than fake behavior.
// ===========================================================================
/**
* WRITE BLOCKED: the tx-log belongs to the live store, not a historical view.
*/
public async appendTxLogLine(_line: string): Promise<void> {
throw new Error('Historical storage is read-only. Cannot append to the transaction log.')
}
/**
* Not supported: historical commit views predate the generational tx-log.
*/
public async readTxLogLines(): Promise<string[]> {
throw new Error(
'Transaction-log reads are not supported on a historical commit view. ' +
'Use the live store (or a Db from brain.now()/brain.asOf()).'
)
}
/**
* Not supported: snapshot a live store via db.persist(path) instead.
*/
public async snapshotToDirectory(_targetPath: string): Promise<void> {
throw new Error(
'snapshotToDirectory is not supported on a historical commit view. ' +
'Call db.persist(path) on the live store instead.'
)
}
/**
* WRITE BLOCKED: cannot restore into a read-only historical view.
*/
public async restoreFromDirectory(_sourcePath: string): Promise<void> {
throw new Error('Historical storage is read-only. Cannot restore from a snapshot.')
}
}

18
src/storage/adapters/memoryStorage.ts
View file

@ -362,8 +362,13 @@ export class MemoryStorage extends BaseStorage {
this.statisticsCache = null
this.statisticsModified = false
// Drop the BlobStorage instance — its LRU cache holds deleted blobs.
// initializeBlobStorage() re-creates it (brain.clear() does this before
// the VFS is rebuilt).
this.blobStorage = undefined
// Clear write-through cache (inherited from BaseStorage)
// Without this, readWithInheritance() would return stale cached data
// Without this, readCanonicalObject() would return stale cached data
// after clear(), causing "ghost" entities to appear
this.clearWriteCache()
}
@ -390,17 +395,6 @@ export class MemoryStorage extends BaseStorage {
}
}
/**
* Check if COW has been explicitly disabled via clear()
* No-op for MemoryStorage (doesn't persist)
* @returns Always false (marker doesn't persist in memory)
* @protected
*/
/**
* Removed checkClearMarker() and createClearMarker() methods
* COW is now always enabled - marker files are no longer used
*/
/**
* Save statistics data to storage
* @param statistics The statistics data to save

600
src/storage/baseStorage.ts
View file

File diff suppressed because it is too large Load diff

77
src/storage/binaryDataCodec.ts Normal file
View file

@ -0,0 +1,77 @@
/**
* @module storage/binaryDataCodec
* @description Single source of truth for wrapping/unwrapping binary payloads
* stored in JSON-based object storage. Binary bytes (compressed blobs, raw
* buffers) are persisted as `{ _binary: true, data: "<base64>" }` so they can
* travel through adapters whose object primitive is JSON. `unwrapBinaryData`
* reverses that wrapping — and MUST be used everywhere bytes come back out,
* because content hashes are computed over the original bytes, not the
* wrapper.
*
* Used by `BaseStorage`'s blob-store bridge and by `BlobStorage`'s
* defense-in-depth verification path.
*/
/**
* @description Wrapped binary data format used when storing binary data in
* JSON-based storage.
*/
export interface WrappedBinaryData {
_binary: true
data: string // base64-encoded
}
/**
* @description Type guard for the wrapped binary format.
* @param data - Candidate value.
* @returns True when `data` is a `{ _binary: true, data: string }` wrapper.
*/
export function isWrappedBinary(data: any): data is WrappedBinaryData {
return (
typeof data === 'object' &&
data !== null &&
data._binary === true &&
typeof data.data === 'string'
)
}
/**
* @description Unwrap binary data from its JSON wrapper.
*
* Handles:
* - `Buffer` → `Buffer` (pass-through)
* - `{ _binary: true, data: "<base64>" }` → `Buffer` (unwrap)
* - Plain object → `Buffer` (JSON stringify — defensive)
* - String → `Buffer`
*
* @param data - Data to unwrap (Buffer, wrapped object, plain object, or string).
* @returns The unwrapped bytes.
* @throws Error when the value cannot be interpreted as binary data.
*/
export function unwrapBinaryData(data: any): Buffer {
// Case 1: Already a Buffer (no unwrapping needed)
if (Buffer.isBuffer(data)) {
return data
}
// Case 2: Wrapped binary data {_binary: true, data: "base64..."}
if (isWrappedBinary(data)) {
return Buffer.from(data.data, 'base64')
}
// Case 3: Plain object (shouldn't happen for binary blobs, but handle gracefully)
if (typeof data === 'object' && data !== null) {
return Buffer.from(JSON.stringify(data))
}
// Case 4: String (convert to Buffer)
if (typeof data === 'string') {
return Buffer.from(data)
}
// Case 5: Invalid type
throw new Error(
`Invalid data type for unwrap: ${typeof data}. ` +
`Expected Buffer or {_binary: true, data: "base64..."}`
)
}

498
src/storage/blobStorage.ts Normal file
View file

@ -0,0 +1,498 @@
/**
* @module storage/blobStorage
* @description Content-addressed blob store. Backs VFS file content with
* SHA-256 addressing (automatic deduplication), reference counting, zstd
* compression where it pays (MIME-aware: already-compressed media is stored
* raw), and an LRU read cache.
*
* The store persists through a narrow key-value bridge
* ({@link BlobStoreAdapter}) provided by `BaseStorage`, which roots all keys
* under the `_cas/` storage area. Key naming is an explicit type contract:
* `blob:<hash>` keys hold binary bytes, `blob-meta:<hash>` keys hold JSON
* metadata — the key format decides how bytes are encoded, never content
* sniffing.
*/
import { createHash } from 'crypto'
import { unwrapBinaryData } from './binaryDataCodec.js'
/**
* @description Key-value bridge the blob store persists through. Implemented
* by `BaseStorage.initializeBlobStorage()` over the adapter's raw object
* primitives.
*/
export interface BlobStoreAdapter {
/** Read the bytes stored under `key`, or `undefined` when absent. */
get(key: string): Promise<Buffer | undefined>
/** Persist `data` under `key` (overwrites). */
put(key: string, data: Buffer): Promise<void>
/** Delete the value under `key`. Missing keys are ignored. */
delete(key: string): Promise<void>
/** List all keys starting with `prefix`. */
list(prefix: string): Promise<string[]>
}
/**
* @description Metadata persisted alongside each blob (under
* `blob-meta:<hash>`).
*/
export interface BlobMetadata {
/** SHA-256 content hash (the blob's identity). */
hash: string
/** Original (uncompressed) size in bytes. */
size: number
/** Stored size in bytes (after compression, if any). */
compressedSize: number
/** Compression applied to the stored bytes. */
compression: 'none' | 'zstd'
/** Creation timestamp (epoch ms). */
createdAt: number
/** Number of logical references to this blob (deduplicated writes). */
refCount: number
}
/**
* @description Options for {@link BlobStorage.write}.
*/
export interface BlobWriteOptions {
/**
* Compression strategy. `'auto'` (default) compresses payloads above 1 KB
* with zstd unless the MIME type says the bytes are already compressed.
* Explicit `'none'`/`'zstd'` is honoured as asserted by the caller.
*/
compression?: 'none' | 'zstd' | 'auto'
/**
* Content type of the payload (e.g. `image/jpeg`, `video/mp4`).
*
* When set on an `auto`-compression write, the store skips zstd for MIME
* types that are already heavily compressed (JPEG, PNG, WebP, MP4, WebM,
* MP3, ZIP, PDF, etc.). zstd over these formats wastes CPU and rarely
* shaves more than a single-digit percent — usually it actually grows the
* payload because the entropy is already maximised by the format itself.
*
* Has no effect when `compression` is `'none'` or `'zstd'` explicitly —
* the caller is asserting the choice and the store honours it.
*/
mimeType?: string
}
/**
* MIME types whose payload is already heavily compressed. zstd over these is
* almost always a CPU-only loss — the bytes are already near entropy-maximal,
* so the output is the same size or slightly larger plus the cost of running
* the compressor. Used by `BlobStorage.selectCompression()` in `auto` mode.
*
* Conservative denylist (well-known formats only). Anything not in this set
* goes through the size heuristic. False negatives (compressing something we
* should have skipped) waste CPU; false positives (skipping something we
* could have compressed) waste a few percent of bytes. The denylist favours
* CPU-cycle safety because the formats listed here are the ones where
* gzip/zstd is reliably a net loss.
*/
const ALREADY_COMPRESSED_MIME_TYPES = new Set<string>([
// Images
'image/jpeg', 'image/jpg', 'image/png', 'image/gif', 'image/webp',
'image/avif', 'image/heic', 'image/heif', 'image/jp2',
// Video
'video/mp4', 'video/webm', 'video/x-matroska', 'video/quicktime',
'video/x-msvideo', 'video/mpeg', 'video/3gpp', 'video/x-ms-wmv',
// Audio
'audio/mpeg', 'audio/mp4', 'audio/aac', 'audio/ogg', 'audio/webm',
'audio/opus', 'audio/flac', 'audio/x-ms-wma',
// Archives
'application/zip', 'application/gzip', 'application/x-gzip',
'application/x-bzip2', 'application/x-7z-compressed',
'application/x-rar-compressed', 'application/x-xz', 'application/x-zstd',
'application/x-compress', 'application/vnd.rar',
// Documents with internal compression
'application/pdf', 'application/epub+zip',
// Office formats (zip-based)
'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
'application/vnd.openxmlformats-officedocument.presentationml.presentation',
'application/vnd.oasis.opendocument.text',
'application/vnd.oasis.opendocument.spreadsheet',
'application/vnd.oasis.opendocument.presentation'
])
/**
* @description True when the MIME type names a payload format known to be
* already heavily compressed. Strips any `;charset=…` / `;boundary=…`
* parameters and lowercases the bare type/subtype before lookup, so
* `'IMAGE/JPEG; charset=binary'` and `'image/jpeg'` resolve identically.
* @param mimeType - MIME type string, or `undefined`.
* @returns Whether `auto` compression should skip zstd for this payload.
*/
export function isAlreadyCompressedMimeType(mimeType: string | undefined): boolean {
if (!mimeType) return false
const bare = mimeType.split(';', 1)[0].trim().toLowerCase()
return ALREADY_COMPRESSED_MIME_TYPES.has(bare)
}
/**
* LRU cache entry.
*/
interface CacheEntry {
data: Buffer
metadata: BlobMetadata
lastAccess: number
size: number
}
/**
* @description Content-addressed, deduplicating, reference-counted blob
* store with MIME-aware zstd compression and an LRU read cache. See the
* module doc for the persistence contract.
*
* @example
* const hash = await blobStorage.write(buffer, { mimeType: 'image/png' })
* const bytes = await blobStorage.read(hash) // verified against the hash
* await blobStorage.delete(hash) // decrements refCount first
*/
export class BlobStorage {
private adapter: BlobStoreAdapter
private cache: Map<string, CacheEntry>
private cacheMaxSize: number
private currentCacheSize: number
// Compression (lazily loaded)
private zstdCompress?: (data: Buffer) => Promise<Buffer>
private zstdDecompress?: (data: Buffer) => Promise<Buffer>
private compressionReady = false
// Configuration
private readonly CACHE_MAX_SIZE = 100 * 1024 * 1024 // 100MB default
private readonly COMPRESSION_THRESHOLD = 1024 // 1KB - don't compress smaller
/**
* @param adapter - Key-value bridge to persist through.
* @param options - `cacheMaxSize` bounds the LRU read cache (bytes,
* default 100 MB).
*/
constructor(adapter: BlobStoreAdapter, options?: { cacheMaxSize?: number }) {
this.adapter = adapter
this.cache = new Map()
this.cacheMaxSize = options?.cacheMaxSize ?? this.CACHE_MAX_SIZE
this.currentCacheSize = 0
}
/**
* Lazy-load the zstd compression module. Falls back to uncompressed
* storage when the optional dependency is unavailable.
*/
private async ensureCompressionReady(): Promise<void> {
if (this.compressionReady) return
try {
// Dynamic import to avoid loading if not needed
// @ts-ignore - Optional dependency, gracefully handled if missing
const zstd = await import('@mongodb-js/zstd')
this.zstdCompress = async (data: Buffer) => {
return Buffer.from(await zstd.compress(data, 3)) // Level 3 = fast
}
this.zstdDecompress = async (data: Buffer) => {
return Buffer.from(await zstd.decompress(data))
}
} catch (error) {
console.warn('zstd compression not available, falling back to uncompressed')
this.zstdCompress = undefined
this.zstdDecompress = undefined
}
this.compressionReady = true
}
/**
* @description Compute the SHA-256 content hash of `data`.
* @param data - Bytes to hash.
* @returns Hex-encoded SHA-256 hash.
*/
static hash(data: Buffer): string {
return createHash('sha256').update(data).digest('hex')
}
/**
* @description Write a blob. Content-addressed: the SHA-256 hash of the
* bytes is the storage key, so identical payloads deduplicate (the
* existing blob's reference count is incremented instead of rewriting).
*
* @param data - Blob bytes.
* @param options - Compression strategy and MIME hint (see
* {@link BlobWriteOptions}).
* @returns The blob's SHA-256 hash.
*/
async write(data: Buffer, options: BlobWriteOptions = {}): Promise<string> {
const hash = BlobStorage.hash(data)
// Deduplication: identical content already stored — just add a reference.
if (await this.has(hash)) {
await this.incrementRefCount(hash)
return hash
}
await this.ensureCompressionReady()
// Determine compression strategy
const compression = this.selectCompression(data, options)
// Compress if needed
let finalData = data
let compressedSize = data.length
if (compression === 'zstd' && this.zstdCompress) {
finalData = await this.zstdCompress(data)
compressedSize = finalData.length
}
// Record the ACTUAL compression state, not the intended one — prevents
// corruption if compression failed to initialize.
const actualCompression = finalData === data ? 'none' : compression
const metadata: BlobMetadata = {
hash,
size: data.length,
compressedSize,
compression: actualCompression,
createdAt: Date.now(),
refCount: 1
}
await this.adapter.put(`blob:${hash}`, finalData)
await this.adapter.put(`blob-meta:${hash}`, Buffer.from(JSON.stringify(metadata)))
// Write-through cache (caches the ORIGINAL bytes, not the compressed form)
this.addToCache(hash, data, metadata)
return hash
}
/**
* @description Read a blob: LRU cache first, then storage with
* decompression and integrity verification (the bytes are re-hashed and
* compared against the requested hash).
*
* @param hash - The blob's SHA-256 hash.
* @returns The original (decompressed) blob bytes.
* @throws Error when the blob is missing or fails integrity verification.
*/
async read(hash: string): Promise<Buffer> {
// Check cache first
const cached = this.getFromCache(hash)
if (cached) {
return cached.data
}
const metadataBuffer = await this.adapter.get(`blob-meta:${hash}`)
if (!metadataBuffer) {
throw new Error(`Blob metadata not found: ${hash}`)
}
// Unwrap before parsing (defense-in-depth): metadata should come back as
// JSON bytes, but an adapter might return the wrapped binary format.
const metadata: BlobMetadata = JSON.parse(unwrapBinaryData(metadataBuffer).toString())
const data = await this.adapter.get(`blob:${hash}`)
if (!data) {
throw new Error(`Blob not found: ${hash}`)
}
// Decompress if needed
let finalData = data
if (metadata.compression === 'zstd') {
if (!this.zstdDecompress) {
await this.ensureCompressionReady()
}
if (!this.zstdDecompress) {
throw new Error('zstd decompression not available')
}
finalData = await this.zstdDecompress(data)
}
// Defense-in-depth unwrap: even though the bridge unwraps, verify it
// happened and re-unwrap if needed. Hash verification must run on the
// original content bytes.
const unwrappedData = unwrapBinaryData(finalData)
// Integrity verification (always on — a content-addressed store that
// returns bytes not matching the address is corruption, not a result)
if (BlobStorage.hash(unwrappedData) !== hash) {
throw new Error(`Blob integrity check failed: ${hash}`)
}
this.addToCache(hash, unwrappedData, metadata)
return unwrappedData
}
/**
* @description Whether a blob with this hash exists (cache or storage).
* @param hash - The blob's SHA-256 hash.
* @returns True when the blob exists.
*/
async has(hash: string): Promise<boolean> {
if (this.cache.has(hash)) {
return true
}
const exists = await this.adapter.get(`blob:${hash}`)
return exists !== undefined
}
/**
* @description Drop one reference to the blob. The stored bytes and
* metadata are physically deleted only when the reference count reaches
* zero — deduplicated content shared by other writers survives.
*
* @param hash - The blob's SHA-256 hash.
*/
async delete(hash: string): Promise<void> {
const refCount = await this.decrementRefCount(hash)
// Only delete if no references remain
if (refCount > 0) {
return
}
await this.adapter.delete(`blob:${hash}`)
await this.adapter.delete(`blob-meta:${hash}`)
this.removeFromCache(hash)
}
/**
* @description Read a blob's metadata without reading its bytes.
* @param hash - The blob's SHA-256 hash.
* @returns The metadata, or `undefined` when the blob does not exist.
*/
async getMetadata(hash: string): Promise<BlobMetadata | undefined> {
const data = await this.adapter.get(`blob-meta:${hash}`)
if (data) {
return JSON.parse(unwrapBinaryData(data).toString())
}
return undefined
}
// ========== PRIVATE METHODS ==========
/**
* Select the compression strategy for a write (see
* {@link BlobWriteOptions.compression}).
*/
private selectCompression(
data: Buffer,
options: BlobWriteOptions
): 'none' | 'zstd' {
if (options.compression === 'none') {
return 'none'
}
if (options.compression === 'zstd') {
return this.zstdCompress ? 'zstd' : 'none'
}
// Auto mode
if (data.length < this.COMPRESSION_THRESHOLD) {
return 'none' // Too small to benefit
}
// Content-type policy: skip already-compressed media. zstd over
// JPEG / MP4 / ZIP etc. is a CPU loss for no measurable byte savings,
// and on hot save paths (image / video uploads) it's the difference
// between fast and slow. Applies only to `auto`; explicit `'zstd'` is
// honoured because the caller is asserting the choice.
if (isAlreadyCompressedMimeType(options.mimeType)) {
return 'none'
}
return this.zstdCompress ? 'zstd' : 'none'
}
/**
* Increment the reference count for an existing blob.
*/
private async incrementRefCount(hash: string): Promise<number> {
const metadata = await this.getMetadata(hash)
if (!metadata) {
throw new Error(`Cannot increment ref count, blob not found: ${hash}`)
}
metadata.refCount++
await this.adapter.put(`blob-meta:${hash}`, Buffer.from(JSON.stringify(metadata)))
return metadata.refCount
}
/**
* Decrement the reference count for a blob (floored at zero).
*/
private async decrementRefCount(hash: string): Promise<number> {
const metadata = await this.getMetadata(hash)
if (!metadata) {
return 0
}
metadata.refCount = Math.max(0, metadata.refCount - 1)
await this.adapter.put(`blob-meta:${hash}`, Buffer.from(JSON.stringify(metadata)))
return metadata.refCount
}
/**
* Add a blob to the LRU cache (evicting least-recently-used entries to
* stay under the size bound).
*/
private addToCache(hash: string, data: Buffer, metadata: BlobMetadata): void {
if (data.length > this.cacheMaxSize) {
return // Blob too large for cache
}
while (
this.currentCacheSize + data.length > this.cacheMaxSize &&
this.cache.size > 0
) {
this.evictLRU()
}
this.cache.set(hash, {
data,
metadata,
lastAccess: Date.now(),
size: data.length
})
this.currentCacheSize += data.length
}
/**
* Get a blob from the cache, refreshing its LRU position.
*/
private getFromCache(hash: string): CacheEntry | undefined {
const entry = this.cache.get(hash)
if (entry) {
entry.lastAccess = Date.now() // Update LRU
}
return entry
}
/**
* Remove a blob from the cache.
*/
private removeFromCache(hash: string): void {
const entry = this.cache.get(hash)
if (entry) {
this.cache.delete(hash)
this.currentCacheSize -= entry.size
}
}
/**
* Evict the least-recently-used cache entry.
*/
private evictLRU(): void {
let oldestHash: string | null = null
let oldestTime = Infinity
for (const [hash, entry] of this.cache.entries()) {
if (entry.lastAccess < oldestTime) {
oldestTime = entry.lastAccess
oldestHash = hash
}
}
if (oldestHash) {
this.removeFromCache(oldestHash)
}
}
}

763
src/storage/cow/BlobStorage.ts
View file

@ -1,763 +0,0 @@
/**
* BlobStorage: Content-Addressable Blob Storage for COW (Copy-on-Write)
*
* State-of-the-art implementation featuring:
* - Content-addressable: SHA-256 hashing
* - Type-aware chunking: Separate vectors, metadata, relationships
* - Compression: zstd for JSON, optimized for vectors
* - LRU caching: Hot blob performance
* - Streaming: Multipart upload for large blobs
* - Batch operations: Parallel I/O
* - Integrity: Cryptographic verification
* - Observability: Metrics and tracing
*
* @module storage/cow/BlobStorage
*/
import { createHash } from 'crypto'
import { NULL_HASH, isNullHash } from './constants.js'
import { unwrapBinaryData } from './binaryDataCodec.js'
/**
* Simple key-value storage interface for COW primitives
* This will be implemented by BaseStorage when COW is integrated
*/
export interface COWStorageAdapter {
get(key: string): Promise<Buffer | undefined>
put(key: string, data: Buffer): Promise<void>
delete(key: string): Promise<void>
list(prefix: string): Promise<string[]>
}
/**
* Blob metadata stored alongside blob data
*/
export interface BlobMetadata {
hash: string // SHA-256 hash
size: number // Original size in bytes
compressedSize: number // Compressed size in bytes
compression: 'none' | 'zstd'
type: 'vector' | 'metadata' | 'tree' | 'commit' | 'blob' | 'raw'
createdAt: number // Timestamp
refCount: number // How many objects reference this blob
}
/**
* Blob write options
*/
export interface BlobWriteOptions {
compression?: 'none' | 'zstd' | 'auto' // Auto chooses based on type
type?: 'vector' | 'metadata' | 'tree' | 'commit' | 'blob' | 'raw'
/**
* Content type of the payload (e.g. `image/jpeg`, `video/mp4`).
*
* When set on an `auto`-compression write, BlobStorage skips zstd for MIME
* types that are already heavily compressed (JPEG, PNG, WebP, MP4, WebM,
* MP3, ZIP, PDF, etc.). Gzip/zstd over these formats wastes CPU and rarely
* shaves more than a single-digit percent — usually it actually grows the
* payload because the entropy is already maximised by the format itself.
*
* Has no effect when `compression` is set to `'none'` or `'zstd'` explicitly
* — the caller is asserting the choice and BlobStorage honours it.
*/
mimeType?: string
skipVerification?: boolean // Skip hash verification (faster, less safe)
}
/**
* MIME types whose payload is already heavily compressed. zstd over these is
* almost always a CPU-only loss — the bytes are already near entropy-maximal,
* so the output is the same size or slightly larger plus the cost of running
* the compressor. Used by `BlobStorage.selectCompression()` in `auto` mode.
*
* Conservative denylist (well-known formats only). Anything not in this set
* goes through the existing per-type heuristic. False negatives (compressing
* something we should have skipped) waste CPU; false positives (skipping
* something we could have compressed) waste a few percent of bytes. The
* denylist favours CPU-cycle safety because the formats listed here are the
* ones where gzip/zstd is reliably a net loss.
*/
const ALREADY_COMPRESSED_MIME_TYPES = new Set<string>([
// Images
'image/jpeg', 'image/jpg', 'image/png', 'image/gif', 'image/webp',
'image/avif', 'image/heic', 'image/heif', 'image/jp2',
// Video
'video/mp4', 'video/webm', 'video/x-matroska', 'video/quicktime',
'video/x-msvideo', 'video/mpeg', 'video/3gpp', 'video/x-ms-wmv',
// Audio
'audio/mpeg', 'audio/mp4', 'audio/aac', 'audio/ogg', 'audio/webm',
'audio/opus', 'audio/flac', 'audio/x-ms-wma',
// Archives
'application/zip', 'application/gzip', 'application/x-gzip',
'application/x-bzip2', 'application/x-7z-compressed',
'application/x-rar-compressed', 'application/x-xz', 'application/x-zstd',
'application/x-compress', 'application/vnd.rar',
// Documents with internal compression
'application/pdf', 'application/epub+zip',
// Office formats (zip-based)
'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
'application/vnd.openxmlformats-officedocument.presentationml.presentation',
'application/vnd.oasis.opendocument.text',
'application/vnd.oasis.opendocument.spreadsheet',
'application/vnd.oasis.opendocument.presentation'
])
/**
* @description True when the MIME type names a payload format known to be
* already heavily compressed. Strips any `;charset=…` / `;boundary=…`
* parameters and lowercases the bare type/subtype before lookup, so
* `'IMAGE/JPEG; charset=binary'` and `'image/jpeg'` resolve identically.
*/
export function isAlreadyCompressedMimeType(mimeType: string | undefined): boolean {
if (!mimeType) return false
const bare = mimeType.split(';', 1)[0].trim().toLowerCase()
return ALREADY_COMPRESSED_MIME_TYPES.has(bare)
}
/**
* Blob read options
*/
export interface BlobReadOptions {
skipDecompression?: boolean // Return compressed data
skipCache?: boolean // Don't use cache
skipVerification?: boolean // Skip hash verification (faster, less safe)
}
/**
* Blob statistics for observability
*/
export interface BlobStats {
totalBlobs: number
totalSize: number
compressedSize: number
cacheHits: number
cacheMisses: number
compressionRatio: number
avgBlobSize: number
dedupSavings: number // Bytes saved from deduplication
}
/**
* LRU Cache entry
*/
interface CacheEntry {
data: Buffer
metadata: BlobMetadata
lastAccess: number
size: number
}
/**
* State-of-the-art content-addressable blob storage
*
* Features:
* - Content addressing via SHA-256
* - Type-aware compression (zstd, vector-optimized)
* - LRU caching with memory limits
* - Streaming for large blobs
* - Batch operations
* - Integrity verification
* - Observability metrics
*/
export class BlobStorage {
private adapter: COWStorageAdapter
private cache: Map<string, CacheEntry>
private cacheMaxSize: number
private currentCacheSize: number
private stats: BlobStats
// Compression (lazily loaded)
private zstdCompress?: (data: Buffer) => Promise<Buffer>
private zstdDecompress?: (data: Buffer) => Promise<Buffer>
private compressionReady = false
// Configuration
private readonly CACHE_MAX_SIZE = 100 * 1024 * 1024 // 100MB default
private readonly MULTIPART_THRESHOLD = 5 * 1024 * 1024 // 5MB
private readonly COMPRESSION_THRESHOLD = 1024 // 1KB - don't compress smaller
constructor(adapter: COWStorageAdapter, options?: {
cacheMaxSize?: number
enableCompression?: boolean
}) {
this.adapter = adapter
this.cache = new Map()
this.cacheMaxSize = options?.cacheMaxSize ?? this.CACHE_MAX_SIZE
this.currentCacheSize = 0
this.stats = {
totalBlobs: 0,
totalSize: 0,
compressedSize: 0,
cacheHits: 0,
cacheMisses: 0,
compressionRatio: 1.0,
avgBlobSize: 0,
dedupSavings: 0
}
// Lazy load compression (only if needed)
if (options?.enableCompression !== false) {
this.initCompression()
}
}
/**
* Lazy load zstd compression module
* (Avoids loading if not needed)
*/
private async initCompression(): Promise<void> {
try {
// Dynamic import to avoid loading if not needed
// @ts-ignore - Optional dependency, gracefully handled if missing
const zstd = await import('@mongodb-js/zstd')
this.zstdCompress = async (data: Buffer) => {
return Buffer.from(await zstd.compress(data, 3)) // Level 3 = fast
}
this.zstdDecompress = async (data: Buffer) => {
return Buffer.from(await zstd.decompress(data))
}
} catch (error) {
console.warn('zstd compression not available, falling back to uncompressed')
this.zstdCompress = undefined
this.zstdDecompress = undefined
}
}
/**
* Ensure compression is ready before write operations
* Fixes race condition where write happens before async compression init completes
*/
private async ensureCompressionReady(): Promise<void> {
if (this.compressionReady) return
await this.initCompression()
this.compressionReady = true
}
/**
* Compute SHA-256 hash of data
*
* @param data - Data to hash
* @returns SHA-256 hash as hex string
*/
static hash(data: Buffer): string {
return createHash('sha256').update(data).digest('hex')
}
/**
* Write a blob to storage
*
* Features:
* - Content-addressable: hash determines storage key
* - Deduplication: existing blob not rewritten
* - Compression: auto-compress based on type
* - Multipart: for large blobs (>5MB)
* - Verification: hash verification
* - Caching: write-through cache
*
* @param data - Blob data to write
* @param options - Write options
* @returns Blob hash
*/
async write(data: Buffer, options: BlobWriteOptions = {}): Promise<string> {
const hash = BlobStorage.hash(data)
// Deduplication: Check if blob already exists
if (await this.has(hash)) {
// Update ref count
await this.incrementRefCount(hash)
this.stats.dedupSavings += data.length
return hash
}
// Ensure compression is initialized before writing
// Fixes race condition where write happens before async init completes
await this.ensureCompressionReady()
// Determine compression strategy
const compression = this.selectCompression(data, options)
// Compress if needed
let finalData = data
let compressedSize = data.length
if (compression === 'zstd' && this.zstdCompress) {
finalData = await this.zstdCompress(data)
compressedSize = finalData.length
}
// Create metadata
// Store ACTUAL compression state, not intended
// Prevents corruption if compression failed to initialize
const actualCompression = finalData === data ? 'none' : compression
const metadata: BlobMetadata = {
hash,
size: data.length,
compressedSize,
compression: actualCompression,
type: options.type || 'blob', // CRITICAL FIX: Use 'blob' default to match storage prefix
createdAt: Date.now(),
refCount: 1
}
// Write blob data
if (finalData.length > this.MULTIPART_THRESHOLD) {
// Large blob: use streaming/multipart
await this.writeMultipart(hash, finalData, metadata)
} else {
// Small blob: single write
const prefix = options.type || 'blob'
await this.adapter.put(`${prefix}:${hash}`, finalData)
}
// Write metadata
const prefix = options.type || 'blob'
await this.adapter.put(`${prefix}-meta:${hash}`, Buffer.from(JSON.stringify(metadata)))
// Update cache (write-through)
this.addToCache(hash, data, metadata)
// Update stats
this.stats.totalBlobs++
this.stats.totalSize += data.length
this.stats.compressedSize += compressedSize
this.stats.compressionRatio = this.stats.totalSize / (this.stats.compressedSize || 1)
this.stats.avgBlobSize = this.stats.totalSize / this.stats.totalBlobs
return hash
}
/**
* Read a blob from storage
*
* Features:
* - Cache lookup first (LRU)
* - Decompression (if compressed)
* - Verification (optional hash check)
* - Streaming for large blobs
*
* @param hash - Blob hash
* @param options - Read options
* @returns Blob data
*/
async read(hash: string, options: BlobReadOptions = {}): Promise<Buffer> {
// Guard against NULL hash (sentinel value)
// NULL_HASH ('0000...0000') is used as a sentinel for "no parent" or "empty tree"
// It should NEVER be read as actual blob data
if (isNullHash(hash)) {
throw new Error(
`Cannot read NULL hash (${NULL_HASH}): ` +
`This is a sentinel value indicating "no parent commit" or "empty tree". ` +
`If you're seeing this error from CommitObject.walk(), there's a bug in commit traversal logic. ` +
`If you're seeing this from TreeObject operations, there's a bug in tree handling.`
)
}
// Check cache first
if (!options.skipCache) {
const cached = this.getFromCache(hash)
if (cached) {
this.stats.cacheHits++
return cached.data
}
this.stats.cacheMisses++
}
// Try to read metadata to determine type (for backward compatibility)
// Try commit, tree, then blob prefixes
let prefix: string | null = null
let metadataBuffer: Buffer | undefined
let metadata: BlobMetadata | undefined
for (const tryPrefix of ['commit', 'tree', 'blob']) {
metadataBuffer = await this.adapter.get(`${tryPrefix}-meta:${hash}`)
if (metadataBuffer) {
prefix = tryPrefix
// Unwrap metadata before parsing (defense-in-depth)
// Metadata should be JSON, but adapter might return wrapped format
const unwrappedMetadata = unwrapBinaryData(metadataBuffer)
metadata = JSON.parse(unwrappedMetadata.toString())
break
}
}
if (!prefix || !metadata) {
throw new Error(`Blob metadata not found: ${hash}`)
}
// Read from storage using determined prefix
const data = await this.adapter.get(`${prefix}:${hash}`)
if (!data) {
throw new Error(`Blob not found: ${hash}`)
}
// Decompress if needed
let finalData = data
if (metadata.compression === 'zstd' && !options.skipDecompression) {
if (!this.zstdDecompress) {
throw new Error('zstd decompression not available')
}
finalData = await this.zstdDecompress(data)
}
// Defense-in-depth unwrap (CRITICAL FIX for blob integrity regression)
// Even though COW adapter should unwrap, verify it happened and re-unwrap if needed
// This prevents "Blob integrity check failed" errors if adapter returns wrapped data
// Uses shared binaryDataCodec utility (single source of truth for unwrap logic)
const unwrappedData = unwrapBinaryData(finalData)
// Verify hash (on unwrapped data)
if (!options.skipVerification && BlobStorage.hash(unwrappedData) !== hash) {
throw new Error(`Blob integrity check failed: ${hash}`)
}
// Add to cache (only if not skipped)
if (!options.skipCache) {
this.addToCache(hash, unwrappedData, metadata)
}
return unwrappedData
}
/**
* Check if blob exists
*
* @param hash - Blob hash
* @returns True if blob exists
*/
async has(hash: string): Promise<boolean> {
// Check cache first
if (this.cache.has(hash)) {
return true
}
// Check storage - try all prefixes for backward compatibility
for (const prefix of ['commit', 'tree', 'blob']) {
const exists = await this.adapter.get(`${prefix}:${hash}`)
if (exists !== undefined) {
return true
}
}
return false
}
/**
* Delete a blob from storage
*
* Features:
* - Reference counting: only delete if refCount = 0
* - Cascade: delete metadata too
* - Cache invalidation
*
* @param hash - Blob hash
*/
async delete(hash: string): Promise<void> {
// Decrement ref count
const refCount = await this.decrementRefCount(hash)
// Only delete if no references remain
if (refCount > 0) {
return
}
// Determine prefix by checking which one exists
let prefix = 'blob'
for (const tryPrefix of ['commit', 'tree', 'blob', 'metadata', 'vector', 'raw']) {
const exists = await this.adapter.get(`${tryPrefix}:${hash}`)
if (exists !== undefined) {
prefix = tryPrefix
break
}
}
// Delete blob data
await this.adapter.delete(`${prefix}:${hash}`)
// Delete metadata
await this.adapter.delete(`${prefix}-meta:${hash}`)
// Remove from cache
this.removeFromCache(hash)
// Update stats
this.stats.totalBlobs--
}
/**
* Get blob metadata without reading full blob
*
* @param hash - Blob hash
* @returns Blob metadata
*/
async getMetadata(hash: string): Promise<BlobMetadata | undefined> {
// Try to read metadata with type-aware prefix (backward compatible)
// Check all valid blob types: commit, tree, blob, metadata, vector, raw
for (const prefix of ['commit', 'tree', 'blob', 'metadata', 'vector', 'raw']) {
const data = await this.adapter.get(`${prefix}-meta:${hash}`)
if (data) {
return JSON.parse(data.toString())
}
}
return undefined
}
/**
* Batch write multiple blobs in parallel
*
* @param blobs - Array of [data, options] tuples
* @returns Array of blob hashes
*/
async writeBatch(blobs: Array<[Buffer, BlobWriteOptions?]>): Promise<string[]> {
return Promise.all(
blobs.map(([data, options]) => this.write(data, options))
)
}
/**
* Batch read multiple blobs in parallel
*
* @param hashes - Array of blob hashes
* @param options - Read options
* @returns Array of blob data
*/
async readBatch(hashes: string[], options?: BlobReadOptions): Promise<Buffer[]> {
return Promise.all(
hashes.map(hash => this.read(hash, options))
)
}
/**
* List all blobs (for garbage collection, debugging)
*
* @returns Array of blob hashes
*/
async listBlobs(): Promise<string[]> {
// List all types of blobs
const hashes = new Set<string>()
for (const prefix of ['commit', 'tree', 'blob']) {
const keys = await this.adapter.list(`${prefix}:`)
keys.forEach((key: string) => {
const hash = key.replace(new RegExp(`^${prefix}:`), '')
hashes.add(hash)
})
}
return Array.from(hashes)
}
/**
* Get storage statistics
*
* @returns Blob statistics
*/
getStats(): BlobStats {
return { ...this.stats }
}
/**
* Clear cache (useful for testing, memory pressure)
*/
clearCache(): void {
this.cache.clear()
this.currentCacheSize = 0
}
/**
* Garbage collect unreferenced blobs
*
* @param referencedHashes - Set of hashes that should be kept
* @returns Number of blobs deleted
*/
async garbageCollect(referencedHashes: Set<string>): Promise<number> {
const allBlobs = await this.listBlobs()
let deleted = 0
for (const hash of allBlobs) {
if (!referencedHashes.has(hash)) {
// Check ref count
const metadata = await this.getMetadata(hash)
if (metadata && metadata.refCount === 0) {
await this.delete(hash)
deleted++
}
}
}
return deleted
}
// ========== PRIVATE METHODS ==========
/**
* Select compression strategy based on data and options
*/
private selectCompression(
data: Buffer,
options: BlobWriteOptions
): 'none' | 'zstd' {
if (options.compression === 'none') {
return 'none'
}
if (options.compression === 'zstd') {
return this.zstdCompress ? 'zstd' : 'none'
}
// Auto mode
if (data.length < this.COMPRESSION_THRESHOLD) {
return 'none' // Too small to benefit
}
// Content-type policy (2.5.0 #32): skip already-compressed media. zstd
// over JPEG / MP4 / ZIP etc. is a CPU loss for no measurable byte savings,
// and on hot save paths (image / video uploads) it's the difference
// between fast and slow. Applies only to `auto`; explicit `'zstd'` is
// honoured because the caller is asserting the choice.
if (isAlreadyCompressedMimeType(options.mimeType)) {
return 'none'
}
// Compress metadata, trees, commits (text/JSON)
if (options.type === 'metadata' || options.type === 'tree' || options.type === 'commit') {
return this.zstdCompress ? 'zstd' : 'none'
}
// Don't compress vectors (already dense)
if (options.type === 'vector') {
return 'none'
}
// Default: compress
return this.zstdCompress ? 'zstd' : 'none'
}
/**
* Write large blob using multipart upload
* (Future enhancement: stream to adapter if supported)
*/
private async writeMultipart(
hash: string,
data: Buffer,
metadata: BlobMetadata
): Promise<void> {
// Single-blob write. Brainy 8.0 ships filesystem + memory only; the
// multipart-upload concern that motivated this method (S3/R2/GCS) is gone
// with the cloud adapters per BR-BRAINY-80-STORAGE-SIMPLIFY.
const prefix = metadata.type || 'blob'
await this.adapter.put(`${prefix}:${hash}`, data)
}
/**
* Increment reference count for a blob
*/
private async incrementRefCount(hash: string): Promise<number> {
const metadata = await this.getMetadata(hash)
if (!metadata) {
throw new Error(`Cannot increment ref count, blob not found: ${hash}`)
}
metadata.refCount++
const prefix = metadata.type || 'blob'
await this.adapter.put(
`${prefix}-meta:${hash}`,
Buffer.from(JSON.stringify(metadata))
)
return metadata.refCount
}
/**
* Decrement reference count for a blob
*/
private async decrementRefCount(hash: string): Promise<number> {
const metadata = await this.getMetadata(hash)
if (!metadata) {
return 0
}
metadata.refCount = Math.max(0, metadata.refCount - 1)
const prefix = metadata.type || 'blob'
await this.adapter.put(
`${prefix}-meta:${hash}`,
Buffer.from(JSON.stringify(metadata))
)
return metadata.refCount
}
/**
* Add blob to LRU cache
*/
private addToCache(hash: string, data: Buffer, metadata: BlobMetadata): void {
// Check if adding would exceed cache size
if (data.length > this.cacheMaxSize) {
return // Blob too large for cache
}
// Evict old entries if needed
while (
this.currentCacheSize + data.length > this.cacheMaxSize &&
this.cache.size > 0
) {
this.evictLRU()
}
// Add to cache
this.cache.set(hash, {
data,
metadata,
lastAccess: Date.now(),
size: data.length
})
this.currentCacheSize += data.length
}
/**
* Get blob from cache
*/
private getFromCache(hash: string): CacheEntry | undefined {
const entry = this.cache.get(hash)
if (entry) {
entry.lastAccess = Date.now() // Update LRU
}
return entry
}
/**
* Remove blob from cache
*/
private removeFromCache(hash: string): void {
const entry = this.cache.get(hash)
if (entry) {
this.cache.delete(hash)
this.currentCacheSize -= entry.size
}
}
/**
* Evict least recently used entry from cache
*/
private evictLRU(): void {
let oldestHash: string | null = null
let oldestTime = Infinity
for (const [hash, entry] of this.cache.entries()) {
if (entry.lastAccess < oldestTime) {
oldestTime = entry.lastAccess
oldestHash = hash
}
}
if (oldestHash) {
this.removeFromCache(oldestHash)
}
}
}

531
src/storage/cow/CommitLog.ts
View file

@ -1,531 +0,0 @@
/**
* CommitLog: Commit history traversal and querying for COW (Copy-on-Write)
*
* Provides efficient commit history operations:
* - Walk commit graph (DAG traversal)
* - Find commits by time, author, operation
* - Time-travel queries (asOf)
* - Commit statistics and analytics
*
* Optimizations:
* - Commit index for fast timestamp lookups
* - Parent cache for efficient traversal
* - Lazy loading (only read commits when needed)
*
* @module storage/cow/CommitLog
*/
import { BlobStorage } from './BlobStorage.js'
import { CommitObject } from './CommitObject.js'
import { RefManager } from './RefManager.js'
/**
* Commit index entry (for fast lookups)
*/
interface CommitIndexEntry {
hash: string
timestamp: number
parentHash: string | null
}
/**
* Commit log statistics
*/
export interface CommitLogStats {
totalCommits: number
oldestCommit: number // Timestamp
newestCommit: number // Timestamp
authors: Set<string>
operations: Set<string>
avgCommitInterval: number // Average time between commits (ms)
}
/**
* CommitLog: Efficient commit history traversal and querying
*
* Pure implementation - modern, clean, fast
*/
export class CommitLog {
private blobStorage: BlobStorage
private refManager: RefManager
private index: Map<string, CommitIndexEntry>
private indexValid: boolean
constructor(blobStorage: BlobStorage, refManager: RefManager) {
this.blobStorage = blobStorage
this.refManager = refManager
this.index = new Map()
this.indexValid = false
}
/**
* Walk commit history from a starting point
*
* Yields commits in reverse chronological order (newest first)
*
* @param startRef - Starting ref/commit (e.g., 'main', commit hash)
* @param options - Walk options
*/
async *walk(
startRef: string = 'main',
options?: {
maxDepth?: number
until?: number
stopAt?: string
filter?: (commit: CommitObject) => boolean
}
): AsyncIterableIterator<CommitObject> {
// Resolve ref to commit hash
let startHash: string
if (/^[a-f0-9]{64}$/.test(startRef)) {
// Already a commit hash
startHash = startRef
} else {
// Resolve ref
const commitHash = await this.refManager.resolveRef(startRef)
if (!commitHash) {
throw new Error(`Ref not found: ${startRef}`)
}
startHash = commitHash
}
// Walk using CommitObject (delegates to it)
yield* CommitObject.walk(this.blobStorage, startHash, options)
}
/**
* Find commit at or before a specific timestamp
*
* Uses index for fast O(log n) lookup
*
* @param ref - Starting ref (e.g., 'main')
* @param timestamp - Target timestamp
* @returns Commit at or before timestamp, or null
*/
async findAtTime(ref: string, timestamp: number): Promise<CommitObject | null> {
// Build index if needed
await this.buildIndex(ref)
// Binary search in index
const entries = Array.from(this.index.values()).sort(
(a, b) => b.timestamp - a.timestamp // Newest first
)
let left = 0
let right = entries.length - 1
let result: CommitIndexEntry | null = null
while (left <= right) {
const mid = Math.floor((left + right) / 2)
const entry = entries[mid]
if (entry.timestamp <= timestamp) {
result = entry
right = mid - 1 // Look for newer commit
} else {
left = mid + 1 // Look for older commit
}
}
if (!result) {
return null
}
// Read full commit
return CommitObject.read(this.blobStorage, result.hash)
}
/**
* Get commit by hash
*
* @param hash - Commit hash
* @returns Commit object
*/
async getCommit(hash: string): Promise<CommitObject> {
return CommitObject.read(this.blobStorage, hash)
}
/**
* Get commits in time range
*
* @param ref - Starting ref
* @param startTime - Start of time range
* @param endTime - End of time range
* @returns Array of commits in range (newest first)
*/
async getInTimeRange(
ref: string,
startTime: number,
endTime: number
): Promise<CommitObject[]> {
const commits: CommitObject[] = []
for await (const commit of this.walk(ref, { until: startTime })) {
if (commit.timestamp >= startTime && commit.timestamp <= endTime) {
commits.push(commit)
}
}
return commits
}
/**
* Get commits by author
*
* @param ref - Starting ref
* @param author - Author name
* @param options - Additional options
* @returns Array of commits by author
*/
async getByAuthor(
ref: string,
author: string,
options?: { maxCount?: number; since?: number }
): Promise<CommitObject[]> {
const commits: CommitObject[] = []
let count = 0
for await (const commit of this.walk(ref, { until: options?.since })) {
if (commit.author === author) {
commits.push(commit)
count++
if (options?.maxCount && count >= options.maxCount) {
break
}
}
}
return commits
}
/**
* Get commits by operation type
*
* @param ref - Starting ref
* @param operation - Operation type (e.g., 'add', 'update', 'delete')
* @param options - Additional options
* @returns Array of commits by operation
*/
async getByOperation(
ref: string,
operation: string,
options?: { maxCount?: number; since?: number }
): Promise<CommitObject[]> {
const commits: CommitObject[] = []
let count = 0
for await (const commit of this.walk(ref, { until: options?.since })) {
if (commit.metadata?.operation === operation) {
commits.push(commit)
count++
if (options?.maxCount && count >= options.maxCount) {
break
}
}
}
return commits
}
/**
* Get commit history as array
*
* @param ref - Starting ref
* @param options - Walk options
* @returns Array of commits (newest first)
*/
async getHistory(
ref: string,
options?: {
maxCount?: number
since?: number
until?: number
}
): Promise<CommitObject[]> {
const commits: CommitObject[] = []
let count = 0
for await (const commit of this.walk(ref, {
maxDepth: options?.maxCount,
until: options?.until
})) {
if (options?.since && commit.timestamp < options.since) {
continue
}
commits.push(commit)
count++
if (options?.maxCount && count >= options.maxCount) {
break
}
}
return commits
}
/**
* Stream commit history (memory-efficient for large histories)
*
* Yields commits one at a time without accumulating in memory.
* Use this for large commit histories (1000s of commits) where
* memory efficiency is important.
*
* @param ref - Starting ref
* @param options - Walk options
* @yields Commits in reverse chronological order (newest first)
*
* @example
* ```typescript
* // Stream all commits without memory accumulation
* for await (const commit of commitLog.streamHistory('main', { maxCount: 10000 })) {
* console.log(commit.message)
* }
* ```
*/
async *streamHistory(
ref: string,
options?: {
maxCount?: number
since?: number
until?: number
}
): AsyncIterableIterator<CommitObject> {
let count = 0
for await (const commit of this.walk(ref, {
maxDepth: options?.maxCount,
until: options?.until
})) {
// Filter by since timestamp if provided
if (options?.since && commit.timestamp < options.since) {
continue
}
yield commit
count++
// Stop after maxCount commits
if (options?.maxCount && count >= options.maxCount) {
break
}
}
}
/**
* Count commits between two commits
*
* @param fromRef - Starting ref/commit
* @param toRef - Ending ref/commit (optional, defaults to fromRef's parent)
* @returns Number of commits between
*/
async countBetween(fromRef: string, toRef?: string): Promise<number> {
const fromHash = await this.resolveToHash(fromRef)
const toHash = toRef ? await this.resolveToHash(toRef) : null
return CommitObject.countBetween(this.blobStorage, fromHash, toHash!)
}
/**
* Find common ancestor of two commits (merge base)
*
* @param ref1 - First ref/commit
* @param ref2 - Second ref/commit
* @returns Common ancestor commit or null
*/
async findCommonAncestor(
ref1: string,
ref2: string
): Promise<CommitObject | null> {
const hash1 = await this.resolveToHash(ref1)
const hash2 = await this.resolveToHash(ref2)
return CommitObject.findCommonAncestor(this.blobStorage, hash1, hash2)
}
/**
* Get commit log statistics
*
* @param ref - Starting ref
* @param options - Options
* @returns Commit log statistics
*/
async getStats(
ref: string = 'main',
options?: { maxDepth?: number }
): Promise<CommitLogStats> {
const authors = new Set<string>()
const operations = new Set<string>()
const timestamps: number[] = []
let totalCommits = 0
for await (const commit of this.walk(ref, options)) {
totalCommits++
authors.add(commit.author)
timestamps.push(commit.timestamp)
if (commit.metadata?.operation) {
operations.add(commit.metadata.operation)
}
}
// Calculate average interval
let avgInterval = 0
if (timestamps.length > 1) {
const intervals: number[] = []
for (let i = 0; i < timestamps.length - 1; i++) {
intervals.push(timestamps[i] - timestamps[i + 1])
}
avgInterval = intervals.reduce((a, b) => a + b, 0) / intervals.length
}
return {
totalCommits,
oldestCommit: timestamps[timestamps.length - 1] ?? 0,
newestCommit: timestamps[0] ?? 0,
authors,
operations,
avgCommitInterval: avgInterval
}
}
/**
* Check if commit is ancestor of another commit
*
* @param ancestorRef - Potential ancestor ref/commit
* @param descendantRef - Descendant ref/commit
* @returns True if ancestor is in descendant's history
*/
async isAncestor(ancestorRef: string, descendantRef: string): Promise<boolean> {
const ancestorHash = await this.resolveToHash(ancestorRef)
const descendantHash = await this.resolveToHash(descendantRef)
// Walk from descendant, check if we encounter ancestor
for await (const commit of this.walk(descendantHash)) {
const commitHash = CommitObject.hash(commit)
if (commitHash === ancestorHash) {
return true
}
}
return false
}
/**
* Get recent commits (last N)
*
* @param ref - Starting ref
* @param count - Number of commits to retrieve
* @returns Array of recent commits
*/
async getRecent(ref: string, count: number = 10): Promise<CommitObject[]> {
return this.getHistory(ref, { maxCount: count })
}
/**
* Find commits with tag
*
* @param ref - Starting ref
* @param tag - Tag to search for
* @returns Array of commits with tag
*/
async findWithTag(ref: string, tag: string): Promise<CommitObject[]> {
const commits: CommitObject[] = []
for await (const commit of this.walk(ref)) {
if (CommitObject.hasTag(commit, tag)) {
commits.push(commit)
}
}
return commits
}
/**
* Get first (oldest) commit
*
* @param ref - Starting ref
* @returns Oldest commit
*/
async getFirstCommit(ref: string): Promise<CommitObject | null> {
let oldest: CommitObject | null = null
for await (const commit of this.walk(ref)) {
oldest = commit
}
return oldest
}
/**
* Get latest commit
*
* @param ref - Starting ref
* @returns Latest commit
*/
async getLatestCommit(ref: string): Promise<CommitObject | null> {
for await (const commit of this.walk(ref, { maxDepth: 1 })) {
return commit
}
return null
}
/**
* Clear index (useful for testing, after new commits)
*/
clearIndex(): void {
this.index.clear()
this.indexValid = false
}
// ========== PRIVATE METHODS ==========
/**
* Build commit index for fast lookups
*
* @param ref - Starting ref
*/
private async buildIndex(ref: string): Promise<void> {
if (this.indexValid) {
return // Already built
}
this.index.clear()
for await (const commit of this.walk(ref)) {
const hash = CommitObject.hash(commit)
this.index.set(hash, {
hash,
timestamp: commit.timestamp,
parentHash: commit.parent
})
}
this.indexValid = true
}
/**
* Resolve ref or hash to commit hash
*
* @param refOrHash - Ref name or commit hash
* @returns Commit hash
*/
private async resolveToHash(refOrHash: string): Promise<string> {
if (/^[a-f0-9]{64}$/.test(refOrHash)) {
return refOrHash // Already a hash
}
const hash = await this.refManager.resolveRef(refOrHash)
if (!hash) {
throw new Error(`Ref not found: ${refOrHash}`)
}
return hash
}
}

563
src/storage/cow/CommitObject.ts
View file

@ -1,563 +0,0 @@
/**
* CommitObject: Snapshot metadata for COW (Copy-on-Write)
*
* Similar to Git commits, represents a point-in-time snapshot of a Brainy instance.
* Commits reference tree objects and parent commits, forming a directed acyclic graph (DAG).
*
* Structure:
* - tree: hash of root tree object (contains all data)
* - parent: hash of parent commit (null for initial commit)
* - message: human-readable commit message
* - author: who/what created this commit
* - timestamp: when this commit was created
* - metadata: additional commit metadata (tags, etc.)
*
* @module storage/cow/CommitObject
*/
import { BlobStorage } from './BlobStorage.js'
import { isNullHash } from './constants.js'
/**
* Commit object structure
*/
export interface CommitObject {
tree: string // Root tree hash
parent: string | null // Parent commit hash (null for initial commit)
message: string // Commit message
author: string // Author (user, system, augmentation)
timestamp: number // Unix timestamp (milliseconds)
metadata?: {
tags?: string[] // Tags (e.g., ['v1.0.0', 'production'])
branch?: string // Branch name (e.g., 'main', 'experiment')
operation?: string // Operation type (e.g., 'add', 'update', 'delete', 'merge')
entityCount?: number // Number of entities at this commit
relationshipCount?: number // Number of relationships at this commit
[key: string]: any // Custom metadata
}
}
/**
* CommitBuilder: Fluent API for building commit objects
*
* Example:
* ```typescript
* const commit = await CommitBuilder.create(blobStorage)
* .tree(treeHash)
* .parent(parentHash)
* .message('Add user entities')
* .author('system')
* .tag('v1.0.0')
* .build()
* ```
*/
export class CommitBuilder {
private _tree?: string
private _parent?: string | null
private _message: string = 'Auto-commit'
private _author: string = 'system'
private _timestamp: number = Date.now()
private _metadata: NonNullable<CommitObject['metadata']> = {}
private blobStorage: BlobStorage
constructor(blobStorage: BlobStorage) {
this.blobStorage = blobStorage
}
static create(blobStorage: BlobStorage): CommitBuilder {
return new CommitBuilder(blobStorage)
}
/**
* Set tree hash
*/
tree(hash: string): this {
this._tree = hash
return this
}
/**
* Set parent commit hash (null for initial commit)
*/
parent(hash: string | null): this {
this._parent = hash
return this
}
/**
* Set commit message
*/
message(message: string): this {
this._message = message
return this
}
/**
* Set commit author
*/
author(author: string): this {
this._author = author
return this
}
/**
* Set commit timestamp (defaults to now)
*/
timestamp(timestamp: number | Date): this {
this._timestamp = typeof timestamp === 'number' ? timestamp : timestamp.getTime()
return this
}
/**
* Add tag to commit
*/
tag(tag: string): this {
if (!this._metadata.tags) {
this._metadata.tags = []
}
this._metadata.tags.push(tag)
return this
}
/**
* Set branch name
*/
branch(branch: string): this {
this._metadata.branch = branch
return this
}
/**
* Set operation type
*/
operation(operation: string): this {
this._metadata.operation = operation
return this
}
/**
* Set entity count
*/
entityCount(count: number): this {
this._metadata.entityCount = count
return this
}
/**
* Set relationship count
*/
relationshipCount(count: number): this {
this._metadata.relationshipCount = count
return this
}
/**
* Set custom metadata
*/
meta(key: string, value: any): this {
this._metadata[key] = value
return this
}
/**
* Build and persist the commit object
*
* @returns Commit hash
*/
async build(): Promise<string> {
if (!this._tree) {
throw new Error('CommitBuilder: tree hash is required')
}
const commit: CommitObject = {
tree: this._tree,
parent: this._parent ?? null,
message: this._message,
author: this._author,
timestamp: this._timestamp,
metadata: Object.keys(this._metadata).length > 0 ? this._metadata : undefined
}
return CommitObject.write(this.blobStorage, commit)
}
}
/**
* CommitObject: Represents a point-in-time snapshot in COW storage
*/
export class CommitObject {
/**
* Serialize commit object to Buffer
*
* Format: JSON (simple, debuggable)
* Future: Could use protobuf for efficiency
*
* @param commit - Commit object
* @returns Serialized commit
*/
static serialize(commit: CommitObject): Buffer {
return Buffer.from(JSON.stringify(commit, null, 0)) // Compact JSON
}
/**
* Deserialize commit object from Buffer
*
* @param data - Serialized commit
* @returns Commit object
*/
static deserialize(data: Buffer): CommitObject {
const commit = JSON.parse(data.toString())
// Validate structure
if (typeof commit.tree !== 'string' || commit.tree.length !== 64) {
throw new Error('Invalid commit object: tree must be 64-char SHA-256')
}
if (commit.parent !== null && (typeof commit.parent !== 'string' || commit.parent.length !== 64)) {
throw new Error('Invalid commit object: parent must be 64-char SHA-256 or null')
}
if (typeof commit.message !== 'string') {
throw new Error('Invalid commit object: message must be string')
}
if (typeof commit.author !== 'string') {
throw new Error('Invalid commit object: author must be string')
}
if (typeof commit.timestamp !== 'number') {
throw new Error('Invalid commit object: timestamp must be number')
}
if (commit.metadata !== undefined && typeof commit.metadata !== 'object') {
throw new Error('Invalid commit object: metadata must be object')
}
return commit
}
/**
* Compute hash of commit object
*
* @param commit - Commit object
* @returns SHA-256 hash
*/
static hash(commit: CommitObject): string {
const data = CommitObject.serialize(commit)
return BlobStorage.hash(data)
}
/**
* Write commit object to blob storage
*
* @param blobStorage - Blob storage instance
* @param commit - Commit object
* @returns Commit hash
*/
static async write(blobStorage: BlobStorage, commit: CommitObject): Promise<string> {
const data = CommitObject.serialize(commit)
return blobStorage.write(data, { type: 'commit', compression: 'auto' })
}
/**
* Read commit object from blob storage
*
* @param blobStorage - Blob storage instance
* @param hash - Commit hash
* @returns Commit object
*/
static async read(blobStorage: BlobStorage, hash: string): Promise<CommitObject> {
const data = await blobStorage.read(hash)
return CommitObject.deserialize(data)
}
/**
* Check if commit is initial commit (has no parent)
*
* @param commit - Commit object
* @returns True if initial commit
*/
static isInitial(commit: CommitObject): boolean {
return commit.parent === null
}
/**
* Check if commit is merge commit (has multiple parents)
* (Future enhancement: support merge commits with multiple parents)
*
* @param commit - Commit object
* @returns True if merge commit
*/
static isMerge(commit: CommitObject): boolean {
// For now, we only support single-parent commits
// Future: extend to support multiple parents
return commit.metadata?.merge !== undefined
}
/**
* Check if commit has a specific tag
*
* @param commit - Commit object
* @param tag - Tag to check
* @returns True if commit has tag
*/
static hasTag(commit: CommitObject, tag: string): boolean {
return commit.metadata?.tags?.includes(tag) ?? false
}
/**
* Get all tags from commit
*
* @param commit - Commit object
* @returns Array of tags
*/
static getTags(commit: CommitObject): string[] {
return commit.metadata?.tags ?? []
}
/**
* Walk commit history (traverse DAG)
*
* Yields commits in reverse chronological order (newest first)
*
* @param blobStorage - Blob storage instance
* @param startHash - Starting commit hash
* @param options - Walk options
*/
static async *walk(
blobStorage: BlobStorage,
startHash: string,
options?: {
maxDepth?: number // Maximum depth to walk
until?: number // Stop at this timestamp
stopAt?: string // Stop at this commit hash
filter?: (commit: CommitObject) => boolean
}
): AsyncIterableIterator<CommitObject> {
let currentHash: string | null = startHash
let depth = 0
// Guard against NULL hash (sentinel for "no parent")
// The initial commit has parent = null or NULL_HASH ('0000...0000')
// We must stop walking when we reach it, not try to read it
while (currentHash && !isNullHash(currentHash)) {
// Check max depth
if (options?.maxDepth && depth >= options.maxDepth) {
break
}
// Read commit
const commit = await CommitObject.read(blobStorage, currentHash)
// Check filter
if (options?.filter && !options.filter(commit)) {
currentHash = commit.parent
depth++
continue
}
// Yield commit
yield commit
// Check stop conditions
if (options?.until && commit.timestamp < options.until) {
break
}
if (options?.stopAt && currentHash === options.stopAt) {
break
}
// Move to parent (can be null or NULL_HASH for initial commit)
currentHash = commit.parent
depth++
}
}
/**
* Find commit at or before a specific timestamp
*
* @param blobStorage - Blob storage instance
* @param startHash - Starting commit hash (e.g., 'main' ref)
* @param timestamp - Target timestamp
* @returns Commit at or before timestamp, or null if not found
*/
static async findAtTime(
blobStorage: BlobStorage,
startHash: string,
timestamp: number
): Promise<CommitObject | null> {
for await (const commit of CommitObject.walk(blobStorage, startHash)) {
if (commit.timestamp <= timestamp) {
return commit
}
}
return null
}
/**
* Get commit history as array (newest first)
*
* @param blobStorage - Blob storage instance
* @param startHash - Starting commit hash
* @param options - Walk options
* @returns Array of commits
*/
static async getHistory(
blobStorage: BlobStorage,
startHash: string,
options?: {
maxDepth?: number
until?: number
stopAt?: string
}
): Promise<CommitObject[]> {
const commits: CommitObject[] = []
for await (const commit of CommitObject.walk(blobStorage, startHash, options)) {
commits.push(commit)
}
return commits
}
/**
* Find common ancestor of two commits (merge base)
*
* Useful for diff/merge operations
*
* @param blobStorage - Blob storage instance
* @param hash1 - First commit hash
* @param hash2 - Second commit hash
* @returns Common ancestor commit, or null if not found
*/
static async findCommonAncestor(
blobStorage: BlobStorage,
hash1: string,
hash2: string
): Promise<CommitObject | null> {
// Build set of all ancestors of commit1
const ancestors1 = new Set<string>()
for await (const commit of CommitObject.walk(blobStorage, hash1)) {
ancestors1.add(CommitObject.hash(commit))
}
// Walk commit2 history, find first commit in ancestors1
for await (const commit of CommitObject.walk(blobStorage, hash2)) {
const commitHash = CommitObject.hash(commit)
if (ancestors1.has(commitHash)) {
return commit
}
}
return null
}
/**
* Count commits between two commits
*
* @param blobStorage - Blob storage instance
* @param fromHash - Starting commit hash
* @param toHash - Ending commit hash
* @returns Number of commits between (inclusive)
*/
static async countBetween(
blobStorage: BlobStorage,
fromHash: string,
toHash: string
): Promise<number> {
let count = 0
for await (const commit of CommitObject.walk(blobStorage, fromHash, {
stopAt: toHash
})) {
count++
}
return count
}
/**
* Get commits in time range
*
* @param blobStorage - Blob storage instance
* @param startHash - Starting commit hash
* @param startTime - Start of time range
* @param endTime - End of time range
* @returns Array of commits in range
*/
static async getInTimeRange(
blobStorage: BlobStorage,
startHash: string,
startTime: number,
endTime: number
): Promise<CommitObject[]> {
const commits: CommitObject[] = []
for await (const commit of CommitObject.walk(blobStorage, startHash, {
until: startTime
})) {
if (commit.timestamp >= startTime && commit.timestamp <= endTime) {
commits.push(commit)
}
}
return commits
}
/**
* Get commits by author
*
* @param blobStorage - Blob storage instance
* @param startHash - Starting commit hash
* @param author - Author name
* @param options - Walk options
* @returns Array of commits by author
*/
static async getByAuthor(
blobStorage: BlobStorage,
startHash: string,
author: string,
options?: { maxDepth?: number }
): Promise<CommitObject[]> {
const commits: CommitObject[] = []
for await (const commit of CommitObject.walk(blobStorage, startHash, {
...options,
filter: c => c.author === author
})) {
commits.push(commit)
}
return commits
}
/**
* Get commits by operation type
*
* @param blobStorage - Blob storage instance
* @param startHash - Starting commit hash
* @param operation - Operation type (e.g., 'add', 'update', 'delete')
* @param options - Walk options
* @returns Array of commits by operation
*/
static async getByOperation(
blobStorage: BlobStorage,
startHash: string,
operation: string,
options?: { maxDepth?: number }
): Promise<CommitObject[]> {
const commits: CommitObject[] = []
for await (const commit of CommitObject.walk(blobStorage, startHash, {
...options,
filter: c => c.metadata?.operation === operation
})) {
commits.push(commit)
}
return commits
}
}

547
src/storage/cow/RefManager.ts
View file

@ -1,547 +0,0 @@
/**
* RefManager: Branch and reference management for COW (Copy-on-Write)
*
* Similar to Git refs, manages symbolic names (branches, tags) that point to commits.
*
* Structure:
* - refs/heads/main → commit hash (main branch)
* - refs/heads/experiment → commit hash (experiment branch)
* - refs/tags/v1.0.0 → commit hash (version tag)
* - HEAD → ref name (current branch)
*
* Features:
* - Branch management (create, delete, list)
* - Tag management
* - HEAD pointer (current branch)
* - Fast-forward and force updates
* - Atomic operations
*
* @module storage/cow/RefManager
*/
import type { COWStorageAdapter } from './BlobStorage.js'
import { compareCodePoints } from '../../utils/collation.js'
/**
* Reference type
*/
export type RefType = 'branch' | 'tag' | 'remote'
/**
* Reference object
*/
export interface Ref {
name: string // Full ref name (e.g., 'refs/heads/main')
commitHash: string // Commit hash this ref points to
type: RefType // Reference type
createdAt: number // When ref was created
updatedAt: number // When ref was last updated
metadata?: {
description?: string
author?: string
[key: string]: any
}
}
/**
* Ref update options
*/
export interface RefUpdateOptions {
force?: boolean // Force update (allow non-fast-forward)
createOnly?: boolean // Only create, fail if exists
updateOnly?: boolean // Only update, fail if doesn't exist
expectedOldValue?: string // CAS: only update if current value matches
}
/**
* RefManager: Manages branches, tags, and HEAD pointer
*
* Pure implementation - no backward compatibility
*/
export class RefManager {
private adapter: COWStorageAdapter
private cache: Map<string, Ref>
private cacheValid: boolean
constructor(adapter: COWStorageAdapter) {
this.adapter = adapter
this.cache = new Map()
this.cacheValid = false
}
/**
* Get reference by name
*
* @param name - Reference name (e.g., 'main', 'refs/heads/main')
* @returns Reference object or undefined
*/
async getRef(name: string): Promise<Ref | undefined> {
const fullName = this.normalizeRefName(name)
// Check cache
if (this.cacheValid && this.cache.has(fullName)) {
return this.cache.get(fullName)
}
// Read from storage
const data = await this.adapter.get(`ref:${fullName}`)
if (!data) {
return undefined
}
const ref = JSON.parse(data.toString()) as Ref
// Update cache
this.cache.set(fullName, ref)
return ref
}
/**
* Set reference to point to commit
*
* @param name - Reference name
* @param commitHash - Commit hash to point to
* @param options - Update options
*/
async setRef(
name: string,
commitHash: string,
options: RefUpdateOptions = {}
): Promise<void> {
const fullName = this.normalizeRefName(name)
// Validate commit hash format
if (!/^[a-f0-9]{64}$/.test(commitHash)) {
throw new Error(`Invalid commit hash: ${commitHash}`)
}
// Check if ref exists
const existing = await this.getRef(fullName)
// Handle createOnly
if (options.createOnly && existing) {
throw new Error(`Ref already exists: ${fullName}`)
}
// Handle updateOnly
if (options.updateOnly && !existing) {
throw new Error(`Ref does not exist: ${fullName}`)
}
// Handle CAS (Compare-And-Swap)
if (options.expectedOldValue !== undefined) {
if (!existing || existing.commitHash !== options.expectedOldValue) {
throw new Error(
`Ref update failed: expected ${options.expectedOldValue}, ` +
`got ${existing?.commitHash ?? 'none'}`
)
}
}
// Fast-forward verification is intentionally permissive: COW is a
// single-writer subsystem locked by the storage adapter's writer lock,
// so out-of-order updates can't reach this point. If multi-writer COW
// is ever introduced, replace with a commit-ancestry walk.
// Create/update ref
const ref: Ref = {
name: fullName,
commitHash,
type: this.getRefType(fullName),
createdAt: existing?.createdAt ?? Date.now(),
updatedAt: Date.now(),
metadata: existing?.metadata
}
// Write to storage
await this.adapter.put(`ref:${fullName}`, Buffer.from(JSON.stringify(ref)))
// Update cache
this.cache.set(fullName, ref)
this.cacheValid = false // Invalidate for listRefs
}
/**
* Delete reference
*
* @param name - Reference name
*/
async deleteRef(name: string): Promise<void> {
const fullName = this.normalizeRefName(name)
// Don't allow deleting HEAD
if (fullName === 'HEAD') {
throw new Error('Cannot delete HEAD')
}
// Don't allow deleting main if it's the only branch
if (fullName === 'refs/heads/main') {
const branches = await this.listRefs('branch')
if (branches.length === 1) {
throw new Error('Cannot delete last branch')
}
}
// Delete from storage
await this.adapter.delete(`ref:${fullName}`)
// Update cache
this.cache.delete(fullName)
this.cacheValid = false
}
/**
* List all references
*
* @param type - Filter by type (optional)
* @returns Array of references
*/
async listRefs(type?: RefType): Promise<Ref[]> {
// Get all ref keys
const keys = await this.adapter.list('ref:')
const refs: Ref[] = []
for (const key of keys) {
const refName = key.replace(/^ref:/, '')
// Skip HEAD in listings (it's special)
if (refName === 'HEAD') {
continue
}
const ref = await this.getRef(refName)
if (ref) {
// Filter by type if requested
if (!type || ref.type === type) {
refs.push(ref)
}
}
}
// Mark cache as valid
this.cacheValid = true
return refs.sort((a, b) => compareCodePoints(a.name, b.name))
}
/**
* Copy reference (create branch from existing ref)
*
* @param sourceName - Source reference name
* @param targetName - Target reference name
* @param options - Update options
*/
async copyRef(
sourceName: string,
targetName: string,
options: RefUpdateOptions = {}
): Promise<void> {
const sourceRef = await this.getRef(sourceName)
if (!sourceRef) {
throw new Error(`Source ref not found: ${sourceName}`)
}
// Set target ref to same commit as source
await this.setRef(targetName, sourceRef.commitHash, options)
}
/**
* Get current HEAD (current branch)
*
* @returns HEAD reference or undefined
*/
async getHead(): Promise<Ref | undefined> {
const data = await this.adapter.get('ref:HEAD')
if (!data) {
return undefined
}
const head = JSON.parse(data.toString()) as { ref: string }
// Resolve symbolic ref
return this.getRef(head.ref)
}
/**
* Set HEAD to point to branch
*
* @param branchName - Branch name (e.g., 'main', 'refs/heads/experiment')
*/
async setHead(branchName: string): Promise<void> {
const fullName = this.normalizeRefName(branchName)
// Verify branch exists
const branch = await this.getRef(fullName)
if (!branch) {
throw new Error(`Branch not found: ${fullName}`)
}
if (branch.type !== 'branch') {
throw new Error(`Cannot set HEAD to non-branch ref: ${fullName}`)
}
// Set HEAD (symbolic ref)
const head = { ref: fullName }
await this.adapter.put('ref:HEAD', Buffer.from(JSON.stringify(head)))
}
/**
* Get current commit hash (resolves HEAD)
*
* @returns Current commit hash or undefined
*/
async getCurrentCommit(): Promise<string | undefined> {
const head = await this.getHead()
return head?.commitHash
}
/**
* Create branch
*
* @param name - Branch name (e.g., 'experiment')
* @param commitHash - Commit hash to point to
* @param options - Create options
*/
async createBranch(
name: string,
commitHash: string,
options?: { description?: string; author?: string }
): Promise<void> {
const fullName = this.normalizeRefName(name, 'branch')
await this.setRef(fullName, commitHash, { createOnly: true })
// Update metadata if provided
if (options?.description || options?.author) {
const ref = await this.getRef(fullName)
if (ref) {
ref.metadata = {
...ref.metadata,
description: options.description,
author: options.author
}
await this.adapter.put(`ref:${fullName}`, Buffer.from(JSON.stringify(ref)))
}
}
}
/**
* Delete branch
*
* @param name - Branch name
*/
async deleteBranch(name: string): Promise<void> {
const fullName = this.normalizeRefName(name, 'branch')
await this.deleteRef(fullName)
}
/**
* List all branches
*
* @returns Array of branch references
*/
async listBranches(): Promise<Ref[]> {
return this.listRefs('branch')
}
/**
* Create tag
*
* @param name - Tag name (e.g., 'v1.0.0')
* @param commitHash - Commit hash to point to
* @param options - Create options
*/
async createTag(
name: string,
commitHash: string,
options?: { description?: string; author?: string }
): Promise<void> {
const fullName = this.normalizeRefName(name, 'tag')
await this.setRef(fullName, commitHash, { createOnly: true })
// Update metadata if provided
if (options?.description || options?.author) {
const ref = await this.getRef(fullName)
if (ref) {
ref.metadata = {
...ref.metadata,
description: options.description,
author: options.author
}
await this.adapter.put(`ref:${fullName}`, Buffer.from(JSON.stringify(ref)))
}
}
}
/**
* Delete tag
*
* @param name - Tag name
*/
async deleteTag(name: string): Promise<void> {
const fullName = this.normalizeRefName(name, 'tag')
await this.deleteRef(fullName)
}
/**
* List all tags
*
* @returns Array of tag references
*/
async listTags(): Promise<Ref[]> {
return this.listRefs('tag')
}
/**
* Check if reference exists
*
* @param name - Reference name
* @returns True if reference exists
*/
async hasRef(name: string): Promise<boolean> {
const ref = await this.getRef(name)
return ref !== undefined
}
/**
* Update reference to new commit (with validation)
*
* @param name - Reference name
* @param newCommitHash - New commit hash
* @param oldCommitHash - Expected old commit hash (for safety)
*/
async updateRef(
name: string,
newCommitHash: string,
oldCommitHash?: string
): Promise<void> {
const options: RefUpdateOptions = {}
if (oldCommitHash) {
options.expectedOldValue = oldCommitHash
}
await this.setRef(name, newCommitHash, options)
}
/**
* Update metadata on an existing ref (merge semantics).
*
* @param name - Reference name
* @param metadata - Metadata fields to merge into the ref
*/
async updateRefMetadata(name: string, metadata: Record<string, unknown>): Promise<void> {
const fullName = this.normalizeRefName(name)
const ref = await this.getRef(fullName)
if (!ref) throw new Error(`Ref not found: ${fullName}`)
ref.metadata = { ...ref.metadata, ...metadata }
ref.updatedAt = Date.now()
await this.adapter.put(`ref:${fullName}`, Buffer.from(JSON.stringify(ref)))
this.cache.set(fullName, ref)
}
/**
* Get commit hash for reference
*
* @param name - Reference name
* @returns Commit hash or undefined
*/
async resolveRef(name: string): Promise<string | undefined> {
const ref = await this.getRef(name)
return ref?.commitHash
}
/**
* Find references pointing to commit
*
* @param commitHash - Commit hash
* @returns Array of references pointing to this commit
*/
async findRefsPointingTo(commitHash: string): Promise<Ref[]> {
const allRefs = await this.listRefs()
return allRefs.filter(ref => ref.commitHash === commitHash)
}
/**
* Clear cache (useful for testing)
*/
clearCache(): void {
this.cache.clear()
this.cacheValid = false
}
// ========== PRIVATE METHODS ==========
/**
* Normalize reference name to full format
*
* Examples:
* - 'main' → 'refs/heads/main'
* - 'v1.0.0' (with type='tag') → 'refs/tags/v1.0.0'
* - 'refs/heads/experiment' → 'refs/heads/experiment'
*
* @param name - Reference name
* @param type - Reference type hint
* @returns Full reference name
*/
private normalizeRefName(name: string, type?: RefType): string {
// Already full format
if (name.startsWith('refs/')) {
return name
}
// HEAD is special
if (name === 'HEAD') {
return 'HEAD'
}
// Infer type from name if not provided
if (!type) {
// Tags usually start with 'v' or contain dots
if (name.startsWith('v') && /\d/.test(name)) {
type = 'tag'
} else {
type = 'branch' // Default to branch
}
}
// Add prefix
switch (type) {
case 'branch':
return `refs/heads/${name}`
case 'tag':
return `refs/tags/${name}`
case 'remote':
return `refs/remotes/${name}`
default:
return name
}
}
/**
* Get reference type from full name
*
* @param fullName - Full reference name
* @returns Reference type
*/
private getRefType(fullName: string): RefType {
if (fullName.startsWith('refs/heads/')) {
return 'branch'
} else if (fullName.startsWith('refs/tags/')) {
return 'tag'
} else if (fullName.startsWith('refs/remotes/')) {
return 'remote'
} else {
return 'branch' // Default
}
}
}

359
src/storage/cow/TreeObject.ts
View file

@ -1,359 +0,0 @@
/**
* TreeObject: Directory structure for COW (Copy-on-Write)
*
* Similar to Git trees, represents the structure of a Brainy instance at a point in time.
* Trees contain entries mapping names to blob hashes.
*
* Structure:
* - entities/ → tree hash (entity blobs)
* - indexes/nouns → blob hash (HNSW noun index)
* - indexes/metadata → blob hash (metadata index)
* - indexes/graph → blob hash (graph adjacency index)
* - indexes/deleted → blob hash (deleted items index)
*
* @module storage/cow/TreeObject
*/
import { BlobStorage } from './BlobStorage.js'
import { compareCodePoints } from '../../utils/collation.js'
/**
* Tree entry: name → blob hash mapping
*/
export interface TreeEntry {
name: string
hash: string
type: 'blob' | 'tree' // blob = leaf, tree = subtree
size: number // Original size (uncompressed)
}
/**
* Tree object structure
*/
export interface TreeObject {
entries: TreeEntry[]
createdAt: number
}
/**
* TreeBuilder: Fluent API for building tree objects
*
* Example:
* ```typescript
* const tree = await TreeBuilder.create(blobStorage)
* .addBlob('entities/abc123', entityHash, size)
* .addBlob('indexes/nouns', nounsHash, size)
* .build()
* ```
*/
export class TreeBuilder {
private entries: TreeEntry[] = []
private blobStorage: BlobStorage
constructor(blobStorage: BlobStorage) {
this.blobStorage = blobStorage
}
static create(blobStorage: BlobStorage): TreeBuilder {
return new TreeBuilder(blobStorage)
}
/**
* Add a blob entry to the tree
*
* @param name - Entry name (e.g., 'entities/abc123')
* @param hash - Blob hash
* @param size - Original blob size
*/
addBlob(name: string, hash: string, size: number): this {
this.entries.push({
name,
hash,
type: 'blob',
size
})
return this
}
/**
* Add a subtree entry to the tree
*
* @param name - Subtree name (e.g., 'entities/')
* @param treeHash - Tree hash
* @param size - Total size of subtree
*/
addTree(name: string, treeHash: string, size: number): this {
this.entries.push({
name,
hash: treeHash,
type: 'tree',
size
})
return this
}
/**
* Build and persist the tree object
*
* @returns Tree hash
*/
async build(): Promise<string> {
const tree: TreeObject = {
// Code-point (UTF-8 byte) order so the serialized tree — and therefore its
// content hash — is reproducible across environments (localeCompare's default
// locale varies by OS/Node/ICU). Sorting happens only here at write time;
// deserialize() preserves stored order, so existing trees keep their hashes
// and stay valid — no migration needed.
entries: this.entries.sort((a, b) => compareCodePoints(a.name, b.name)),
createdAt: Date.now()
}
return TreeObject.write(this.blobStorage, tree)
}
}
/**
* TreeObject: Represents directory structure in COW storage
*/
export class TreeObject {
/**
* Serialize tree object to Buffer
*
* Format: JSON (simple, debuggable)
* Future: Could use protobuf for efficiency
*
* @param tree - Tree object
* @returns Serialized tree
*/
static serialize(tree: TreeObject): Buffer {
return Buffer.from(JSON.stringify(tree, null, 0)) // Compact JSON
}
/**
* Deserialize tree object from Buffer
*
* @param data - Serialized tree
* @returns Tree object
*/
static deserialize(data: Buffer): TreeObject {
const tree = JSON.parse(data.toString())
// Validate structure
if (!tree.entries || !Array.isArray(tree.entries)) {
throw new Error('Invalid tree object: missing entries array')
}
if (typeof tree.createdAt !== 'number') {
throw new Error('Invalid tree object: missing or invalid createdAt')
}
// Validate entries
for (const entry of tree.entries) {
if (typeof entry.name !== 'string') {
throw new Error(`Invalid tree entry: name must be string`)
}
if (typeof entry.hash !== 'string' || entry.hash.length !== 64) {
throw new Error(`Invalid tree entry: hash must be 64-char SHA-256`)
}
if (entry.type !== 'blob' && entry.type !== 'tree') {
throw new Error(`Invalid tree entry: type must be 'blob' or 'tree'`)
}
if (typeof entry.size !== 'number' || entry.size < 0) {
throw new Error(`Invalid tree entry: size must be non-negative number`)
}
}
return tree
}
/**
* Compute hash of tree object
*
* @param tree - Tree object
* @returns SHA-256 hash
*/
static hash(tree: TreeObject): string {
const data = TreeObject.serialize(tree)
return BlobStorage.hash(data)
}
/**
* Write tree object to blob storage
*
* @param blobStorage - Blob storage instance
* @param tree - Tree object
* @returns Tree hash
*/
static async write(blobStorage: BlobStorage, tree: TreeObject): Promise<string> {
const data = TreeObject.serialize(tree)
return blobStorage.write(data, { type: 'tree', compression: 'auto' })
}
/**
* Read tree object from blob storage
*
* @param blobStorage - Blob storage instance
* @param hash - Tree hash
* @returns Tree object
*/
static async read(blobStorage: BlobStorage, hash: string): Promise<TreeObject> {
const data = await blobStorage.read(hash)
return TreeObject.deserialize(data)
}
/**
* Get specific entry from tree
*
* @param tree - Tree object
* @param name - Entry name
* @returns Tree entry or undefined
*/
static getEntry(tree: TreeObject, name: string): TreeEntry | undefined {
return tree.entries.find(e => e.name === name)
}
/**
* Get all blob entries from tree (non-recursive)
*
* @param tree - Tree object
* @returns Array of blob entries
*/
static getBlobs(tree: TreeObject): TreeEntry[] {
return tree.entries.filter(e => e.type === 'blob')
}
/**
* Get all subtree entries from tree (non-recursive)
*
* @param tree - Tree object
* @returns Array of tree entries
*/
static getSubtrees(tree: TreeObject): TreeEntry[] {
return tree.entries.filter(e => e.type === 'tree')
}
/**
* Walk tree recursively, yielding all blob entries
*
* @param blobStorage - Blob storage instance
* @param tree - Tree object
*/
static async *walk(
blobStorage: BlobStorage,
tree: TreeObject
): AsyncIterableIterator<TreeEntry> {
for (const entry of tree.entries) {
if (entry.type === 'blob') {
yield entry
} else {
// Recurse into subtree
const subtree = await TreeObject.read(blobStorage, entry.hash)
yield* TreeObject.walk(blobStorage, subtree)
}
}
}
/**
* Compute total size of tree (recursive)
*
* @param blobStorage - Blob storage instance
* @param tree - Tree object
* @returns Total size in bytes
*/
static async getTotalSize(blobStorage: BlobStorage, tree: TreeObject): Promise<number> {
let totalSize = 0
for await (const entry of TreeObject.walk(blobStorage, tree)) {
totalSize += entry.size
}
return totalSize
}
/**
* Create a new tree by updating a single entry
* (Copy-on-write: creates new tree, doesn't modify original)
*
* @param blobStorage - Blob storage instance
* @param tree - Original tree
* @param name - Entry name to update
* @param hash - New blob/tree hash
* @param size - New size
* @returns New tree hash
*/
static async updateEntry(
blobStorage: BlobStorage,
tree: TreeObject,
name: string,
hash: string,
size: number
): Promise<string> {
const builder = TreeBuilder.create(blobStorage)
// Copy all entries except the one being updated
for (const entry of tree.entries) {
if (entry.name === name) {
// Replace with new entry
if (entry.type === 'blob') {
builder.addBlob(name, hash, size)
} else {
builder.addTree(name, hash, size)
}
} else {
// Keep existing entry
if (entry.type === 'blob') {
builder.addBlob(entry.name, entry.hash, entry.size)
} else {
builder.addTree(entry.name, entry.hash, entry.size)
}
}
}
// If entry didn't exist, add it
if (!TreeObject.getEntry(tree, name)) {
builder.addBlob(name, hash, size)
}
return builder.build()
}
/**
* Diff two trees, return changed/added/deleted entries
*
* @param tree1 - First tree (base)
* @param tree2 - Second tree (comparison)
* @returns Diff result
*/
static diff(tree1: TreeObject, tree2: TreeObject): {
added: TreeEntry[]
modified: TreeEntry[]
deleted: TreeEntry[]
} {
const entries1 = new Map(tree1.entries.map(e => [e.name, e]))
const entries2 = new Map(tree2.entries.map(e => [e.name, e]))
const added: TreeEntry[] = []
const modified: TreeEntry[] = []
const deleted: TreeEntry[] = []
// Find added and modified
for (const [name, entry2] of entries2) {
const entry1 = entries1.get(name)
if (!entry1) {
added.push(entry2)
} else if (entry1.hash !== entry2.hash) {
modified.push(entry2)
}
}
// Find deleted
for (const [name, entry1] of entries1) {
if (!entries2.has(name)) {
deleted.push(entry1)
}
}
return { added, modified, deleted }
}
}

100
src/storage/cow/binaryDataCodec.ts
View file

@ -1,100 +0,0 @@
/**
* Binary Data Codec: Single Source of Truth for Wrap/Unwrap Operations
*
* This module provides the ONLY implementation of binary data encoding/decoding
* used across all storage adapters and blob storage.
*
* Design Principles:
* - DRY: One implementation, used everywhere
* - Single Responsibility: Only handles binary ↔ JSON conversion
* - Type-Safe: Proper TypeScript types
* - Defensive: Handles all edge cases
*
* Used by:
* - BaseStorage COW adapter (write/read operations)
* - BlobStorage (defense-in-depth verification)
* - All storage adapters (via BaseStorage)
*
* @module storage/cow/binaryDataCodec
*/
/**
* Wrapped binary data format
* Used when storing binary data in JSON-based storage
*/
export interface WrappedBinaryData {
_binary: true
data: string // base64-encoded
}
/**
* Check if data is wrapped binary format
*/
export function isWrappedBinary(data: any): data is WrappedBinaryData {
return (
typeof data === 'object' &&
data !== null &&
data._binary === true &&
typeof data.data === 'string'
)
}
/**
* Unwrap binary data from JSON wrapper
*
* This is the SINGLE SOURCE OF TRUTH for unwrapping binary data.
* All storage operations MUST use this function.
*
* Handles:
* - Buffer → Buffer (pass-through)
* - {_binary: true, data: "base64..."} → Buffer (unwrap)
* - Plain object → Buffer (JSON stringify)
* - Other types → Error
*
* @param data - Data to unwrap (may be Buffer, wrapped object, or plain object)
* @returns Unwrapped Buffer
* @throws Error if data type is invalid
*/
export function unwrapBinaryData(data: any): Buffer {
// Case 1: Already a Buffer (no unwrapping needed)
if (Buffer.isBuffer(data)) {
return data
}
// Case 2: Wrapped binary data {_binary: true, data: "base64..."}
if (isWrappedBinary(data)) {
return Buffer.from(data.data, 'base64')
}
// Case 3: Plain object (shouldn't happen for binary blobs, but handle gracefully)
if (typeof data === 'object' && data !== null) {
return Buffer.from(JSON.stringify(data))
}
// Case 4: String (convert to Buffer)
if (typeof data === 'string') {
return Buffer.from(data)
}
// Case 5: Invalid type
throw new Error(
`Invalid data type for unwrap: ${typeof data}. ` +
`Expected Buffer or {_binary: true, data: "base64..."}`
)
}
/**
* Ensure data is a Buffer
*
* Convenience function that combines type checking and unwrapping.
* Use this when you need to ensure you have a Buffer.
*
* @param data - Data that should be or can be converted to Buffer
* @returns Buffer
*/
export function ensureBuffer(data: any): Buffer {
if (Buffer.isBuffer(data)) {
return data
}
return unwrapBinaryData(data)
}

59
src/storage/cow/constants.ts
View file

@ -1,59 +0,0 @@
/**
* COW Storage Constants
*
* Sentinel values and utilities for Copy-On-Write storage system.
*
* @module storage/cow/constants
*/
/**
* NULL_HASH - Sentinel value for "no parent commit" or "empty tree"
*
* In Git-like COW systems, we need a way to represent:
* - Initial commit (has no parent)
* - Empty tree (contains no files)
*
* We use a 64-character zero hash as a sentinel value.
* This should NEVER be used as an actual content hash.
*
* @constant
* @example
* ```typescript
* const builder = CommitBuilder.create(storage)
* .tree(NULL_HASH) // Empty tree
* .parent(null) // No parent (use null, not NULL_HASH)
* .build()
* ```
*/
export const NULL_HASH = '0000000000000000000000000000000000000000000000000000000000000000'
/**
* Check if a hash is the NULL sentinel value
*
* @param hash - Hash to check (can be string or null)
* @returns true if hash is null or NULL_HASH
*
* @example
* ```typescript
* if (isNullHash(commit.parent)) {
* console.log('This is the initial commit')
* }
* ```
*/
export function isNullHash(hash: string | null | undefined): boolean {
return hash === null || hash === undefined || hash === NULL_HASH
}
/**
* Check if a hash is valid (non-null, non-empty, proper format)
*
* @param hash - Hash to check
* @returns true if hash is a valid SHA-256 hash
*/
export function isValidHash(hash: string | null | undefined): boolean {
if (isNullHash(hash)) {
return false
}
// SHA-256 hash must be exactly 64 hexadecimal characters
return typeof hash === 'string' && /^[a-f0-9]{64}$/.test(hash)
}

24
src/storage/storageFactory.ts
View file

@ -52,30 +52,10 @@ export interface StorageOptions {
[key: string]: any
}
/** Branch name for COW storage (filesystem only). */
branch?: string
/** Whether to enable COW blob compression. Default `true`. */
enableCompression?: boolean
/** Operation tuning (timeouts, retry budgets) passed through to the adapter. */
operationConfig?: OperationConfig
}
/**
* Attach the COW (copy-on-write) options to the adapter so the storage's
* branch + compression settings are honored when the adapter's
* `initializeCOW()` hook fires later in `brain.init()`.
*/
function configureCOW(storage: any, options?: StorageOptions): void {
if (typeof storage?.initializeCOW === 'function') {
storage._cowOptions = {
branch: options?.branch || 'main',
enableCompression: options?.enableCompression !== false
}
}
}
/**
* Resolve `StorageOptions` to a concrete storage adapter.
*
@ -85,9 +65,7 @@ function configureCOW(storage: any, options?: StorageOptions): void {
* - `forceFileSystemStorage: true` returns `FileSystemStorage`.
*/
export async function createStorage(options: StorageOptions = {}): Promise<StorageAdapter> {
const storage = await pickAdapter(options)
configureCOW(storage, options)
return storage
return pickAdapter(options)
}
async function pickAdapter(options: StorageOptions): Promise<StorageAdapter> {