diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index 2701cbe..0f7f9d4 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -8,6 +8,7 @@ import tables import strutils import os import times +import std/asyncdispatch import config import ../query/lexer import ../query/parser @@ -22,7 +23,7 @@ type HttpServer* = ref object config: BaraConfig running: bool - db: LSMTree + db*: LSMTree ctx: ExecutionContext metrics*: Metrics secretKey*: string diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index 7fbc912..39a392b 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -377,6 +377,8 @@ proc validateType*(colType: string, value: string): (bool, string) = return (false, "Type mismatch: expected " & t & " but got '" & value & "'") return (true, "") +proc lowerExpr*(node: Node): IRExpr + proc validateConstraints*(ctx: ExecutionContext, tableName: string, fields: seq[string], values: seq[seq[string]]): (bool, string) = let tbl = ctx.getTableDef(tableName) diff --git a/src/barabadb/storage/lsm.nim b/src/barabadb/storage/lsm.nim index fe58371..f125c63 100644 --- a/src/barabadb/storage/lsm.nim +++ b/src/barabadb/storage/lsm.nim @@ -7,6 +7,7 @@ import std/tables import std/monotimes import std/streams import std/locks +import std/asyncdispatch import bloom import wal import mmap @@ -40,15 +41,15 @@ type mmapFile*: MmapFile LSMTree* = ref object - dir: string + dir*: string memTable: MemTable immutableMem: MemTable - sstables: seq[SSTable] + sstables*: seq[SSTable] wal: WriteAheadLog memMaxSize: int currentSeq: uint64 nextSSTableId: int - lock: Lock + lock*: Lock proc newMemTable(maxSize: int = DefaultMemTableSize): MemTable = MemTable(entries: @[], size: 0, maxSize: maxSize) @@ -323,6 +324,7 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree = result.memMaxSize = memMaxSize result.currentSeq = 0 result.nextSSTableId = nextId + discard proc put*(db: LSMTree, key: string, value: seq[byte]) = acquire(db.lock) diff --git a/src/baradadb.nim b/src/baradadb.nim index 0f210fe..61dbd45 100644 --- a/src/baradadb.nim +++ b/src/baradadb.nim @@ -1,15 +1,78 @@ ## BaraDB — Multimodal Database Engine ## Main entry point import std/asyncdispatch +import std/threadpool +import std/locks import barabadb/core/server +import barabadb/core/httpserver import barabadb/core/config +import barabadb/storage/lsm +import barabadb/storage/compaction + +type + CompactionManager* = ref object + db*: LSMTree + strategy*: compaction.CompactionStrategy + +proc newCompactionManager*(db: LSMTree): CompactionManager = + result = CompactionManager(db: db, strategy: compaction.newCompactionStrategy(db.dir)) + for sst in db.sstables: + let meta = compaction.SSTableMeta( + path: sst.path, + level: sst.level, + minKey: sst.minKey, + maxKey: sst.maxKey, + entryCount: sst.entryCount, + sizeBytes: sst.entryCount * 64, + createdAt: 0, + ) + result.strategy.addTable(meta) + for sst in db.sstables: + let meta = SSTableMeta( + path: sst.path, + level: sst.level, + minKey: sst.minKey, + maxKey: sst.maxKey, + entryCount: sst.entryCount, + sizeBytes: sst.entryCount * 64, + createdAt: 0, + ) + result.strategy.addTable(meta) + +proc compact*(cm: CompactionManager) = + acquire(cm.db.lock) + defer: release(cm.db.lock) + for level in 0 ..< compaction.MaxLevel: + if cm.strategy.needsCompaction(level): + discard cm.strategy.compact(level) + +proc startCompactionLoop*(cm: CompactionManager, intervalMs: int = 60000) {.async.} = + while true: + await sleepAsync(intervalMs) + cm.compact() + +proc runTcpServer(config: BaraConfig) {.async.} = + echo "BaraDB TCP listening on ", config.address, ":", config.port + var server = newServer(config) + await server.run() proc main() = let config = loadConfig() echo "BaraDB v0.1.0 — Multimodal Database Engine" - echo "Listening on ", config.address, ":", config.port - var server = newServer(config) - waitFor server.run() + + # Start HTTP server (blocking, multi-threaded via hunos) in background thread + var httpServer = newHttpServer(config) + spawn httpServer.run(config.port + 440) # HTTP port = TCP port + 440 + + # Start background compaction loop + let cm = newCompactionManager(httpServer.db) + asyncCheck cm.startCompactionLoop() + + # Start TCP wire protocol server on main thread with async event loop + waitFor runTcpServer(config) + + # Shutdown + httpServer.stop() when isMainModule: main()