Make message processing safe to repeat

Suppose your application gives a user 10 reward points when they sign up. Crediting those points twice would give the user a larger reward than intended.

A producer’s IdempotencyKey prevents repeated produce calls from storing duplicate messages. It does not prevent a consumer handler from processing the same stored message again.

For example, a consumer handler processes message 1 and commits the 10 points to Postgres. Its process then crashes before SQLStreams records success. After the lease expires, another consumer instance can process message 1 again. No second message was produced, so the producer’s idempotency key cannot prevent the points from being added again.

This guide records a reward’s id and updates the user’s points in one transaction. A repeated reward id skips the update. This makes the database change idempotent, meaning repeating the request has the same effect as applying it once.

Before you begin

Use the Go project and running database from the Quickstart. Replace its producer and consumer with the examples below. They use a new rewards.earned stream and a credit-rewards consumer group.

1. Create the reward tables

Run this from the Quickstart project directory:

docker compose exec -T db psql -U username -d quickstart <<'SQL'
CREATE TABLE reward_credits (
    reward_id text PRIMARY KEY,
    user_id text NOT NULL,
    points integer NOT NULL
);

CREATE TABLE user_rewards (
    user_id text PRIMARY KEY,
    points integer NOT NULL
);
SQL

reward_credits records which rewards have been applied. Its primary key prevents two transactions from crediting the same reward. user_rewards stores each user’s total points.

2. Produce a reward request

Replace cmd/produce/main.go with:

package main

import (
  "fmt"
  "os"

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

type RewardEarnedV1 struct {
  RewardId string `json:"reward_id"`
  UserId   string `json:"user_id"`
  Points   int    `json:"points"`
}

func (RewardEarnedV1) 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, "username", "password", "localhost", "quickstart", nil)
  if err != nil {
    return err
  }
  defer pool.Close()

  client, err := sqlstreams.NewClient(ctx, pool, nil)
  if err != nil {
    return err
  }

  rewards := client.Stream[RewardEarnedV1]("rewards.earned")
  if _, err := rewards.Register(ctx, nil); err != nil {
    return err
  }
  producer, err := rewards.Producer().Register(ctx, nil)
  if err != nil {
    return err
  }

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

Keep RewardId unchanged when requesting the same reward again. A different reward needs a different id, even when it belongs to the same user.

3. Record the reward and points together

Replace cmd/consume/main.go with:

package main

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

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

type RewardEarnedV1 struct {
  RewardId string `json:"reward_id"`
  UserId   string `json:"user_id"`
  Points   int    `json:"points"`
}

func (RewardEarnedV1) 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, "username", "password", "localhost", "quickstart", nil)
  if err != nil {
    return err
  }
  defer pool.Close()

  client, err := sqlstreams.NewClient(ctx, pool, nil)
  if err != nil {
    return err
  }

  rewards := client.Stream[RewardEarnedV1]("rewards.earned")
  registered, err := rewards.Get(ctx)
  if err != nil {
    return err
  }
  if registered == nil {
    return errors.New("stream not registered -- run the producer first")
  }
  consumer, err := rewards.Consumer("credit-rewards").Register(ctx, nil)
  if err != nil {
    return err
  }

  return consumer.Consume(ctx, func(ctx context.Context, reward *RewardEarnedV1) error {
    err := client.InTransaction(ctx, func(ctx context.Context, tx sqlstreams.Tx) error {
      inserted, err := tx.Exec(ctx,
        `INSERT INTO reward_credits (reward_id, user_id, points)
         VALUES ($1, $2, $3) ON CONFLICT (reward_id) DO NOTHING`,
        reward.RewardId, reward.UserId, reward.Points)
      if err != nil {
        return err
      }
      if inserted.RowsAffected() == 0 {
        return nil
      }

      _, err = tx.Exec(ctx,
        `INSERT INTO user_rewards (user_id, points) VALUES ($1, $2)
         ON CONFLICT (user_id) DO UPDATE
         SET points = user_rewards.points + EXCLUDED.points`,
        reward.UserId, reward.Points)
      return err
    })
    if err != nil {
      return err
    }
    fmt.Printf("reward recorded: %s\n", reward.RewardId)
    return nil
  }, nil)
}

So what does this InTransaction do?

On new reward: The insert into reward_credits adds a row, so RowsAffected() is 1. The code adds the points and commits both writes.
Already credited: The reward id exists, so the insert adds no row and RowsAffected() is 0. The code skips the points update and returns nil, allowing SQLStreams to record success for this message.
Points update fails: Returning the error rolls back the transaction, including the new reward record. A retry can then insert that record and attempt the points update again.

4. Run the same reward twice

Produce two messages requesting the same signup reward:

go run ./cmd/produce
go run ./cmd/produce

These are separate messages with the same RewardId. This checks the application’s duplicate protection without needing to crash a process.

Start the consumer:

go run ./cmd/consume

The consumer prints this line for each message, including a duplicate whose reward was already recorded:

reward recorded: signup:user-123

After both messages have been processed, press Ctrl-C once to stop the consumer. Check the points:

docker compose exec -T db psql -U username -d quickstart \
  -c "SELECT user_id, points FROM user_rewards WHERE user_id = 'user-123';"

The result is user-123 with 10 points, rather than 20. Keep the reward records for as long as the same rewards may be requested again. Deleting a record allows that reward to add points again.

Consumer groups — Why a consumer handler can run again after completing its work.
Run a database transaction — Commit and rollback behavior of InTransaction.
Consumer handler results — How the consumer handler reports success or asks SQLStreams to retry.
Produce messages in a database transaction — Save an application change and its message together.