open-brainy/src/graph/lsm/LSMTree.ts

/**
 * @module graph/lsm/LSMTree
 * @description Log-Structured Merge tree for the JS (open-core) graph store — the
 * fallback used when no native graph provider is registered. Verb-id postings are
 * buffered in an in-memory MemTable, flushed to immutable sorted SSTables, and
 * background-compacted; bloom filters give fast negative lookups.
 *
 * Architecture:
 * - MemTable: in-memory write buffer (flush threshold configurable, default 100K)
 * - SSTables: immutable sorted segments, persisted via the StorageAdapter
 * - Bloom filters: in-memory membership pre-checks
 * - Compaction: background merge of SSTables (reclaims superseded segments)
 *
 * Complexity: O(1) MemTable writes; O(log n) reads with bloom-filter pre-filtering.
 * Note: this JS implementation loads its SSTables into memory after open (it is the
 * fallback engine). Billion-scale, on-disk-resident operation is the native graph
 * provider's role behind the provider boundary, not this fallback's; no absolute
 * memory/latency figures are claimed here without a cited benchmark.
 */

import { StorageAdapter } from '../../coreTypes.js'
import { SSTable, SSTableEntry } from './SSTable.js'
import { prodLog } from '../../utils/logger.js'

/**
 * LSMTree configuration
 */
export interface LSMTreeConfig {
  /**
   * MemTable flush threshold (number of relationships)
   * Default: 100000 (100K relationships, ~24MB RAM)
   */
  memTableThreshold?: number

  /**
   * Maximum number of SSTables at each level before compaction
   * Default: 10
   */
  maxSSTablesPerLevel?: number

  /**
   * Storage key prefix for SSTables
   * Default: 'graph-lsm'
   */
  storagePrefix?: string

  /**
   * Enable background compaction
   * Default: true
   */
  enableCompaction?: boolean

  /**
   * Compaction interval in milliseconds
   * Default: 60000 (1 minute)
   */
  compactionInterval?: number
}

/**
 * In-memory write buffer (MemTable)
 * Stores recent writes before flushing to SSTable
 */
class MemTable {
  /**
   * sourceId → targetIds
   */
  private data: Map<string, Set<string>>

  /**
   * Number of relationships in MemTable
   */
  private count: number

  constructor() {
    this.data = new Map()
    this.count = 0
  }

  /**
   * Add a relationship
   */
  add(sourceId: string, targetId: string): void {
    if (!this.data.has(sourceId)) {
      this.data.set(sourceId, new Set())
    }

    const targets = this.data.get(sourceId)!
    if (!targets.has(targetId)) {
      targets.add(targetId)
      this.count++
    }
  }

  /**
   * Get targets for a sourceId
   */
  get(sourceId: string): string[] | null {
    const targets = this.data.get(sourceId)
    return targets ? Array.from(targets) : null
  }

  /**
   * Get all entries as Map for flushing
   */
  getAll(): Map<string, Set<string>> {
    return this.data
  }

  /**
   * Get number of relationships
   */
  size(): number {
    return this.count
  }

  /**
   * Check if empty
   */
  isEmpty(): boolean {
    return this.count === 0
  }

  /**
   * Clear all data
   */
  clear(): void {
    this.data.clear()
    this.count = 0
  }

  /**
   * Estimate memory usage
   */
  estimateMemoryUsage(): number {
    let bytes = 0
    this.data.forEach((targets, sourceId) => {
      bytes += sourceId.length * 2 // UTF-16
      bytes += targets.size * 40 // ~40 bytes per UUID
    })
    return bytes
  }
}

/**
 * Manifest - Tracks all SSTables and their levels
 */
/**
 * Persisted manifest payload as written by `saveManifest()` (the `data` field
 * of the manifest metadata record, after a JSON round-trip).
 */
type PersistedManifestData = {
  sstables?: Record<string, number>
  lastCompaction?: number
  totalRelationships?: number
}

/**
 * Persisted SSTable payload as written by `flushMemTable()`/`compact()` (the
 * `data` field of an SSTable metadata record): serialized bytes stored as a
 * plain number array so they survive JSON round-trips.
 */
type PersistedSSTableData = {
  type: string
  data: number[]
}

interface Manifest {
  /**
   * Map of SSTable ID to level
   */
  sstables: Map<string, number>

  /**
   * Last compaction time
   */
  lastCompaction: number

  /**
   * Total number of relationships
   */
  totalRelationships: number
}

/**
 * LSMTree - Main LSM-tree implementation
 *
 * Provides efficient graph storage with:
 * - Fast writes via MemTable
 * - Efficient reads via bloom filters and binary search
 * - Automatic compaction to maintain performance
 * - Integration with any StorageAdapter
 */
export class LSMTree {
  /**
   * Storage adapter for persistence
   */
  private storage: StorageAdapter

  /**
   * Configuration
   */
  private config: Required<LSMTreeConfig>

  /**
   * In-memory write buffer
   */
  private memTable: MemTable

  /**
   * Loaded SSTables grouped by level
   * Level 0: Fresh from MemTable (smallest, most recent)
   * Level 1-6: Progressively larger, older, merged files
   */
  private sstablesByLevel: Map<number, SSTable[]>

  /**
   * Manifest tracking all SSTables
   */
  private manifest: Manifest

  /**
   * Compaction timer
   */
  private compactionTimer?: NodeJS.Timeout

  /**
   * Whether compaction is currently running
   */
  private isCompacting: boolean

  /**
   * Whether LSMTree has been initialized
   */
  private initialized: boolean

  constructor(storage: StorageAdapter, config: LSMTreeConfig = {}) {
    this.storage = storage
    this.config = {
      memTableThreshold: config.memTableThreshold ?? 100000,
      maxSSTablesPerLevel: config.maxSSTablesPerLevel ?? 10,
      storagePrefix: config.storagePrefix ?? 'graph-lsm',
      enableCompaction: config.enableCompaction ?? true,
      compactionInterval: config.compactionInterval ?? 60000
    }

    this.memTable = new MemTable()
    this.sstablesByLevel = new Map()
    this.manifest = {
      sstables: new Map(),
      lastCompaction: Date.now(),
      totalRelationships: 0
    }
    this.isCompacting = false
    this.initialized = false
  }

  /**
   * Initialize the LSMTree
   * Loads manifest and prepares for operations
   */
  async init(): Promise<void> {
    if (this.initialized) {
      return
    }

    try {
      // Load manifest from storage
      await this.loadManifest()

      // Start compaction timer if enabled
      if (this.config.enableCompaction) {
        this.startCompactionTimer()
      }

      this.initialized = true
      prodLog.info('LSMTree: Initialized successfully')
    } catch (error) {
      prodLog.error('LSMTree: Initialization failed', error)
      throw error
    }
  }

  /**
   * Add a relationship to the LSM-tree
   * @param sourceId Source node ID
   * @param targetId Target node ID
   */
  async add(sourceId: string, targetId: string): Promise<void> {
    const startTime = performance.now()

    // Add to MemTable
    this.memTable.add(sourceId, targetId)
    this.manifest.totalRelationships++

    // Check if MemTable needs flushing
    if (this.memTable.size() >= this.config.memTableThreshold) {
      await this.flushMemTable()
    }

    const elapsed = performance.now() - startTime

    // Performance assertion - writes should be fast
    if (elapsed > 10.0) {
      prodLog.warn(`LSMTree: Slow write operation: ${elapsed.toFixed(2)}ms`)
    }
  }

  /**
   * Get targets for a sourceId
   * Checks MemTable first, then SSTables with bloom filter optimization
   *
   * @param sourceId Source node ID
   * @returns Array of target IDs, or null if not found
   */
  async get(sourceId: string): Promise<string[] | null> {
    const startTime = performance.now()

    // Merge results from MemTable AND SSTables
    // Data can span both after a flush (old data in SSTables, new in MemTable)
    const allTargets = new Set<string>()

    // Check MemTable (hot data)
    const memResult = this.memTable.get(sourceId)
    if (memResult !== null) {
      for (const target of memResult) {
        allTargets.add(target)
      }
    }

    // Check SSTables from newest to oldest
    const maxLevel = Math.max(...Array.from(this.sstablesByLevel.keys()), 0)

    for (let level = 0; level <= maxLevel; level++) {
      const sstables = this.sstablesByLevel.get(level) || []

      for (const sstable of sstables) {
        // Quick check: Is sourceId in range?
        if (!sstable.isInRange(sourceId)) {
          continue
        }

        // Quick check: Does bloom filter say it might be here?
        if (!sstable.mightContain(sourceId)) {
          continue
        }

        // Binary search in SSTable
        const targets = sstable.get(sourceId)
        if (targets) {
          for (const target of targets) {
            allTargets.add(target)
          }
        }
      }
    }

    const elapsed = performance.now() - startTime

    // Performance assertion - reads should be fast
    if (elapsed > 5.0) {
      prodLog.warn(`LSMTree: Slow read operation for ${sourceId}: ${elapsed.toFixed(2)}ms`)
    }

    return allTargets.size > 0 ? Array.from(allTargets) : null
  }

  /**
   * Flush MemTable to a new L0 SSTable
   */
  private async flushMemTable(): Promise<void> {
    if (this.memTable.isEmpty()) {
      return
    }

    const startTime = Date.now()
    prodLog.info(`LSMTree: Flushing MemTable (${this.memTable.size()} relationships)`)

    try {
      // Create SSTable from MemTable
      const sstable = SSTable.fromMap(this.memTable.getAll(), 0)

      // Serialize and save to storage
      const data = sstable.serialize()
      const storageKey = `${this.config.storagePrefix}-${sstable.metadata.id}`

      await this.storage.saveMetadata(storageKey, {
        noun: 'thing', // Required for NounMetadata
        data: {
          type: 'lsm-sstable',
          data: Array.from(data) // Convert Uint8Array to number[] for JSON storage
        }
      })

      // Add to L0 SSTables
      if (!this.sstablesByLevel.has(0)) {
        this.sstablesByLevel.set(0, [])
      }
      this.sstablesByLevel.get(0)!.push(sstable)

      // Update manifest
      this.manifest.sstables.set(sstable.metadata.id, 0)
      await this.saveManifest()

      // Clear MemTable
      this.memTable.clear()

      const elapsed = Date.now() - startTime
      prodLog.info(`LSMTree: MemTable flushed in ${elapsed}ms`)

      // Check if L0 needs compaction
      const l0Count = this.sstablesByLevel.get(0)?.length || 0
      if (l0Count >= this.config.maxSSTablesPerLevel) {
        // Trigger compaction asynchronously
        setImmediate(() => this.compact(0))
      }
    } catch (error) {
      prodLog.error('LSMTree: Failed to flush MemTable', error)
      throw error
    }
  }

  /**
   * Compact a level by merging SSTables
   * @param level Level to compact
   */
  private async compact(level: number): Promise<void> {
    if (this.isCompacting) {
      prodLog.debug('LSMTree: Compaction already in progress, skipping')
      return
    }

    this.isCompacting = true
    const startTime = Date.now()

    try {
      const sstables = this.sstablesByLevel.get(level) || []
      if (sstables.length < this.config.maxSSTablesPerLevel) {
        this.isCompacting = false
        return
      }

      prodLog.info(`LSMTree: Compacting L${level} (${sstables.length} SSTables)`)

      // Merge all SSTables at this level
      const merged = SSTable.merge(sstables, level + 1)

      // Serialize and save merged SSTable
      const data = merged.serialize()
      const storageKey = `${this.config.storagePrefix}-${merged.metadata.id}`

      await this.storage.saveMetadata(storageKey, {
        noun: 'thing', // Required for NounMetadata
        data: {
          type: 'lsm-sstable',
          data: Array.from(data)
        }
      })

      // Reclaim the compacted-away SSTables: drop them from the manifest AND
      // delete their persisted payloads, so the system channel does not grow
      // unbounded with graph write volume. (deleteMetadata is idempotent, so a
      // never-persisted memtable-only SSTable is a harmless no-op.)
      for (const sstable of sstables) {
        const oldKey = `${this.config.storagePrefix}-${sstable.metadata.id}`
        this.manifest.sstables.delete(sstable.metadata.id)
        try {
          await this.storage.deleteMetadata(oldKey)
        } catch (error) {
          // A reclaim failure must not abort compaction (the merged SSTable is
          // already durable and the manifest no longer references the old one);
          // surface it so a persistent leak is visible rather than silent.
          prodLog.warn(`LSMTree: failed to delete compacted SSTable ${sstable.metadata.id}`, error)
        }
      }

      // Update in-memory structures
      this.sstablesByLevel.set(level, [])

      if (!this.sstablesByLevel.has(level + 1)) {
        this.sstablesByLevel.set(level + 1, [])
      }
      this.sstablesByLevel.get(level + 1)!.push(merged)

      // Update manifest
      this.manifest.sstables.set(merged.metadata.id, level + 1)
      this.manifest.lastCompaction = Date.now()
      await this.saveManifest()

      const elapsed = Date.now() - startTime
      prodLog.info(`LSMTree: Compaction complete in ${elapsed}ms`)

      // Check if next level needs compaction
      const nextLevelCount = this.sstablesByLevel.get(level + 1)?.length || 0
      if (nextLevelCount >= this.config.maxSSTablesPerLevel && level < 6) {
        // Trigger next level compaction
        setImmediate(() => this.compact(level + 1))
      }
    } catch (error) {
      prodLog.error(`LSMTree: Compaction failed for L${level}`, error)
    } finally {
      this.isCompacting = false
    }
  }

  /**
   * Start background compaction timer
   */
  private startCompactionTimer(): void {
    this.compactionTimer = setInterval(() => {
      // Check each level for compaction needs
      for (let level = 0; level < 6; level++) {
        const count = this.sstablesByLevel.get(level)?.length || 0
        if (count >= this.config.maxSSTablesPerLevel) {
          this.compact(level)
          break // Only compact one level per interval
        }
      }
    }, this.config.compactionInterval)
    // Background compaction must never keep the host process alive —
    // close() compacts/flushes deterministically; this interval is best-effort.
    if (this.compactionTimer && typeof this.compactionTimer.unref === 'function') {
      this.compactionTimer.unref()
    }
  }

  /**
   * Stop background compaction timer
   */
  private stopCompactionTimer(): void {
    if (this.compactionTimer) {
      clearInterval(this.compactionTimer)
      this.compactionTimer = undefined
    }
  }

  /**
   * Load manifest from storage
   */
  private async loadManifest(): Promise<void> {
    try {
      const metadata = await this.storage.getMetadata(`${this.config.storagePrefix}-manifest`)

      if (metadata && metadata.data) {
        // Storage boundary: `data` is the JSON manifest payload written by
        // saveManifest(); re-typed from the metadata channel's `unknown`.
        const data = metadata.data as PersistedManifestData
        this.manifest.sstables = new Map(Object.entries(data.sstables || {}))
        this.manifest.lastCompaction = data.lastCompaction || Date.now()
        this.manifest.totalRelationships = data.totalRelationships || 0

        // Load SSTables from storage
        await this.loadSSTables()
      }
    } catch (error) {
      prodLog.debug('LSMTree: No existing manifest found, starting fresh')
    }
  }

  /**
   * Load SSTables from storage based on manifest
   */
  private async loadSSTables(): Promise<void> {
    const loadPromises: Promise<void>[] = []

    this.manifest.sstables.forEach((level, sstableId) => {
      const loadPromise = (async () => {
        try {
          const storageKey = `${this.config.storagePrefix}-${sstableId}`
          const metadata = await this.storage.getMetadata(storageKey)

          if (metadata && metadata.data) {
            // Storage boundary: `data` is the JSON SSTable payload written by
            // flushMemTable()/compact(); re-typed from `unknown`.
            const data = metadata.data as PersistedSSTableData
            if (data.type === 'lsm-sstable') {
              // Convert number[] back to Uint8Array
              const uint8Data = new Uint8Array(data.data)
              const sstable = SSTable.deserialize(uint8Data)

              if (!this.sstablesByLevel.has(level)) {
                this.sstablesByLevel.set(level, [])
              }
              this.sstablesByLevel.get(level)!.push(sstable)
            }
          }
        } catch (error) {
          prodLog.warn(`LSMTree: Failed to load SSTable ${sstableId}`, error)
        }
      })()

      loadPromises.push(loadPromise)
    })

    await Promise.all(loadPromises)
    prodLog.info(`LSMTree: Loaded ${this.manifest.sstables.size} SSTables`)
  }

  /**
   * Save manifest to storage
   */
  private async saveManifest(): Promise<void> {
    try {
      await this.storage.saveMetadata(
        `${this.config.storagePrefix}-manifest`,
        {
          noun: 'thing', // Required for NounMetadata
          data: {
            sstables: Object.fromEntries(this.manifest.sstables),
            lastCompaction: this.manifest.lastCompaction,
            totalRelationships: this.manifest.totalRelationships
          }
        }
      )
    } catch (error) {
      prodLog.error('LSMTree: Failed to save manifest', error)
      throw error
    }
  }

  /**
   * Get statistics about the LSM-tree
   */
  getStats(): {
    memTableSize: number
    memTableMemory: number
    sstableCount: number
    sstablesByLevel: Record<number, number>
    totalRelationships: number
    lastCompaction: number
  } {
    const sstablesByLevel: Record<number, number> = {}
    this.sstablesByLevel.forEach((sstables, level) => {
      sstablesByLevel[level] = sstables.length
    })

    return {
      memTableSize: this.memTable.size(),
      memTableMemory: this.memTable.estimateMemoryUsage(),
      sstableCount: this.manifest.sstables.size,
      sstablesByLevel,
      totalRelationships: this.manifest.totalRelationships,
      lastCompaction: this.manifest.lastCompaction
    }
  }

  /**
   * Flush MemTable to SSTables without closing
   * Called by GraphAdjacencyIndex.flush() and brain.close()
   */
  async flush(): Promise<void> {
    if (!this.memTable.isEmpty()) {
      await this.flushMemTable()
    }
  }

  async close(): Promise<void> {
    this.stopCompactionTimer()

    // Final MemTable flush
    await this.flush()

    prodLog.info('LSMTree: Closed successfully')
  }

  /**
   * Get total relationship count
   */
  size(): number {
    return this.manifest.totalRelationships
  }

  /**
   * Check if LSM-tree is healthy
   */
  isHealthy(): boolean {
    return this.initialized && !this.isCompacting
  }
}