From aa4ab112104fe30e6beb6a8e0a934527c25802e8 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Thu, 18 Jun 2026 21:29:39 +0300 Subject: [PATCH] feat: canonical Nim client refactor with typed rows, pool, and allographer wrapper - Extract wire protocol into clients/nim/src/baradb/wire.nim - Add BaraError exception hierarchy - Refactor BaraClient/SyncClient with typedRows, AsyncLock request queue, timeouts, TLS config - Add BaraPool and optional HTTP fallback - Add mock-server wire and pool unit tests - Bump baradb nimble package to 1.2.0 - Make nim-allographer depend on canonical baradb client - Use typed rows in allographer toJson - Deprecate src/barabadb/client/client.nim - Update docs/en/clients.md and clients/nim/README.md --- clients/nim-allographer/allographer.nimble | 1 + .../libs/baradb/baradb_client.nim | 740 +------- .../models/baradb/baradb_exec.nim | 73 +- clients/nim/README.md | 59 +- clients/nim/baradb.nimble | 2 +- clients/nim/src/baradb/client.nim | 591 +++--- clients/nim/src/baradb/errors.nim | 10 + clients/nim/src/baradb/http.nim | 38 + clients/nim/src/baradb/pool.nim | 161 ++ clients/nim/src/baradb/wire.nim | 244 +++ clients/nim/tests/test_client.nim | 7 + clients/nim/tests/test_integration.nim | 153 +- clients/nim/tests/test_pool.nim | 19 + clients/nim/tests/test_wire.nim | 68 + docs/en/clients.md | 58 +- .../plans/2026-06-18-good-nim-client-plan.md | 1579 +++++++++++++++++ .../2026-06-18-good-nim-client-design.md | 274 +++ src/barabadb/client/client.nim | 1 + 18 files changed, 2813 insertions(+), 1265 deletions(-) create mode 100644 clients/nim/src/baradb/errors.nim create mode 100644 clients/nim/src/baradb/http.nim create mode 100644 clients/nim/src/baradb/pool.nim create mode 100644 clients/nim/src/baradb/wire.nim create mode 100644 clients/nim/tests/test_pool.nim create mode 100644 clients/nim/tests/test_wire.nim create mode 100644 docs/superpowers/plans/2026-06-18-good-nim-client-plan.md create mode 100644 docs/superpowers/specs/2026-06-18-good-nim-client-design.md diff --git a/clients/nim-allographer/allographer.nimble b/clients/nim-allographer/allographer.nimble index a817665..2b61188 100644 --- a/clients/nim-allographer/allographer.nimble +++ b/clients/nim-allographer/allographer.nimble @@ -16,6 +16,7 @@ srcDir = "src" requires "nim >= 2.0.0" requires "db_connector >= 0.1.0" requires "checksums >= 0.1.0" +requires "baradb >= 1.2.0" import strformat, os diff --git a/clients/nim-allographer/src/allographer/query_builder/libs/baradb/baradb_client.nim b/clients/nim-allographer/src/allographer/query_builder/libs/baradb/baradb_client.nim index 3c4048d..26c7140 100644 --- a/clients/nim-allographer/src/allographer/query_builder/libs/baradb/baradb_client.nim +++ b/clients/nim-allographer/src/allographer/query_builder/libs/baradb/baradb_client.nim @@ -1,420 +1,13 @@ -## BaraDB Client — Self-contained Nim client library -## No dependency on BaraDB server code. -## Communicates via the BaraDB Wire Protocol (binary, big-endian). - +## BaraDB driver glue for nim-allographer. +## All wire/socket logic lives in the canonical `baradb/client` package. import std/asyncdispatch -import std/asyncnet -import std/net as netmod -import std/locks -import std/strutils -import std/endians +import baradb/client +export client -# === Wire Protocol (self-contained, no server dependency) === - -const - ProtocolMagic* = 0x42415241'u32 - -type - FieldKind* = enum - fkNull = 0x00 - fkBool = 0x01 - fkInt8 = 0x02 - fkInt16 = 0x03 - fkInt32 = 0x04 - fkInt64 = 0x05 - fkFloat32 = 0x06 - fkFloat64 = 0x07 - fkString = 0x08 - fkBytes = 0x09 - fkArray = 0x0A - fkObject = 0x0B - fkVector = 0x0C - fkJson = 0x0D - - MsgKind* = enum - # Client messages - mkClientHandshake = 0x01 - mkQuery = 0x02 - mkQueryParams = 0x03 - mkExecute = 0x04 - mkBatch = 0x05 - mkTransaction = 0x06 - mkClose = 0x07 - mkPing = 0x08 - mkAuth = 0x09 - # Server messages - mkServerHandshake = 0x80 - mkReady = 0x81 - mkData = 0x82 - mkComplete = 0x83 - mkError = 0x84 - mkAuthChallenge = 0x85 - mkAuthOk = 0x86 - mkSchemaChange = 0x87 - mkPong = 0x88 - mkTransactionState = 0x89 - - ResultFormat* = enum - rfBinary = 0x00 - rfJson = 0x01 - rfText = 0x02 - - WireValue* = object - case kind*: FieldKind - of fkNull: discard - of fkBool: boolVal*: bool - of fkInt8: int8Val*: int8 - of fkInt16: int16Val*: int16 - of fkInt32: int32Val*: int32 - of fkInt64: int64Val*: int64 - of fkFloat32: float32Val*: float32 - of fkFloat64: float64Val*: float64 - of fkString: strVal*: string - of fkBytes: bytesVal*: seq[byte] - of fkArray: arrayVal*: seq[WireValue] - of fkObject: objVal*: seq[(string, WireValue)] - of fkVector: vecVal*: seq[float32] - of fkJson: jsonVal*: string - -proc writeUint32(buf: var seq[byte], val: uint32) = - var bytes: array[4, byte] - bigEndian32(addr bytes, unsafeAddr val) - buf.add(bytes) - -proc writeUint64(buf: var seq[byte], val: uint64) = - var bytes: array[8, byte] - bigEndian64(addr bytes, unsafeAddr val) - buf.add(bytes) - -proc writeString(buf: var seq[byte], s: string) = - buf.writeUint32(uint32(s.len)) - for ch in s: - buf.add(byte(ch)) - -proc readUint32(buf: openArray[byte], pos: var int): uint32 = - var bytes: array[4, byte] - for i in 0..3: bytes[i] = buf[pos + i] - bigEndian32(addr result, unsafeAddr bytes) - pos += 4 - -proc readUint64(buf: openArray[byte], pos: var int): uint64 = - var bytes: array[8, byte] - for i in 0..7: bytes[i] = buf[pos + i] - bigEndian64(addr result, unsafeAddr bytes) - pos += 8 - -proc readString(buf: openArray[byte], pos: var int): string = - let len = int(readUint32(buf, pos)) - result = newString(len) - for i in 0.." - of fkArray: return "" - of fkObject: return "" - of fkVector: return "" - of fkJson: return wv.jsonVal - -proc readQueryResponse*(client: BaraClient): Future[QueryResult] {.async.} = - let headerData = await client.socket.recv(12) - if headerData.len < 12: - raise newException(IOError, "Connection closed") - - var pos = 0 - let hdrData = toBytes(headerData) - let kind = MsgKind(readUint32(hdrData, pos)) - let payloadLen = int(readUint32(hdrData, pos)) - discard readUint32(hdrData, pos) - - let payloadStr = await client.socket.recv(payloadLen) - var payload = toBytes(payloadStr) - - result = QueryResult(columns: @[], rows: @[], rowCount: 0, affectedRows: 0) - - if kind == mkReady: - return - if kind == mkError and payload.len >= 8: - var epos = 0 - let code = readUint32(payload, epos) - let emsg = readString(payload, epos) - raise newException(IOError, "Error " & $code & ": " & emsg) - if kind == mkData: - var dpos = 0 - let colCount = int(readUint32(payload, dpos)) - var cols: seq[string] = @[] - for i in 0..= 12: - var chPos = 0 - let chData = toBytes(compHeader) - let compKind = MsgKind(readUint32(chData, chPos)) - let compLen = int(readUint32(chData, chPos)) - discard readUint32(chData, chPos) - let compPayloadStr = await client.socket.recv(compLen) - if compKind == mkComplete: - var cpPos = 0 - result.affectedRows = int(readUint32(toBytes(compPayloadStr), cpPos)) - return - if kind == mkComplete: - var rpos = 0 - result.affectedRows = int(readUint32(payload, rpos)) - return - -proc query*(client: BaraClient, sql: string): Future[QueryResult] {.async.} = - if not client.connected: - raise newException(IOError, "Not connected") - - let msg = makeQueryMessage(client.nextId(), sql) - let msgStr = toString(msg) - await client.socket.send(msgStr) - - return await client.readQueryResponse() - -proc query*(client: BaraClient, sql: string, params: seq[WireValue]): Future[QueryResult] {.async.} = - if not client.connected: - raise newException(IOError, "Not connected") - - let msg = makeQueryParamsMessage(client.nextId(), sql, params) - let msgStr = toString(msg) - await client.socket.send(msgStr) - - return await client.readQueryResponse() - -proc exec*(client: BaraClient, sql: string): Future[int] {.async.} = - let qr = await client.query(sql) - return qr.affectedRows - -# === Migration API (BaraQL native) === +# === Migration helpers (allographer-specific) === proc createMigration*(client: BaraClient, name: string, upBody: string, downBody: string = ""): Future[QueryResult] {.async.} = - ## Send CREATE MIGRATION via BaraQL. Server handles checksums, locking, rollback. var sql = "CREATE MIGRATION " & name & " { UP: " & upBody & ";" if downBody.len > 0: sql &= " DOWN: " & downBody & ";" @@ -438,326 +31,3 @@ proc migrationStatus*(client: BaraClient): Future[QueryResult] {.async.} = proc migrationDryRun*(client: BaraClient, name: string): Future[QueryResult] {.async.} = return await client.query("MIGRATION DRY RUN " & name) - -proc auth*(client: BaraClient, token: string) {.async.} = - if not client.connected: - raise newException(IOError, "Not connected") - - let msg = makeAuthMessage(client.nextId(), token) - let msgStr = toString(msg) - await client.socket.send(msgStr) - - let headerData = await client.socket.recv(12) - if headerData.len < 12: - raise newException(IOError, "Connection closed") - - var pos = 0 - let hdrData = toBytes(headerData) - let kind = MsgKind(readUint32(hdrData, pos)) - let payloadLen = int(readUint32(hdrData, pos)) - discard readUint32(hdrData, pos) - - if kind == mkAuthOk: - return - elif kind == mkError: - let payloadStr = await client.socket.recv(payloadLen) - var epos = 0 - let emsg = readString(toBytes(payloadStr), epos) - raise newException(IOError, "Auth failed: " & emsg) - else: - raise newException(IOError, "Unexpected auth response: 0x" & toHex(uint32(kind), 2)) - -proc ping*(client: BaraClient): Future[bool] {.async.} = - if not client.connected: - return false - let msg = buildMessage(mkPing, client.nextId(), @[]) - let msgStr = toString(msg) - await client.socket.send(msgStr) - - let headerData = await client.socket.recv(12) - if headerData.len < 12: - return false - - var pos = 0 - let hdrData = toBytes(headerData) - let kind = MsgKind(readUint32(hdrData, pos)) - return kind == mkPong - -# === Fluent Query Builder === - -type - QueryBuilder* = ref object - client: BaraClient - selectCols: seq[string] - fromTable: string - whereClauses: seq[string] - joinClauses: seq[string] - groupByCols: seq[string] - havingClause: string - orderCols: seq[string] - orderDirs: seq[string] - limitVal: int - offsetVal: int - -proc newQueryBuilder*(client: BaraClient): QueryBuilder = - QueryBuilder(client: client, limitVal: 0, offsetVal: 0) - -proc select*(qb: QueryBuilder, cols: varargs[string]): QueryBuilder = - for c in cols: qb.selectCols.add(c) - return qb - -proc `from`*(qb: QueryBuilder, table: string): QueryBuilder = - qb.fromTable = table - return qb - -proc where*(qb: QueryBuilder, clause: string): QueryBuilder = - qb.whereClauses.add(clause) - return qb - -proc join*(qb: QueryBuilder, table: string, on: string): QueryBuilder = - qb.joinClauses.add("JOIN " & table & " ON " & on) - return qb - -proc leftJoin*(qb: QueryBuilder, table: string, on: string): QueryBuilder = - qb.joinClauses.add("LEFT JOIN " & table & " ON " & on) - return qb - -proc groupBy*(qb: QueryBuilder, cols: varargs[string]): QueryBuilder = - for c in cols: qb.groupByCols.add(c) - return qb - -proc having*(qb: QueryBuilder, clause: string): QueryBuilder = - qb.havingClause = clause - return qb - -proc orderBy*(qb: QueryBuilder, col: string, dir: string = "ASC"): QueryBuilder = - qb.orderCols.add(col) - qb.orderDirs.add(dir) - return qb - -proc limit*(qb: QueryBuilder, n: int): QueryBuilder = - qb.limitVal = n - return qb - -proc offset*(qb: QueryBuilder, n: int): QueryBuilder = - qb.offsetVal = n - return qb - -proc build*(qb: QueryBuilder): string = - result = "SELECT " & (if qb.selectCols.len > 0: qb.selectCols.join(", ") else: "*") - result &= " FROM " & qb.fromTable - for j in qb.joinClauses: result &= " " & j - if qb.whereClauses.len > 0: result &= " WHERE " & qb.whereClauses.join(" AND ") - if qb.groupByCols.len > 0: result &= " GROUP BY " & qb.groupByCols.join(", ") - if qb.havingClause.len > 0: result &= " HAVING " & qb.havingClause - if qb.orderCols.len > 0: - result &= " ORDER BY " - for i, col in qb.orderCols: - if i > 0: result &= ", " - result &= col & " " & qb.orderDirs[i] - if qb.limitVal > 0: result &= " LIMIT " & $qb.limitVal - if qb.offsetVal > 0: result &= " OFFSET " & $qb.offsetVal - -proc exec*(qb: QueryBuilder): Future[QueryResult] {.async.} = - return await qb.client.query(qb.build()) - -# === Blocking Sync Client (production-grade, no waitFor) === - -type - SyncClient* = ref object - config: ClientConfig - socket: netmod.Socket - connected: bool - requestId: uint32 - lock: Lock - -proc newSyncClient*(config: ClientConfig = defaultConfig()): SyncClient = - result = SyncClient(config: config, connected: false, requestId: 0) - result.socket = netmod.newSocket() - initLock(result.lock) - -proc recvExact(sock: netmod.Socket, size: int): string = - result = "" - while result.len < size: - let chunk = sock.recv(size - result.len) - if chunk.len == 0: - raise newException(IOError, "Connection closed") - result.add(chunk) - -proc readQueryResponseBlocking(client: SyncClient): QueryResult = - let headerData = client.socket.recvExact(12) - var pos = 0 - let hdrData = toBytes(headerData) - let kind = MsgKind(readUint32(hdrData, pos)) - let payloadLen = int(readUint32(hdrData, pos)) - discard readUint32(hdrData, pos) - - let payloadStr = client.socket.recvExact(payloadLen) - var payload = toBytes(payloadStr) - - result = QueryResult(columns: @[], rows: @[], rowCount: 0, affectedRows: 0) - - if kind == mkReady: - return - if kind == mkError and payload.len >= 8: - var epos = 0 - let code = readUint32(payload, epos) - let emsg = readString(payload, epos) - raise newException(IOError, "Error " & $code & ": " & emsg) - if kind == mkData: - var dpos = 0 - let colCount = int(readUint32(payload, dpos)) - var cols: seq[string] = @[] - for i in 0.. 0: - sql &= " DOWN: " & downBody & ";" - sql &= " }" - return client.query(sql) - -proc applyMigration*(client: SyncClient, name: string): QueryResult = - return client.query("APPLY MIGRATION " & name) - -proc migrateUp*(client: SyncClient, count: int = 0): QueryResult = - var sql = "MIGRATION UP" - if count > 0: - sql &= " " & $count - return client.query(sql) - -proc migrateDown*(client: SyncClient, count: int = 1): QueryResult = - return client.query("MIGRATION DOWN " & $count) - -proc migrationStatus*(client: SyncClient): QueryResult = - return client.query("MIGRATION STATUS") - -proc migrationDryRun*(client: SyncClient, name: string): QueryResult = - return client.query("MIGRATION DRY RUN " & name) - -proc auth*(client: SyncClient, token: string) = - acquire(client.lock) - try: - if not client.connected: - raise newException(IOError, "Not connected") - let msg = makeAuthMessage(0, token) - netmod.send(client.socket, toString(msg)) - let headerData = client.socket.recvExact(12) - var pos = 0 - let hdrData = toBytes(headerData) - let kind = MsgKind(readUint32(hdrData, pos)) - let payloadLen = int(readUint32(hdrData, pos)) - discard readUint32(hdrData, pos) - if kind == mkAuthOk: - return - elif kind == mkError: - let payloadStr = client.socket.recvExact(payloadLen) - var epos = 0 - let emsg = readString(toBytes(payloadStr), epos) - raise newException(IOError, "Auth failed: " & emsg) - else: - raise newException(IOError, "Unexpected auth response") - finally: - release(client.lock) - -proc ping*(client: SyncClient): bool = - acquire(client.lock) - try: - if not client.connected: - return false - let msg = buildMessage(mkPing, 0, @[]) - netmod.send(client.socket, toString(msg)) - let headerData = client.socket.recvExact(12) - var pos = 0 - let hdrData = toBytes(headerData) - let kind = MsgKind(readUint32(hdrData, pos)) - return kind == mkPong - except: - return false - finally: - release(client.lock) - -proc `$`*(qr: QueryResult): string = - if qr.columns.len == 0: return "(no results)" - result = "" - for i, col in qr.columns: - result &= col - if i < qr.columns.len - 1: result &= ", " - result &= "\n" - for row in qr.rows: - result &= row.join(", ") & "\n" - result &= "(" & $qr.rowCount & " rows)" diff --git a/clients/nim-allographer/src/allographer/query_builder/models/baradb/baradb_exec.nim b/clients/nim-allographer/src/allographer/query_builder/models/baradb/baradb_exec.nim index f0d3f08..88fad28 100644 --- a/clients/nim-allographer/src/allographer/query_builder/models/baradb/baradb_exec.nim +++ b/clients/nim-allographer/src/allographer/query_builder/models/baradb/baradb_exec.nim @@ -213,40 +213,53 @@ proc placeholdersToWireValuesRaw*(args: seq[JsonNode]): seq[WireValue] = # toJson # ================================================================================ +proc wireValueToJson*(wv: WireValue): JsonNode = + case wv.kind + of fkNull: + result = newJNull() + of fkBool: + result = newJBool(wv.boolVal) + of fkInt8: + result = newJInt(int(wv.int8Val)) + of fkInt16: + result = newJInt(int(wv.int16Val)) + of fkInt32: + result = newJInt(int(wv.int32Val)) + of fkInt64: + result = newJInt(int(wv.int64Val)) + of fkFloat32: + result = newJFloat(float(wv.float32Val)) + of fkFloat64: + result = newJFloat(wv.float64Val) + of fkString: + result = newJString(wv.strVal) + of fkBytes: + result = newJString("") + of fkArray: + result = newJArray() + for item in wv.arrayVal: + result.add(wireValueToJson(item)) + of fkObject: + result = newJObject() + for (name, val) in wv.objVal: + result[name] = wireValueToJson(val) + of fkVector: + result = newJArray() + for f in wv.vecVal: + result.add(newJFloat(float(f))) + of fkJson: + try: + result = parseJson(wv.jsonVal) + except JsonParsingError: + result = newJString(wv.jsonVal) + proc toJson*(resultSet: QueryResult): seq[JsonNode] = var response_table = newSeq[JsonNode](resultSet.rowCount) for r in 0 ..< resultSet.rowCount: var response_row = newJObject() for c in 0 ..< resultSet.columns.len: let key = resultSet.columns[c] - let val = resultSet.rows[r][c] - let colType = if c < resultSet.columnTypes.len: resultSet.columnTypes[c] else: "fkString" - if val.len == 0: - response_row[key] = newJNull() - else: - case colType - of "fkNull": - response_row[key] = newJNull() - of "fkBool": - response_row[key] = newJBool(val == "t" or val == "true" or val == "1") - of "fkInt8", "fkInt16", "fkInt32", "fkInt64": - try: - response_row[key] = newJInt(val.parseInt) - except ValueError: - response_row[key] = newJString(val) - of "fkFloat32", "fkFloat64": - try: - response_row[key] = newJFloat(val.parseFloat) - except ValueError: - response_row[key] = newJString(val) - of "fkJson": - try: - response_row[key] = parseJson(val) - except JsonParsingError: - response_row[key] = newJString(val) - else: - # fkString, fkBytes, fkArray, fkObject, fkVector, and unknown types - response_row[key] = newJString(val) + response_row[key] = wireValueToJson(resultSet.typedRows[r][c]) response_table[r] = response_row return response_table @@ -832,7 +845,9 @@ proc first*(self: RawBaradbQuery): Future[Option[JsonNode]] {.async.} = proc firstPlain*(self: RawBaradbQuery): Future[seq[string]] {.async.} = self.log.logger(self.queryString) - return await self.getRowPlain(self.queryString, self.placeHolder) + let row = await self.getRowPlain(self.queryString, self.placeHolder) + if row.isSome: return row.get() + return @[] # ================================================================================ diff --git a/clients/nim/README.md b/clients/nim/README.md index 9028702..279073d 100644 --- a/clients/nim/README.md +++ b/clients/nim/README.md @@ -15,7 +15,7 @@ Official Nim client for **BaraDB** — a multimodal database engine. Add to your `.nimble` file: ```nim -requires "baradb >= 1.1.6" +requires "baradb >= 1.2.0" ``` Or clone locally: @@ -95,6 +95,63 @@ proc main() {.async.} = waitFor main() ``` +## Connection Pool + +```nim +import asyncdispatch, baradb/client, baradb/pool + +proc main() {.async.} = + let cfg = ClientConfig(host: "127.0.0.1", port: 9472) + let pool = newBaraPool(cfg, minConnections = 2, maxConnections = 10) + withClient(pool): + let r = await c.query("SELECT name FROM users WHERE id = ?", + @[WireValue(kind: fkInt64, int64Val: 1)]) + echo r.typedRows + +waitFor main() +``` + +## Typed Rows + +`QueryResult` now carries both a legacy string view (`rows`) and a typed view (`typedRows`): + +```nim +let r = await client.query("SELECT * FROM vectors") +for row in r.typedRows: + if row[0].kind == fkVector: + echo row[0].vecVal +``` + +## TLS + +TLS for the synchronous client is available via `when defined(ssl)`. The async binary client requires a user-supplied `sslContext` because `asyncnet` does not provide native TLS; alternatively use the HTTP fallback. + +## Error Handling + +All client errors inherit from `BaraError`: + +- `BaraIoError` — connection / timeout issues +- `BaraServerError` — server returned an error frame +- `BaraAuthError` — authentication failure +- `BaraProtocolError` — unexpected wire response +- `BaraPoolTimeoutError` — no connection available in time + +## HTTP Fallback + +For environments where only the HTTP port is open: + +```nim +import asyncdispatch, baradb/http + +proc main() {.async.} = + let c = newBaraHttpClient() + let result = await c.query("SELECT * FROM users") + echo result + c.close() + +waitFor main() +``` + ## Running Tests Unit tests (no server): diff --git a/clients/nim/baradb.nimble b/clients/nim/baradb.nimble index 142e748..4610ef2 100644 --- a/clients/nim/baradb.nimble +++ b/clients/nim/baradb.nimble @@ -1,6 +1,6 @@ # Package -version = "1.1.6" +version = "1.2.0" author = "BaraDB Team" description = "Official Nim client for BaraDB — async binary protocol client" license = "Apache-2.0" diff --git a/clients/nim/src/baradb/client.nim b/clients/nim/src/baradb/client.nim index 6c695e3..9a4d8ca 100644 --- a/clients/nim/src/baradb/client.nim +++ b/clients/nim/src/baradb/client.nim @@ -1,261 +1,48 @@ -## BaraDB Client — Self-contained Nim client library -## No dependency on BaraDB server code. -## Communicates via the BaraDB Wire Protocol (binary, big-endian). - +## BaraDB Client — canonical Nim client library. +## Self-contained; depends only on Nim stdlib. import std/asyncdispatch import std/asyncnet import std/net as netmod import std/locks import std/strutils -import std/endians -# === Wire Protocol (self-contained, no server dependency) === +import ./wire +export wire +import ./errors +export errors -const - ProtocolMagic* = 0x42415241'u32 +# === AsyncLock (stdlib-only serialization primitive) === type - FieldKind* = enum - fkNull = 0x00 - fkBool = 0x01 - fkInt8 = 0x02 - fkInt16 = 0x03 - fkInt32 = 0x04 - fkInt64 = 0x05 - fkFloat32 = 0x06 - fkFloat64 = 0x07 - fkString = 0x08 - fkBytes = 0x09 - fkArray = 0x0A - fkObject = 0x0B - fkVector = 0x0C - fkJson = 0x0D + AsyncLockObj = object + locked: bool + waiters: seq[Future[void]] - MsgKind* = enum - # Client messages - mkClientHandshake = 0x01 - mkQuery = 0x02 - mkQueryParams = 0x03 - mkExecute = 0x04 - mkBatch = 0x05 - mkTransaction = 0x06 - mkClose = 0x07 - mkPing = 0x08 - mkAuth = 0x09 - # Server messages - mkServerHandshake = 0x80 - mkReady = 0x81 - mkData = 0x82 - mkComplete = 0x83 - mkError = 0x84 - mkAuthChallenge = 0x85 - mkAuthOk = 0x86 - mkSchemaChange = 0x87 - mkPong = 0x88 - mkTransactionState = 0x89 + AsyncLock* = ref AsyncLockObj - ResultFormat* = enum - rfBinary = 0x00 - rfJson = 0x01 - rfText = 0x02 +proc initAsyncLock*(): AsyncLock = + new(result) + result.locked = false + result.waiters = @[] - WireValue* = object - case kind*: FieldKind - of fkNull: discard - of fkBool: boolVal*: bool - of fkInt8: int8Val*: int8 - of fkInt16: int16Val*: int16 - of fkInt32: int32Val*: int32 - of fkInt64: int64Val*: int64 - of fkFloat32: float32Val*: float32 - of fkFloat64: float64Val*: float64 - of fkString: strVal*: string - of fkBytes: bytesVal*: seq[byte] - of fkArray: arrayVal*: seq[WireValue] - of fkObject: objVal*: seq[(string, WireValue)] - of fkVector: vecVal*: seq[float32] - of fkJson: jsonVal*: string +proc acquire*(lock: AsyncLock): Future[void] = + var fut = newFuture[void]("AsyncLock.acquire") + if not lock.locked: + lock.locked = true + fut.complete() + else: + lock.waiters.add(fut) + return fut -proc writeUint32(buf: var seq[byte], val: uint32) = - var bytes: array[4, byte] - bigEndian32(addr bytes, unsafeAddr val) - buf.add(bytes) +proc release*(lock: AsyncLock) = + if lock.waiters.len > 0: + let next = lock.waiters[0] + lock.waiters.delete(0) + next.complete() + else: + lock.locked = false -proc writeUint64(buf: var seq[byte], val: uint64) = - var bytes: array[8, byte] - bigEndian64(addr bytes, unsafeAddr val) - buf.add(bytes) - -proc writeString(buf: var seq[byte], s: string) = - buf.writeUint32(uint32(s.len)) - for ch in s: - buf.add(byte(ch)) - -proc readUint32(buf: openArray[byte], pos: var int): uint32 = - var bytes: array[4, byte] - for i in 0..3: bytes[i] = buf[pos + i] - bigEndian32(addr result, unsafeAddr bytes) - pos += 4 - -proc readUint64(buf: openArray[byte], pos: var int): uint64 = - var bytes: array[8, byte] - for i in 0..7: bytes[i] = buf[pos + i] - bigEndian64(addr result, unsafeAddr bytes) - pos += 8 - -proc readString(buf: openArray[byte], pos: var int): string = - let len = int(readUint32(buf, pos)) - result = newString(len) - for i in 0.." of fkJson: return wv.jsonVal -proc readQueryResponse(client: BaraClient): Future[QueryResult] {.async.} = - let headerData = await client.socket.recv(12) - if headerData.len < 12: - raise newException(IOError, "Connection closed") - +proc readResponsePayload(client: BaraClient): Future[(MsgKind, seq[byte])] {.async.} = + let headerStr = await recvExact(client.socket, 12, client.config.timeoutMs) var pos = 0 - let hdrData = toBytes(headerData) + let hdrData = toBytes(headerStr) let kind = MsgKind(readUint32(hdrData, pos)) let payloadLen = int(readUint32(hdrData, pos)) discard readUint32(hdrData, pos) + let payloadStr = await recvExact(client.socket, payloadLen, client.config.timeoutMs) + return (kind, toBytes(payloadStr)) - let payloadStr = await client.socket.recv(payloadLen) - var payload = toBytes(payloadStr) - - result = QueryResult(columns: @[], rows: @[], rowCount: 0, affectedRows: 0) - +proc parseQueryResponse(client: BaraClient, kind: MsgKind, payload: seq[byte]): Future[QueryResult] {.async.} = + result = QueryResult(columns: @[], rows: @[], typedRows: @[], rowCount: 0, affectedRows: 0) if kind == mkReady: return if kind == mkError and payload.len >= 8: var epos = 0 let code = readUint32(payload, epos) let emsg = readString(payload, epos) - raise newException(IOError, "Error " & $code & ": " & emsg) + var err = newException(BaraServerError, "Error " & $code & ": " & emsg) + err.code = code + raise err if kind == mkData: var dpos = 0 let colCount = int(readUint32(payload, dpos)) - var cols: seq[string] = @[] for i in 0..= 12: - var chPos = 0 - let chData = toBytes(compHeader) - let compKind = MsgKind(readUint32(chData, chPos)) - let compLen = int(readUint32(chData, chPos)) - discard readUint32(chData, chPos) - let compPayloadStr = await client.socket.recv(compLen) - if compKind == mkComplete: - var cpPos = 0 - result.affectedRows = int(readUint32(toBytes(compPayloadStr), cpPos)) + let (compKind, compPayload) = await client.readResponsePayload() + if compKind == mkComplete and compPayload.len >= 4: + var cpPos = 0 + result.affectedRows = int(readUint32(compPayload, cpPos)) return if kind == mkComplete: var rpos = 0 result.affectedRows = int(readUint32(payload, rpos)) return + raise newException(BaraProtocolError, "Unexpected response kind: 0x" & toHex(uint32(kind), 2)) + +proc doQuery(client: BaraClient, msg: seq[byte]): Future[QueryResult] {.async.} = + if not client.connected: + raise newException(BaraIoError, "Not connected") + await client.sendLock.acquire() + try: + await client.socket.send(toString(msg)) + let (kind, payload) = await client.readResponsePayload() + return await client.parseQueryResponse(kind, payload) + finally: + client.sendLock.release() proc query*(client: BaraClient, sql: string): Future[QueryResult] {.async.} = - if not client.connected: - raise newException(IOError, "Not connected") - let msg = makeQueryMessage(client.nextId(), sql) - let msgStr = toString(msg) - await client.socket.send(msgStr) - - return await client.readQueryResponse() + return await client.doQuery(msg) proc query*(client: BaraClient, sql: string, params: seq[WireValue]): Future[QueryResult] {.async.} = - if not client.connected: - raise newException(IOError, "Not connected") - let msg = makeQueryParamsMessage(client.nextId(), sql, params) - let msgStr = toString(msg) - await client.socket.send(msgStr) - - return await client.readQueryResponse() + return await client.doQuery(msg) proc exec*(client: BaraClient, sql: string): Future[int] {.async.} = let qr = await client.query(sql) return qr.affectedRows proc auth*(client: BaraClient, token: string) {.async.} = - if not client.connected: - raise newException(IOError, "Not connected") - let msg = makeAuthMessage(client.nextId(), token) - let msgStr = toString(msg) - await client.socket.send(msgStr) - - let headerData = await client.socket.recv(12) - if headerData.len < 12: - raise newException(IOError, "Connection closed") - - var pos = 0 - let hdrData = toBytes(headerData) - let kind = MsgKind(readUint32(hdrData, pos)) - let payloadLen = int(readUint32(hdrData, pos)) - discard readUint32(hdrData, pos) - - if kind == mkAuthOk: - return - elif kind == mkError: - let payloadStr = await client.socket.recv(payloadLen) - var epos = 0 - let emsg = readString(toBytes(payloadStr), epos) - raise newException(IOError, "Auth failed: " & emsg) - else: - raise newException(IOError, "Unexpected auth response: 0x" & toHex(uint32(kind), 2)) + await client.sendLock.acquire() + try: + await client.socket.send(toString(msg)) + let (kind, payload) = await client.readResponsePayload() + case kind + of mkAuthOk: + return + of mkError: + var epos = 0 + discard readUint32(payload, epos) + let emsg = readString(payload, epos) + raise newException(BaraAuthError, "Auth failed: " & emsg) + else: + raise newException(BaraProtocolError, "Unexpected auth response: 0x" & toHex(uint32(kind), 2)) + finally: + client.sendLock.release() proc ping*(client: BaraClient): Future[bool] {.async.} = if not client.connected: return false let msg = buildMessage(mkPing, client.nextId(), @[]) - let msgStr = toString(msg) - await client.socket.send(msgStr) - - let headerData = await client.socket.recv(12) - if headerData.len < 12: + await client.sendLock.acquire() + try: + await client.socket.send(toString(msg)) + let (kind, _) = await client.readResponsePayload() + return kind == mkPong + except: return false + finally: + client.sendLock.release() - var pos = 0 - let hdrData = toBytes(headerData) - let kind = MsgKind(readUint32(hdrData, pos)) - return kind == mkPong +proc readQueryResponse*(client: BaraClient): Future[QueryResult] {.async.} = + ## Read and parse the next server response. Does NOT acquire sendLock; + ## callers that already sent a message manually can use this. + let (kind, payload) = await client.readResponsePayload() + return await client.parseQueryResponse(kind, payload) # === Fluent Query Builder === @@ -532,7 +356,7 @@ proc build*(qb: QueryBuilder): string = proc exec*(qb: QueryBuilder): Future[QueryResult] {.async.} = return await qb.client.query(qb.build()) -# === Blocking Sync Client (production-grade, no waitFor) === +# === Blocking Sync Client === type SyncClient* = ref object @@ -547,70 +371,64 @@ proc newSyncClient*(config: ClientConfig = defaultConfig()): SyncClient = result.socket = netmod.newSocket() initLock(result.lock) -proc recvExact(sock: netmod.Socket, size: int): string = +proc recvExactBlocking(sock: netmod.Socket, size: int): string = result = "" while result.len < size: let chunk = sock.recv(size - result.len) if chunk.len == 0: - raise newException(IOError, "Connection closed") + raise newException(BaraIoError, "Connection closed") result.add(chunk) -proc readQueryResponseBlocking(client: SyncClient): QueryResult = - let headerData = client.socket.recvExact(12) +proc readResponsePayloadBlocking(client: SyncClient): (MsgKind, seq[byte]) = + let headerData = client.socket.recvExactBlocking(12) var pos = 0 let hdrData = toBytes(headerData) let kind = MsgKind(readUint32(hdrData, pos)) let payloadLen = int(readUint32(hdrData, pos)) discard readUint32(hdrData, pos) + let payloadStr = client.socket.recvExactBlocking(payloadLen) + return (kind, toBytes(payloadStr)) - let payloadStr = client.socket.recvExact(payloadLen) - var payload = toBytes(payloadStr) - - result = QueryResult(columns: @[], rows: @[], rowCount: 0, affectedRows: 0) - +proc parseQueryResponseBlocking(client: SyncClient, kind: MsgKind, payload: seq[byte]): QueryResult = + result = QueryResult(columns: @[], rows: @[], typedRows: @[], rowCount: 0, affectedRows: 0) if kind == mkReady: return if kind == mkError and payload.len >= 8: var epos = 0 let code = readUint32(payload, epos) let emsg = readString(payload, epos) - raise newException(IOError, "Error " & $code & ": " & emsg) + var err = newException(BaraServerError, "Error " & $code & ": " & emsg) + err.code = code + raise err if kind == mkData: var dpos = 0 let colCount = int(readUint32(payload, dpos)) - var cols: seq[string] = @[] for i in 0..= 4: var cpPos = 0 - result.affectedRows = int(readUint32(toBytes(compPayloadStr), cpPos)) + result.affectedRows = int(readUint32(compPayload, cpPos)) return if kind == mkComplete: var rpos = 0 result.affectedRows = int(readUint32(payload, rpos)) return + raise newException(BaraProtocolError, "Unexpected response kind: 0x" & toHex(uint32(kind), 2)) proc connect*(client: SyncClient) = netmod.connect(client.socket, client.config.host, Port(client.config.port)) @@ -630,10 +448,11 @@ proc query*(client: SyncClient, sql: string): QueryResult = acquire(client.lock) try: if not client.connected: - raise newException(IOError, "Not connected") + raise newException(BaraIoError, "Not connected") let msg = makeQueryMessage(0, sql) netmod.send(client.socket, toString(msg)) - return readQueryResponseBlocking(client) + let (kind, payload) = client.readResponsePayloadBlocking() + return client.parseQueryResponseBlocking(kind, payload) finally: release(client.lock) @@ -641,10 +460,11 @@ proc query*(client: SyncClient, sql: string, params: seq[WireValue]): QueryResul acquire(client.lock) try: if not client.connected: - raise newException(IOError, "Not connected") + raise newException(BaraIoError, "Not connected") let msg = makeQueryParamsMessage(0, sql, params) netmod.send(client.socket, toString(msg)) - return readQueryResponseBlocking(client) + let (kind, payload) = client.readResponsePayloadBlocking() + return client.parseQueryResponseBlocking(kind, payload) finally: release(client.lock) @@ -656,24 +476,20 @@ proc auth*(client: SyncClient, token: string) = acquire(client.lock) try: if not client.connected: - raise newException(IOError, "Not connected") + raise newException(BaraIoError, "Not connected") let msg = makeAuthMessage(0, token) netmod.send(client.socket, toString(msg)) - let headerData = client.socket.recvExact(12) - var pos = 0 - let hdrData = toBytes(headerData) - let kind = MsgKind(readUint32(hdrData, pos)) - let payloadLen = int(readUint32(hdrData, pos)) - discard readUint32(hdrData, pos) - if kind == mkAuthOk: + let (kind, payload) = client.readResponsePayloadBlocking() + case kind + of mkAuthOk: return - elif kind == mkError: - let payloadStr = client.socket.recvExact(payloadLen) + of mkError: var epos = 0 - let emsg = readString(toBytes(payloadStr), epos) - raise newException(IOError, "Auth failed: " & emsg) + discard readUint32(payload, epos) + let emsg = readString(payload, epos) + raise newException(BaraAuthError, "Auth failed: " & emsg) else: - raise newException(IOError, "Unexpected auth response") + raise newException(BaraProtocolError, "Unexpected auth response") finally: release(client.lock) @@ -684,10 +500,7 @@ proc ping*(client: SyncClient): bool = return false let msg = buildMessage(mkPing, 0, @[]) netmod.send(client.socket, toString(msg)) - let headerData = client.socket.recvExact(12) - var pos = 0 - let hdrData = toBytes(headerData) - let kind = MsgKind(readUint32(hdrData, pos)) + let (kind, _) = client.readResponsePayloadBlocking() return kind == mkPong except: return false diff --git a/clients/nim/src/baradb/errors.nim b/clients/nim/src/baradb/errors.nim new file mode 100644 index 0000000..56b4236 --- /dev/null +++ b/clients/nim/src/baradb/errors.nim @@ -0,0 +1,10 @@ +## BaraDB client exception hierarchy + +type + BaraError* = object of CatchableError + BaraProtocolError* = object of BaraError + BaraServerError* = object of BaraError + code*: uint32 + BaraAuthError* = object of BaraError + BaraIoError* = object of BaraError + BaraPoolTimeoutError* = object of BaraError diff --git a/clients/nim/src/baradb/http.nim b/clients/nim/src/baradb/http.nim new file mode 100644 index 0000000..6295cb4 --- /dev/null +++ b/clients/nim/src/baradb/http.nim @@ -0,0 +1,38 @@ +## Optional HTTP/REST fallback client for BaraDB. +import std/asyncdispatch +import std/httpclient +import std/json +import std/strformat +import ./errors + +type + BaraHttpClient* = ref object + baseUrl*: string + token*: string + http: AsyncHttpClient + +proc newBaraHttpClient*(host = "127.0.0.1", port = 9912, token = ""): BaraHttpClient = + BaraHttpClient( + baseUrl: fmt"http://{host}:{port}/api", + token: token, + http: newAsyncHttpClient(), + ) + +proc close*(client: BaraHttpClient) = + client.http.close() + +proc query*(client: BaraHttpClient, sql: string): Future[JsonNode] {.async.} = + var headers = newHttpHeaders({"Content-Type": "application/json"}) + if client.token.len > 0: + headers["Authorization"] = "Bearer " & client.token + let body = %*{ "query": sql } + let response = await client.http.request( + client.baseUrl & "/query", + httpMethod = HttpPost, + body = $body, + headers = headers, + ) + let text = await response.body + if response.code.int != 200: + raise newException(BaraServerError, "HTTP error " & $response.code.int & ": " & text) + return parseJson(text) diff --git a/clients/nim/src/baradb/pool.nim b/clients/nim/src/baradb/pool.nim new file mode 100644 index 0000000..322dcf3 --- /dev/null +++ b/clients/nim/src/baradb/pool.nim @@ -0,0 +1,161 @@ +## Async connection pool for BaraDB. +import std/asyncdispatch +import std/deques +import std/monotimes +import std/times +import ./client +import ./errors + +type + PoolConnection = ref object + client: BaraClient + inUse: bool + createdAt: int64 + lastUsedAt: int64 + + PoolConfig* = object + minConnections*: int + maxConnections*: int + maxIdleTimeMs*: int + maxLifetimeMs*: int + + BaraPool* = ref object + clientConfig: ClientConfig + poolConfig: PoolConfig + connections: seq[PoolConnection] + waiters: Deque[Future[void]] + lock: AsyncLock + +proc defaultPoolConfig*(): PoolConfig = + PoolConfig( + minConnections: 2, + maxConnections: 10, + maxIdleTimeMs: 300_000, + maxLifetimeMs: 3_600_000, + ) + +proc nowUnix(): int64 = getTime().toUnix() + +proc newBaraPool*(clientConfig: ClientConfig, + minConnections = 2, + maxConnections = 10, + poolConfig = defaultPoolConfig()): BaraPool = + result = BaraPool( + clientConfig: clientConfig, + poolConfig: poolConfig, + connections: @[], + waiters: initDeque[Future[void]](), + lock: initAsyncLock(), + ) + result.poolConfig.minConnections = minConnections + result.poolConfig.maxConnections = maxConnections + +proc isExpired(cfg: PoolConfig, conn: PoolConnection): bool = + let now = nowUnix() + if cfg.maxLifetimeMs > 0 and (now - conn.createdAt) * 1000 >= cfg.maxLifetimeMs: + return true + if cfg.maxIdleTimeMs > 0 and conn.lastUsedAt > 0 and (now - conn.lastUsedAt) * 1000 >= cfg.maxIdleTimeMs: + return true + return false + +proc openConnection(pool: BaraPool): Future[BaraClient] {.async.} = + let client = newClient(pool.clientConfig) + try: + await client.connect() + except BaraError: + raise + except CatchableError as e: + raise newException(BaraIoError, "Failed to open connection: " & e.msg) + return client + +proc closeConnection(conn: PoolConnection) = + if not conn.client.isNil: + conn.client.close() + +proc wakeOneWaiter(pool: BaraPool) = + while pool.waiters.len > 0: + let w = pool.waiters.popFirst() + if not w.finished: + w.complete() + break + +proc acquireConnection(pool: BaraPool): Future[BaraClient] {.async.} = + let deadline = getMonoTime() + initDuration(milliseconds = pool.clientConfig.timeoutMs) + while true: + await pool.lock.acquire() + # Reuse idle, non-expired connection + var i = 0 + while i < pool.connections.len: + let conn = pool.connections[i] + if not conn.inUse: + if pool.poolConfig.isExpired(conn): + pool.connections.del(i) + pool.lock.release() + closeConnection(conn) + await pool.lock.acquire() + continue + conn.inUse = true + conn.lastUsedAt = nowUnix() + pool.lock.release() + return conn.client + inc i + # Create new if under max + if pool.connections.len < pool.poolConfig.maxConnections: + pool.lock.release() + let client = await pool.openConnection() + await pool.lock.acquire() + let conn = PoolConnection( + client: client, + inUse: true, + createdAt: nowUnix(), + lastUsedAt: nowUnix(), + ) + pool.connections.add(conn) + pool.lock.release() + return client + pool.lock.release() + # Wait for a connection to be released + if getMonoTime() >= deadline: + raise newException(BaraPoolTimeoutError, "Timed out waiting for a free connection") + let w = newFuture[void]("pool.wait") + await pool.lock.acquire() + pool.waiters.addLast(w) + pool.lock.release() + let ok = await withTimeout(w, pool.clientConfig.timeoutMs) + if not ok: + await pool.lock.acquire() + var kept = initDeque[Future[void]]() + while pool.waiters.len > 0: + let x = pool.waiters.popFirst() + if x != w: + kept.addLast(x) + pool.waiters = move(kept) + pool.lock.release() + raise newException(BaraPoolTimeoutError, "Timed out waiting for a free connection") + +proc releaseConnection(pool: BaraPool, client: BaraClient) {.async.} = + await pool.lock.acquire() + for conn in pool.connections: + if conn.client == client: + conn.inUse = false + conn.lastUsedAt = nowUnix() + break + pool.lock.release() + wakeOneWaiter(pool) + +template withClient*(pool: BaraPool, body: untyped): untyped = + let c = await pool.acquireConnection() + try: + body + finally: + await pool.releaseConnection(c) + +proc stats*(pool: BaraPool): Future[(int, int, int)] {.async.} = + await pool.lock.acquire() + let total = pool.connections.len + var inUse = 0 + for c in pool.connections: + if c.inUse: + inc inUse + pool.lock.release() + return (total, total - inUse, inUse) diff --git a/clients/nim/src/baradb/wire.nim b/clients/nim/src/baradb/wire.nim new file mode 100644 index 0000000..cdc59b6 --- /dev/null +++ b/clients/nim/src/baradb/wire.nim @@ -0,0 +1,244 @@ +## BaraDB binary wire protocol — shared between client and server. +import std/endians + +const + ProtocolMagic* = 0x42415241'u32 + +type + FieldKind* = enum + fkNull = 0x00 + fkBool = 0x01 + fkInt8 = 0x02 + fkInt16 = 0x03 + fkInt32 = 0x04 + fkInt64 = 0x05 + fkFloat32 = 0x06 + fkFloat64 = 0x07 + fkString = 0x08 + fkBytes = 0x09 + fkArray = 0x0A + fkObject = 0x0B + fkVector = 0x0C + fkJson = 0x0D + + MsgKind* = enum + mkClientHandshake = 0x01 + mkQuery = 0x02 + mkQueryParams = 0x03 + mkExecute = 0x04 + mkBatch = 0x05 + mkTransaction = 0x06 + mkClose = 0x07 + mkPing = 0x08 + mkAuth = 0x09 + mkServerHandshake = 0x80 + mkReady = 0x81 + mkData = 0x82 + mkComplete = 0x83 + mkError = 0x84 + mkAuthChallenge = 0x85 + mkAuthOk = 0x86 + mkSchemaChange = 0x87 + mkPong = 0x88 + mkTransactionState = 0x89 + + ResultFormat* = enum + rfBinary = 0x00 + rfJson = 0x01 + rfText = 0x02 + + WireValue* = object + case kind*: FieldKind + of fkNull: discard + of fkBool: boolVal*: bool + of fkInt8: int8Val*: int8 + of fkInt16: int16Val*: int16 + of fkInt32: int32Val*: int32 + of fkInt64: int64Val*: int64 + of fkFloat32: float32Val*: float32 + of fkFloat64: float64Val*: float64 + of fkString: strVal*: string + of fkBytes: bytesVal*: seq[byte] + of fkArray: arrayVal*: seq[WireValue] + of fkObject: objVal*: seq[(string, WireValue)] + of fkVector: vecVal*: seq[float32] + of fkJson: jsonVal*: string + +proc writeUint32*(buf: var seq[byte], val: uint32) = + var bytes: array[4, byte] + bigEndian32(addr bytes, unsafeAddr val) + buf.add(bytes) + +proc writeUint64(buf: var seq[byte], val: uint64) = + var bytes: array[8, byte] + bigEndian64(addr bytes, unsafeAddr val) + buf.add(bytes) + +proc writeString*(buf: var seq[byte], s: string) = + buf.writeUint32(uint32(s.len)) + for ch in s: + buf.add(byte(ch)) + +proc readUint32*(buf: openArray[byte], pos: var int): uint32 = + var bytes: array[4, byte] + for i in 0..3: bytes[i] = buf[pos + i] + bigEndian32(addr result, unsafeAddr bytes) + pos += 4 + +proc readUint64*(buf: openArray[byte], pos: var int): uint64 = + var bytes: array[8, byte] + for i in 0..7: bytes[i] = buf[pos + i] + bigEndian64(addr result, unsafeAddr bytes) + pos += 8 + +proc readString*(buf: openArray[byte], pos: var int): string = + let len = int(readUint32(buf, pos)) + result = newString(len) + for i in 0.." + +suite "Typed rows": + test "QueryResult carries typed rows for string and int": + let client = newClient() + let qb = newQueryBuilder(client) + discard qb + check compiles(client.query("SELECT 1")) diff --git a/clients/nim/tests/test_integration.nim b/clients/nim/tests/test_integration.nim index 53075b1..8d895c4 100644 --- a/clients/nim/tests/test_integration.nim +++ b/clients/nim/tests/test_integration.nim @@ -4,7 +4,7 @@ import std/unittest import std/asyncdispatch -import std/asyncnet +import std/net as netmod import std/strutils import std/os import baradb/client @@ -15,8 +15,8 @@ const proc serverAvailable(): bool = try: - var socket = newAsyncSocket() - waitFor socket.connect(TestHost, Port(TestPort)) + var socket = netmod.newSocket() + socket.connect(TestHost, Port(TestPort), timeout = 1000) socket.close() return true except: @@ -26,96 +26,103 @@ let hasServer = serverAvailable() suite "Integration: Connection": test "Connect and close": - if not hasServer: + if hasServer: + var client = newClient(ClientConfig(host: TestHost, port: TestPort)) + check not client.isConnected + waitFor client.connect() + check client.isConnected + client.close() + check not client.isConnected + else: skip() - var client = newClient(ClientConfig(host: TestHost, port: TestPort)) - check not client.isConnected - waitFor client.connect() - check client.isConnected - client.close() - check not client.isConnected test "Ping": - if not hasServer: + if hasServer: + var client = newClient(ClientConfig(host: TestHost, port: TestPort)) + waitFor client.connect() + check (waitFor client.ping()) == true + client.close() + else: skip() - var client = newClient(ClientConfig(host: TestHost, port: TestPort)) - waitFor client.connect() - check (waitFor client.ping()) == true - client.close() suite "Integration: Query": test "Simple SELECT": - if not hasServer: + if hasServer: + var client = newClient(ClientConfig(host: TestHost, port: TestPort)) + waitFor client.connect() + let result = waitFor client.query("SELECT 1 as one") + check result.rowCount >= 0 + client.close() + else: skip() - var client = newClient(ClientConfig(host: TestHost, port: TestPort)) - waitFor client.connect() - let result = waitFor client.query("SELECT 1 as one") - check result.rowCount >= 0 - client.close() test "Parameterized query": - if not hasServer: + if hasServer: + var client = newClient(ClientConfig(host: TestHost, port: TestPort)) + waitFor client.connect() + let result = waitFor client.query( + "SELECT $1 as num, $2 as txt", + @[WireValue(kind: fkInt64, int64Val: 42), WireValue(kind: fkString, strVal: "hello")] + ) + check result.rowCount >= 0 + client.close() + else: skip() - var client = newClient(ClientConfig(host: TestHost, port: TestPort)) - waitFor client.connect() - let result = waitFor client.query( - "SELECT $1 as num, $2 as txt", - @[WireValue(kind: fkInt64, int64Val: 42), WireValue(kind: fkString, strVal: "hello")] - ) - check result.rowCount >= 0 - client.close() suite "Integration: DDL & DML": test "Create table, insert, select, drop": - if not hasServer: + if hasServer: + var client = newClient(ClientConfig(host: TestHost, port: TestPort)) + waitFor client.connect() + + try: + discard waitFor client.exec("DROP TABLE IF EXISTS nim_test_users") + except: + discard + + discard waitFor client.exec("CREATE TABLE nim_test_users (id INT PRIMARY KEY, name STRING, age INT)") + let affected = waitFor client.exec("INSERT INTO nim_test_users (id, name, age) VALUES (1, 'Alice', 30)") + check affected >= 0 + + let result = waitFor client.query("SELECT name, age FROM nim_test_users WHERE id = 1") + check result.rowCount == 1 + client.close() + else: skip() - var client = newClient(ClientConfig(host: TestHost, port: TestPort)) - waitFor client.connect() - - try: - discard waitFor client.exec("DROP TABLE IF EXISTS nim_test_users") - except: - discard - - discard waitFor client.exec("CREATE TABLE nim_test_users (id INT PRIMARY KEY, name STRING, age INT)") - let affected = waitFor client.exec("INSERT INTO nim_test_users (id, name, age) VALUES (1, 'Alice', 30)") - check affected >= 0 - - let result = waitFor client.query("SELECT name, age FROM nim_test_users WHERE id = 1") - check result.rowCount == 1 - client.close() suite "Integration: QueryBuilder": test "Builder exec": - if not hasServer: + if hasServer: + var client = newClient(ClientConfig(host: TestHost, port: TestPort)) + waitFor client.connect() + + try: + discard waitFor client.exec("DROP TABLE IF EXISTS nim_test_products") + except: + discard + + discard waitFor client.exec("CREATE TABLE nim_test_products (id INT PRIMARY KEY, name STRING, price FLOAT)") + discard waitFor client.exec("INSERT INTO nim_test_products (id, name, price) VALUES (1, 'Widget', 9.99)") + + let result = waitFor newQueryBuilder(client) + .select("name", "price") + .from("nim_test_products") + .where("id = 1") + .exec() + check result.rowCount == 1 + + discard waitFor client.exec("DROP TABLE nim_test_products") + client.close() + else: skip() - var client = newClient(ClientConfig(host: TestHost, port: TestPort)) - waitFor client.connect() - - try: - discard waitFor client.exec("DROP TABLE IF EXISTS nim_test_products") - except: - discard - - discard waitFor client.exec("CREATE TABLE nim_test_products (id INT PRIMARY KEY, name STRING, price FLOAT)") - discard waitFor client.exec("INSERT INTO nim_test_products (id, name, price) VALUES (1, 'Widget', 9.99)") - - let result = waitFor newQueryBuilder(client) - .select("name", "price") - .from("nim_test_products") - .where("id = 1") - .exec() - check result.rowCount == 1 - - discard waitFor client.exec("DROP TABLE nim_test_products") - client.close() suite "Integration: SyncClient": test "Sync query": - if not hasServer: + if hasServer: + var client = newSyncClient(ClientConfig(host: TestHost, port: TestPort)) + client.connect() + let result = client.query("SELECT 1 as one") + check result.rowCount >= 0 + client.close() + else: skip() - var client = newSyncClient(ClientConfig(host: TestHost, port: TestPort)) - client.connect() - let result = client.query("SELECT 1 as one") - check result.rowCount >= 0 - client.close() diff --git a/clients/nim/tests/test_pool.nim b/clients/nim/tests/test_pool.nim new file mode 100644 index 0000000..d9a0a04 --- /dev/null +++ b/clients/nim/tests/test_pool.nim @@ -0,0 +1,19 @@ +import std/unittest +import std/asyncdispatch +import baradb/client +import baradb/pool + +suite "BaraPool": + test "pool stats with one acquired connection": + proc run() {.async.} = + let cfg = ClientConfig(host: "127.0.0.1", port: 9472, timeoutMs: 100) + let pool = newBaraPool(cfg, minConnections = 0, maxConnections = 2) + # Without a server, acquire should fail cleanly (timeout or connection refused) + var failedCleanly = false + try: + withClient(pool): + discard + except BaraError: + failedCleanly = true + check failedCleanly + waitFor run() diff --git a/clients/nim/tests/test_wire.nim b/clients/nim/tests/test_wire.nim new file mode 100644 index 0000000..f86c2cd --- /dev/null +++ b/clients/nim/tests/test_wire.nim @@ -0,0 +1,68 @@ +import std/unittest +import std/asyncdispatch +import std/asyncnet +import std/json +import baradb/wire +import baradb/client + +proc buildDataResponse(cols: seq[string], rows: seq[seq[WireValue]], affected: int): seq[byte] = + var payload: seq[byte] = @[] + payload.writeUint32(uint32(cols.len)) + for c in cols: + payload.writeString(c) + for c in cols: + payload.add(byte(fkString)) + payload.writeUint32(uint32(rows.len)) + for row in rows: + for wv in row: + payload.serializeValue(wv) + result = buildMessage(mkData, 1'u32, payload) + var completePayload: seq[byte] = @[] + completePayload.writeUint32(uint32(affected)) + result.add(buildMessage(mkComplete, 1'u32, completePayload)) + +suite "Wire protocol": + test "buildMessage header is 12 bytes + payload": + let msg = buildMessage(mkQuery, 7'u32, toBytes("SELECT 1")) + check msg.len == 12 + 8 + + test "serialize/deserialize round-trip for WireValue": + let original = WireValue(kind: fkInt64, int64Val: 42) + var buf: seq[byte] = @[] + buf.serializeValue(original) + var pos = 0 + let decoded = deserializeValue(buf, pos) + check decoded.kind == fkInt64 + check decoded.int64Val == 42 + + test "client query against mock server returns typedRows and rows": + proc run() {.async.} = + var server = newAsyncSocket() + server.setSockOpt(OptReuseAddr, true) + server.bindAddr(Port(0), "127.0.0.1") + let port = server.getLocalAddr()[1] + server.listen() + + proc serve() {.async.} = + let s = await server.accept() + let data = buildDataResponse( + @["name", "age"], + @[ + @[WireValue(kind: fkString, strVal: "Alice"), WireValue(kind: fkInt32, int32Val: 30)], + ], + 0, + ) + await s.send(toString(data)) + s.close() + + asyncCheck serve() + + let client = newClient(ClientConfig(host: "127.0.0.1", port: int(port), timeoutMs: 5000)) + await client.connect() + let qr = await client.query("SELECT name, age FROM users") + check qr.rowCount == 1 + check qr.typedRows[0][1].int32Val == 30 + check qr.rows[0][1] == "30" + client.close() + server.close() + waitFor run() diff --git a/docs/en/clients.md b/docs/en/clients.md index 72c1ad1..643fbfd 100644 --- a/docs/en/clients.md +++ b/docs/en/clients.md @@ -123,60 +123,44 @@ async def main(): asyncio.run(main()) ``` -## Nim (Embedded Mode) +## Nim -### Add Dependency +Install the official client: -```nim -# In your .nimble file -requires "barabadb >= 0.1.0" +```bash +nimble install baradb ``` -### Embedded Usage +### Async with connection pool ```nim -import barabadb/storage/lsm -import barabadb/storage/btree -import barabadb/vector/engine -import barabadb/graph/engine +import asyncdispatch, baradb/client, baradb/pool -# Key-Value store -var db = newLSMTree("./data") -db.put("user:1", cast[seq[byte]]("Alice")) -let (found, value) = db.get("user:1") -db.close() +proc main() {.async.} = + let cfg = ClientConfig(host: "127.0.0.1", port: 9472) + let pool = newBaraPool(cfg, minConnections = 2, maxConnections = 10) + withClient(pool): + let r = await c.query("SELECT name FROM users WHERE id = ?", + @[WireValue(kind: fkInt64, int64Val: 1)]) + echo r.typedRows -# B-Tree index -var btree = newBTreeIndex[string, int]() -btree.insert("Alice", 30) -let ages = btree.get("Alice") - -# Vector search -var idx = newHNSWIndex(dimensions = 128) -idx.insert(1, @[0.1'f32, 0.2, 0.3], {"category": "A"}.toTable) -let results = idx.search(@[0.1'f32, 0.2, 0.3], k = 10) - -# Graph -var g = newGraph() -let alice = g.addNode("Person", {"name": "Alice"}.toTable) -let bob = g.addNode("Person", {"name": "Bob"}.toTable) -discard g.addEdge(alice, bob, "knows") -let path = g.shortestPath(alice, bob) +waitFor main() ``` -### Client Library +### Sync client ```nim -import barabadb/client/client +import baradb/client -var c = newBaraClient("localhost", 9472) +let c = newSyncClient() c.connect() -let result = c.query("SELECT name FROM users") -for row in result.rows: - echo row["name"] +let r = c.query("SELECT * FROM users") +echo r.rows c.close() ``` +For Laravel-style query building, use `nim-allographer` with the `Baradb` driver. + ## Rust ### Add Dependency diff --git a/docs/superpowers/plans/2026-06-18-good-nim-client-plan.md b/docs/superpowers/plans/2026-06-18-good-nim-client-plan.md new file mode 100644 index 0000000..81d05c7 --- /dev/null +++ b/docs/superpowers/plans/2026-06-18-good-nim-client-plan.md @@ -0,0 +1,1579 @@ +# Good Nim Client for BaraDB — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:subagent-driven-development` (recommended) or `superpowers:executing-plans` to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Refactor `clients/nim` into a canonical, production-ready BaraDB client and make `clients/nim-allographer` a thin wrapper around it, removing duplicated wire-protocol code. + +**Architecture:** Move all wire-protocol and socket handling into `clients/nim`. Add typed rows, an async request queue, a connection pool, TLS, and a proper error hierarchy there. `clients/nim-allographer` adds `requires "baradb"` and replaces its copied client with re-exports plus its own query-builder and migration glue. + +**Tech Stack:** Nim 2.0+, `asyncdispatch`, `asyncnet`, `asynclocks`, `std/endians`, `std/httpclient` (optional HTTP module), `unittest`. + +--- + +## File Structure + +### New / heavily changed in `clients/nim` + +| File | What it becomes | +|------|-----------------| +| `src/baradb/wire.nim` | New. Protocol constants, `WireValue`, serialize/deserialize, message builders. | +| `src/baradb/errors.nim` | New. `BaraError` hierarchy. | +| `src/baradb/client.nim` | Refactored. Async/sync client, typed `QueryResult`, request queue, timeouts, reconnect. | +| `src/baradb/pool.nim` | New. Async connection pool `BaraPool` + `withClient`. | +| `src/baradb/http.nim` | New. Optional HTTP/REST fallback client. | +| `tests/test_wire.nim` | New. Round-trip and mock-server tests for the protocol. | +| `tests/test_pool.nim` | New. Pool acquire/release/timeout/eviction tests. | +| `tests/test_client.nim` | Updated. QueryBuilder tests still valid; add typed-row tests. | +| `baradb.nimble` | Bump version to `1.2.0`; add `srcDir = "src"` stays. | +| `README.md` | Document pool, typed rows, TLS, HTTP fallback. | + +### Changed in `clients/nim-allographer` + +| File | What changes | +|------|--------------| +| `allographer.nimble` | Add `requires "baradb >= 1.2.0"`. | +| `src/allographer/query_builder/libs/baradb/baradb_client.nim` | Remove wire/sync/query-builder code; re-export from `baradb/client`; keep migration helpers. | +| `src/allographer/query_builder/models/baradb/baradb_exec.nim` | Use `resultSet.typedRows` in `toJson`; keep parameterized query path. | + +### Changed in server tree + +| File | What changes | +|------|--------------| +| `src/barabadb/client/client.nim` | Add `{.deprecated: "use baradb/client from clients/nim".`} pragma. | +| `docs/en/clients.md` | Update Nim section to point to `clients/nim` and mention pool/typed rows. | + +--- + +## Task 1: Extract wire protocol into `src/baradb/wire.nim` + +**Files:** +- Create: `clients/nim/src/baradb/wire.nim` +- Modify: `clients/nim/src/baradb/client.nim` (remove duplicated protocol constants/procs) +- Test: `clients/nim/tests/test_wire.nim` + +The current `client.nim` contains `FieldKind`, `MsgKind`, `WireValue`, serialization, and message builders. Move all of it to `wire.nim` and re-export from `client.nim` so existing `import baradb/client` callers keep working. + +- [ ] **Step 1: Create `clients/nim/src/baradb/wire.nim`** + +```nim +## BaraDB binary wire protocol — shared between client and server. +import std/endians +import std/strutils + +const + ProtocolMagic* = 0x42415241'u32 + +type + FieldKind* = enum + fkNull = 0x00 + fkBool = 0x01 + fkInt8 = 0x02 + fkInt16 = 0x03 + fkInt32 = 0x04 + fkInt64 = 0x05 + fkFloat32 = 0x06 + fkFloat64 = 0x07 + fkString = 0x08 + fkBytes = 0x09 + fkArray = 0x0A + fkObject = 0x0B + fkVector = 0x0C + fkJson = 0x0D + + MsgKind* = enum + mkClientHandshake = 0x01 + mkQuery = 0x02 + mkQueryParams = 0x03 + mkExecute = 0x04 + mkBatch = 0x05 + mkTransaction = 0x06 + mkClose = 0x07 + mkPing = 0x08 + mkAuth = 0x09 + mkServerHandshake = 0x80 + mkReady = 0x81 + mkData = 0x82 + mkComplete = 0x83 + mkError = 0x84 + mkAuthChallenge = 0x85 + mkAuthOk = 0x86 + mkSchemaChange = 0x87 + mkPong = 0x88 + mkTransactionState = 0x89 + + ResultFormat* = enum + rfBinary = 0x00 + rfJson = 0x01 + rfText = 0x02 + + WireValue* = object + case kind*: FieldKind + of fkNull: discard + of fkBool: boolVal*: bool + of fkInt8: int8Val*: int8 + of fkInt16: int16Val*: int16 + of fkInt32: int32Val*: int32 + of fkInt64: int64Val*: int64 + of fkFloat32: float32Val*: float32 + of fkFloat64: float64Val*: float64 + of fkString: strVal*: string + of fkBytes: bytesVal*: seq[byte] + of fkArray: arrayVal*: seq[WireValue] + of fkObject: objVal*: seq[(string, WireValue)] + of fkVector: vecVal*: seq[float32] + of fkJson: jsonVal*: string + +proc writeUint32(buf: var seq[byte], val: uint32) = + var bytes: array[4, byte] + bigEndian32(addr bytes, unsafeAddr val) + buf.add(bytes) + +proc writeUint64(buf: var seq[byte], val: uint64) = + var bytes: array[8, byte] + bigEndian64(addr bytes, unsafeAddr val) + buf.add(bytes) + +proc writeString(buf: var seq[byte], s: string) = + buf.writeUint32(uint32(s.len)) + for ch in s: + buf.add(byte(ch)) + +proc readUint32(buf: openArray[byte], pos: var int): uint32 = + var bytes: array[4, byte] + for i in 0..3: bytes[i] = buf[pos + i] + bigEndian32(addr result, unsafeAddr bytes) + pos += 4 + +proc readUint64(buf: openArray[byte], pos: var int): uint64 = + var bytes: array[8, byte] + for i in 0..7: bytes[i] = buf[pos + i] + bigEndian64(addr result, unsafeAddr bytes) + pos += 8 + +proc readString(buf: openArray[byte], pos: var int): string = + let len = int(readUint32(buf, pos)) + result = newString(len) + for i in 0.." + of fkArray: return "" + of fkObject: return "" + of fkVector: return "" + of fkJson: return wv.jsonVal + +proc readResponsePayload(client: BaraClient): Future[(MsgKind, seq[byte])] {.async.} = + let headerStr = await recvExact(client.socket, 12, client.config.timeoutMs) + var pos = 0 + let hdrData = toBytes(headerStr) + let kind = MsgKind(readUint32(hdrData, pos)) + let payloadLen = int(readUint32(hdrData, pos)) + discard readUint32(hdrData, pos) + let payloadStr = await recvExact(client.socket, payloadLen, client.config.timeoutMs) + return (kind, toBytes(payloadStr)) + +proc parseQueryResponse(client: BaraClient, kind: MsgKind, payload: seq[byte]): Future[QueryResult] {.async.} = + result = QueryResult(columns: @[], rows: @[], typedRows: @[], rowCount: 0, affectedRows: 0) + if kind == mkReady: + return + if kind == mkError and payload.len >= 8: + var epos = 0 + let code = readUint32(payload, epos) + let emsg = readString(payload, epos) + var err = newException(BaraServerError, "Error " & $code & ": " & emsg) + err.code = code + raise err + if kind == mkData: + var dpos = 0 + let colCount = int(readUint32(payload, dpos)) + for i in 0..= 4: + var cpPos = 0 + result.affectedRows = int(readUint32(compPayload, cpPos)) + return + if kind == mkComplete: + var rpos = 0 + result.affectedRows = int(readUint32(payload, rpos)) + return + raise newException(BaraProtocolError, "Unexpected response kind: 0x" & toHex(uint32(kind), 2)) + +proc doQuery(client: BaraClient, msg: seq[byte]): Future[QueryResult] {.async.} = + if not client.connected: + raise newException(BaraIoError, "Not connected") + await client.sendLock.acquire() + try: + await client.socket.send(toString(msg)) + let (kind, payload) = await client.readResponsePayload() + return await client.parseQueryResponse(kind, payload) + finally: + client.sendLock.release() + +proc query*(client: BaraClient, sql: string): Future[QueryResult] {.async.} = + let msg = makeQueryMessage(client.nextId(), sql) + return await client.doQuery(msg) + +proc query*(client: BaraClient, sql: string, params: seq[WireValue]): Future[QueryResult] {.async.} = + let msg = makeQueryParamsMessage(client.nextId(), sql, params) + return await client.doQuery(msg) + +proc exec*(client: BaraClient, sql: string): Future[int] {.async.} = + let qr = await client.query(sql) + return qr.affectedRows + +proc auth*(client: BaraClient, token: string) {.async.} = + let msg = makeAuthMessage(client.nextId(), token) + await client.sendLock.acquire() + try: + await client.socket.send(toString(msg)) + let (kind, payload) = await client.readResponsePayload() + case kind + of mkAuthOk: + return + of mkError: + var epos = 0 + discard readUint32(payload, epos) + let emsg = readString(payload, epos) + raise newException(BaraAuthError, "Auth failed: " & emsg) + else: + raise newException(BaraProtocolError, "Unexpected auth response: 0x" & toHex(uint32(kind), 2)) + finally: + client.sendLock.release() + +proc ping*(client: BaraClient): Future[bool] {.async.} = + if not client.connected: + return false + let msg = buildMessage(mkPing, client.nextId(), @[]) + await client.sendLock.acquire() + try: + await client.socket.send(toString(msg)) + let (kind, _) = await client.readResponsePayload() + return kind == mkPong + except: + return false + finally: + client.sendLock.release() + +proc readQueryResponse*(client: BaraClient): Future[QueryResult] {.async.} = + ## Read and parse the next server response. Does NOT acquire sendLock; + ## callers that already sent a message manually can use this. + let (kind, payload) = await client.readResponsePayload() + return await client.parseQueryResponse(kind, payload) + +# === Fluent Query Builder === + +type + QueryBuilder* = ref object + client: BaraClient + selectCols: seq[string] + fromTable: string + whereClauses: seq[string] + joinClauses: seq[string] + groupByCols: seq[string] + havingClause: string + orderCols: seq[string] + orderDirs: seq[string] + limitVal: int + offsetVal: int + +proc newQueryBuilder*(client: BaraClient): QueryBuilder = + QueryBuilder(client: client, limitVal: 0, offsetVal: 0) + +proc select*(qb: QueryBuilder, cols: varargs[string]): QueryBuilder = + for c in cols: qb.selectCols.add(c) + return qb + +proc `from`*(qb: QueryBuilder, table: string): QueryBuilder = + qb.fromTable = table + return qb + +proc where*(qb: QueryBuilder, clause: string): QueryBuilder = + qb.whereClauses.add(clause) + return qb + +proc join*(qb: QueryBuilder, table: string, on: string): QueryBuilder = + qb.joinClauses.add("JOIN " & table & " ON " & on) + return qb + +proc leftJoin*(qb: QueryBuilder, table: string, on: string): QueryBuilder = + qb.joinClauses.add("LEFT JOIN " & table & " ON " & on) + return qb + +proc groupBy*(qb: QueryBuilder, cols: varargs[string]): QueryBuilder = + for c in cols: qb.groupByCols.add(c) + return qb + +proc having*(qb: QueryBuilder, clause: string): QueryBuilder = + qb.havingClause = clause + return qb + +proc orderBy*(qb: QueryBuilder, col: string, dir: string = "ASC"): QueryBuilder = + qb.orderCols.add(col) + qb.orderDirs.add(dir) + return qb + +proc limit*(qb: QueryBuilder, n: int): QueryBuilder = + qb.limitVal = n + return qb + +proc offset*(qb: QueryBuilder, n: int): QueryBuilder = + qb.offsetVal = n + return qb + +proc build*(qb: QueryBuilder): string = + result = "SELECT " & (if qb.selectCols.len > 0: qb.selectCols.join(", ") else: "*") + result &= " FROM " & qb.fromTable + for j in qb.joinClauses: result &= " " & j + if qb.whereClauses.len > 0: result &= " WHERE " & qb.whereClauses.join(" AND ") + if qb.groupByCols.len > 0: result &= " GROUP BY " & qb.groupByCols.join(", ") + if qb.havingClause.len > 0: result &= " HAVING " & qb.havingClause + if qb.orderCols.len > 0: + result &= " ORDER BY " + for i, col in qb.orderCols: + if i > 0: result &= ", " + result &= col & " " & qb.orderDirs[i] + if qb.limitVal > 0: result &= " LIMIT " & $qb.limitVal + if qb.offsetVal > 0: result &= " OFFSET " & $qb.offsetVal + +proc exec*(qb: QueryBuilder): Future[QueryResult] {.async.} = + return await qb.client.query(qb.build()) + +# === Blocking Sync Client === + +type + SyncClient* = ref object + config: ClientConfig + socket: netmod.Socket + connected: bool + requestId: uint32 + lock: Lock + +proc newSyncClient*(config: ClientConfig = defaultConfig()): SyncClient = + result = SyncClient(config: config, connected: false, requestId: 0) + result.socket = netmod.newSocket() + initLock(result.lock) + +proc recvExactBlocking(sock: netmod.Socket, size: int): string = + result = "" + while result.len < size: + let chunk = sock.recv(size - result.len) + if chunk.len == 0: + raise newException(BaraIoError, "Connection closed") + result.add(chunk) + +proc readResponsePayloadBlocking(client: SyncClient): (MsgKind, seq[byte]) = + let headerData = client.socket.recvExactBlocking(12) + var pos = 0 + let hdrData = toBytes(headerData) + let kind = MsgKind(readUint32(hdrData, pos)) + let payloadLen = int(readUint32(hdrData, pos)) + discard readUint32(hdrData, pos) + let payloadStr = client.socket.recvExactBlocking(payloadLen) + return (kind, toBytes(payloadStr)) + +proc parseQueryResponseBlocking(client: SyncClient, kind: MsgKind, payload: seq[byte]): QueryResult = + result = QueryResult(columns: @[], rows: @[], typedRows: @[], rowCount: 0, affectedRows: 0) + if kind == mkReady: + return + if kind == mkError and payload.len >= 8: + var epos = 0 + let code = readUint32(payload, epos) + let emsg = readString(payload, epos) + var err = newException(BaraServerError, "Error " & $code & ": " & emsg) + err.code = code + raise err + if kind == mkData: + var dpos = 0 + let colCount = int(readUint32(payload, dpos)) + for i in 0..= 4: + var cpPos = 0 + result.affectedRows = int(readUint32(compPayload, cpPos)) + return + if kind == mkComplete: + var rpos = 0 + result.affectedRows = int(readUint32(payload, rpos)) + return + raise newException(BaraProtocolError, "Unexpected response kind: 0x" & toHex(uint32(kind), 2)) + +proc connect*(client: SyncClient) = + netmod.connect(client.socket, client.config.host, Port(client.config.port)) + client.connected = true + +proc close*(client: SyncClient) = + if client.connected: + try: + let msg = buildMessage(mkClose, 0, @[]) + netmod.send(client.socket, toString(msg)) + except: discard + netmod.close(client.socket) + client.connected = false + deinitLock(client.lock) + +proc query*(client: SyncClient, sql: string): QueryResult = + acquire(client.lock) + try: + if not client.connected: + raise newException(BaraIoError, "Not connected") + let msg = makeQueryMessage(0, sql) + netmod.send(client.socket, toString(msg)) + let (kind, payload) = client.readResponsePayloadBlocking() + return client.parseQueryResponseBlocking(kind, payload) + finally: + release(client.lock) + +proc query*(client: SyncClient, sql: string, params: seq[WireValue]): QueryResult = + acquire(client.lock) + try: + if not client.connected: + raise newException(BaraIoError, "Not connected") + let msg = makeQueryParamsMessage(0, sql, params) + netmod.send(client.socket, toString(msg)) + let (kind, payload) = client.readResponsePayloadBlocking() + return client.parseQueryResponseBlocking(kind, payload) + finally: + release(client.lock) + +proc exec*(client: SyncClient, sql: string): int = + let qr = client.query(sql) + return qr.affectedRows + +proc auth*(client: SyncClient, token: string) = + acquire(client.lock) + try: + if not client.connected: + raise newException(BaraIoError, "Not connected") + let msg = makeAuthMessage(0, token) + netmod.send(client.socket, toString(msg)) + let (kind, payload) = client.readResponsePayloadBlocking() + case kind + of mkAuthOk: + return + of mkError: + var epos = 0 + discard readUint32(payload, epos) + let emsg = readString(payload, epos) + raise newException(BaraAuthError, "Auth failed: " & emsg) + else: + raise newException(BaraProtocolError, "Unexpected auth response") + finally: + release(client.lock) + +proc ping*(client: SyncClient): bool = + acquire(client.lock) + try: + if not client.connected: + return false + let msg = buildMessage(mkPing, 0, @[]) + netmod.send(client.socket, toString(msg)) + let (kind, _) = client.readResponsePayloadBlocking() + return kind == mkPong + except: + return false + finally: + release(client.lock) + +proc `$`*(qr: QueryResult): string = + if qr.columns.len == 0: return "(no results)" + result = "" + for i, col in qr.columns: + result &= col + if i < qr.columns.len - 1: result &= ", " + result &= "\n" + for row in qr.rows: + result &= row.join(", ") & "\n" + result &= "(" & $qr.rowCount & " rows)" +``` + +- [ ] **Step 2: Update `clients/nim/tests/test_client.nim` to add typed-row test** + +Append to the existing file: + +```nim +suite "Typed rows": + test "QueryResult carries typed rows for string and int": + let client = newClient() + let qb = newQueryBuilder(client) + discard qb # builder itself doesn't touch the network + # Verify the API surface exists + check compiles(client.query("SELECT 1")) +``` + +- [ ] **Step 3: Compile and run unit tests** + +```bash +cd clients/nim +nim c -r tests/test_client.nim +``` + +Expected: tests pass. + +--- + +## Task 4: Add `src/baradb/pool.nim` — async connection pool + +**Files:** +- Create: `clients/nim/src/baradb/pool.nim` +- Create: `clients/nim/tests/test_pool.nim` + +- [ ] **Step 1: Create `clients/nim/src/baradb/pool.nim`** + +```nim +## Async connection pool for BaraDB. +import std/asyncdispatch +import std/deques +import std/asynclocks +import std/monotimes +import std/times +import ./client +import ./errors + +type + PoolConnection = ref object + client: BaraClient + inUse: bool + createdAt: int64 + lastUsedAt: int64 + + PoolConfig* = object + minConnections*: int + maxConnections*: int + maxIdleTimeMs*: int + maxLifetimeMs*: int + + BaraPool* = ref object + clientConfig: ClientConfig + poolConfig: PoolConfig + connections: seq[PoolConnection] + waiters: Deque[Future[void]] + lock: AsyncLock + +proc defaultPoolConfig*(): PoolConfig = + PoolConfig( + minConnections: 2, + maxConnections: 10, + maxIdleTimeMs: 300_000, + maxLifetimeMs: 3_600_000, + ) + +proc nowUnix(): int64 = getTime().toUnix() + +proc newBaraPool*(clientConfig: ClientConfig, + minConnections = 2, + maxConnections = 10, + poolConfig = defaultPoolConfig()): BaraPool = + result = BaraPool( + clientConfig: clientConfig, + poolConfig: poolConfig, + connections: @[], + waiters: initDeque[Future[void]](), + lock: initAsyncLock(), + ) + result.poolConfig.minConnections = minConnections + result.poolConfig.maxConnections = maxConnections + +proc isExpired(cfg: PoolConfig, conn: PoolConnection): bool = + let now = nowUnix() + if cfg.maxLifetimeMs > 0 and (now - conn.createdAt) * 1000 >= cfg.maxLifetimeMs: + return true + if cfg.maxIdleTimeMs > 0 and conn.lastUsedAt > 0 and (now - conn.lastUsedAt) * 1000 >= cfg.maxIdleTimeMs: + return true + return false + +proc openConnection(pool: BaraPool): Future[BaraClient] {.async.} = + let client = newClient(pool.clientConfig) + await client.connect() + return client + +proc closeConnection(conn: PoolConnection) = + if not conn.client.isNil: + conn.client.close() + +proc wakeOneWaiter(pool: BaraPool) = + while pool.waiters.len > 0: + let w = pool.waiters.popFirst() + if not w.finished: + w.complete() + break + +proc acquireConnection(pool: BaraPool): Future[BaraClient] {.async.} = + let deadline = getMonoTime() + initDuration(milliseconds = pool.clientConfig.timeoutMs) + while true: + await pool.lock.acquire() + # Reuse idle, non-expired connection + var i = 0 + while i < pool.connections.len: + let conn = pool.connections[i] + if not conn.inUse: + if pool.isExpired(conn): + pool.connections.del(i) + pool.lock.release() + closeConnection(conn) + await pool.lock.acquire() + continue + conn.inUse = true + conn.lastUsedAt = nowUnix() + pool.lock.release() + return conn.client + inc i + # Create new if under max + if pool.connections.len < pool.poolConfig.maxConnections: + pool.lock.release() + let client = await pool.openConnection() + await pool.lock.acquire() + let conn = PoolConnection( + client: client, + inUse: true, + createdAt: nowUnix(), + lastUsedAt: nowUnix(), + ) + pool.connections.add(conn) + pool.lock.release() + return client + pool.lock.release() + # Wait for a connection to be released + if getMonoTime() >= deadline: + raise newException(BaraPoolTimeoutError, "Timed out waiting for a free connection") + let w = newFuture[void]("pool.wait") + await pool.lock.acquire() + pool.waiters.addLast(w) + pool.lock.release() + let ok = await withTimeout(w, pool.clientConfig.timeoutMs) + if not ok: + await pool.lock.acquire() + var kept = initDeque[Future[void]]() + while pool.waiters.len > 0: + let x = pool.waiters.popFirst() + if x != w: + kept.addLast(x) + pool.waiters = move(kept) + pool.lock.release() + raise newException(BaraPoolTimeoutError, "Timed out waiting for a free connection") + +proc releaseConnection(pool: BaraPool, client: BaraClient) {.async.} = + await pool.lock.acquire() + for conn in pool.connections: + if conn.client == client: + conn.inUse = false + conn.lastUsedAt = nowUnix() + break + pool.lock.release() + wakeOneWaiter(pool) + +template withClient*(pool: BaraPool, body: untyped): untyped = + let c = await pool.acquireConnection() + try: + body + finally: + await pool.releaseConnection(c) + +proc stats*(pool: BaraPool): Future[(int, int, int)] {.async.} = + await pool.lock.acquire() + let total = pool.connections.len + var inUse = 0 + for c in pool.connections: + if c.inUse: + inc inUse + pool.lock.release() + return (total, total - inUse, inUse) +``` + +- [ ] **Step 2: Create `clients/nim/tests/test_pool.nim`** + +```nim +import std/unittest +import std/asyncdispatch +import baradb/client +import baradb/pool + +suite "BaraPool": + test "pool stats with one acquired connection": + proc run() {.async.} = + let cfg = ClientConfig(host: "127.0.0.1", port: 9472, timeoutMs: 100) + let pool = newBaraPool(cfg, minConnections = 0, maxConnections = 2) + # Without a server, acquire should timeout cleanly + var timedOut = false + try: + withClient(pool): + discard + except BaraPoolTimeoutError: + timedOut = true + check timedOut + waitFor run() +``` + +- [ ] **Step 3: Compile pool and tests** + +```bash +cd clients/nim +nim c src/baradb/pool.nim +nim c -r tests/test_pool.nim +``` + +Expected: compilation succeeds; the unit test passes because it expects a timeout. + +--- + +## Task 5: Add optional HTTP fallback `src/baradb/http.nim` + +**Files:** +- Create: `clients/nim/src/baradb/http.nim` +- Test: manual curl/HTTP test (no unit test required for optional module) + +- [ ] **Step 1: Create `clients/nim/src/baradb/http.nim`** + +```nim +## Optional HTTP/REST fallback client for BaraDB. +import std/asyncdispatch +import std/httpclient +import std/json +import std/strformat +import ./errors + +type + BaraHttpClient* = ref object + baseUrl*: string + token*: string + http: AsyncHttpClient + +proc newBaraHttpClient*(host = "127.0.0.1", port = 9912, token = ""): BaraHttpClient = + BaraHttpClient( + baseUrl: fmt"http://{host}:{port}/api", + token: token, + http: newAsyncHttpClient(), + ) + +proc close*(client: BaraHttpClient) = + client.http.close() + +proc query*(client: BaraHttpClient, sql: string): Future[JsonNode] {.async.} = + var headers = newHttpHeaders({"Content-Type": "application/json"}) + if client.token.len > 0: + headers["Authorization"] = "Bearer " & client.token + let body = %*{ "query": sql } + let response = await client.http.request( + client.baseUrl & "/query", + httpMethod = HttpPost, + body = $body, + headers = headers, + ) + let text = await response.body + if response.code.int != 200: + raise newException(BaraServerError, "HTTP error " & $response.code.int & ": " & text) + return parseJson(text) +``` + +- [ ] **Step 2: Compile** + +```bash +cd clients/nim +nim c src/baradb/http.nim +``` + +Expected: successful compilation. + +--- + +## Task 6: Wire-protocol unit tests with a mock async server + +**Files:** +- Create: `clients/nim/tests/test_wire.nim` + +- [ ] **Step 1: Create `clients/nim/tests/test_wire.nim`** + +```nim +import std/unittest +import std/asyncdispatch +import std/asyncnet +import std/json +import baradb/wire +import baradb/client + +proc buildDataResponse(cols: seq[string], rows: seq[seq[WireValue]], affected: int): seq[byte] = + var payload: seq[byte] = @[] + payload.writeUint32(uint32(cols.len)) + for c in cols: + payload.writeString(c) + for c in cols: + payload.add(byte(fkString)) # fake common type + payload.writeUint32(uint32(rows.len)) + for row in rows: + for wv in row: + payload.serializeValue(wv) + result = buildMessage(mkData, 1'u32, payload) + var completePayload: seq[byte] = @[] + completePayload.writeUint32(uint32(affected)) + result.add(buildMessage(mkComplete, 1'u32, completePayload)) + +suite "Wire protocol": + test "buildMessage header is 12 bytes + payload": + let msg = buildMessage(mkQuery, 7'u32, toBytes("SELECT 1")) + check msg.len == 12 + 8 + 1 # header + len("SELECT 1") + format byte + + test "serialize/deserialize round-trip for WireValue": + let original = WireValue(kind: fkInt64, int64Val: 42) + var buf: seq[byte] = @[] + buf.serializeValue(original) + var pos = 0 + let decoded = deserializeValue(buf, pos) + check decoded.kind == fkInt64 + check decoded.int64Val == 42 + + test "client query against mock server returns typedRows and rows": + proc run() {.async.} = + var server = newAsyncSocket() + server.setSockOpt(OptReuseAddr, true) + server.bindAddr(Port(0), "127.0.0.1") + let port = server.getLocalAddr()[1] + server.listen() + + proc serve() {.async.} = + let s = await server.accept() + let data = buildDataResponse( + @["name", "age"], + @[ + @[WireValue(kind: fkString, strVal: "Alice"), WireValue(kind: fkInt32, int32Val: 30)], + ], + 0, + ) + await s.send(toString(data)) + s.close() + + asyncCheck serve() + + let client = newClient(ClientConfig(host: "127.0.0.1", port: int(port))) + await client.connect() + let qr = await client.query("SELECT name, age FROM users") + check qr.rowCount == 1 + check qr.typedRows[0][1].int32Val == 30 + check qr.rows[0][1] == "30" + client.close() + server.close() + waitFor run() +``` + +Note: The last test reuses the helper payload builder; adjust indices if the helper appends both messages. + +- [ ] **Step 2: Compile and run** + +```bash +cd clients/nim +nim c -r tests/test_wire.nim +``` + +Expected: tests pass. + +--- + +## Task 7: Bump version and update `clients/nim/README.md` + +**Files:** +- Modify: `clients/nim/baradb.nimble` +- Modify: `clients/nim/README.md` + +- [ ] **Step 1: Bump version in `clients/nim/baradb.nimble`** + +```nim +version = "1.2.0" +``` + +- [ ] **Step 2: Update `clients/nim/README.md` to document new features** + +Add sections for: + +- `QueryResult.typedRows` +- `BaraPool` and `withClient` +- TLS via `ssl: true` +- Optional `baradb/http` +- New exception hierarchy + +Keep the existing quick-start examples. + +- [ ] **Step 3: Run the full client test suite** + +```bash +cd clients/nim +nimble test_unit +``` + +Expected: all unit tests pass. + +--- + +## Task 8: Make `clients/nim-allographer` depend on the canonical client + +**Files:** +- Modify: `clients/nim-allographer/allographer.nimble` +- Modify: `clients/nim-allographer/src/allographer/query_builder/libs/baradb/baradb_client.nim` + +- [ ] **Step 1: Add dependency to `clients/nim-allographer/allographer.nimble`** + +```nim +requires "baradb >= 1.2.0" +``` + +- [ ] **Step 2: Replace `baradb_client.nim` with a thin wrapper** + +```nim +## BaraDB driver glue for nim-allographer. +## All wire/socket logic lives in the canonical `baradb/client` package. +import std/asyncdispatch +import std/json +import baradb/client +export client + +# === Migration helpers (allographer-specific) === + +proc createMigration*(client: BaraClient, name: string, upBody: string, + downBody: string = ""): Future[QueryResult] {.async.} = + var sql = "CREATE MIGRATION " & name & " { UP: " & upBody & ";" + if downBody.len > 0: + sql &= " DOWN: " & downBody & ";" + sql &= " }" + return await client.query(sql) + +proc applyMigration*(client: BaraClient, name: string): Future[QueryResult] {.async.} = + return await client.query("APPLY MIGRATION " & name) + +proc migrateUp*(client: BaraClient, count: int = 0): Future[QueryResult] {.async.} = + var sql = "MIGRATION UP" + if count > 0: + sql &= " " & $count + return await client.query(sql) + +proc migrateDown*(client: BaraClient, count: int = 1): Future[QueryResult] {.async.} = + return await client.query("MIGRATION DOWN " & $count) + +proc migrationStatus*(client: BaraClient): Future[QueryResult] {.async.} = + return await client.query("MIGRATION STATUS") + +proc migrationDryRun*(client: BaraClient, name: string): Future[QueryResult] {.async.} = + return await client.query("MIGRATION DRY RUN " & name) +``` + +- [ ] **Step 3: Compile allographer baradb driver** + +```bash +cd clients/nim-allographer +nim c src/allographer/query_builder/libs/baradb/baradb_client.nim +``` + +Expected: successful compilation. + +--- + +## Task 9: Use typed rows in allographer `baradb_exec.nim` + +**Files:** +- Modify: `clients/nim-allographer/src/allographer/query_builder/models/baradb/baradb_exec.nim` + +- [ ] **Step 1: Rewrite `toJson` to read `typedRows`** + +Replace the existing `toJson` proc with: + +```nim +proc wireValueToJson*(wv: WireValue): JsonNode = + case wv.kind + of fkNull: newJNull() + of fkBool: newJBool(wv.boolVal) + of fkInt8: newJInt(int(wv.int8Val)) + of fkInt16: newJInt(int(wv.int16Val)) + of fkInt32: newJInt(int(wv.int32Val)) + of fkInt64: newJInt(int(wv.int64Val)) + of fkFloat32: newJFloat(float(wv.float32Val)) + of fkFloat64: newJFloat(wv.float64Val) + of fkString: newJString(wv.strVal) + of fkBytes: newJString("") + of fkArray: + result = newJArray() + for item in wv.arrayVal: + result.add(wireValueToJson(item)) + of fkObject: + result = newJObject() + for (name, val) in wv.objVal: + result[name] = wireValueToJson(val) + of fkVector: + result = newJArray() + for f in wv.vecVal: + result.add(newJFloat(float(f))) + of fkJson: + try: + result = parseJson(wv.jsonVal) + except JsonParsingError: + result = newJString(wv.jsonVal) + +proc toJson*(resultSet: QueryResult): seq[JsonNode] = + var response_table = newSeq[JsonNode](resultSet.rowCount) + for r in 0 ..< resultSet.rowCount: + var response_row = newJObject() + for c in 0 ..< resultSet.columns.len: + let key = resultSet.columns[c] + response_row[key] = wireValueToJson(resultSet.typedRows[r][c]) + response_table[r] = response_row + return response_table +``` + +- [ ] **Step 2: Compile** + +```bash +cd clients/nim-allographer +nim c src/allographer/query_builder/models/baradb/baradb_exec.nim +``` + +Expected: successful compilation. + +--- + +## Task 10: Deprecate the old server-side embedded client + +**Files:** +- Modify: `src/barabadb/client/client.nim` + +- [ ] **Step 1: Add a deprecated module-level pragma** + +At the very top of `src/barabadb/client/client.nim`, add: + +```nim +{.deprecated: "Use the canonical baradb/client from clients/nim instead.".} +``` + +- [ ] **Step 2: Compile check** + +```bash +nim c src/barabadb/client/client.nim +``` + +Expected: deprecation warning, but compilation succeeds. + +--- + +## Task 11: Update `docs/en/clients.md` + +**Files:** +- Modify: `docs/en/clients.md` + +- [ ] **Step 1: Rewrite the Nim client section** + +Replace the "Nim (Embedded Mode)" and "Client Library" subsections under Nim with: + +```markdown +## Nim + +Install the official client: + +```bash +nimble install baradb +``` + +### Async with connection pool + +```nim +import asyncdispatch, baradb/client, baradb/pool + +proc main() {.async.} = + let cfg = ClientConfig(host: "127.0.0.1", port: 9472) + let pool = newBaraPool(cfg, minConnections = 2, maxConnections = 10) + withClient(pool): + let r = await c.query("SELECT name FROM users WHERE id = ?", + @[WireValue(kind: fkInt64, int64Val: 1)]) + echo r.typedRows + +waitFor main() +``` + +### Sync client + +```nim +import baradb/client + +let c = newSyncClient() +c.connect() +let r = c.query("SELECT * FROM users") +echo r.rows +c.close() +``` + +For Laravel-style query building, use `nim-allographer` with the `Baradb` driver. +``` +``` + +- [ ] **Step 2: Verify markdown renders** + +No build step required; visually inspect the file. + +--- + +## Task 12: Integration & regression testing + +**Files:** +- Run: `clients/nim/tests/test_integration.nim` +- Run: `clients/nim-allographer/tests/baradb/*` + +- [ ] **Step 1: Run standalone integration tests if a server is available** + +```bash +cd clients/nim +nim c -r tests/test_integration.nim +``` + +Expected: passes if BaraDB is running on `localhost:9472`; otherwise skips. + +- [ ] **Step 2: Run allographer baradb tests if a server is available** + +```bash +cd clients/nim-allographer +testament p 'tests/baradb/test_*.nim' +``` + +Expected: existing tests pass. + +- [ ] **Step 3: Run the standalone unit tests one final time** + +```bash +cd clients/nim +nimble test_unit +``` + +Expected: all unit tests pass. + +--- + +## Self-Review Checklist + +### Spec coverage + +| Spec requirement | Implementing task | +|------------------|-------------------| +| Single source of truth for wire protocol | Task 1 | +| Exception hierarchy | Task 2 | +| Typed rows | Task 3, Task 9 | +| Request queue / concurrent safety | Task 3 (`AsyncLock`) | +| Timeouts | Task 3 (`recvExact` + `withTimeout`) | +| Connection pool | Task 4 | +| TLS option | Task 3 (`ssl` config) | +| HTTP fallback | Task 5 | +| Allographer wrapper | Task 8 | +| Deprecate old client | Task 10 | +| Docs update | Task 7, Task 11 | +| Tests | Tasks 1–12 | + +### Placeholder scan + +No `TBD`, `TODO`, "implement later", or "add appropriate error handling" strings remain. Every step includes file paths, code, or exact commands. + +### Type consistency + +- `ClientConfig` is defined only in `clients/nim/src/baradb/client.nim`. +- `WireValue`, `FieldKind`, `MsgKind`, `buildMessage`, `makeQueryMessage`, `makeQueryParamsMessage` live in `clients/nim/src/baradb/wire.nim` and are re-exported. +- `QueryResult` always has both `rows` and `typedRows`. +- `BaraPool.withClient` returns the canonical `BaraClient`. +- All exceptions inherit from `BaraError`. + +### Known gaps / next iteration + +- Native `mkBatch`/`mkTransaction` messages are left for a follow-up once server semantics are stable. +- Replacing allographer's custom pool with `BaraPool` is Phase 2 and intentionally out of scope for this plan. diff --git a/docs/superpowers/specs/2026-06-18-good-nim-client-design.md b/docs/superpowers/specs/2026-06-18-good-nim-client-design.md new file mode 100644 index 0000000..275ca54 --- /dev/null +++ b/docs/superpowers/specs/2026-06-18-good-nim-client-design.md @@ -0,0 +1,274 @@ +# Design: A Good Nim Client for BaraDB + +**Date:** 2026-06-18 +**Status:** Approved (approach B) +**Scope:** `clients/nim` and `clients/nim-allographer` + +## 1. Goal + +Turn the existing Nim client code into a production-grade, easy-to-use client for BaraDB with: + +- A single source of truth for the binary wire protocol. +- Async + sync APIs that are safe under concurrent use. +- Connection pooling, timeouts, TLS, and reconnect support. +- Typed values (not only strings) for vectors, JSON, bytes, etc. +- Clean integration with `nim-allographer` so the Laravel-style query builder keeps working. +- Good unit and integration test coverage without requiring a live server for every test. + +## 2. Current State + +- `clients/nim` (`baradb` nimble package) is a self-contained, stdlib-only async/sync client. It duplicates the wire protocol to avoid depending on the server source. +- `clients/nim-allographer` is a fork of `itsumura-h/nim-allographer`. It copy-pastes the same client into `src/allographer/query_builder/libs/baradb/baradb_client.nim` and adds a connection pool, query builder integration, migrations, and prepared-statement helpers. +- `src/barabadb/client/client.nim` is an incomplete embedded client bundled with the server and should not be used by applications. +- The Python, JavaScript, and Rust clients already have internal request queues so that concurrent operations on one TCP connection do not interleave frames on the wire. The Nim clients do not. +- The Nim clients convert every `WireValue` to `string`, which loses type information for vectors, JSON, bytes, arrays, and objects. +- `timeoutMs` and `maxRetries` exist in `ClientConfig` but are not honored. + +## 3. Design Principles + +1. **Canonical low-level package.** `clients/nim` owns the wire protocol, socket handling, typed values, request serialization, pooling, TLS, and timeouts. +2. **Thin allographer wrapper.** `clients/nim-allographer` imports the canonical package and only adds allographer-specific glue (types, `dbOpen`, query builder, migrations, transactions). +3. **No new runtime dependencies for the standalone client.** It must stay stdlib-only so it can be used in embedded and restricted environments. +4. **Backward compatibility.** Existing `dbOpen(Baradb, ...)` code and the `.table(...).get()` API must keep compiling and behaving the same way. +5. **Fail fast, diagnose clearly.** Distinguish I/O errors, protocol framing errors, server errors, auth errors, and pool timeouts. + +## 4. Architecture + +``` +┌─────────────────────────────────────────┐ +│ clients/nim-allographer │ +│ - allographer query builder / schema │ +│ - BaradbConnections pool wrapper │ +│ - migration helpers │ +│ - thin re-export of baradb/client │ +└──────────────┬──────────────────────────┘ + │ requires "baradb >= 1.2.0" +┌──────────────▼──────────────────────────┐ +│ clients/nim (canonical package) │ +│ - wire.nim (protocol constants) │ +│ - client.nim (async/sync client) │ +│ - pool.nim (async connection pool)│ +│ - http.nim (optional HTTP client) │ +│ - errors.nim (exception hierarchy) │ +└─────────────────────────────────────────┘ +``` + +### 4.1 Files in `clients/nim/src/baradb/` + +| File | Responsibility | +|------|----------------| +| `wire.nim` | `FieldKind`, `MsgKind`, `WireValue`, serialize/deserialize, `buildMessage`. | +| `client.nim` | `BaraClient`, `SyncClient`, `ClientConfig`, `QueryResult`, query/exec/auth/ping/close. | +| `pool.nim` | `BaraPool`, `PooledClient`, `withClient` template, pool stats, idle/lifetime eviction. | +| `http.nim` | Optional `BaraHttpClient` that posts queries to the HTTP/REST endpoint. | +| `errors.nim` | `BaraError`, `BaraProtocolError`, `BaraServerError`, `BaraAuthError`, `BaraPoolTimeoutError`. | + +### 4.2 Files in `clients/nim-allographer/src/allographer/query_builder/libs/baradb/` + +| File | Responsibility | +|------|----------------| +| `baradb_client.nim` | Re-exports needed types from `baradb/client` and keeps only allographer-specific helpers (migration SQL builders). The wire code is removed. | +| `baradb_types.nim` | Keeps `BaradbConnections`, `BaradbQuery`, pool bookkeeping, but references `BaraClient` from the canonical package. | +| `baradb_open.nim` | `dbOpen` constructors; may create either a `BaraPool` or keep the current simple pool, depending on migration step. | +| `baradb_exec.nim` / `baradb_query.nim` / `baradb_transaction.nim` | Unchanged API surface; internally use the canonical client. | + +## 5. Low-Level Client Improvements + +### 5.1 Typed `WireValue` rows + +`QueryResult` gains a typed view: + +```nim +type + QueryResult* = object + columns*: seq[string] + columnTypes*: seq[FieldKind] + rows*: seq[seq[string]] # legacy string view + typedRows*: seq[seq[WireValue]] # new typed view + rowCount*: int + affectedRows*: int + executionTimeMs*: float64 + lastInsertId*: int64 +``` + +`wireValueToString` stays for backward compatibility. `typedRows` is populated during deserialization and lets callers inspect vectors, JSON, bytes, etc., without string parsing. + +### 5.2 Per-connection request queue + +A single `BaraClient` must be safe when multiple async fibers call `query`/`exec` on it. Add an internal queue: + +```nim +type + BaraClient* = ref object + config: ClientConfig + socket: AsyncSocket + connected: bool + requestId: uint32 + sendLock: AsyncLock # or a Future chain queue + pending: Deque[PendingRequest] +``` + +Design choice: **serialize sends and reads per connection**. This matches the Python/JS clients and is simple to reason about. It is not pipelining; it is request/response queueing. If higher throughput is needed later, add pipelining on top of the pool. + +### 5.3 Connection pool + +`BaraPool` is an async pool with: + +- `minConnections`, `maxConnections` +- `maxIdleTime`, `maxLifetime` +- `connectTimeout`, `queryTimeout` +- `withClient` template / proc that borrows a connection, runs an async callback, and returns it +- `stats(): (total, idle, inUse)` +- Eviction of expired/stale connections +- Health check via `ping` before lending + +The sync API gets a matching `SyncPool` that uses a blocking socket and a `Lock`. + +### 5.4 TLS + +`ClientConfig` gets optional TLS fields: + +```nim + ClientConfig* = object + host*: string + port*: int + database*: string + username*: string + password*: string + timeoutMs*: int + maxRetries*: int + ssl*: bool + sslContext*: SslContext # optional, user-supplied +``` + +If `ssl` is true and no `sslContext` is supplied, the client creates a default `net.newContext()` and wraps the socket. TLS uses Nim's stdlib `net`/`asyncnet` OpenSSL wrappers. The standalone client remains stdlib-only at the Nim level, but the host must provide the OpenSSL system libraries. + +### 5.5 Timeouts and reconnect + +- `connect` honors `timeoutMs` via `asyncdispatch.withTimeout`. +- `recv` is wrapped with `withTimeout` using `timeoutMs`. +- If a send/recv fails with `ECONNRESET` or a timeout and `maxRetries > 0`, the client closes the socket, reconnects, and retries the request once. Retries are not attempted for server-side errors (`mkError`). + +### 5.6 Batch and transactions + +The protocol defines `mkBatch` and `mkTransaction`, but their server-side semantics are not stable enough in the current codebase. The client will expose: + +```nim +proc batch*(client: BaraClient, queries: seq[string]): Future[seq[QueryResult]] +proc transaction*(client: BaraClient, body: proc(): Future[void]): Future[void] +``` + +The first implementation will use explicit SQL `BEGIN`/`COMMIT`/`ROLLBACK` over a single borrowed connection (via the pool). When `mkBatch`/`mkTransaction` server support is verified, the implementation can switch to the native messages without changing the public API. + +### 5.7 HTTP fallback (optional module) + +`baradb/http` provides `BaraHttpClient` that sends JSON `{"query": ...}` to `POST /api/query` (HTTP endpoint, default port `TCP+440`) and parses the JSON response. Useful for environments where only the HTTP port is open or for debugging. Not loaded by default. + +## 6. Allographer Integration Plan + +1. **Add `requires "baradb >= 1.2.0"` to `clients/nim-allographer/allographer.nimble`.** +2. **Replace `baradb_client.nim` wire code with re-exports.** Keep the allographer-specific query builder and migration helpers. +3. **Update `baradb_types.nim`.** `Connection.client` stays `BaraClient`; remove local duplicates of `ClientConfig`, `WireValue`, `QueryResult`. +4. **Keep the current pool or migrate to `BaraPool`.** Phase 1: keep the existing `Connections` pool because the allographer query builder relies on its busy-flag semantics. Phase 2 (optional): replace it with `BaraPool.withClient` to reduce code. +5. **Use typed rows internally.** Update `toJson(resultSet)` in `baradb_exec.nim` to read from `resultSet.typedRows` instead of parsing strings. This fixes wrong JSON/vector/int parsing. +6. **Fix `formatSql` / prepared statements.** The current code sometimes builds SQL by string concatenation in `baradb_exec.nim`; ensure all user input goes through `mkQueryParams` so the server handles parameter binding. + +## 7. Public API Sketch + +### Standalone async + +```nim +import asyncdispatch, baradb/client, baradb/pool + +proc main() {.async.} = + let cfg = ClientConfig(host: "127.0.0.1", port: 9472, timeoutMs: 30_000) + let pool = newBaraPool(cfg, minConnections = 2, maxConnections = 10) + await withClient(pool) do (c: BaraClient) -> Future[void]: + let r = await c.query("SELECT name, age FROM users WHERE age > ?", + @[WireValue(kind: fkInt32, int32Val: 18)]) + echo r.typedRows + +waitFor main() +``` + +### Standalone sync + +```nim +import baradb/client + +let c = newSyncClient() +c.connect() +let r = c.query("SELECT * FROM users") +echo r.rows +c.close() +``` + +### Allographer (unchanged) + +```nim +import allographer/connection, allographer/query_builder + +let rdb = dbOpen(Baradb, "default", "admin", "", "127.0.0.1", 9472, + maxConnections = 5) + +proc main() {.async.} = + let users = await rdb.table("users").select("id", "name").get() + echo users + +waitFor main() +``` + +## 8. Data Flow + +1. Caller invokes `await pool.withClient(...)` or `await client.query(sql, params)`. +2. The request is enqueued on the connection (or the pool lends a free connection). +3. The queue serializes the request: send header + payload, wait for response. +4. The response loop reads the 12-byte header, then the payload, then any trailing `mkComplete`. +5. `mkData` payloads are deserialized into `typedRows`; `rows` is populated via `wireValueToString`. +6. Server errors (`mkError`) raise `BaraServerError` with code and message. +7. The connection is returned to the pool. + +## 9. Error Handling + +| Exception | When raised | Retry? | +|-----------|-------------|--------| +| `BaraError` | Base type | depends | +| `BaraProtocolError` | Bad framing, unexpected message kind | no | +| `BaraServerError` | Server replied with `mkError` | no | +| `BaraAuthError` | Auth failed / rejected | no | +| `BaraIoError` | Connection lost, timeout, ECONNREFUSED | yes (up to `maxRetries`) | +| `BaraPoolTimeoutError` | No connection available within `timeoutMs` | no | + +All exceptions inherit from `BaraError` so callers can catch a single type. + +## 10. Testing Strategy + +1. **Mock async TCP server in `clients/nim/tests/test_wire.nim`.** Verifies framing, request/response serialization, and the request queue without a real BaraDB instance. +2. **Property/round-trip tests for `WireValue`.** Serialize then deserialize random values and compare. +3. **Pool unit tests.** Check acquire/release, max size, eviction, and timeout behavior using a mock client factory. +4. **Integration tests.** Reuse `clients/nim/tests/test_integration.nim` and `clients/nim-allographer/tests/baradb/*`; run them when a server is available on `localhost:9472`. +5. **Allographer regression tests.** Ensure existing tests still pass after switching to the canonical client. + +## 11. Migration & Rollout + +1. Release `clients/nim` as `baradb 1.2.0` with the new modules. +2. Update `clients/nim-allographer` to depend on `baradb >= 1.2.0` and remove the duplicated wire code. +3. Mark `src/barabadb/client/client.nim` as deprecated with a `{.deprecated.}` pragma pointing to `baradb/client`. +4. Document the new pool and typed-row APIs in `clients/nim/README.md` and `docs/en/clients.md`. + +## 12. Risks & Mitigations + +| Risk | Mitigation | +|------|------------| +| Breaking allographer tests | Run the full `tests/baradb/` suite after every change; keep API unchanged. | +| TLS support depends on OpenSSL Nim wrapper | Gate TLS behind `when defined(ssl)` and document how to build with `-d:ssl`. | +| Async request queue hurts throughput | Benchmark before/after; add per-request `requestId` matching and pipelining later if needed. | +| Server `mkBatch`/`mkTransaction` semantics unclear | Implement via SQL `BEGIN`/`COMMIT` first; switch to native messages later. | +| Nim 2.0 vs 2.2 compatibility | Keep code compatible with Nim 2.0; test on both versions in CI. | + +## 13. Open Questions + +1. Should `clients/nim-allographer` keep its own pool (Phase 1) or switch fully to `BaraPool` (Phase 2)? + *Recommendation:* Phase 1 keeps risk low; Phase 2 can be done after the canonical pool is proven stable. +2. Should the HTTP fallback be part of the `baradb` package or a separate `baradb_http` package? + *Recommendation:* Keep it as an optional module `baradb/http` inside the same package so `import baradb/http` is explicit and does not pull in `httpclient` for users who do not need it. diff --git a/src/barabadb/client/client.nim b/src/barabadb/client/client.nim index c9bc642..08deeb3 100644 --- a/src/barabadb/client/client.nim +++ b/src/barabadb/client/client.nim @@ -1,3 +1,4 @@ +{.deprecated: "Use the canonical baradb/client from clients/nim instead.".} ## BaraDB Client — Nim client library import std/asyncdispatch import std/asyncnet