Client

sqlstreams.NewClient(ctx, pool, cfg) is the one constructor a program calls. It builds every registration object over your *pgxpool.Pool and pings the pool once, so a wrong address or credential fails here instead of at the first produce. The pool stays yours: SQLStreams never closes it, and the client has no Close.

ctx, stop := sqlstreams.LifecycleContext(nil)
defer stop()

pool, err := sqlstreams.NewPostgresPool(ctx, "app_user", os.Getenv("PGPASSWORD"), "localhost", "app_db", nil)
if err != nil {
	return err
}
defer pool.Close()

client, err := sqlstreams.NewClient(ctx, pool, &sqlstreams.ClientConfig{
	Schema:       "payments",
	AllowDestroy: false,
})
if err != nil {
	return err
}

Verbs

verb	returns	notes
Streams(ctx)	[]*Stream	every registered stream, ordered by name
Stream[T](name)	*StreamHandle[T]	no I/O; Stream
Schedulers(ctx)	[]*Schedule	every registered schedule, ordered by name
Scheduler(name)	*SchedulerHandle	no I/O; Scheduler
System()	*SystemHandle	no I/O; System
Manager()	*ManagerHandle	no I/O; Manager
InTransaction(ctx, transactionFunc)	error	one transaction, your closure, commit; the closure receives a sqlstreams.Tx that ProduceInTx accepts

InTransaction does not retry. A transient blip or an ambiguous commit surfaces as-is, and only you know what is safe to rerun in your closure. Rerunning the whole closure is dedup-safe only under caller-supplied IdempotencyKey values; unset keys mint fresh per call, so a rerun produces twice (transactional produce).

sqlstreams.LifecycleContext(logger) returns a context cancelled on the first SIGINT or SIGTERM, which starts graceful shutdown of every blocking verb; a second signal during the drain force-exits with status 128 plus the signal number. nil uses the default logger.

Config

ClientConfig

field	default	what it decides
Schema	"sqlstreams"	the Postgres namespace holding every table; one schema is one installation
AllowDestroy	false	whether any Destroy verb on this client may run
DisableManager	false	whether Consume skips running the system manager beside its session; Manager().Run is unaffected
Logger	text lines to stderr, warn and up	your *slog.Logger or anything satisfying logging.Logger; held once
Retry	common.NewDefaultRetryPolicy()	the retry curve for every Postgres call the client makes, never a message’s redelivery

NewClient captures the values and copies Retry; editing the config after it returns changes nothing. Every config below the client inherits Logger and Retry, so none of them carries either field. Two clients over the same pool is how one consumer runs without upkeep while another runs with it.

The schema

Two clients on two schemas in the same database each get their own catalog, streams, and sequences; orders registers in both and they are different streams. client.System().Register creates the schema if it is missing, which needs CREATE on the database; without that you get SQL0064, and the fix is to create the schema yourself and grant the role USAGE and CREATE on it.

The pool sets no search_path. Every statement SQLStreams runs names its schema outright, sqlstreams.message_log_4 and never message_log_4, so your own SQL inside InTransaction resolves the way it would on any other connection: a CREATE TABLE orders in that closure lands in your default schema, not SQLStreams’s. The diagnose queries on every error page carry a {schema} placeholder for the same reason.

Gotchas

• A first program imports sqlstreams and nothing else of SQLStreams’s. Every type a user spells is declared or aliased there; the implementation packages are importable without a stability commitment (the supported API).
• DisableManager is for consumer pods beside a dedicated sqlstreams manager run. It does not remove DDL from the process: the consumer’s own stream janitor still runs there and creates and drops partitions, so its database role needs DDL rights on that stream’s tables.