update core modules and project config

baradadb.nimble

@@ -7,6 +7,8 @@ srcDir        = "src"
 bin           = @["baradadb"]
 binDir        = "build"
 
+switch("define", "ssl")
+
 # Dependencies
 requires "nim >= 2.2.0"
 requires "https://github.com/katehonz/hunos >= 1.2.0"

src/barabadb/core/httpserver.nim

@@ -58,10 +58,10 @@ proc newHttpServer*(config: BaraConfig): HttpServer =
 proc createToken*(server: HttpServer, userId, role: string): string =
   let header = %*{"alg": "HS256", "typ": "JWT"}
   var claims = newTable[string, Claim]()
-  claims["sub"] = newStringClaim(userId)
-  claims["role"] = newStringClaim(role)
-  claims["iat"] = newTimeClaim(getTime())
-  claims["exp"] = newTimeClaim(getTime() + 24.hours)
+  claims["sub"] = newSUB(%userId)
+  claims["role"] = newClaim(GENERAL, %role)
+  claims["iat"] = newIAT(getTime().toUnix())
+  claims["exp"] = newEXP((getTime() + 24.hours).toUnix())
   var token = initJWT(header, claims)
   token.sign(server.secretKey)
   return $token

src/barabadb/core/replication.nim

@@ -2,6 +2,9 @@
 import std/tables
 import std/sets
 import std/locks
+import std/net
+import std/strutils
+
 
 type
   ReplicationMode* = enum
@@ -69,40 +72,59 @@ proc connectReplica*(rm: ReplicationManager, id: string) =
     rm.replicas[id].connected = true
   release(rm.lock)
 
-proc disconnectReplica*(rm: ReplicationManager, id: string) =
-  acquire(rm.lock)
-  if id in rm.replicas:
-    rm.replicas[id].state = rsDisconnected
-    rm.replicas[id].connected = false
-  release(rm.lock)
+proc shipToReplica(replica: Replica, lsn: uint64, data: seq[byte]): bool =
+  ## Send replication data to a replica via TCP.
+  ## Protocol: "REP <lsn> <dataLen>\n<data>"
+  ## Response: "ACK <lsn>\n" on success
+  try:
+    var sock = newSocket()
+    sock.connect(replica.host, Port(replica.port))
+    let header = "REP " & $lsn & " " & $data.len & "\n"
+    sock.send(header)
+    if data.len > 0:
+      sock.send(cast[string](data))
+    var response = ""
+    sock.readLine(response)
+    sock.close()
+    let parts = response.strip().split(" ")
+    return parts.len >= 2 and parts[0] == "ACK"
+  except:
+    return false
 
 proc writeLsn*(rm: ReplicationManager, data: seq[byte]): uint64 =
   acquire(rm.lock)
   inc rm.currentLsn
   let lsn = rm.currentLsn
 
+  var replicasToShip: seq[Replica]
+  for id, replica in rm.replicas:
+    if replica.connected and replica.host.len > 0 and replica.port > 0:
+      replicasToShip.add(replica)
+
   case rm.mode
   of rmAsync:
-    # Fire and forget — don't wait
     release(rm.lock)
+    for replica in replicasToShip:
+      discard shipToReplica(replica, lsn, data)
     return lsn
   of rmSync:
-    # Wait for all replicas
     rm.pendingAcks[lsn] = initHashSet[string]()
-    for id, replica in rm.replicas:
-      if replica.connected:
-        rm.pendingAcks[lsn].incl(id)
+    for replica in replicasToShip:
+      rm.pendingAcks[lsn].incl(replica.id)
     release(rm.lock)
+    for replica in replicasToShip:
+      discard shipToReplica(replica, lsn, data)
     return lsn
   of rmSemiSync:
-    # Wait for N replicas
     rm.pendingAcks[lsn] = initHashSet[string]()
     var count = 0
-    for id, replica in rm.replicas:
-      if replica.connected and count < rm.syncReplicaCount:
-        rm.pendingAcks[lsn].incl(id)
+    for replica in replicasToShip:
+      if count < rm.syncReplicaCount:
+        rm.pendingAcks[lsn].incl(replica.id)
         inc count
     release(rm.lock)
+    for replica in replicasToShip:
+      discard shipToReplica(replica, lsn, data)
     return lsn
 
 proc ackLsn*(rm: ReplicationManager, replicaId: string, lsn: uint64) =

src/barabadb/core/server.nim

@@ -18,6 +18,7 @@ import ../query/executor
 import ../storage/lsm
 import ../core/mvcc
 import ../core/disttxn
+import ../core/replication
 import jwt as jwtlib
 
 type
@@ -28,6 +29,7 @@ type
     ctx*: ExecutionContext
     txnManager*: TxnManager
     distTxnManager*: DistTxnManager
+    replicationManager*: ReplicationManager
     tls*: TLSContext
     activeConnections*: int
 
@@ -46,7 +48,8 @@ proc newServer*(config: BaraConfig): Server =
     let tlsConfig = newTLSConfig(config.certFile, config.keyFile)
     tls = newTLSContext(tlsConfig)
   Server(config: config, running: false, db: db, ctx: ctx,
-         txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(), tls: tls)
+         txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(),
+         replicationManager: newReplicationManager(), tls: tls)
 
 # ----------------------------------------------------------------------
 # Wire Protocol Helpers
@@ -111,7 +114,8 @@ proc valueToWire(val: string, colType: string): WireValue =
     return WireValue(kind: fkJson, jsonVal: val)
   return WireValue(kind: fkString, strVal: val)
 
-proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq[WireValue] = @[]): (bool, QueryResult, string) =
+proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq[WireValue] = @[],
+                   replication: ReplicationManager = nil): (bool, QueryResult, string) =
   try:
     let tokens = tokenize(query)
     let astNode = parse(tokens)
@@ -121,6 +125,14 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq
 
     let result = executor.executeQuery(ctx, astNode, params)
     if result.success:
+      # Ship written key-value pairs to replicas
+      if replication != nil and result.keyValuePairs.len > 0:
+        for (key, value) in result.keyValuePairs:
+          var data = newSeq[byte](key.len + 1 + value.len)
+          for i, c in key: data[i] = byte(c)
+          data[key.len] = byte(0)
+          for i, c in value: data[key.len + 1 + i] = c
+          discard replication.writeLsn(data)
       var qr = QueryResult(affectedRows: result.affectedRows, rowCount: result.rows.len)
       qr.columns = result.columns
 
@@ -304,6 +316,38 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
           await client.send("ERR invalid message\n")
         continue
 
+      # Detect replication data (starts with "REP ")
+      if headerData.len >= 4 and headerData[0..3] == "REP ":
+        var rest = headerData[4..^1]
+        while '\n' notin rest:
+          let more = await client.recv(1024)
+          if more.len == 0: break
+          rest.add(more)
+        let parts = rest.strip().split(" ")
+        if parts.len >= 2:
+          let lsn = try: parseUInt(parts[0]) except: 0'u64
+          let dataLen = try: parseInt(parts[1]) except: 0
+          if dataLen > 0:
+            var data = ""
+            while data.len < dataLen:
+              let chunk = await client.recv(dataLen - data.len)
+              if chunk.len == 0: break
+              data.add(chunk)
+            # Apply replicated data to database
+            if data.len > 0:
+              let nullPos = data.find('\0')
+              if nullPos >= 0:
+                let key = data[0..<nullPos]
+                let value = data[nullPos+1..^1]
+                if value.len > 0:
+                  server.db.put(key, cast[seq[byte]](value))
+                else:
+                  server.db.delete(key)
+          await client.send("ACK " & $lsn & "\n")
+        else:
+          await client.send("ERR\n")
+        continue
+
       let (ok, header) = parseHeader(headerData)
       if not ok:
         break
@@ -338,7 +382,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
         info("[" & $clientId & "] Query: " & queryStr)
 
         let startTicks = getMonoTime().ticks()
-        let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr)
+        let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, replication=server.replicationManager)
         let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000)
 
         if durationMs >= slowThreshold:
@@ -359,7 +403,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
         info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)")
 
         let startTicks = getMonoTime().ticks()
-        let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, params)
+        let (success, result, errorMsg) = executeQuery(server.db, connCtx, queryStr, params, replication=server.replicationManager)
         let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000)
 
         if durationMs >= slowThreshold:

src/barabadb/query/executor.nim

@@ -118,9 +118,12 @@ type
     rows*: seq[Row]
     affectedRows*: int
     message*: string
+    keyValuePairs*: seq[(string, seq[byte])]
 
-proc okResult*(rows: seq[Row] = @[], cols: seq[string] = @[], affected: int = 0, msg: string = ""): ExecResult =
-  ExecResult(success: true, columns: cols, rows: rows, affectedRows: affected, message: msg)
+proc okResult*(rows: seq[Row] = @[], cols: seq[string] = @[], affected: int = 0, msg: string = "",
+               kvPairs: seq[(string, seq[byte])] = @[]): ExecResult =
+  ExecResult(success: true, columns: cols, rows: rows, affectedRows: affected, message: msg,
+             keyValuePairs: kvPairs)
 
 proc errResult*(msg: string): ExecResult =
   ExecResult(success: false, columns: @[], rows: @[], affectedRows: 0, message: msg)
@@ -688,7 +691,8 @@ proc execPointRead(ctx: ExecutionContext, table: string, key: string): seq[Row]
     return @[row]
   return @[]
 
-proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], values: seq[seq[string]]): int =
+proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], values: seq[seq[string]],
+                  kvPairs: var seq[(string, seq[byte])]): int =
   if not hasPrivilege(ctx, table, "INSERT"):
     return 0
   var count = 0
@@ -720,6 +724,7 @@ proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], valu
       discard ctx.txnManager.write(ctx.pendingTxn, fullKey, cast[seq[byte]](valStr))
     else:
       ctx.db.put(fullKey, cast[seq[byte]](valStr))
+      kvPairs.add((fullKey, cast[seq[byte]](valStr)))
 
     for colName in ctx.btrees.keys.toSeq():
       if colName.startsWith(table & "."):
@@ -746,7 +751,8 @@ proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], valu
     inc count
   return count
 
-proc execDelete*(ctx: ExecutionContext, table: string, key: string): int =
+proc execDelete*(ctx: ExecutionContext, table: string, key: string,
+                  kvPairs: var seq[(string, seq[byte])]): int =
   if not hasPrivilege(ctx, table, "DELETE"):
     return 0
   let fullKey = table & "." & key
@@ -763,6 +769,7 @@ proc execDelete*(ctx: ExecutionContext, table: string, key: string): int =
       discard ctx.txnManager.delete(ctx.pendingTxn, fullKey)
     else:
       ctx.db.delete(fullKey)
+      kvPairs.add((fullKey, @[]))
     # Update FTS indexes
     for ftsKey, ftsIdx in ctx.ftsIndexes:
       if ftsKey.startsWith(table & "."):
@@ -773,7 +780,8 @@ proc execDelete*(ctx: ExecutionContext, table: string, key: string): int =
     return 1
   return 0
 
-proc execUpdateRow*(ctx: ExecutionContext, table: string, key: string, sets: Table[string, string]): int =
+proc execUpdateRow*(ctx: ExecutionContext, table: string, key: string, sets: Table[string, string],
+                     kvPairs: var seq[(string, seq[byte])]): int =
   if not hasPrivilege(ctx, table, "UPDATE"):
     return 0
   let fullKey = table & "." & key
@@ -819,6 +827,7 @@ proc execUpdateRow*(ctx: ExecutionContext, table: string, key: string, sets: Tab
     discard ctx.txnManager.write(ctx.pendingTxn, fullKey, cast[seq[byte]](newVal))
   else:
     ctx.db.put(fullKey, cast[seq[byte]](newVal))
+    kvPairs.add((fullKey, cast[seq[byte]](newVal)))
   # Update FTS indexes: remove old doc, add new
   for ftsKey, ftsIdx in ctx.ftsIndexes:
     if ftsKey.startsWith(table & "."):
@@ -1953,7 +1962,8 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue]
         row[f] = mutableValues[0][i]
     fireTriggers(ctx, stmt.insTarget, "before", "insert", row)
 
-    let count = execInsert(ctx, stmt.insTarget, mutableFields, mutableValues)
+    var kvPairs: seq[(string, seq[byte])]
+    let count = execInsert(ctx, stmt.insTarget, mutableFields, mutableValues, kvPairs)
 
     # Fire AFTER INSERT triggers
     fireTriggers(ctx, stmt.insTarget, "after", "insert", row)
@@ -1961,7 +1971,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue]
     if ctx.onChange != nil:
       for i in 0..<count:
         ctx.onChange(ChangeEvent(table: stmt.insTarget, kind: ckInsert, key: "", data: ""))
-    return okResult(affected=count)
+    return okResult(affected=count, kvPairs=kvPairs)
 
   of nkUpdate:
     if stmt.updSet.len == 0: return okResult()
@@ -1979,6 +1989,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue]
     # Scan and apply
     let rows = execScan(ctx, stmt.updTarget)
     var count = 0
+    var kvPairs: seq[(string, seq[byte])]
     for row in rows:
       # Check WHERE
       if stmt.updWhere != nil and stmt.updWhere.whereExpr != nil:
@@ -2007,19 +2018,20 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue]
           newRow[col] = val
         fireTriggers(ctx, stmt.updTarget, "before", "update", oldRow)
 
-        count += execUpdateRow(ctx, stmt.updTarget, row["$key"], sets)
+        count += execUpdateRow(ctx, stmt.updTarget, row["$key"], sets, kvPairs)
 
         # Fire AFTER UPDATE triggers
         fireTriggers(ctx, stmt.updTarget, "after", "update", newRow)
 
         if ctx.onChange != nil:
           ctx.onChange(ChangeEvent(table: stmt.updTarget, kind: ckUpdate, key: old, data: ""))
-    return okResult(affected=count)
+    return okResult(affected=count, kvPairs=kvPairs)
 
   of nkDelete:
     # Delete all rows matching WHERE
     let rows = execScan(ctx, stmt.delTarget)
     var count = 0
+    var kvPairs: seq[(string, seq[byte])]
     for row in rows:
       if stmt.delWhere != nil and stmt.delWhere.whereExpr != nil:
         let whereExpr = lowerExpr(stmt.delWhere.whereExpr)
@@ -2029,14 +2041,14 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue]
         # Fire BEFORE DELETE triggers
         fireTriggers(ctx, stmt.delTarget, "before", "delete", row)
 
-        count += execDelete(ctx, stmt.delTarget, row["$key"])
+        count += execDelete(ctx, stmt.delTarget, row["$key"], kvPairs)
 
         # Fire AFTER DELETE triggers
         fireTriggers(ctx, stmt.delTarget, "after", "delete", row)
 
         if ctx.onChange != nil:
           ctx.onChange(ChangeEvent(table: stmt.delTarget, kind: ckDelete, key: old, data: ""))
-    return okResult(affected=count)
+    return okResult(affected=count, kvPairs=kvPairs)
 
   of nkCreateTable:
     var tbl = TableDef(name: stmt.crtName, columns: @[], pkColumns: @[],
@@ -2125,12 +2137,14 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue]
 
   of nkCommitTxn:
     if ctx.pendingTxn != nil and ctx.pendingTxn.state == tsActive:
+      var kvPairs: seq[(string, seq[byte])]
       for key, version in ctx.pendingTxn.writeSet:
         if version.value == @[]: ctx.db.delete(key)
         else: ctx.db.put(key, version.value)
+        kvPairs.add((key, version.value))
       discard ctx.txnManager.commit(ctx.pendingTxn)
       ctx.pendingTxn = nil
-      return okResult(msg="Transaction committed")
+      return okResult(msg="Transaction committed", kvPairs=kvPairs)
     return errResult("No active transaction to commit")
 
   of nkRollbackTxn: