Function openPgxV5DBPool

cmd/river/rivercli/command.go:133–179

(ctx context.Context, databaseURL string, statementTimeout *time.Duration)

}

func openPgxV5DBPool(ctx context.Context, databaseURL string, statementTimeout *time.Duration) (*pgxpool.Pool, error) {
	const (
		defaultIdleInTransactionSessionTimeout = 11 * time.Second // should be greater than statement timeout because statements count towards idle-in-transaction
		defaultStatementTimeout                = 10 * time.Second
	)

	pgxConfig, err := pgxpool.ParseConfig(databaseURL)
	if err != nil {
		return nil, fmt.Errorf("error parsing database URL: %w", err)
	}

	// Sets a parameter in a parameter map (aimed at a Postgres connection
	// configuration map), but only if that parameter wasn't already set.
	setParamIfUnset := func(runtimeParams map[string]string, name, val string) {
		if currentVal := runtimeParams[name]; currentVal != "" {
			return
		}

		runtimeParams[name] = val
	}

	runtimeParams := pgxConfig.ConnConfig.RuntimeParams
	if runtimeParams == nil {
		runtimeParams = make(map[string]string)
		pgxConfig.ConnConfig.RuntimeParams = runtimeParams
	}

	var statementTimeoutMilliseconds string
	if statementTimeout != nil {
		statementTimeoutMilliseconds = strconv.Itoa(int(statementTimeout.Milliseconds()))
	}

	setParamIfUnset(runtimeParams, "application_name", "river CLI")
	setParamIfUnset(runtimeParams, "idle_in_transaction_session_timeout", strconv.Itoa(int(defaultIdleInTransactionSessionTimeout.Milliseconds())))
	runtimeParams["statement_timeout"] = cmp.Or(
		statementTimeoutMilliseconds,
		runtimeParams["statement_timeout"],
		strconv.Itoa(int(defaultStatementTimeout.Milliseconds())),
	)

	dbPool, err := pgxpool.NewWithConfig(ctx, pgxConfig)
	if err != nil {
		return nil, fmt.Errorf("error connecting to Postgres database: %w", err)
	}

	return dbPool, nil
}

func openSQLitePool(protocol, urlWithoutProtocol string) (*sql.DB, error) {
	dbPool, err := sql.Open(protocol, urlWithoutProtocol)