Documentation
¶
Overview ¶
Package postgres connects Vukem to PostgreSQL through pgx. It is a separate module so apps that only use SQLite never download pgx.
vukem.New(vukem.WithDatabaseDrivers(postgres.Driver())) // DATABASE_URL=postgres://user:pass@localhost:5432/app?sslmode=disable
Once Driver is registered, everything built on the database package (ORM, schema builder, migrations, database-backed stores) speaks the PostgreSQL dialect.
NewBroker is the second half of the module: an sse.Broker on LISTEN/NOTIFY, so server-sent events cross processes (web service to queue:work service) on the database the app already has.
broker, err := postgres.NewBroker(db) hub := sse.NewHub(broker)
See https://github.com/AlexGalhardo/vukem/blob/main/docs/database.md.
Example ¶
An app registers the driver once; DATABASE_URL then selects it by scheme.
package main
import (
"fmt"
"log"
"github.com/AlexGalhardo/vukem"
"github.com/AlexGalhardo/vukem/drivers/postgres"
)
func main() {
// DATABASE_URL=postgres://user:pass@localhost:5432/app?sslmode=disable
app, err := vukem.New(vukem.WithDatabaseDrivers(postgres.Driver()))
if err != nil {
log.Fatal(err)
}
db, err := app.DB()
if err != nil {
log.Fatal(err)
}
fmt.Println(db.Dialect().Placeholder(1))
}
Output:
Example (Open) ¶
Outside an app, pass the driver to database.Open.
package main
import (
"context"
"log"
"os"
"github.com/AlexGalhardo/vukem/database"
"github.com/AlexGalhardo/vukem/drivers/postgres"
)
func main() {
ctx := context.Background()
db, err := database.Open(ctx, os.Getenv("DATABASE_URL"), database.WithDriver(postgres.Driver()))
if err != nil {
log.Fatal(err)
}
defer db.Close()
if err := db.Ping(ctx); err != nil {
log.Print(err)
}
}
Output:
Index ¶
Examples ¶
Constants ¶
const MaxNotifyPayload = 7999
MaxNotifyPayload is the largest payload, in bytes, that Publish accepts.
PostgreSQL rejects a notification payload of 8000 bytes or more, so this is the last size it takes. The server derives that bound from its block size, so a server built with a smaller BLCKSZ accepts less and answers with its own error; the default build accepts exactly this much. Events on an SSE topic are tens of bytes, so in practice the limit only catches a mistake such as streaming a whole document instead of a progress update.
Variables ¶
var ErrNotPostgres = errors.New("postgres: the broker needs a PostgreSQL connection through pgx")
ErrNotPostgres is returned by NewBroker when the database is not PostgreSQL, and by Broker.Subscribe when it is not reached through pgx. Both are configuration mistakes, and failing loudly beats a broker that accepts every publish and delivers nothing.
var ErrPayloadTooLarge = errors.New("postgres: payload over the NOTIFY limit")
ErrPayloadTooLarge is returned by Broker.Publish for a payload over MaxNotifyPayload. An event that large does not belong on a notification channel, so it is refused rather than truncated: a truncated event would reach the browser as valid-looking, unparseable data.
Functions ¶
func Driver ¶
Driver returns the PostgreSQL driver for database.Open and vukem.WithDatabaseDrivers. It accepts postgres:// and postgresql:// URLs.
Example ¶
package main
import (
"fmt"
"github.com/AlexGalhardo/vukem/drivers/postgres"
)
func main() {
d := postgres.Driver()
fmt.Println(d.SQLName, d.Schemes, d.Dialect.Placeholder(2))
}
Output: pgx [postgres postgresql] $2
Types ¶
type Broker ¶ added in v0.5.0
type Broker struct {
// contains filtered or unexported fields
}
Broker carries server-sent events between processes over PostgreSQL LISTEN/NOTIFY. It implements sse.Broker, so an app whose web service and queue:work service are separate processes streams progress from the worker to the browser on the database it already has, without adding Redis purely as a message bus.
broker, err := postgres.NewBroker(db) hub := sse.NewHub(broker)
What it costs:
- A payload must stay at or under MaxNotifyPayload bytes (PostgreSQL refuses a NOTIFY payload of 8000 bytes or more); Publish refuses a larger one with ErrPayloadTooLarge rather than truncate it.
- LISTEN occupies a connection for the life of the subscription, so each one is taken out of the pool (sse.Hub holds one per topic that has local subscribers and releases it when the last one leaves, so the count follows watched topics, not watching browsers). Size database.Config.MaxOpen (10 by default) for the topics an instance watches at once, plus the connections its requests need.
- A notification is not persisted and is not replayed: a subscriber that is not connected when the event is published never sees it. The stream is the fast path; durable state belongs in a table.
func NewBroker ¶ added in v0.5.0
func NewBroker(db *database.DB, opts ...BrokerOption) (*Broker, error)
NewBroker returns a broker on db, which must be PostgreSQL. It fails rather than degrade on any other database: a broker that accepts every publish and delivers nothing looks like working code and breaks only in production, under load, in the one path nobody watches. Whether the connection really goes through pgx is only visible on a connection, so that half of ErrNotPostgres comes from Broker.Subscribe.
Example ¶
The broker carries events between processes, so the queue:work service can stream progress to a browser connected to the web service. Both processes build the hub the same way, on the same database.
package main
import (
"context"
"log"
"os"
"github.com/AlexGalhardo/vukem/database"
"github.com/AlexGalhardo/vukem/drivers/postgres"
"github.com/AlexGalhardo/vukem/sse"
)
func main() {
ctx := context.Background()
db, err := database.Open(ctx, os.Getenv("DATABASE_URL"), database.WithDriver(postgres.Driver()))
if err != nil {
log.Fatal(err)
}
// Not PostgreSQL: the broker would accept every publish and deliver
// nothing, so an app fails to boot instead.
broker, err := postgres.NewBroker(db)
if err != nil {
log.Fatal(err)
}
defer db.Close()
hub := sse.NewHub(broker)
defer hub.Close()
// In the worker: progress for one conversion, at most postgres.MaxNotifyPayload bytes.
if err := hub.Publish(ctx, "conversions:42", sse.Event{Event: "progress", Data: `{"percent":10}`}); err != nil {
log.Print(err)
}
}
Output:
func (*Broker) Publish ¶ added in v0.5.0
Publish sends payload to every listener of channel. A payload over MaxNotifyPayload bytes is refused with ErrPayloadTooLarge.
func (*Broker) Subscribe ¶ added in v0.5.0
Subscribe delivers the payloads published on channels until ctx is done, then closes the returned channel. It returns once every LISTEN is in place, so an event published after it returns is delivered.
The subscription holds one connection of the pool until ctx is done. A notification that the returned channel has no room for is dropped rather than allowed to block that connection, which serves every other subscriber of the same channels: sse.Hub already treats a gap as a reason for the client to reconnect, and a blocked listener would also stop the server from truncating its notification queue.
type BrokerOption ¶ added in v0.5.0
type BrokerOption func(*Broker)
BrokerOption configures a Broker.
func WithBrokerLogger ¶ added in v0.5.0
func WithBrokerLogger(log *slog.Logger) BrokerOption
WithBrokerLogger sets the logger a Broker reports a subscription that ended unexpectedly on; the default is slog.Default. A nil logger is ignored.
Source Files
Directories