108 lines
3.6 KiB
Go
108 lines
3.6 KiB
Go
// Package pgkit is the small piece of internal/db/db.go that wasn't
|
|
// specific to any one bounded context: connecting with a startup retry
|
|
// loop (docker compose's "healthy" and "actually accepting
|
|
// connections" can be a beat apart) and applying an idempotent
|
|
// schema.sql-style migration under an advisory lock so two replicas of
|
|
// the same service starting at once don't race each other.
|
|
//
|
|
// Each data-owning service (iam, billing, conversation, knowledge) gets
|
|
// its own *pgkit.DB over its own database — see each service's
|
|
// migrations/0001_init.sql — never a shared one. That split is the
|
|
// difference between "microservices" and "one database with several
|
|
// front ends" (see ARCHITECTURE.md, "Database-per-service").
|
|
package pgkit
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/lib/pq"
|
|
)
|
|
|
|
// Sentinel errors every repository package in this repo returns, so
|
|
// application-layer use cases can errors.Is against one shared set
|
|
// instead of each service inventing its own (matching internal/db's
|
|
// original ErrNotFound/ErrEmailTaken/ErrInvalidToken).
|
|
var (
|
|
ErrNotFound = errors.New("pgkit: not found")
|
|
ErrAlreadyTaken = errors.New("pgkit: unique constraint violated")
|
|
ErrInvalid = errors.New("pgkit: invalid or expired")
|
|
)
|
|
|
|
// DB wraps *sql.DB exactly as internal/db.DB did.
|
|
type DB struct {
|
|
*sql.DB
|
|
}
|
|
|
|
// Open connects to Postgres and retries pinging it for up to ~30s.
|
|
func Open(dsn string) (*DB, error) {
|
|
sqlDB, err := sql.Open("postgres", dsn)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("pgkit: open: %w", err)
|
|
}
|
|
|
|
// Pool sizing: each microservice is still a single logical writer
|
|
// per table, same reasoning as the monolith's own comment here —
|
|
// a handful of connections covers concurrent gRPC calls without
|
|
// exhausting Postgres' default max_connections (100) once you
|
|
// multiply by four data-owning services.
|
|
sqlDB.SetMaxOpenConns(20)
|
|
sqlDB.SetMaxIdleConns(5)
|
|
sqlDB.SetConnMaxLifetime(30 * time.Minute)
|
|
sqlDB.SetConnMaxIdleTime(5 * time.Minute)
|
|
|
|
var lastErr error
|
|
deadline := time.Now().Add(30 * time.Second)
|
|
for {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
lastErr = sqlDB.PingContext(ctx)
|
|
cancel()
|
|
if lastErr == nil {
|
|
return &DB{sqlDB}, nil
|
|
}
|
|
if time.Now().After(deadline) {
|
|
_ = sqlDB.Close()
|
|
return nil, fmt.Errorf("pgkit: ping (gave up after 30s): %w", lastErr)
|
|
}
|
|
time.Sleep(time.Second)
|
|
}
|
|
}
|
|
|
|
// migrationLockID is an arbitrary constant used with pg_advisory_lock.
|
|
// Shared across services since each connects to its own database (the
|
|
// lock is scoped per-database by Postgres), so there's no cross-service
|
|
// collision risk in reusing one constant.
|
|
const migrationLockID = 7_402_113_915
|
|
|
|
// Migrate applies schema (expected to be idempotent — CREATE ... IF NOT
|
|
// EXISTS throughout, as with the monolith's schema.sql) under an
|
|
// advisory lock.
|
|
func (d *DB) Migrate(ctx context.Context, schema string) error {
|
|
conn, err := d.Conn(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("pgkit: migrate: acquire connection: %w", err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
if _, err := conn.ExecContext(ctx, `SELECT pg_advisory_lock($1)`, migrationLockID); err != nil {
|
|
return fmt.Errorf("pgkit: migrate: lock: %w", err)
|
|
}
|
|
defer func() {
|
|
_, _ = conn.ExecContext(context.Background(), `SELECT pg_advisory_unlock($1)`, migrationLockID)
|
|
}()
|
|
|
|
if _, err := conn.ExecContext(ctx, schema); err != nil {
|
|
return fmt.Errorf("pgkit: migrate: apply schema: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// IsUniqueViolation reports whether err is Postgres' unique_violation (23505).
|
|
func IsUniqueViolation(err error) bool {
|
|
var pqErr *pq.Error
|
|
return errors.As(err, &pqErr) && pqErr.Code == "23505"
|
|
}
|