| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556 |
- // Package postgres wraps pgx for the few queries the services need.
- // M0 doesn't use Postgres; M1 needs it for routing + delivery.
- //
- // We use the pgxpool connection pool so concurrent queries don't
- // step on each other.
- package postgres
- import (
- "context"
- "fmt"
- "time"
- "github.com/jackc/pgx/v5/pgxpool"
- )
- // Pool is a thin alias for *pgxpool.Pool so the call sites don't
- // import pgx directly.
- type Pool = pgxpool.Pool
- // Connect dials Postgres with retries. The first migration run will
- // hit this before the DB is up, so we wait.
- func Connect(ctx context.Context, dsn string) (*Pool, error) {
- cfg, err := pgxpool.ParseConfig(dsn)
- if err != nil {
- return nil, fmt.Errorf("parse dsn: %w", err)
- }
- cfg.MaxConns = 20
- cfg.MinConns = 2
- cfg.MaxConnIdleTime = 5 * time.Minute
- var pool *Pool
- deadline, hasDeadline := ctx.Deadline()
- if !hasDeadline {
- deadline = time.Now().Add(30 * time.Second)
- }
- for {
- pool, err = pgxpool.NewWithConfig(ctx, cfg)
- if err == nil {
- pingCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
- err = pool.Ping(pingCtx)
- cancel()
- if err == nil {
- return pool, nil
- }
- pool.Close()
- }
- if time.Now().After(deadline) {
- return nil, fmt.Errorf("postgres connect timeout: %w", err)
- }
- select {
- case <-ctx.Done():
- return nil, ctx.Err()
- case <-time.After(1 * time.Second):
- }
- }
- }
|