Produce messages in a database transaction

Suppose you are building a user signup flow where you want to send a welcome email after successful signup.

Seems simple, but what happens if your application saves the account but fails before sending the email?

The transactional outbox pattern addresses this gap between saving data and requesting downstream work. And SQLStreams solves this simply.

Use ProduceFunc to save the user and a message requesting the welcome email in one PostgreSQL transaction. Both writes commit together, or neither does.

Before you begin

Use the Go project and running database from the Quickstart. Keep its WelcomeEmailV1 payload and signup.welcome-email stream. You will change the producer in cmd/produce/main.go.

1. Create the application table

Run this from the Quickstart project directory:

docker compose exec -T db psql -U username -d quickstart <<'SQL'
CREATE TABLE users (
    id text PRIMARY KEY
);
SQL

This table holds the application’s users in the same database as SQLStreams.

2. Write the user and message together

Add "context" to the imports in cmd/produce/main.go.

Replace the producer.Produce call, its error check, and the following fmt.Printf with this block. Keep the final return nil in run.

produced, err := producer.ProduceFunc(ctx,
  func(ctx context.Context, tx sqlstreams.Tx) (*WelcomeEmailV1, error) {
    if _, err := tx.Exec(ctx,
      `INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, "user-123"); err != nil {
      return nil, err
    }
    return &WelcomeEmailV1{UserId: "user-123"}, nil
  }, &sqlstreams.ProduceOptions{IdempotencyKey: "welcome-email:user-123"})
if err != nil {
  return err
}
if produced.Duplicate {
  fmt.Println("welcome-email request already produced for user-123")
} else {
  fmt.Printf("committed user=user-123 message_id=%d\n", produced.Id)
}

Use the supplied tx for your database writes. When the callback returns a payload and nil, SQLStreams produces the message and commits the transaction.

If the callback returns an error, SQLStreams rolls back the transaction, including the user insert.

3. Run the producer

go run ./cmd/produce

A successful first run prints a line like this. The message id depends on what you have already produced:

committed user=user-123 message_id=1

Check the saved user:

docker compose exec -T db psql -U username -d quickstart \
  -c "SELECT id FROM users WHERE id = 'user-123';"

The query returns user-123. The successful ProduceFunc call confirms that the user insert and message committed together.

The callback can run again during database recovery. ON CONFLICT makes repeating this user insert harmless, and the stable IdempotencyKey protects the message insert from duplication.

Produce a message from database work — Callback retries, return values, and errors for ProduceFunc.
Run a database transaction — Use InTransaction when several operations need to share one transaction.
Produce within a transaction — Add messages to that transaction with ProduceInTx or ProduceFuncInTx.
Make message processing safe to repeat — Protect the consumer side from applying the same change twice.