first commit
This commit is contained in:
122
go-api/internal/db/db.go
Normal file
122
go-api/internal/db/db.go
Normal file
@@ -0,0 +1,122 @@
|
||||
// Package db owns the PostgreSQL connection pool and the database health check.
|
||||
package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/krow/krow-backend/go-api/internal/config"
|
||||
)
|
||||
|
||||
// DB wraps the pgx pool together with the schema the application is pinned to.
|
||||
type DB struct {
|
||||
Pool *pgxpool.Pool
|
||||
Schema string
|
||||
}
|
||||
|
||||
// Open builds the pool and verifies it can actually reach the database.
|
||||
//
|
||||
// pgxpool.New is lazy — it returns a usable pool without having connected —
|
||||
// so a bad host or a wrong password would otherwise not surface until the
|
||||
// first request. Acquiring and pinging once here turns a misconfiguration into
|
||||
// a startup failure instead of a runtime surprise.
|
||||
func Open(ctx context.Context, cfg config.DBConfig) (*DB, error) {
|
||||
poolCfg, err := pgxpool.ParseConfig(cfg.DSN())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse database config: %w", err)
|
||||
}
|
||||
poolCfg.MaxConns = cfg.MaxOpenConns
|
||||
poolCfg.MinIdleConns = cfg.MinIdleConns
|
||||
poolCfg.MaxConnLifetime = cfg.ConnMaxLifetime
|
||||
|
||||
pool, err := pgxpool.NewWithConfig(ctx, poolCfg)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create connection pool: %w", err)
|
||||
}
|
||||
|
||||
pingCtx, cancel := context.WithTimeout(ctx, cfg.ConnectTimeout)
|
||||
defer cancel()
|
||||
if err := pool.Ping(pingCtx); err != nil {
|
||||
pool.Close()
|
||||
return nil, fmt.Errorf("connect to %s: %w", cfg.Redacted(), err)
|
||||
}
|
||||
|
||||
return &DB{Pool: pool, Schema: cfg.Schema}, nil
|
||||
}
|
||||
|
||||
// Close releases every pooled connection.
|
||||
func (d *DB) Close() {
|
||||
if d != nil && d.Pool != nil {
|
||||
d.Pool.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// Health is what the /health endpoint reports about the database.
|
||||
type Health struct {
|
||||
Reachable bool `json:"reachable"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Version string `json:"version,omitempty"`
|
||||
Database string `json:"database,omitempty"`
|
||||
Schema string `json:"schema,omitempty"`
|
||||
SchemaPresent bool `json:"schema_present"`
|
||||
AppliedMigration *int64 `json:"applied_migration,omitempty"`
|
||||
MigrationDirty bool `json:"migration_dirty"`
|
||||
TableCount int `json:"table_count"`
|
||||
LatencyMS int64 `json:"latency_ms"`
|
||||
}
|
||||
|
||||
// Check answers "can the API serve requests against this database right now".
|
||||
//
|
||||
// It reports more than a ping because a reachable database with no schema in it
|
||||
// is a different failure from an unreachable one, and both are worth telling
|
||||
// apart at a glance during Phase 1. Reads are confined to the configured schema
|
||||
// via to_regclass and a count over information_schema, which is the standard
|
||||
// SQL view rather than a pg_catalog table.
|
||||
func (d *DB) Check(ctx context.Context) Health {
|
||||
started := time.Now()
|
||||
h := Health{Schema: d.Schema}
|
||||
|
||||
conn, err := d.Pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
h.Error = err.Error()
|
||||
h.LatencyMS = time.Since(started).Milliseconds()
|
||||
return h
|
||||
}
|
||||
defer conn.Release()
|
||||
|
||||
if err := conn.QueryRow(ctx,
|
||||
`SELECT current_database(), current_setting('server_version')`,
|
||||
).Scan(&h.Database, &h.Version); err != nil {
|
||||
h.Error = err.Error()
|
||||
h.LatencyMS = time.Since(started).Milliseconds()
|
||||
return h
|
||||
}
|
||||
h.Reachable = true
|
||||
|
||||
if err := conn.QueryRow(ctx,
|
||||
`SELECT count(*)::int FROM information_schema.tables
|
||||
WHERE table_schema = $1 AND table_type = 'BASE TABLE'`,
|
||||
d.Schema,
|
||||
).Scan(&h.TableCount); err != nil {
|
||||
h.Error = err.Error()
|
||||
h.LatencyMS = time.Since(started).Milliseconds()
|
||||
return h
|
||||
}
|
||||
h.SchemaPresent = h.TableCount > 0
|
||||
|
||||
// golang-migrate's bookkeeping table. Absent before the first migration,
|
||||
// which is a legitimate state and not an error.
|
||||
var version int64
|
||||
var dirty bool
|
||||
err = conn.QueryRow(ctx, `SELECT version, dirty FROM schema_migrations LIMIT 1`).Scan(&version, &dirty)
|
||||
if err == nil {
|
||||
h.AppliedMigration = &version
|
||||
h.MigrationDirty = dirty
|
||||
}
|
||||
|
||||
h.LatencyMS = time.Since(started).Milliseconds()
|
||||
return h
|
||||
}
|
||||
Reference in New Issue
Block a user