open-brainy/src/storage/cow/BlobStorage.ts

/**
 * 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'

/**
 * 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' | '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' | 'raw'
  skipVerification?: boolean  // Skip hash verification (faster, less safe)
}

/**
 * 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>

  // 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
    }
  }

  /**
   * 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
    }

    // 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
    const metadata: BlobMetadata = {
      hash,
      size: data.length,
      compressedSize,
      compression,
      type: options.type || 'raw',
      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> {
    // v5.3.4 fix: 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
        metadata = JSON.parse(metadataBuffer.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)
    }

    // Verify hash (optional, expensive)
    if (!options.skipVerification && BlobStorage.hash(finalData) !== hash) {
      throw new Error(`Blob integrity check failed: ${hash}`)
    }

    // Add to cache (only if not skipped)
    if (!options.skipCache) {
      this.addToCache(hash, finalData, metadata)
    }

    return finalData
  }

  /**
   * 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']) {
      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)
    // Try commit, tree, then blob prefixes
    for (const prefix of ['commit', 'tree', 'blob']) {
      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
    }

    // 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> {
    // For now, just write as single blob
    // TODO: Implement actual multipart upload for S3/R2/GCS
    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)
    }
  }
}