mirror of
https://github.com/tiennm99/keepalive.git
synced 2026-10-15 04:18:46 +00:00
feat: initial db-keepalive — pluggable adapters for 6 free-tier DBs
Consolidates the *-keepalive family (redis, valkey, postgresql, mysql, mongodb, couchbase) into a single binary with a runtime DB_TYPE dispatch and an Adapter interface. Original repos archived with a pointer here.
This commit is contained in:
commit
e3b7ea5011
16 files changed
+1080
No files matched your search
@@ -0,0 +1,46 @@
|
||||
// Package adapter defines the contract every database driver implements.
|
||||
//
|
||||
// An Adapter connects to a single backing store, increments a counter on every
|
||||
// keepalive tick (the cheapest write that proves the cluster is live), and
|
||||
// releases its resources on Close. New databases plug in by adding a file in
|
||||
// this package and registering a factory in Registry.
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
type Adapter interface {
|
||||
Connect(ctx context.Context) error
|
||||
Increment(ctx context.Context) (int64, error)
|
||||
Close(ctx context.Context) error
|
||||
}
|
||||
|
||||
type Factory func() (Adapter, error)
|
||||
|
||||
var Registry = map[string]Factory{}
|
||||
|
||||
func New(dbType string) (Adapter, error) {
|
||||
f, ok := Registry[dbType]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("unknown DB_TYPE %q (known: %v)", dbType, Known())
|
||||
}
|
||||
return f()
|
||||
}
|
||||
|
||||
func Known() []string {
|
||||
out := make([]string, 0, len(Registry))
|
||||
for k := range Registry {
|
||||
out = append(out, k)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func envOrFail(name string) (string, error) {
|
||||
v, ok := lookupEnv(name)
|
||||
if !ok || v == "" {
|
||||
return "", fmt.Errorf("env %s is required", name)
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/couchbase/gocb/v2"
|
||||
)
|
||||
|
||||
func init() {
|
||||
Registry["couchbase"] = func() (Adapter, error) { return &couchbaseAdapter{}, nil }
|
||||
}
|
||||
|
||||
type couchbaseAdapter struct {
|
||||
cluster *gocb.Cluster
|
||||
coll *gocb.Collection
|
||||
docID string
|
||||
}
|
||||
|
||||
func (a *couchbaseAdapter) Connect(_ context.Context) error {
|
||||
conn, err := envOrFail("COUCHBASE_CONNECTION_STRING")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
user, err := envOrFail("COUCHBASE_USERNAME")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
pass, err := envOrFail("COUCHBASE_PASSWORD")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
bucket, err := envOrFail("COUCHBASE_BUCKET_NAME")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
scope, err := envOrFail("COUCHBASE_SCOPE_NAME")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
collName, err := envOrFail("COUCHBASE_COLLECTION_NAME")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
opts := gocb.ClusterOptions{
|
||||
Authenticator: gocb.PasswordAuthenticator{Username: user, Password: pass},
|
||||
}
|
||||
if err := opts.ApplyProfile(gocb.ClusterConfigProfileWanDevelopment); err != nil {
|
||||
return err
|
||||
}
|
||||
cluster, err := gocb.Connect(conn, opts)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
b := cluster.Bucket(bucket)
|
||||
if err := b.WaitUntilReady(5*time.Second, nil); err != nil {
|
||||
return err
|
||||
}
|
||||
a.cluster = cluster
|
||||
a.coll = b.Scope(scope).Collection(collName)
|
||||
a.docID = envOr("COUNTER_KEY", "counter")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *couchbaseAdapter) Increment(_ context.Context) (int64, error) {
|
||||
docOut, err := a.coll.Get(a.docID, &gocb.GetOptions{})
|
||||
if err != nil {
|
||||
// On first run the doc may not exist; seed at 1.
|
||||
if _, upErr := a.coll.Upsert(a.docID, uint64(1), &gocb.UpsertOptions{}); upErr != nil {
|
||||
return 0, upErr
|
||||
}
|
||||
return 1, nil
|
||||
}
|
||||
var current uint64
|
||||
if err := docOut.Content(¤t); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
current++
|
||||
if _, err := a.coll.Upsert(a.docID, current, &gocb.UpsertOptions{}); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return int64(current), nil
|
||||
}
|
||||
|
||||
func (a *couchbaseAdapter) Close(_ context.Context) error {
|
||||
return a.cluster.Close(nil)
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
package adapter
|
||||
|
||||
import "os"
|
||||
|
||||
func lookupEnv(name string) (string, bool) {
|
||||
return os.LookupEnv(name)
|
||||
}
|
||||
|
||||
func envOr(name, def string) string {
|
||||
if v, ok := os.LookupEnv(name); ok && v != "" {
|
||||
return v
|
||||
}
|
||||
return def
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||||
)
|
||||
|
||||
func init() {
|
||||
Registry["mongodb"] = func() (Adapter, error) { return &mongoAdapter{}, nil }
|
||||
Registry["mongo"] = Registry["mongodb"]
|
||||
}
|
||||
|
||||
type mongoAdapter struct {
|
||||
client *mongo.Client
|
||||
coll *mongo.Collection
|
||||
docID string
|
||||
}
|
||||
|
||||
func (a *mongoAdapter) Connect(ctx context.Context) error {
|
||||
uri, err := envOrFail("MONGODB_URI")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
dbName, err := envOrFail("MONGODB_DATABASE")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
collName, err := envOrFail("MONGODB_COLLECTION")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
client, err := mongo.Connect(options.Client().ApplyURI(uri))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
a.client = client
|
||||
a.coll = client.Database(dbName).Collection(collName)
|
||||
a.docID = envOr("COUNTER_KEY", "counter")
|
||||
return client.Ping(ctx, nil)
|
||||
}
|
||||
|
||||
func (a *mongoAdapter) Increment(ctx context.Context) (int64, error) {
|
||||
update := bson.M{"$inc": bson.M{"count": 1}}
|
||||
opts := options.FindOneAndUpdate().SetUpsert(true).SetReturnDocument(options.After)
|
||||
var out struct {
|
||||
Count int64 `bson:"count"`
|
||||
}
|
||||
err := a.coll.FindOneAndUpdate(ctx, bson.M{"_id": a.docID}, update, opts).Decode(&out)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return out.Count, nil
|
||||
}
|
||||
|
||||
func (a *mongoAdapter) Close(ctx context.Context) error {
|
||||
return a.client.Disconnect(ctx)
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"time"
|
||||
|
||||
_ "github.com/go-sql-driver/mysql"
|
||||
)
|
||||
|
||||
func init() {
|
||||
Registry["mysql"] = func() (Adapter, error) { return &mysqlAdapter{}, nil }
|
||||
}
|
||||
|
||||
type mysqlAdapter struct {
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
func (a *mysqlAdapter) Connect(ctx context.Context) error {
|
||||
dsn, err := envOrFail("DATA_SOURCE_NAME")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
db, err := sql.Open("mysql", dsn)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
db.SetConnMaxLifetime(3 * time.Minute)
|
||||
db.SetMaxOpenConns(10)
|
||||
db.SetMaxIdleConns(10)
|
||||
a.db = db
|
||||
return a.db.PingContext(ctx)
|
||||
}
|
||||
|
||||
func (a *mysqlAdapter) Increment(ctx context.Context) (int64, error) {
|
||||
tx, err := a.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx,
|
||||
"UPDATE `keepalive` SET `value` = `value` + 1 WHERE `key` = 'counter'",
|
||||
); err != nil {
|
||||
tx.Rollback()
|
||||
return 0, err
|
||||
}
|
||||
var value int64
|
||||
if err := tx.QueryRowContext(ctx,
|
||||
"SELECT `value` FROM `keepalive` WHERE `key` = 'counter'",
|
||||
).Scan(&value); err != nil {
|
||||
tx.Rollback()
|
||||
return 0, err
|
||||
}
|
||||
return value, tx.Commit()
|
||||
}
|
||||
|
||||
func (a *mysqlAdapter) Close(_ context.Context) error {
|
||||
return a.db.Close()
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
|
||||
_ "github.com/lib/pq"
|
||||
)
|
||||
|
||||
func init() {
|
||||
Registry["postgresql"] = func() (Adapter, error) { return &postgresAdapter{}, nil }
|
||||
Registry["postgres"] = Registry["postgresql"]
|
||||
}
|
||||
|
||||
type postgresAdapter struct {
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
func (a *postgresAdapter) Connect(ctx context.Context) error {
|
||||
uri, err := envOrFail("SERVICE_URI")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
db, err := sql.Open("postgres", uri)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
a.db = db
|
||||
return a.db.PingContext(ctx)
|
||||
}
|
||||
|
||||
func (a *postgresAdapter) Increment(ctx context.Context) (int64, error) {
|
||||
tx, err := a.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var value int64
|
||||
if err := tx.QueryRowContext(ctx,
|
||||
`UPDATE keepalive SET value = value + 1 WHERE key = 'counter' RETURNING value`,
|
||||
).Scan(&value); err != nil {
|
||||
tx.Rollback()
|
||||
return 0, err
|
||||
}
|
||||
return value, tx.Commit()
|
||||
}
|
||||
|
||||
func (a *postgresAdapter) Close(_ context.Context) error {
|
||||
return a.db.Close()
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
func init() {
|
||||
Registry["redis"] = func() (Adapter, error) { return &redisAdapter{}, nil }
|
||||
}
|
||||
|
||||
type redisAdapter struct {
|
||||
client *redis.Client
|
||||
key string
|
||||
}
|
||||
|
||||
func (a *redisAdapter) Connect(ctx context.Context) error {
|
||||
url, err := envOrFail("REDIS_URL")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
opt, err := redis.ParseURL(url)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
a.client = redis.NewClient(opt)
|
||||
a.key = envOr("COUNTER_KEY", "counter")
|
||||
return a.client.Ping(ctx).Err()
|
||||
}
|
||||
|
||||
func (a *redisAdapter) Increment(ctx context.Context) (int64, error) {
|
||||
return a.client.Incr(ctx, a.key).Result()
|
||||
}
|
||||
|
||||
func (a *redisAdapter) Close(_ context.Context) error {
|
||||
return a.client.Close()
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/valkey-io/valkey-go"
|
||||
)
|
||||
|
||||
func init() {
|
||||
Registry["valkey"] = func() (Adapter, error) { return &valkeyAdapter{}, nil }
|
||||
}
|
||||
|
||||
type valkeyAdapter struct {
|
||||
client valkey.Client
|
||||
key string
|
||||
}
|
||||
|
||||
func (a *valkeyAdapter) Connect(_ context.Context) error {
|
||||
url, err := envOrFail("VALKEY_URL")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
opt, err := valkey.ParseURL(url)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
client, err := valkey.NewClient(opt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
a.client = client
|
||||
a.key = envOr("COUNTER_KEY", "counter")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *valkeyAdapter) Increment(ctx context.Context) (int64, error) {
|
||||
return a.client.Do(ctx, a.client.B().Incr().Key(a.key).Build()).AsInt64()
|
||||
}
|
||||
|
||||
func (a *valkeyAdapter) Close(_ context.Context) error {
|
||||
a.client.Close()
|
||||
return nil
|
||||
}
|
||||
Reference in new issue
Block a user