diff --git a/baradadb.nimble b/baradadb.nimble index 45e0c78..796bb09 100644 --- a/baradadb.nimble +++ b/baradadb.nimble @@ -9,6 +9,8 @@ binDir = "build" # Dependencies requires "nim >= 2.2.0" +requires "hunos >= 1.2.0" +requires "jwt >= 2.1.0" # Tasks task build_debug, "Build debug version": diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index d120f28..bb41739 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -1,18 +1,21 @@ -## BaraDB HTTP Server — REST API with JSON responses -import std/asynchttpserver -import std/asyncdispatch -import std/strutils -import std/json -import std/tables -import std/os +## BaraDB HTTP Server — REST API using Hunos +import hunos +import hunos/router +import hunos/middleware +import hunos/context +import json +import tables +import strutils +import os +import times import config import ../query/lexer import ../query/parser import ../query/executor import ../storage/lsm import ../core/mvcc -import ../protocol/auth import ../protocol/ratelimit +import jwt as jwtlib type HttpServer* = ref object @@ -20,9 +23,8 @@ type running: bool db: LSMTree ctx: ExecutionContext - authManager: AuthManager - rateLimiter*: RateLimiter metrics*: Metrics + secretKey*: string Metrics* = ref object queriesTotal*: int @@ -37,98 +39,75 @@ proc newHttpServer*(config: BaraConfig): HttpServer = let ctx = newExecutionContext(db) ctx.txnManager = newTxnManager() HttpServer(config: config, running: false, db: db, ctx: ctx, - authManager: newAuthManager(), - rateLimiter: newRateLimiter(perClientRate=100), - metrics: Metrics(queriesTotal: 0, queryErrors: 0, - insertCount: 0, selectCount: 0, activeConnections: 0)) - -proc newMetrics*(): Metrics = - Metrics() + secretKey: "baradb-default-secret-change-in-production!", + metrics: Metrics()) # ---------------------------------------------------------------------- -# Authentication middleware +# JWT helpers # ---------------------------------------------------------------------- -proc extractToken(headers: HttpHeaders): string = - let authHeader = headers.getOrDefault("Authorization") - if authHeader.startsWith("Bearer "): - return authHeader[7..^1] - return "" +proc createToken*(server: HttpServer, userId, role: string): string = + let header = %*{"alg": "HS256", "typ": "JWT"} + var claims = newTable[string, Claim]() + claims["sub"] = newStringClaim(userId) + claims["role"] = newStringClaim(role) + claims["iat"] = newTimeClaim(getTime()) + claims["exp"] = newTimeClaim(getTime() + 24.hours) + var token = initJWT(header, claims) + token.sign(server.secretKey) + return $token -proc authorize(server: HttpServer, headers: HttpHeaders): (bool, JWTClaims) = - let token = extractToken(headers) - if token.len == 0: - return (true, JWTClaims(role: "anonymous")) - let (ok, claims) = server.authManager.verifyToken(token) - return (ok, claims) +proc verifyToken*(server: HttpServer, tokenStr: string): (bool, string, string) = + try: + let token = tokenStr.toJWT() + if not token.verify(server.secretKey, HS256): + return (false, "", "") + let userId = token.claims["sub"].node.str + let role = if "role" in token.claims: token.claims["role"].node.str else: "user" + return (true, userId, role) + except: + return (false, "", "") # ---------------------------------------------------------------------- -# JSON helpers +# Handlers # ---------------------------------------------------------------------- -proc jsonError(code: int, message: string): JsonNode = - %* {"error": {"code": code, "message": message}} +proc queryHandler(server: HttpServer): RequestHandler = + return proc(request: Request) {.gcsafe.} = + {.cast(gcsafe).}: + let ctx = newContext(request) + server.metrics.queriesTotal += 1 -# ---------------------------------------------------------------------- -# Request handler -# ---------------------------------------------------------------------- + # Auth check + let authHeader = request.headers["Authorization"] + if authHeader.len > 0 and authHeader.startsWith("Bearer "): + let tokenStr = authHeader[7..^1] + let (valid, userId, role) = server.verifyToken(tokenStr) + if not valid: + ctx.json(%*{"error": "Unauthorized"}, 401) + return -proc corsHeaders(): HttpHeaders = - newHttpHeaders([ - ("Content-Type", "application/json"), - ("Access-Control-Allow-Origin", "*"), - ("Access-Control-Allow-Methods", "GET, POST, OPTIONS"), - ("Access-Control-Allow-Headers", "Content-Type, Authorization"), - ("Access-Control-Max-Age", "86400") - ]) - -proc handleRequest(server: HttpServer, req: Request) {.async, gcsafe.} = - {.cast(gcsafe).}: - inc server.metrics.queriesTotal - inc server.metrics.activeConnections - - # Rate limiting - let clientIp = req.hostname - if not server.rateLimiter.allowRequest(clientIp): - await req.respond(Http429, $jsonError(429, "Rate limit exceeded"), corsHeaders()) - dec server.metrics.activeConnections - return - - # CORS preflight - if req.reqMethod == HttpOptions: - await req.respond(Http200, "", corsHeaders()) - dec server.metrics.activeConnections - return - - case req.url.path - of "/query": - if req.reqMethod != HttpPost: - await req.respond(Http405, $jsonError(405, "Method not allowed. Use POST."), corsHeaders()) + let body = parseJson(request.body) + if body == nil or "query" notin body: + ctx.json(%*{"error": "Missing 'query' in request body"}, 400) return - let (authed, claims) = server.authorize(req.headers) - if not authed: - await req.respond(Http401, $jsonError(401, "Unauthorized"), corsHeaders()) - return - - let queryStr = req.body + let queryStr = body["query"].getStr() if queryStr.len == 0: - await req.respond(Http400, $jsonError(400, "Missing query in request body"), corsHeaders()) + ctx.json(%*{"error": "Empty query"}, 400) return var reqCtx = cloneForConnection(server.ctx) - let tokens = tokenize(queryStr) let astNode = parse(tokens) if astNode.stmts.len == 0: - await req.respond(Http200, $ %* {"rows": [], "affectedRows": 0, "columns": []}, corsHeaders()) + ctx.json(%*{"rows": [], "affectedRows": 0, "columns": []}) return let result = executor.executeQuery(reqCtx, astNode) if result.success: - inc server.metrics.selectCount var jsonRows = newJArray() for row in result.rows: var jsonRow = newJObject() @@ -141,38 +120,95 @@ proc handleRequest(server: HttpServer, req: Request) {.async, gcsafe.} = var jsonCols = newJArray() for c in result.columns: jsonCols.add(%c) - await req.respond(Http200, $ %* { + ctx.json(%*{ "rows": jsonRows, "affectedRows": result.affectedRows, "columns": jsonCols, "message": if result.message.len > 0: %result.message else: newJNull() - }, corsHeaders()) + }) else: - inc server.metrics.queryErrors - await req.respond(Http400, $jsonError(400, result.message), corsHeaders()) + server.metrics.queryErrors += 1 + ctx.json(%*{"error": result.message}, 400) - of "/health": - await req.respond(Http200, $ %* {"status": "ok", "version": "0.1.0"}, corsHeaders()) +proc healthHandler(): RequestHandler = + return proc(request: Request) {.gcsafe.} = + let ctx = newContext(request) + ctx.json(%*{"status": "ok", "version": "0.1.0"}) - of "/metrics": - let prometheus = "baradb_queries_total " & $server.metrics.queriesTotal & "\n" & - "baradb_query_errors_total " & $server.metrics.queryErrors & "\n" & - "baradb_inserts_total " & $server.metrics.insertCount & "\n" & - "baradb_selects_total " & $server.metrics.selectCount & "\n" & - "baradb_connections_active " & $server.metrics.activeConnections & "\n" - await req.respond(Http200, prometheus, newHttpHeaders([("Content-Type", "text/plain")])) +proc metricsHandler(server: HttpServer): RequestHandler = + return proc(request: Request) {.gcsafe.} = + let prometheus = "baradb_queries_total " & $server.metrics.queriesTotal & "\n" & + "baradb_query_errors_total " & $server.metrics.queryErrors & "\n" & + "baradb_inserts_total " & $server.metrics.insertCount & "\n" & + "baradb_selects_total " & $server.metrics.selectCount & "\n" & + "baradb_connections_active " & $server.metrics.activeConnections & "\n" + request.respond(200, @[("Content-Type", "text/plain")], prometheus) - else: - await req.respond(Http404, $jsonError(404, "Not found: " & req.url.path), corsHeaders()) +proc authHandler(server: HttpServer): RequestHandler = + return proc(request: Request) {.gcsafe.} = + {.cast(gcsafe).}: + let ctx = newContext(request) + let body = parseJson(request.body) + if body == nil or "username" notin body or "password" notin body: + ctx.json(%*{"error": "Missing username or password"}, 400) + return - dec server.metrics.activeConnections + let username = body["username"].getStr() + let token = server.createToken(username, "user") + ctx.json(%*{ + "token": token, + "user": username, + "role": "user" + }) -proc run*(server: HttpServer, port: int = 8080) {.async.} = +proc openApiHandler(): RequestHandler = + return proc(request: Request) {.gcsafe.} = + let ctx = newContext(request) + ctx.json(%*{ + "openapi": "3.0.0", + "info": {"title": "BaraDB API", "version": "0.1.0"}, + "paths": { + "/query": { + "post": { + "summary": "Execute SQL query", + "requestBody": { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"] + } + } + } + } + } + }, + "/health": {"get": {"summary": "Health check"}}, + "/metrics": {"get": {"summary": "Prometheus metrics"}}, + "/auth": {"post": {"summary": "Authenticate and get JWT token"}} + } + }) + +# ---------------------------------------------------------------------- +# Server lifecycle +# ---------------------------------------------------------------------- + +proc run*(server: HttpServer, port: int = 8080) = + var router = newRouter() + router.post("/query", server.queryHandler()) + router.get("/health", healthHandler()) + router.get("/metrics", server.metricsHandler()) + router.post("/auth", server.authHandler()) + router.get("/api", openApiHandler()) + + var stack = newMiddlewareStack(router) + stack.use(corsMiddleware()) + + let hunosServer = newServer(stack) + echo "BaraDB HTTP listening on port ", port server.running = true - var httpServer = newAsyncHttpServer() - let serverRef = server - await httpServer.serve(Port(port), proc (req: Request): Future[void] {.gcsafe.} = - serverRef.handleRequest(req)) + hunosServer.serve(Port(port)) proc stop*(server: HttpServer) = server.running = false diff --git a/src/barabadb/storage/compaction.nim b/src/barabadb/storage/compaction.nim index 170768c..eb8abc1 100644 --- a/src/barabadb/storage/compaction.nim +++ b/src/barabadb/storage/compaction.nim @@ -3,6 +3,7 @@ import std/tables import std/algorithm import std/os import std/math +import ../storage/lsm const MaxLevel* = 7 @@ -68,25 +69,63 @@ proc compact*(cs: CompactionStrategy, level: int): CompactionResult = return CompactionResult() var entriesRead = 0 - var allEntries: seq[(string, seq[byte], uint64, bool)] = @[] + var allEntries: seq[Entry] = @[] + # Read all entries from SSTable files for t in tables: - entriesRead += t.entryCount - # In real impl, would read SSTable file and merge - # For now, simulate the merge + try: + let sst = loadSSTable(t.path) + for key, offset in sst.index: + let (found, entry) = readSSTableEntry(sst, key) + if found: + allEntries.add(entry) + inc entriesRead + except: + discard + # Sort by key, then by timestamp (newest first for dedup) + allEntries.sort(proc(a, b: Entry): int = + let kc = cmp(a.key, b.key) + if kc != 0: return kc + return cmp(b.timestamp, a.timestamp) # newest first + ) + + # Deduplicate: keep only the newest version of each key + var merged: seq[Entry] = @[] + var lastKey = "" + for entry in allEntries: + if entry.key != lastKey: + merged.add(entry) + lastKey = entry.key + + # Filter out tombstones (deleted entries) + var final: seq[Entry] = @[] + for entry in merged: + if not entry.deleted: + final.add(entry) + + # Write merged SSTable let outputPath = cs.dataDir / "sstables" / ("level_" & $level & "_" & $tables[0].createdAt & ".sst") + var sst = writeSSTable(final, outputPath, level + 1) + let outputMeta = SSTableMeta( path: outputPath, level: level + 1, - minKey: tables[0].minKey, - maxKey: tables[^1].maxKey, - entryCount: entriesRead, - sizeBytes: entriesRead * 64, # estimate + minKey: sst.minKey, + maxKey: sst.maxKey, + entryCount: final.len, + sizeBytes: final.len * 64, createdAt: tables[^1].createdAt, ) - # Remove old tables from level + # Remove old SSTable files + for t in tables: + try: + removeFile(t.path) + except: + discard + + # Update level arrays var newTables: seq[SSTableMeta] = @[] for t in cs.levels[level]: var found = false @@ -98,7 +137,6 @@ proc compact*(cs: CompactionStrategy, level: int): CompactionResult = newTables.add(t) cs.levels[level] = newTables - # Add to next level if level + 1 < MaxLevel: cs.levels[level + 1].add(outputMeta) @@ -106,7 +144,7 @@ proc compact*(cs: CompactionStrategy, level: int): CompactionResult = inputTables: tables, outputTables: @[outputMeta], entriesRead: entriesRead, - entriesWritten: entriesRead, + entriesWritten: final.len, ) proc levelCount*(cs: CompactionStrategy): int = diff --git a/src/barabadb/storage/lsm.nim b/src/barabadb/storage/lsm.nim index 934448f..fe58371 100644 --- a/src/barabadb/storage/lsm.nim +++ b/src/barabadb/storage/lsm.nim @@ -31,13 +31,13 @@ type SSTable* = object path*: string - index: Table[string, int64] # key -> file offset in data block - bloom: BloomFilter - level: int - minKey: string - maxKey: string - entryCount: int - mmapFile: MmapFile + index*: Table[string, int64] + bloom*: BloomFilter + level*: int + minKey*: string + maxKey*: string + entryCount*: int + mmapFile*: MmapFile LSMTree* = ref object dir: string diff --git a/src/barabadb/storage/recovery.nim b/src/barabadb/storage/recovery.nim index fb4a329..195b502 100644 --- a/src/barabadb/storage/recovery.nim +++ b/src/barabadb/storage/recovery.nim @@ -2,6 +2,7 @@ import std/streams import std/os import ../storage/wal +import ../storage/lsm type RecoveryState* = enum @@ -50,7 +51,6 @@ proc scanWAL*(rec: CrashRecovery): seq[RecoveredEntry] = if stream == nil: return - # Read magic and version var magic: uint32 = 0 var version: uint32 = 0 if stream.readData(addr magic, 4) != 4: @@ -67,7 +67,6 @@ proc scanWAL*(rec: CrashRecovery): seq[RecoveredEntry] = var txnId: uint64 = 0 var entryCount = 0 - # Read entries while not stream.atEnd(): var kind: uint8 = 0 var timestamp: uint64 = 0 @@ -110,16 +109,11 @@ proc analyze*(rec: CrashRecovery): RecoveryResult = rec.result = RecoveryResult(state: recDone, applied: false) return rec.result - # Find last successful checkpoint or committed txn var lastCommitted: uint64 = 0 - var uncommittedEntries: seq[RecoveredEntry] = @[] - for entry in rec.entries: if entry.txnId > lastCommitted: lastCommitted = entry.txnId - # Entries with txnId < lastCommitted are committed -> redo - # Entries with txnId == lastCommitted and no commit seen -> undo var redoCount = 0 var undoCount = 0 @@ -141,15 +135,35 @@ proc analyze*(rec: CrashRecovery): RecoveryResult = return rec.result -proc recover*(rec: CrashRecovery): RecoveryResult = +proc recover*(rec: CrashRecovery, db: LSMTree = nil): RecoveryResult = let result = rec.analyze() if not result.applied: return result - # In a real system, would redo committed entries and undo uncommitted ones - # For now, provide the analysis result - return result + if db == nil: + return result + + # REDO: apply committed entries (txnId < lastCommitted) to LSM-Tree + var redoCount = 0 + var undoCount = 0 + for entry in rec.entries: + if entry.txnId < result.lastTxn: + # Committed — redo + if entry.isDelete: + db.delete(entry.key) + else: + db.put(entry.key, entry.value) + inc redoCount + else: + # Uncommitted — skip (undo) + inc undoCount + + rec.result.redone = redoCount + rec.result.undone = undoCount + rec.result.state = recDone + + return rec.result proc totalEntries*(rec: CrashRecovery): int = rec.entries.len