hidb_en.md

HiDB Package (pkg/hidb)

The hidb package provides the database abstraction layer and connection manager for the HypGo framework, supporting multiple database engines, read-write separation (Master-Replica), connection pooling, transaction management, Redis caching, and an extensible plugin system.

Key Features

• Multi-Database Support: Abstracts MySQL / PostgreSQL driver differences through the Dialect interface, using Bun ORM as the query builder layer.
• Read-Write Separation: Write operations always route to Master, read operations are distributed to Replicas via ReplicaPool using Round-Robin, with automatic fallback to Master when no Replicas are available.
• Lock-Free Round-Robin: Uses atomic.Pointer + atomic.Uint64 for a completely lock-free read path (copy-on-write), zero mutex contention at 100k+ QPS.
• Connection Pool Management: Supports MaxIdleConns / MaxOpenConns / ConnMaxLifetime (default 30 minutes) configuration, with Master and each Replica managing connection pools independently.
• Transaction Management: Provides both native sql.Tx and Bun ORM bun.Tx transaction interfaces.
• Redis Integration: Built-in go-redis/v9 client, unified lifecycle management with SQL databases.
• Plugin System: Supports dynamic registration and loading of non-SQL databases (e.g., Cassandra) through the DatabasePlugin interface.
• Graceful Shutdown: Shuts down in order: Replica -> Master -> Redis -> Plugins, with accumulated errors reported together.
• Backward Compatible: HypDB() / SQL() always return Master; existing code requires no changes.

Basic Usage

package main

import (
	"context"
	"log"

	"github.com/maoxiaoyue/hypgo/pkg/hidb"
	"github.com/maoxiaoyue/hypgo/pkg/hidb/pg"
)

func main() {
	// 1. Create database manager (PostgreSQL)
	db, err := hidb.NewWithInterface(
		appConfig.Database,           // Implements config.DatabaseConfigInterface
		hidb.WithDialect(pg.New()),   // Specify SQL dialect
	)
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	// 2. Health check
	if err := db.HealthCheck(context.Background()); err != nil {
		log.Fatal(err)
	}

	// 3. Read operations -> Replica (falls back to Master when no Replicas)
	readDB := db.ReadHypDB()
	var users []User
	readDB.NewSelect().Model(&users).Scan(context.Background())

	// 4. Write operations -> Always uses Master
	writeDB := db.WriteHypDB()
	writeDB.NewInsert().Model(&newUser).Exec(context.Background())
}

Read-Write Separation (Master-Replica)

Configuration

Enable read-write separation by implementing the config.ReplicaConfigProvider interface:

type ReplicaConfigProvider interface {
	GetReplicas() []ReplicaConfig
}

type ReplicaConfig struct {
	DSN          string // Replica connection string
	MaxIdleConns int    // Maximum idle connections
	MaxOpenConns int    // Maximum connections
}

Query Routing

Method	Target	Description
ReadHypDB()	Replica (ORM)	Round-Robin distribution, falls back to Master
ReadSQL()	Replica (Raw SQL)	Round-Robin distribution, falls back to Master
WriteHypDB()	Master (ORM)	Always uses Master
WriteSQL()	Master (Raw SQL)	Always uses Master
HypDB()	Master (ORM)	Backward compatible, equivalent to WriteHypDB()
SQL()	Master (Raw SQL)	Backward compatible, equivalent to WriteSQL()

Round-Robin Load Balancing

// Read path is completely lock-free: atomic.Pointer + atomic.Uint64
// Write path uses copy-on-write: Mutex protection + create new slice + atomic.Store
replicas := *rp.replicas.Load()  // Zero lock contention
idx := rp.counter.Add(1) - 1
replica := replicas[idx % uint64(len(replicas))]

ConnMaxLifetime

Master and all Replicas uniformly set ConnMaxLifetime = 30 minutes, preventing long-lived connections from holding stale state (e.g., DNS changes, password rotation).

Status Query

db.HasReplicas()   // bool -- Whether Replicas are configured
db.ReplicaCount()  // int  -- Number of Replicas

Transaction Management

Native SQL Transactions

err := db.Transaction(ctx, func(tx *sql.Tx) error {
	_, err := tx.ExecContext(ctx, "INSERT INTO users (name) VALUES (?)", "Alice")
	if err != nil {
		return err // Auto Rollback
	}
	return nil // Auto Commit
})

Bun ORM Transactions

err := db.HypDBTransaction(ctx, func(ctx context.Context, tx bun.Tx) error {
	_, err := tx.NewInsert().Model(&user).Exec(ctx)
	if err != nil {
		return err // Auto Rollback
	}
	return nil // Auto Commit
})

Redis / KeyDB Integration

HiDB includes a complete Redis wrapper, also compatible with KeyDB (Redis protocol compatible). Beyond the raw go-redis/v9 client, it provides high-level methods covering all common data structures.

Configuration

database:
  driver: "redis"      # Or used alongside SQL
  redis:
    addr: "localhost:6379"    # KeyDB uses the same settings
    password: "your_password"
    db: 0

type RedisConfigInterface interface {
    GetAddr() string
    GetPassword() string
    GetDB() int
}

Basic Key-Value

ctx := context.Background()

// Set (with TTL, 0 = no expiry)
db.RedisSet(ctx, "user:1:name", "Alice", 10*time.Minute)

// Get
name, err := db.RedisGet(ctx, "user:1:name")
if hidb.RedisIsNil(err) {
    // Key does not exist
}

// Delete
db.RedisDel(ctx, "user:1:name", "user:2:name")

// Check existence
count, _ := db.RedisExists(ctx, "user:1:name")

// TTL management
db.RedisExpire(ctx, "session:abc", 30*time.Minute)
ttl, _ := db.RedisTTL(ctx, "session:abc")

JSON Serialization

Store and retrieve Go structs directly with automatic JSON serialization/deserialization:

type GameState struct {
    Players []string `json:"players"`
    Round   int      `json:"round"`
    Status  string   `json:"status"`
}

// Store
state := GameState{Players: []string{"p1", "p2"}, Round: 1, Status: "playing"}
db.RedisSetJSON(ctx, "game:123:state", state, time.Hour)

// Retrieve
var loaded GameState
db.RedisGetJSON(ctx, "game:123:state", &loaded)

Hash

// Set multiple fields
db.RedisHSet(ctx, "user:1", "name", "Alice", "email", "alice@test.com")

// Get single field
name, _ := db.RedisHGet(ctx, "user:1", "name")

// Get all fields
all, _ := db.RedisHGetAll(ctx, "user:1")
// all = map[string]string{"name": "Alice", "email": "alice@test.com"}

// Delete field
db.RedisHDel(ctx, "user:1", "email")

List (Queue)

// Push
db.RedisLPush(ctx, "queue:tasks", "task1", "task2")
db.RedisRPush(ctx, "queue:tasks", "task3")

// Pop
task, _ := db.RedisLPop(ctx, "queue:tasks")

// Range query
tasks, _ := db.RedisLRange(ctx, "queue:tasks", 0, -1)

// Length
length, _ := db.RedisLLen(ctx, "queue:tasks")

Set

// Add members
db.RedisSAdd(ctx, "room:1:players", "player1", "player2")

// Check membership
isMember, _ := db.RedisSIsMember(ctx, "room:1:players", "player1")

// Get all members
members, _ := db.RedisSMembers(ctx, "room:1:players")

// Remove member
db.RedisSRem(ctx, "room:1:players", "player1")

Sorted Set (Leaderboard)

import "github.com/redis/go-redis/v9"

// Add scores
db.RedisZAdd(ctx, "leaderboard", redis.Z{Score: 100, Member: "player1"})

// Get ranking (with scores)
results, _ := db.RedisZRangeWithScores(ctx, "leaderboard", 0, 9) // Top 10

// Query score
score, _ := db.RedisZScore(ctx, "leaderboard", "player1")

// Remove
db.RedisZRem(ctx, "leaderboard", "player1")

Atomic Operations

// Counter
db.RedisIncr(ctx, "api:requests:count")
db.RedisDecr(ctx, "stock:item:42")
db.RedisIncrBy(ctx, "user:1:score", 50)

Distributed Lock (SetNX)

// Attempt to acquire lock (30 second auto-release)
acquired, _ := db.RedisSetNX(ctx, "lock:game:123", "owner-uuid", 30*time.Second)
if acquired {
    defer db.RedisDel(ctx, "lock:game:123")
    // Execute critical section...
}

Pub/Sub

// Publish
db.RedisPublish(ctx, "game:events", "player_joined")

// Subscribe (caller must Close)
sub, _ := db.RedisSubscribe(ctx, "game:events")
defer sub.Close()

for msg := range sub.Channel() {
    fmt.Printf("Channel: %s, Payload: %s\n", msg.Channel, msg.Payload)
}

Pipeline (Batch Operations)

// Regular pipeline (non-transactional)
pipe, _ := db.RedisPipeline()
pipe.Set(ctx, "key1", "val1", 0)
pipe.Set(ctx, "key2", "val2", 0)
pipe.Incr(ctx, "counter")
cmds, err := pipe.Exec(ctx)

// Transactional pipeline (MULTI/EXEC)
txPipe, _ := db.RedisTxPipeline()
txPipe.Set(ctx, "key1", "val1", 0)
txPipe.Set(ctx, "key2", "val2", 0)
cmds, err = txPipe.Exec(ctx)

Scan (Cursor Iteration)

// Non-blocking key scan
var cursor uint64
for {
    keys, nextCursor, _ := db.RedisScan(ctx, cursor, "user:*", 100)
    for _, key := range keys {
        fmt.Println(key)
    }
    cursor = nextCursor
    if cursor == 0 {
        break
    }
}

Raw Client Access

When you need go-redis features not covered by the wrapper:

client := db.Redis() // *redis.Client
// Use the full go-redis/v9 API directly

KeyDB Compatibility

KeyDB uses the same protocol as Redis; just change the connection address:

database:
  driver: "redis"
  redis:
    addr: "keydb-server:6379"  # Point to KeyDB
    password: "password"
    db: 0

KeyDB advantages: multi-threaded architecture, FLASH storage support, Active Replication -- all work transparently with no code changes required.

Dialect System

Interface

type Dialect interface {
	DriverName() string          // "mysql" or "postgres"
	BunDialect() schema.Dialect  // Bun ORM dialect instance
}

Built-in Dialects

MySQL / TiDB:

import "github.com/maoxiaoyue/hypgo/pkg/hidb/mysql"

db, err := hidb.NewWithInterface(cfg, hidb.WithDialect(mysql.New()))

PostgreSQL:

import "github.com/maoxiaoyue/hypgo/pkg/hidb/pg"

db, err := hidb.NewWithInterface(cfg, hidb.WithDialect(pg.New()))

Plugin System

DatabasePlugin Interface

type DatabasePlugin interface {
	Name() string
	Init(config map[string]interface{}) error
	Connect() error
	Close() error
	Ping(ctx context.Context) error
}

Registration and Usage

// Register plugin
cassandraPlugin := cassandra.NewPlugin()
db.RegisterPlugin(cassandraPlugin)

// Dynamic loading (Init + Connect)
db.LoadPlugin("cassandra", map[string]interface{}{
	"hosts":    []string{"127.0.0.1"},
	"keyspace": "my_keyspace",
})

// Get plugin
if plugin, ok := db.GetPlugin("cassandra"); ok {
	plugin.Ping(ctx)
}

Cassandra Plugin

pkg/hidb/cassandra provides a complete Cassandra 5.0+ / ScyllaDB driver wrapper, including a Fluent DDL builder, Model mapping, CRUD builder, Vector/ANN queries, Migration, RBAC, Schema Introspection, and more.

See hidb.cassandra for full documentation.

Health Check and Lifecycle

// Connection status
db.IsConnected()  // bool

// Full health check (Master + Replicas + Redis + Plugins)
err := db.HealthCheck(ctx)

// Database driver type
db.Type()  // "mysql", "postgres", "redis", etc.

// Graceful shutdown (in order: Replicas -> Master -> Redis -> Plugins)
err := db.Close()

Backward Compatible Factory

When the configuration object does not implement config.DatabaseConfigInterface, New() auto-adapts via reflection:

// Legacy struct (uses reflection adapter)
db, err := hidb.New(legacyConfig, hidb.WithDialect(pg.New()))

// Recommended: implement DatabaseConfigInterface
db, err := hidb.NewWithInterface(modernConfig, hidb.WithDialect(pg.New()))

DatabaseConfigAdapter extracts Driver, DSN, MaxIdleConns, MaxOpenConns, Redis, and other fields via reflection, ensuring existing code requires no changes.

File Structure

pkg/hidb/
├── hidb.go              # Core: Database, Dialect, Option, Plugin system,
│                         #   Transaction management, Health check, Graceful shutdown
├── redis.go             # Redis/KeyDB high-level wrapper (KV, Hash, List, Set, ZSet, Pub/Sub, Pipeline)
├── redis_test.go        # Redis nil guard + RedisIsNil tests
├── readwrite.go         # ReplicaPool: Round-Robin load balancing, Read-write separation
├── readwrite_test.go    # 11 tests: Round-Robin, concurrency, fallback, shutdown
├── mysql/
│   └── mysql.go         # MySQL / TiDB dialect implementation
├── pg/
│   └── pg.go            # PostgreSQL dialect implementation
└── cassandra/
    └── cassandra.go     # Cassandra plugin implementation

Dependencies

Package	Purpose
database/sql	Go standard SQL interface
github.com/uptrace/bun	ORM query builder
github.com/redis/go-redis/v9	Redis client
github.com/go-sql-driver/mysql	MySQL driver
github.com/lib/pq	PostgreSQL driver
github.com/gocql/gocql	Cassandra driver