Quickstart

Produce and consume one welcome-email message using two Go programs and an existing Postgres database. The consumer prints the request; it does not send an email.

1. Prepare the example

You need Go 1.27+ and a local Postgres database. These examples use app_db on localhost:5432, with role app_user. The role must be able to create schemas and tables in that database. Substitute your own connection details in both programs if they differ.

Create a directory and install the library:

mkdir sqlstreams-quickstart
cd sqlstreams-quickstart
go mod init example.com/sqlstreams-quickstart
go get github.com/agentstax/sqlstreams
mkdir -p cmd/produce cmd/consume
export PGPASSWORD='your-local-database-password'

Run the commands below from this directory, in the same terminal so both programs receive PGPASSWORD. The first stream registration creates SQLStreams’s shared tables and the stream’s tables in the sqlstreams schema. There is no separate migration command for this example.

2. Produce a message

Save this as cmd/produce/main.go:

package main

import (
	"fmt"
	"os"

	"github.com/agentstax/sqlstreams/client"
)

type WelcomeEmail struct {
	UserId string `json:"user_id"`
}

func (WelcomeEmail) SchemaVersion() int { return 1 }

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

func run() error {
	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, nil)
	if err != nil {
		return err
	}

	emails := client.Stream[WelcomeEmail]("signup.welcome-email")
	if _, err := emails.Register(ctx, nil); err != nil {
		return err
	}

	producer, err := emails.Producer().Register(ctx, nil)
	if err != nil {
		return err
	}

	produced, err := producer.Produce(ctx, &WelcomeEmail{UserId: "user-123"}, nil)
	if err != nil {
		return err
	}
	fmt.Printf("produced message id=%d\n", produced.Id)
	return nil
}

Run it:

go run ./cmd/produce

The output includes produced message id=1 on a fresh stream; the id may be higher if the stream already contains messages. Each run produces another message.

WelcomeEmail is stored as JSONB. Its SchemaVersion method identifies the payload version, which the consumer type must also declare. Schema Versions explains how versions share a stream.

3. Consume the message

Save this as cmd/consume/main.go:

package main

import (
	"context"
	"errors"
	"fmt"
	"os"

	"github.com/agentstax/sqlstreams/client"
)

type WelcomeEmail struct {
	UserId string `json:"user_id"`
}

func (WelcomeEmail) SchemaVersion() int { return 1 }

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

func run() error {
	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, nil)
	if err != nil {
		return err
	}

	emails := client.Stream[WelcomeEmail]("signup.welcome-email")
	registered, err := emails.Get(ctx)
	if err != nil {
		return err
	}
	if registered == nil {
		return errors.New("stream not registered -- run the producer first")
	}

	sender := emails.Consumer("email-sender")
	consumer, err := sender.Register(ctx, nil)
	if err != nil {
		return err
	}

	return consumer.Consume(ctx, receiveEmail, nil)
}

func receiveEmail(ctx context.Context, email *WelcomeEmail) error {
	fmt.Printf("received welcome email request for %s\n", email.UserId)
	return nil
}

Run it:

go run ./cmd/consume

The output includes:

received welcome email request for user-123

Startup also writes logs to stderr. On a fresh installation, alert evidence is insufficient means an alert check has too little metric history to evaluate yet; it does not mean the message failed. It stops once the collector has recorded a couple of minutes of samples.

The process keeps waiting for messages. Press Ctrl-C to stop it: LifecycleContext turns SIGINT and SIGTERM into cancellation, and Consume drains its in-flight handlers before returning.

The new email-sender group reads retained history by default, including the message produced before it started. Restarting the same group resumes its existing cursor. If a previous run already consumed the message, stop the consumer, run the producer again, then restart the consumer.

Returning nil succeeds; an ordinary error schedules a retry. Handler Outcomes covers terminal errors and requested delays. A real email handler must tolerate duplicate delivery and pass its context to blocking calls: crashes and timeouts can cause at-least-once redelivery. See Consumer Timeouts and Side Effects & Retries.

4. Continue with your application

• Transactional Produce puts a business write and its message in one commit.
• Message Lifecycle explains retries, leases, and the rows you can inspect when a handler fails.
• Fan-out, Retention & Replay explains independent groups and how long their source messages remain available.
• Consumer lists configuration fields, defaults, and session options.