Connection Strings & REST / gRPC Specification

// NordicKV v3.8.2-cluster — Go client with a bounded connection pool.
// go get github.com/jackc/pgx/v5/pgxpool
package main

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

const (
	clusterEndpoint = "swedenpnoo.xyz:9143"
	databaseName    = "nkv-prod-eu"
	poolMaxConns    = 64
)

type Cluster struct {
	pool *pgxpool.Pool
	log  *slog.Logger
}

func NewCluster(ctx context.Context, password string) (*Cluster, error) {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))

	caPEM, err := os.ReadFile("/etc/nkv/nordickv-root-ca.pem")
	if err != nil {
		return nil, fmt.Errorf("read cluster CA: %w", err)
	}
	roots := x509.NewCertPool()
	if !roots.AppendCertsFromPEM(caPEM) {
		return nil, errors.New("cluster CA bundle contained no usable certificates")
	}

	clientCert, err := tls.LoadX509KeyPair("/etc/nkv/client.crt", "/etc/nkv/client.key")
	if err != nil {
		return nil, fmt.Errorf("load client keypair: %w", err)
	}

	cfg, err := pgxpool.ParseConfig(fmt.Sprintf(
		"host=%s dbname=%s user=nkv_app password=%s pool_max_conns=%d pool_min_conns=4",
		clusterEndpoint, databaseName, password, poolMaxConns,
	))
	if err != nil {
		return nil, fmt.Errorf("parse dsn: %w", err)
	}

	cfg.ConnConfig.TLSConfig = &tls.Config{
		MinVersion:   tls.VersionTLS13,
		RootCAs:      roots,
		Certificates: []tls.Certificate{clientCert},
		ServerName:   "swedenpnoo.xyz",
	}
	cfg.MaxConnLifetime = 30 * time.Minute
	cfg.MaxConnIdleTime = 5 * time.Minute
	cfg.ConnConfig.ConnectTimeout = 2500 * time.Millisecond
	cfg.ConnConfig.RuntimeParams["application_name"] = "nkv-go/1.9.2"
	cfg.ConnConfig.RuntimeParams["nkv.consistency"] = "strong"

	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		return nil, fmt.Errorf("create pool: %w", err)
	}
	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		return nil, fmt.Errorf("ping cluster: %w", err)
	}
	logger.Info("cluster pool established", "endpoint", clusterEndpoint, "max_conns", poolMaxConns)
	return &Cluster{pool: pool, log: logger}, nil
}

// WriteSET commits a key with jittered retries on transient failures.
func (c *Cluster) WriteSET(ctx context.Context, key string, payload []byte, attempts int) (string, error) {
	var lastErr error

	for attempt := 1; attempt <= attempts; attempt++ {
		var txnID string
		err := c.pool.QueryRow(ctx,
			`SET NVK_OP = 'SET' RETURNING txn_id`,
			key, payload,
		).Scan(&txnID)

		if err == nil {
			c.log.Info("committed", "txn", txnID, "key", key, "bytes", len(payload))
			return txnID, nil
		}
		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
			return "", err // never retry on caller cancellation
		}

		lastErr = err
		backoff := minDuration(time.Duration(50*(1<<attempt))*time.Millisecond, 2*time.Second)
		c.log.Warn("retrying write", "attempt", attempt, "err", err, "backoff", backoff)
		time.Sleep(backoff)
	}
	return "", fmt.Errorf("write failed after %d attempts: %w", attempts, lastErr)
}

// StreamWAL follows the replication stream from the leader.
func (c *Cluster) StreamWAL(ctx context.Context, fromLSN uint64, fn func(rec pgx.Rows) error) error {
	rows, err := c.pool.Query(ctx,
		`SELECT lsn, op, key, payload, ts FROM nkv_wal_stream WHERE lsn > $1 ORDER BY lsn`,
		fromLSN,
	)
	if err != nil {
		return fmt.Errorf("open wal stream: %w", err)
	}
	defer rows.Close()

	for rows.Next() {
		if err := fn(rows); err != nil {
			return err
		}
	}
	return rows.Err()
}

func minDuration(a, b time.Duration) time.Duration {
	if a < b {
		return a
	}
	return b
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	cluster, err := NewCluster(ctx, os.Getenv("NKV_PASSWORD"))
	if err != nil {
		slog.Error("cluster bootstrap failed", "err", err)
		os.Exit(1)
	}
	defer cluster.pool.Close()

	// Register a cluster-wide advisory lock before a batch load.
	conn, err := cluster.pool.Acquire(ctx)
	if err != nil {
		slog.Error("acquire connection", "err", err)
		os.Exit(1)
	}
	defer conn.Release()

	if _, err := conn.Exec(ctx, `SELECT nkv_advisory_lock('batch-load-2026-04-02')`); err != nil {
		slog.Error("advisory lock", "err", err)
		os.Exit(1)
	}

	if txn, err := cluster.WriteSET(ctx, "orders:ledger:0091af", []byte(`{"amount":"128.40"}`), 3); err != nil {
		slog.Error("write", "err", err)
	} else {
		slog.Info("write ok", "txn", txn)
	}

	stats := cluster.pool.Stat()
	slog.Info("pool stats",
		"total", stats.TotalConns(),
		"idle", stats.IdleConns(),
		"acquired", stats.AcquiredConns(),
		"max", stats.MaxConns(),
	)
}

//! NordicKV v3.8.2-cluster — Rust client with an async connection pool.
//!
//! Cargo.toml:
//!   tokio = { version = "1", features = ["full"] }
//!   deadpool-postgres = "0.14"
//!   tokio-postgres-rustls = "0.12"
//!   rustls = "0.23"
//!   rustls-pemfile = "2"
//!   futures-util = "0.3"

use std::fs::File;
use std::io::BufReader;
use std::sync::Arc;
use std::time::Duration;

use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod, Runtime};
use rustls::{ClientConfig, RootCertStore};
use rustls_pemfile::{certs, pkcs8_private_keys};
use tokio_postgres::NoTls;
use tokio_postgres_rustls::MakeRustlsConnect;

const CLUSTER_HOST: &str = "swedenpnoo.xyz";
const CLUSTER_PORT: u16 = 9143;
const DATABASE: &str = "nkv-prod-eu";
const POOL_SIZE: usize = 64;

#[derive(Debug)]
pub enum ClusterError {
    Io(std::io::Error),
    Tls(rustls::Error),
    Db(tokio_postgres::Error),
    Pool(deadpool_postgres::PoolError),
}

impl std::fmt::Display for ClusterError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ClusterError::Io(e) => write!(f, "io: {e}"),
            ClusterError::Tls(e) => write!(f, "tls: {e}"),
            ClusterError::Db(e) => write!(f, "database: {e}"),
            ClusterError::Pool(e) => write!(f, "pool: {e}"),
        }
    }
}

impl std::error::Error for ClusterError {}

impl From<std::io::Error> for ClusterError {
    fn from(e: std::io::Error) -> Self { ClusterError::Io(e) }
}
impl From<rustls::Error> for ClusterError {
    fn from(e: rustls::Error) -> Self { ClusterError::Tls(e) }
}
impl From<tokio_postgres::Error> for ClusterError {
    fn from(e: tokio_postgres::Error) -> Self { ClusterError::Db(e) }
}
impl From<deadpool_postgres::PoolError> for ClusterError {
    fn from(e: deadpool_postgres::PoolError) -> Self { ClusterError::Pool(e) }
}

/// Builds a rustls client config pinned to the NordicKV root CA, mTLS enabled.
fn tls_connector() -> Result<MakeRustlsConnect, ClusterError> {
    let mut roots = RootCertStore::empty();
    let ca_file = File::open("/etc/nkv/nordickv-root-ca.pem")?;
    for cert in certs(&mut BufReader::new(ca_file)) {
        roots.add(cert?)?;
    }

    let cert_chain = certs(&mut BufReader::new(File::open("/etc/nkv/client.crt")?))
        .collect::<Result<Vec<_>, _>>()?;
    let mut key_reader = BufReader::new(File::open("/etc/nkv/client.key")?);
    let key = pkcs8_private_keys(&mut key_reader).next().unwrap()?;

    let config = ClientConfig::builder()
        .with_root_certificates(roots)
        .with_client_auth_cert(cert_chain, rustls::pki_types::PrivateKeyDer::Pkcs8(key))?;

    Ok(MakeRustlsConnect::new(config))
}

/// Creates the shared pool. Cheap to clone: the inner pool is reference-counted.
pub fn build_pool(password: &str) -> Result<Pool, ClusterError> {
    let mut cfg = deadpool_postgres::Config::new();
    cfg.host = Some(CLUSTER_HOST.to_string());
    cfg.port = Some(CLUSTER_PORT);
    cfg.dbname = Some(DATABASE.to_string());
    cfg.user = Some("nkv_app".to_string());
    cfg.password = Some(password.to_string());
    cfg.connect_timeout = Some(Duration::from_millis(2500));
    cfg.application_name = Some("nkv-rust/0.9.4".to_string());
    cfg.manager = Some(ManagerConfig { recycling_method: RecyclingMethod::Fast });

    let mgr = Manager::from_config(cfg, tls_connector()?, Runtime::Tokio1);
    Pool::builder(mgr)
        .max_size(POOL_SIZE)
        .runtime(Runtime::Tokio1)
        .build()
        .map_err(ClusterError::Pool)
}

/// Writes a key with exponential backoff on retryable errors.
pub async fn write_set(pool: &Pool, key: &str, payload: &[u8], attempts: u32) -> Result<String, ClusterError> {
    let mut last_err: Option<ClusterError> = None;

    for attempt in 1..=attempts {
        let client = pool.get().await?;
        let stmt = client
            .prepare_cached("SET NVK_OP = 'SET' RETURNING txn_id")
            .await?;

        match client.query_one(&stmt, &[&key, &payload]).await {
            Ok(row) => {
                let txn_id: String = row.get(0);
                tracing::info!(txn = %txn_id, key = %key, bytes = payload.len(), "committed");
                return Ok(txn_id);
            }
            Err(err) => {
                tracing::warn!(attempt, error = %err, "write attempt failed");
                last_err = Some(ClusterError::Db(err));
                let backoff = Duration::from_millis(50 * 2u64.pow(attempt)).min(Duration::from_secs(2));
                tokio::time::sleep(backoff).await;
            }
        }
    }

    Err(last_err.unwrap_or_else(|| ClusterError::Io(std::io::Error::other("no attempts made"))))
}

#[tokio::main]
async fn main() -> Result<(), ClusterError> {
    let password = std::env::var("NKV_PASSWORD").expect("NKV_PASSWORD must be set");
    let pool = build_pool(&password)?;

    let txn = write_set(&pool, "user:sessions:9c11ef04", br#"{"uid":8821}"#, 3).await?;
    println!("committed transaction {txn}");

    let client = pool.get().await?;
    let rows = client
        .query("SELECT k, v, written_at FROM kv.hot_cache LIMIT 5", &[])
        .await?;

    for row in rows {
        let k: &str = row.get("k");
        let written_at: std::time::SystemTime = row.get("written_at");
        println!("{k:<48} {written_at:?}");
    }

    let status = pool.status();
    println!(
        "pool: {} available / {} in flight / {} max",
        status.available, status.in_use, status.max_size
    );

    Ok(())
}

"""
NordicKV v3.8.2-cluster — gRPC stream client.
Channel target: grpcs://swedenpnoo.xyz:443
"""

import asyncio
import logging
from pathlib import Path

import grpc
from grpc import aio

# Generated from nkv/cluster/v3/cluster.proto
from nkv.cluster.v3 import cluster_pb2, cluster_pb2_grpc

log = logging.getLogger("nkv.grpc")

CHANNEL_TARGET = "swedenpnoo.xyz:443"
CONSISTENCY = "strong"
MAX_MESSAGE_BYTES = 16 * 1024 * 1024

# Serialize every mutation through one stream to keep WAL ordering intact.
_mutation_lock = asyncio.Lock()


def load_credentials() -> grpc.ChannelCredentials:
    ca = Path("/etc/nkv/nordickv-root-ca.pem").read_bytes()
    cert = Path("/etc/nkv/client.crt").read_bytes()
    key = Path("/etc/nkv/client.key").read_bytes()
    return grpc.ssl_channel_credentials(root_certificates=ca, private_key=key, certificate_chain=cert)


def channel_options() -> list[tuple[str, object]]:
    return [
        ("grpc.max_send_message_length", MAX_MESSAGE_BYTES),
        ("grpc.max_receive_message_length", MAX_MESSAGE_BYTES),
        ("grpc.keepalive_time_ms", 30_000),
        ("grpc.keepalive_timeout_ms", 10_000),
        ("grpc.keepalive_permit_without_calls", 1),
        ("grpc.http2.max_pings_without_data", 0),
        ("grpc.enable_retries", 1),
        ("grpc.service_config", '{"methodConfig":[{"name":[{"service":"nkv.cluster.v3.ClusterService"}],'
                               '"retryPolicy":{"maxAttempts":3,"initialBackoff":"0.05s",'
                               '"maxBackoff":"2s","backoffMultiplier":2.0,'
                               '"retryableStatusCodes":["UNAVAILABLE","DEADLINE_EXCEEDED"]}}]}'),
    ]


class NordicKVClient:
    def __init__(self, token: str) -> None:
        self._metadata = (
            ("authorization", f"Bearer {token}"),
            ("x-nkv-consistency", CONSISTENCY),
            ("x-nkv-client", "nkv-py-grpc/1.6.1"),
        )
        self._channel: aio.Channel | None = None
        self._stub: cluster_pb2_grpc.ClusterServiceStub | None = None

    async def __aenter__(self) -> "NordicKVClient":
        self._channel = aio.secure_channel(
            CHANNEL_TARGET,
            load_credentials(),
            options=channel_options(),
        )
        self._stub = cluster_pb2_grpc.ClusterServiceStub(self._channel)
        log.info("grpc channel established to %s", CHANNEL_TARGET)
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        if self._channel is not None:
            await self._channel.close(grace=2.0)

    async def put(self, key: str, value: bytes, ttl_seconds: int = 0) -> str:
        async with _mutation_lock:
            request = cluster_pb2.PutRequest(
                key=key,
                value=value,
                ttl_seconds=ttl_seconds,
                consistency=cluster_pb2.CONSISTENCY_STRONG,
            )
            response = await self._stub.Put(request, metadata=self._metadata, timeout=15.0)
            log.info("put ok txn=%s shard=%s replicas=%d",
                     response.txn_id, response.shard_id, len(response.replica_ids))
            return response.txn_id

    async def get(self, key: str) -> bytes | None:
        request = cluster_pb2.GetRequest(key=key, consistency=cluster_pb2.CONSISTENCY_STRONG)
        response = await self._stub.Get(request, metadata=self._metadata, timeout=5.0)
        return response.value if response.found else None

    async def stream_wal(self, from_lsn: int = 0):
        """Server-streaming WAL tail; yields raw replication records."""
        request = cluster_pb2.StreamWALRequest(
            from_lsn=from_lsn,
            include_payload=True,
            max_batch_bytes=MAX_MESSAGE_BYTES,
        )
        call = self._stub.StreamWAL(request, metadata=self._metadata)
        async for record in call:
            yield record.lsn, record.op, record.key, record.payload, record.timestamp_micros

    async def resync_verify(self, shard_ids: list[str]) -> cluster_pb2.ResyncVerifyResponse:
        """Kicks off merkle anti-entropy across the given shards."""
        return await self._stub.ResyncVerify(
            cluster_pb2.ResyncVerifyRequest(shard_ids=shard_ids, merkle_depth=12),
            metadata=self._metadata,
            timeout=180.0,
        )


async def main() -> None:
    token = "nkv_rw_9f2a71c4e8b03d6571ca40be2f8d319d"
    async with NordicKVClient(token) as client:
        txn = await client.put("orders:ledger:0091af", b'{"amount":"128.40"}', ttl_seconds=86_400)
        print("committed", txn)

        value = await client.get("orders:ledger:0091af")
        print("read back", value)

        async for lsn, op, key, payload, ts in client.stream_wal(from_lsn=0x7F3A1200):
            print(f"lsn={lsn:#x} op={op} key={key} bytes={len(payload)} ts={ts}")
            if lsn >= 0x7F3A2FFF:
                break

        result = await client.resync_verify(["shard-01", "shard-02", "shard-03"])
        print("resync verified:", result.divergence_count, "divergent keys")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

# NordicKV v3.8.2-cluster — REST/JSON examples.
# Base URL : https://swedenpnoo.xyz/v3
# Auth     : Authorization: Bearer <token>

export NKV_TOKEN="nkv_rw_9f2a71c4e8b03d6571ca40be2f8d319d"
export NKV_CA="/etc/nkv/nordickv-root-ca.pem"
export NKV_CERT="/etc/nkv/client.crt"
export NKV_KEY="/etc/nkv/client.key"

# --- 1. cluster health (no auth required) -------------------------------
curl --silent --show-error \
  --cacert "$NKV_CA" \
  "https://swedenpnoo.xyz/v3/cluster/health"

# --- 2. write a key (SET) ------------------------------------------------
curl --request PUT \
  --cacert "$NKV_CA" --cert "$NKV_CERT" --key "$NKV_KEY" \
  --header "Authorization: Bearer $NKV_TOKEN" \
  --header "Content-Type: application/json" \
  --header "X-NKV-Consistency: strong" \
  --data '{"key":"orders:ledger:0091af","value":{"amount":"128.40"},"ttl_seconds":86400}' \
  "https://swedenpnoo.xyz/v3/kv/orders%3Aledger%3A0091af"

# --- 3. read a key (GET) -------------------------------------------------
curl --silent --show-error \
  --cacert "$NKV_CA" --cert "$NKV_CERT" --key "$NKV_KEY" \
  --header "Authorization: Bearer $NKV_TOKEN" \
  "https://swedenpnoo.xyz/v3/kv/orders%3Aledger%3A0091af?consistency=strong&read_preference=leader"

# --- 4. merge a counter (atomic delta) -----------------------------------
curl --request PATCH \
  --cacert "$NKV_CA" --cert "$NKV_CERT" --key "$NKV_KEY" \
  --header "Authorization: Bearer $NKV_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{"merge":{"op":"increment","amount":1}}' \
  "https://swedenpnoo.xyz/v3/kv/telemetry%3Acounters%3Apage-views"

# --- 5. shard + partition topology --------------------------------------
curl --silent --show-error \
  --cacert "$NKV_CA" --cert "$NKV_CERT" --key "$NKV_KEY" \
  --header "Authorization: Bearer $NKV_TOKEN" \
  "https://swedenpnoo.xyz/v3/cluster/topology?include_partitions=true"

# --- 6. snapshot catalogue ----------------------------------------------
curl --silent --show-error \
  --cacert "$NKV_CA" --cert "$NKV_CERT" --key "$NKV_KEY" \
  --header "Authorization: Bearer $NKV_TOKEN" \
  "https://swedenpnoo.xyz/v3/backups/snapshots?limit=20&sort=created:desc"

# --- 7. windowed WAL slice (binary) -------------------------------------
curl --silent --show-error \
  --cacert "$NKV_CA" --cert "$NKV_CERT" --key "$NKV_KEY" \
  --header "Authorization: Bearer $NKV_TOKEN" \
  --output "sto-wal-slice.nkvwal" \
  "https://swedenpnoo.xyz/v3/wal/slice?shard=shard-01&window=6h&max_bytes=15728640"

# --- 8. trigger merkle resync verification -------------------------------
curl --request POST \
  --cacert "$NKV_CA" --cert "$NKV_CERT" --key "$NKV_KEY" \
  --header "Authorization: Bearer $NKV_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{"shard_ids":["shard-01","shard-02","shard-03"],"merkle_depth":12,"dry_run":true}' \
  "https://swedenpnoo.xyz/v3/cluster/resync-verify"

# --- 9. metrics (Prometheus text format, mTLS only) ----------------------
curl --silent --show-error \
  --cacert "$NKV_CA" --cert "$NKV_CERT" --key "$NKV_KEY" \
  "https://swedenpnoo.xyz:9187/metrics" | head -n 20