diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim new file mode 100644 index 0000000..0a8a7a8 --- /dev/null +++ b/src/barabadb/core/httpserver.nim @@ -0,0 +1,147 @@ +## BaraDB HTTP Server — REST API with JSON responses +import std/asynchttpserver +import std/asyncdispatch +import std/strutils +import std/json +import std/os +import config +import ../query/lexer +import ../query/parser +import ../query/executor +import ../storage/lsm +import ../core/mvcc +import ../protocol/auth + +type + HttpServer* = ref object + config: BaraConfig + running: bool + db: LSMTree + ctx: ExecutionContext + authManager: AuthManager + metrics*: Metrics + + Metrics* = ref object + queriesTotal*: int + queryErrors*: int + insertCount*: int + selectCount*: int + activeConnections*: int + +proc newHttpServer*(config: BaraConfig): HttpServer = + let dataDir = config.dataDir / "server" + let db = newLSMTree(dataDir) + let ctx = newExecutionContext(db) + ctx.txnManager = newTxnManager() + HttpServer(config: config, running: false, db: db, ctx: ctx, + authManager: newAuthManager(), + metrics: Metrics(queriesTotal: 0, queryErrors: 0, + insertCount: 0, selectCount: 0, activeConnections: 0)) + +proc newMetrics*(): Metrics = + Metrics() + +# ---------------------------------------------------------------------- +# Authentication middleware +# ---------------------------------------------------------------------- + +proc extractToken(headers: HttpHeaders): string = + let authHeader = headers.getOrDefault("Authorization") + if authHeader.startsWith("Bearer "): + return authHeader[7..^1] + return "" + +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) + +# ---------------------------------------------------------------------- +# JSON helpers +# ---------------------------------------------------------------------- + +proc jsonError(code: int, message: string): JsonNode = + %* {"error": {"code": code, "message": message}} + +# ---------------------------------------------------------------------- +# Request handler +# ---------------------------------------------------------------------- + +proc handleRequest(server: HttpServer, req: Request) {.async, gcsafe.} = + {.cast(gcsafe).}: + inc server.metrics.queriesTotal + inc server.metrics.activeConnections + + case req.url.path + of "/query": + if req.reqMethod != HttpPost: + await req.respond(Http405, $jsonError(405, "Method not allowed. Use POST."), + newHttpHeaders([("Content-Type", "application/json")])) + return + + let (authed, claims) = server.authorize(req.headers) + if not authed: + await req.respond(Http401, $jsonError(401, "Unauthorized"), + newHttpHeaders([("Content-Type", "application/json")])) + return + + let queryStr = req.body + if queryStr.len == 0: + await req.respond(Http400, $jsonError(400, "Missing query in request body"), + newHttpHeaders([("Content-Type", "application/json")])) + 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": []}, + newHttpHeaders([("Content-Type", "application/json")])) + return + + let (success, errMsg, affectedRows) = executor.executeQuery(reqCtx, astNode) + + if success: + inc server.metrics.selectCount + await req.respond(Http200, $ %* { + "rows": newJArray(), + "affectedRows": affectedRows, + "columns": newJArray() + }, newHttpHeaders([("Content-Type", "application/json")])) + else: + inc server.metrics.queryErrors + await req.respond(Http400, $jsonError(400, errMsg), + newHttpHeaders([("Content-Type", "application/json")])) + + of "/health": + await req.respond(Http200, $ %* {"status": "ok", "version": "0.1.0"}, + newHttpHeaders([("Content-Type", "application/json")])) + + 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")])) + + else: + await req.respond(Http404, $jsonError(404, "Not found: " & req.url.path), + newHttpHeaders([("Content-Type", "application/json")])) + + dec server.metrics.activeConnections + +proc run*(server: HttpServer, port: int = 8080) {.async.} = + server.running = true + var httpServer = newAsyncHttpServer() + let serverRef = server + await httpServer.serve(Port(port), proc (req: Request): Future[void] {.gcsafe.} = + serverRef.handleRequest(req)) + +proc stop*(server: HttpServer) = + server.running = false + server.db.close() diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index 46d2610..e79585e 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -4,7 +4,6 @@ import std/asyncnet import std/strutils import std/os import std/endians -import std/tables import config import ../protocol/wire import ../query/lexer diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index e27096a..2bf930c 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -2,6 +2,7 @@ import std/strutils import std/tables import std/hashes +import std/sequtils import ast import ir import ../core/types @@ -46,6 +47,26 @@ proc cloneForConnection*(ctx: ExecutionContext): ExecutionContext = btrees: ctx.btrees, txnManager: ctx.txnManager, pendingTxn: nil) +proc getTableDef(ctx: ExecutionContext, tableName: string): TableDef = + if tableName in ctx.tables: + return ctx.tables[tableName] + var tbl = TableDef(name: tableName, columns: @[], pkColumns: @[]) + return tbl + +proc getColumnDef(tbl: TableDef, colName: string): ColumnDef = + for col in tbl.columns: + if col.name.toLower() == colName.toLower(): + return col + +proc getValue(values: seq[string], fields: seq[string], colName: string): string = + for i, f in fields: + if f.toLower() == colName.toLower() and i < values.len: + return values[i] + return "" + +proc isNull(value: string): bool = + result = value.len == 0 or value.toLower() == "null" + proc execScan(ctx: ExecutionContext, table: string): seq[Row] = ## Full table scan via LSM-Tree memtable scan. ## Rows are stored as: "{table}.{key}" -> value @@ -99,13 +120,13 @@ proc execInsert(ctx: ExecutionContext, table: string, fields: seq[string], value ctx.db.put(fullKey, cast[seq[byte]](valStr)) # Populate B-Tree indexes (always direct, not transactional for now) - for colName, btIdx in ctx.btrees: + for colName in ctx.btrees.keys.toSeq(): if colName.startsWith(table & "."): let colOnly = colName[table.len + 1..^1] let colVal = getValue(rowVals, fields, colOnly) if colVal.len > 0 and not isNull(colVal): let entry = IndexEntry(lsmKey: fullKey, rowValue: valStr) - btIdx.insert(colVal, entry) + ctx.btrees[colName].insert(colVal, entry) inc count return count @@ -140,26 +161,6 @@ proc execUpdateRow(ctx: ExecutionContext, table: string, key: string, sets: seq[ # Constraint Validation # ---------------------------------------------------------------------- -proc getTableDef(ctx: ExecutionContext, tableName: string): TableDef = - if tableName in ctx.tables: - return ctx.tables[tableName] - var tbl = TableDef(name: tableName, columns: @[], pkColumns: @[]) - return tbl - -proc getColumnDef(tbl: TableDef, colName: string): ColumnDef = - for col in tbl.columns: - if col.name.toLower() == colName.toLower(): - return col - -proc getValue(values: seq[string], fields: seq[string], colName: string): string = - for i, f in fields: - if f.toLower() == colName.toLower() and i < values.len: - return values[i] - return "" - -proc isNull(value: string): bool = - result = value.len == 0 or value.toLower() == "null" - proc validateConstraints*(ctx: ExecutionContext, tableName: string, fields: seq[string], values: seq[seq[string]]): (bool, string) = ## Validate INSERT/UPDATE constraints.