pgo

README

pgo (/pɪɡəʊ/): Postgres integrations in Go

This is a piecemeal collection of code snippets, which has evolved from my Postgres+Go projects whenever they were reused, without any plan/structure.

Now I'm (forcibly) moving from applications to this repo the boilerplate / heavy-lifting around:

• net/http.Handler
  • router
  • middleware (authentication, logging, CORS, RequestID, ...)
  • Postgres middleware attaches a pgxpool.Conn to request context for authorized user; useful for RLS
• Retrieval Augmented Generation (RAG)
  • fetch embeddings from LLM APIs (OpenAI, Groq, Anthropic, Google, Ollama, ...)
  • utils for pgvector search, augmented generation
• Pipelines (realtime/batch)
  • Postgres' Logical Replication (optionally LISTEN/NOTIFY)
  • MQTT, Kafka, HTTP, ClickHouse, gRPC (add more by writing plugins yourself)

Usage

• As a standalone binary for publishing Postgres CDC to Kafka, MQTT, ClickHouse, etc, see docs/postgres-cdc.md

It's also possible to import functions, etc. This reusability seems helpful, mostly because I'm learning by being forced to write reliable code that other code might depend on. Most of it actually isn’t dependable, yet. If you're curious, start by browsing the examples, skimming over any doc.go, *.md files.

Contributing

Please see CONTRIBUTING.md.

License

Apache License 2.0

Documentation

Overview

Package pgo simplifies the process of querying PostgreSQL database upon an HTTP request. It uses jackc/pgx(https://pkg.go.dev/github.com/jackc/pgx/v5) as the PostgreSQL driver.

Key Features:

- HTTP Routing: A lightweight router with support for middleware, grouping, and customizable error handling.

- PostgreSQL Integration: Streamlined interactions with PostgreSQL databases using pgx, a powerful PostgreSQL driver for Go.

• Convenience Functions: A collection of helper functions for common database operations (select, insert, update, etc.), designed to work seamlessly within HTTP request contexts.

• Error Handling: Robust error handling mechanisms for both HTTP routing and database operations, ensuring predictable and informative error responses.

Example Usage:

router := pgo.NewRouter(pgo.WithTLS("cert.pem", "key.pem"))
router.Use(middleware.Logger)

// Define routes
router.Handle("GET /users", usersHandler)
router.Handle("POST /users", createUserHandler)

api := router.Group("/api")
api.Handle("GET /products", productsHandler)

// Start the server
router.ListenAndServe(":8080")

Additional Information:

- For detailed information on HTTP routing, see the documentation for the `Router` type and its associated functions.

- For PostgreSQL-specific helpers and utilities, refer to the documentation for the `pgxutil` package.

Index

• Variables
• func BasicAuthUser(r *http.Request) (string, bool)
• func BindOrError(r *http.Request, w http.ResponseWriter, dst interface{}) error
• func Blob(w http.ResponseWriter, statusCode int, data []byte, contentType string)
• func Conn(r *http.Request) (*oidc.IntrospectionResponse, *pgxpool.Conn, *pgconn.PgError)
• func ConnWithRole(r *http.Request) (*oidc.IntrospectionResponse, *pgxpool.Conn, *pgconn.PgError)
• func DefaultPool() (*pgxpool.Pool, error)
• func Error(w http.ResponseWriter, statusCode int, message string)
• func HTML(w http.ResponseWriter, statusCode int, html string)
• func InitDefaultPool(connString string, config ...*pgxpool.Config) (*pgxpool.Pool, error)
• func JSON(w http.ResponseWriter, statusCode int, data interface{})
• func OIDCUser(r *http.Request) (*oidc.IntrospectionResponse, bool)
• func Text(w http.ResponseWriter, statusCode int, text string)
• type ContextKey
• type ErrorResponse
• type Middleware
• type Router
  • func NewRouter(opts ...RouterOptions) *Router
  • func (r *Router) Group(prefix string) *Router
  • func (r *Router) Handle(methodPattern string, handler http.Handler)
  • func (r *Router) ListenAndServe(addr string) error
  • func (r *Router) Shutdown(ctx context.Context) error
  • func (r *Router) Use(mw Middleware)
• type RouterOptions
  • func WithServerOptions(opts ...func(*http.Server)) RouterOptions
  • func WithTLS(certFile, keyFile string) RouterOptions

Variables

var (
	ErrTooManyRows        = errors.New("too many rows")
	ErrPoolNotInitialized = errors.New("default pool not initialized")
)

Functions

func BasicAuthUser

func BasicAuthUser(r *http.Request) (string, bool)

BasicAuthUser retrieves the authenticated username from the context.

func BindOrError

func BindOrError(r *http.Request, w http.ResponseWriter, dst interface{}) error

BindOrError decodes the JSON body of an HTTP request, r, into the given destination object, dst. If decoding fails, it responds with a 400 Bad Request error.

func Blob

func Blob(w http.ResponseWriter, statusCode int, data []byte, contentType string)

Blob writes a binary response with the given status code and data.

func Conn

func Conn(r *http.Request) (*oidc.IntrospectionResponse, *pgxpool.Conn, *pgconn.PgError)

Conn retrieves the OIDC user and a pgxpool.Conn from the request context. It returns an error if the user or connection is not found in the context. Currently it only supports OIDC users. But the authZ middleware chain works, and error occurs here.

func ConnWithRole

func ConnWithRole(r *http.Request) (*oidc.IntrospectionResponse, *pgxpool.Conn, *pgconn.PgError)

ConnWithRole retrieves the OIDC user, a pgxpool.Conn, and checks for a role from the request context. It's designed for use with Row Level Security (RLS) enabled on a table.

This function accomplishes the following:

1. Calls Conn(r) to retrieve the OIDC user and a pgxpool.Conn from the context.
2. Attempts to retrieve the role value from the context using PgRoleCtxKey.
3. Marshals the user's claims to a JSON string.
4. Escapes the JSON string for safe insertion into the SQL query.
5. Constructs a combined query that sets both the role and request claims. - The role is set using the retrieved value from the context. - The request claims are set using either the environment variable PGO_POSTGRES_OIDC_REQUEST_JWT_CLAIMS (if set) or a default value "request.jwt.claims" (aligned with PostgREST behavior). https://docs.postgrest.org/en/v12/references/transactions.html#request-headers-cookies-and-jwt-claims
6. Executes the combined query on the connection.
7. Handles potential errors: - If Conn(r) returns an error, it's propagated directly. - If the role is not found in the context, a custom PgError is returned. - If marshalling claims or executing the query fails, a custom PgError is returned with appropriate details.

ConnWithRole is useful for scenarios where RLS policies rely on user claims to restrict access to specific rows.

Below example RLS policy allows a user to only select rows where user_id (column of the table) matches the OIDC sub claim PostgreSQL: ALTER TABLE wallets ENABLE ROW LEVEL SECURITY; ALTER TABLE wallets FORCE ROW LEVEL SECURITY; DROP POLICY IF EXISTS select_own ON wallets; CREATE POLICY select_own ON wallets FOR SELECT USING (user_id = (current_setting('request.jwt.claims', true)::json->>'sub')::TEXT); ALTER POLICY select_own ON wallets TO authn;

func DefaultPool

func DefaultPool() (*pgxpool.Pool, error)

DefaultPool returns the default PostgreSQL connection pool. If the pool is uninitialized, it returns nil.

func Error

func Error(w http.ResponseWriter, statusCode int, message string)

Error sends a JSON response with an error code and message.

func HTML

func HTML(w http.ResponseWriter, statusCode int, html string)

HTML writes an HTML response with the given status code and HTML content.

func InitDefaultPool

func InitDefaultPool(connString string, config ...*pgxpool.Config) (*pgxpool.Pool, error)

InitDefaultPool initializes the default PostgreSQL connection pool and returns it. It accepts a connection string `connString` and an optional pgxpool.Config. If the connection string fails to parse, the function falls back to using libpq environment variables, such as PGPASSWORD, for establishing the connection.

connString: The libpq connection string. If the connection string contains special characters (like in passwords), use the KEY=VALUE format, as recommended here: https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING-KEYWORD-VALUE

If parsing the provided connection string fails and the PGPASSWORD environment variable is not set, the function terminates the process.

A background goroutine is started to periodically check the health of the connections.

func JSON

func JSON(w http.ResponseWriter, statusCode int, data interface{})

JSON writes a JSON response with the given status code and data.

func OIDCUser

func OIDCUser(r *http.Request) (*oidc.IntrospectionResponse, bool)

OIDCUser extracts the OIDC user from the request context.

func Text

func Text(w http.ResponseWriter, statusCode int, text string)

Text writes a plain text response with the given status code and text content.

Types

type ContextKey

type ContextKey string

const (
	RequestIDCtxKey ContextKey = "RequestID"
	LogEntryCtxKey  ContextKey = "LogEntry"
	OIDCUserCtxKey  ContextKey = "OIDCUser"
	BasicAuthCtxKey ContextKey = "BasicAuth"
	PgConnCtxKey    ContextKey = "PgConn"
	PgRoleCtxKey    ContextKey = "PgRole"
)

type ErrorResponse

type ErrorResponse struct {
	Code    int    `json:"code"`
	Message string `json:"message"`
}

ErrorResponse represents a structured error response.

type Middleware

type Middleware func(http.Handler) http.Handler

Middleware defines a function type that represents a middleware. Middleware functions wrap an http.Handler to modify or enhance its behavior.

type Router

type Router struct {
	// contains filtered or unexported fields
}

Router is the main structure for handling HTTP routing and middleware.

func NewRouter

func NewRouter(opts ...RouterOptions) *Router

NewRouter creates a new instance of Router with the given options.

func (*Router) Group

func (r *Router) Group(prefix string) *Router

Group creates a new sub-router with a specified prefix. The sub-router inherits the middleware from its parent router.

func (*Router) Handle

func (r *Router) Handle(methodPattern string, handler http.Handler)

Handle registers an HTTP handler function for a given method and pattern as introduced in [Routing Enhancements for Go 1.22](https://go.dev/blog/routing-enhancements) The handler `METHOD /pattern` on a route group with a /prefix resolves to `METHOD /prefix/pattern`

func (*Router) ListenAndServe

func (r *Router) ListenAndServe(addr string) error

ListenAndServe starts the server, automatically choosing between HTTP and HTTPS based on TLS config.

func (*Router) Shutdown

func (r *Router) Shutdown(ctx context.Context) error

Shutdown gracefully shuts down the HTTP server.

func (*Router) Use

func (r *Router) Use(mw Middleware)

Use adds middleware to the router. Middleware functions are applied in the order they are added.

type RouterOptions

type RouterOptions func(*Router)

RouterOptions is a function type that represents options to configure a Router.

func WithServerOptions

func WithServerOptions(opts ...func(*http.Server)) RouterOptions

WithServerOptions returns a RouterOptions function that sets custom http.Server options.

func WithTLS

func WithTLS(certFile, keyFile string) RouterOptions

WithTLS provides a simplified way to enable HTTPS in your router.