func (b *backend) QueueStats(ctx context.Context, qq *entroq.QueuesQuery) (map[string]*entroq.QueueStat, error) {
	q := `SELECT
			queue,
			COUNT(*) AS count,
			COUNT(*) FILTER(WHERE at > NOW() AND claimant != '00000000-0000-0000-0000-000000000000') AS claimed,
			COUNT(*) FILTER(WHERE at <= NOW()) as available,
			MAX(claims) AS max_claims
		FROM tasks`
	var values []interface{}
	if len(qq.MatchPrefix) != 0 || len(qq.MatchExact) != 0 {
		q += " WHERE"
	}
	var matchFragments []string
	for _, m := range qq.MatchPrefix {
		matchFragments = append(matchFragments, fmt.Sprintf(" queue LIKE $%d", len(values)+1))
		values = append(values, m+"%")
	}
	for _, m := range qq.MatchExact {
		matchFragments = append(matchFragments, fmt.Sprintf(" queue = $%d", len(values)+1))
		values = append(values, m)
	}
	if len(matchFragments) != 0 {
		q += strings.Join(matchFragments, " OR ")
	}
	if qq.Limit > 0 {
		q += fmt.Sprintf(" LIMIT $%d", len(values)+1)
		values = append(values, qq.Limit)
	}
	q += " GROUP BY queue"
	rows, err := b.db.QueryContext(ctx, q, values...)
	if err != nil {
		return nil, fmt.Errorf("queue names: %w", err)
	}
	defer rows.Close()
	queues := make(map[string]*entroq.QueueStat)
	for rows.Next() {
		var (
			q         string
			count     int
			claimed   int
			available int
			maxClaims int
		)
		if err := rows.Scan(&q, &count, &claimed, &available, &maxClaims); err != nil {
			return nil, fmt.Errorf("row scan: %w", err)
		}
		queues[q] = &entroq.QueueStat{
			Name:      q,
			Size:      count,
			Claimed:   claimed,
			Available: available,
			MaxClaims: maxClaims,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("queue iteration: %w", err)
	}
	return queues, nil
}
QueueStats returns a mapping from queue names to their statistics.
package eqpg
import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"log"
	"math/rand"
	"net/url"
	"strings"
	"time"
	"github.com/google/uuid"
	"github.com/lib/pq"
	"github.com/shiblon/entroq"
	"github.com/shiblon/entroq/subq"
)
func escp(p string) string {
	return "'" + strings.NewReplacer("\\", "\\\\", "'", "\\'").Replace(p) + "'"
}
type SSLMode string
const (
	SSLDisable    SSLMode = "disable"      
	SSLAllow      SSLMode = "allow"        
	SSLPrefer     SSLMode = "prefer"       
	SSLRequire    SSLMode = "require"      
	SSLVerifyCA   SSLMode = "verify-ca"    
	SSLVerifyFull SSLMode = "verify-full"  
)
type pgOptions struct {
	db       string
	user     string
	password string
	attempts int
	nw       entroq.NotifyWaiter
	sslMode SSLMode
	sslClientKeyFile  string
	sslClientCertFile string
	sslServerCAFile   string
}
type PGOpt func(opts *pgOptions)
func WithUsername(name string) PGOpt {
	return func(opts *pgOptions) {
		opts.user = name
	}
}
func WithPassword(pwd string) PGOpt {
	return func(opts *pgOptions) {
		opts.password = pwd
	}
}
func WithDB(db string) PGOpt {
	return func(opts *pgOptions) {
		opts.db = db
	}
}
func WithSSL(mode SSLMode, sslOpts ...PGOpt) PGOpt {
	return func(opts *pgOptions) {
		opts.sslMode = mode
		for _, o := range sslOpts {
			o(opts)
		}
	}
}
func WithSSLClientFiles(certFile, keyFile string) PGOpt {
	return func(opts *pgOptions) {
		opts.sslClientCertFile = certFile
		opts.sslClientKeyFile = keyFile
	}
}
func WithSSLServerCAFile(caFile string) PGOpt {
	return func(opts *pgOptions) {
		opts.sslServerCAFile = caFile
	}
}
func WithConnectAttempts(num int) PGOpt {
	if num < 1 {
		num = 1
	}
	return func(opts *pgOptions) {
		opts.attempts = num
	}
}
func WithNotifyWaiter(nw entroq.NotifyWaiter) PGOpt {
	return func(opts *pgOptions) {
		opts.nw = nw
	}
}
func Opener(hostPort string, opts ...PGOpt) entroq.BackendOpener {
	options := &pgOptions{
		db:       "postgres",
		user:     "postgres",
		password: "password",
		attempts: 1,
		nw:       subq.New(),
		sslMode: SSLDisable,
	}
	for _, o := range opts {
		o(options)
	}
	return func(ctx context.Context) (entroq.Backend, error) {
		u, err := url.Parse("postgres://" + hostPort)
		if err != nil {
			return nil, fmt.Errorf("failed to parse hostport %q: %w", hostPort, err)
		}
		host := u.Hostname()
		port := u.Port()
		if port != "" && host == "" {
			host = "::"
		}
		if host != "" && port == "" {
			port = "5432"
		}
		params := []string{
			"sslmode=" + string(options.sslMode),
		}
		if options.user != "" {
			params = append(params, fmt.Sprintf("user=%s", escp(options.user)))
		}
		if options.password != "" {
			params = append(params, fmt.Sprintf("password=%s", escp(options.password)))
		}
		if host != "" {
			params = append(params, fmt.Sprintf("host=%s", escp(host)))
		}
		if port != "" {
			params = append(params, fmt.Sprintf("port=%s", port))
		}
		if options.sslClientKeyFile != "" {
			params = append(params, "sslkey="+url.QueryEscape(options.sslClientKeyFile))
		}
		if options.sslClientCertFile != "" {
			params = append(params, "sslcert="+url.QueryEscape(options.sslClientCertFile))
		}
		if options.sslServerCAFile != "" {
			params = append(params, "sslrootcert="+url.QueryEscape(options.sslServerCAFile))
		}
		connStr := strings.Join(params, " ")
		db, err := sql.Open("postgres", connStr)
		if err != nil {
			return nil, fmt.Errorf("failed to open postgres DB: %w", err)
		}
		for i := 0; i < options.attempts; i++ {
			if err = db.PingContext(ctx); err == nil {
				return New(ctx, db, options.nw)
			}
			if i < options.attempts-1 {
				select {
				case <-ctx.Done():
					return nil, fmt.Errorf("pg opener: %w", ctx.Err())
				case <-time.After(5 * time.Second):
				}
			}
		}
		return nil, fmt.Errorf("time out postgres init: %w", err)
	}
}
type backend struct {
	db *sql.DB
	nw entroq.NotifyWaiter
}
func New(ctx context.Context, db *sql.DB, nw entroq.NotifyWaiter) (*backend, error) {
	b := &backend{
		db: db,
		nw: nw,
	}
	err := b.initDB(ctx)
	if err == io.EOF {
		return nil, io.EOF
	}
	if err != nil {
		return nil, fmt.Errorf("failed to initialize database: %w", err)
	}
	return b, nil
}
func (b *backend) Close() error {
	if err := b.db.Close(); err != nil {
		return fmt.Errorf("pg backend close: %w", err)
	}
	return nil
}
func (b *backend) initDB(ctx context.Context) error {
	_, err := b.db.ExecContext(ctx, `
		ALTER TABLE IF EXISTS tasks ADD COLUMN IF NOT EXISTS claims INTEGER NOT NULL DEFAULT 0;
		ALTER TABLE IF EXISTS tasks ADD COLUMN IF NOT EXISTS attempt INTEGER NOT NULL DEFAULT 0;
		ALTER TABLE IF EXISTS tasks ADD COLUMN IF NOT EXISTS err TEXT NOT NULL DEFAULT '';
		CREATE TABLE IF NOT EXISTS tasks (
		  id UUID PRIMARY KEY NOT NULL,
		  version INTEGER NOT NULL DEFAULT 0,
		  queue TEXT NOT NULL DEFAULT '',
		  at TIMESTAMP WITH TIME ZONE NOT NULL,
		  created TIMESTAMP WITH TIME ZONE,
		  modified TIMESTAMP WITH TIME ZONE NOT NULL,
		  claimant UUID,
		  value BYTEA,
		  claims INTEGER NOT NULL DEFAULT 0,
		  attempt INTEGER NOT NULL DEFAULT 0,
		  err TEXT NOT NULL DEFAULT ''
		);
		CREATE INDEX IF NOT EXISTS byID ON tasks (id);
		CREATE INDEX IF NOT EXISTS byVersion ON tasks (version);
		CREATE INDEX IF NOT EXISTS byQueue ON tasks (queue);
		CREATE INDEX IF NOT EXISTS byQueueAt ON tasks (queue, at);
	`)
	if err != nil {
		return fmt.Errorf("initDB: %w", err)
	}
	return nil
}
func (b *backend) Queues(ctx context.Context, qq *entroq.QueuesQuery) (map[string]int, error) {
	return entroq.QueuesFromStats(b.QueueStats(ctx, qq))
}