HiDB Package (pkg/hidb)

Key Features

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

type ReplicaConfigProvider interface {
	GetReplicas() []ReplicaConfig
}
type ReplicaConfig struct {
	DSN string // Replica connection string
	MaxIdleConns int // Maximum idle connections
	MaxOpenConns int // Maximum connections
}

Query Routing

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

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

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

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

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

KeyDB Compatibility

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

Dialect System

Interface

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

Built-in Dialects

import "github.com/maoxiaoyue/hypgo/pkg/hidb/mysql"
db, err := hidb.NewWithInterface(cfg, hidb.WithDialect(mysql.New()))

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

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

// 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()))

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