Produce a message from database work

ProduceFunc(ctx context.Context, producerFunc ProducerFunc[T], options *ProduceOptions) (*ProduceResult[T], error)

Opens a transaction, calls producerFunc, and produces the returned payload in that transaction. Database writes made through the supplied tx commit with the message. This call does not use automatic batching.

Example

Use a registered producer for the payments.requested stream and the example payload. The application owns an orders table with an id text PRIMARY KEY column.

result, err := producer.ProduceFunc(ctx,
  func(ctx context.Context, tx sqlstreams.Tx) (*PaymentRequestedV1, error) {
    if _, err := tx.Exec(ctx,
      `INSERT INTO orders (id) VALUES ($1) ON CONFLICT DO NOTHING`, "order-123"); err != nil {
      return nil, err
    }
    return &PaymentRequestedV1{OrderId: "order-123"}, nil
  }, &sqlstreams.ProduceOptions{IdempotencyKey: "payment-request:order-123"})
if err != nil {
  return err
}
fmt.Println(result.Id, result.Duplicate)

Parameters

Parameter	Type	Required	Description
`ctx`	`context.Context`	Yes	Controls the transaction and retries. A commit failure can leave the outcome uncertain.
`producerFunc`	`ProducerFunc[T]`	Yes	Returns a non-`nil`, JSON-encodable payload with `SchemaVersion` at least `1`, or an error. Use its tx for database work.
`options`	`*ProduceOptions`	No	`nil` uses producer defaults. A stable `IdempotencyKey` deduplicates message inserts across calls.

ProducerFunc

type ProducerFunc[T Versioned] func(ctx context.Context, tx Tx) (*T, error)

Returning an error rolls back that attempt. SQLStreams owns transaction completion. Do not commit or roll back the supplied transaction yourself.

Returns

Condition	Return value
Transaction commits a new message	`(result, nil)` with its id.
Idempotency key already exists	`(result, nil)` with `Id == 0` and `Duplicate == true`.
Operation fails	`(nil, err)`.

The ProduceResult contains this call’s payload, including when the message is a duplicate.

Callback retries

The callback can run again during transient-error or missing-partition recovery. Keep its database writes safe to repeat and use the supplied tx. Calls to external services do not roll back with the transaction.

Errors

Error	When
Validation error	`producerFunc` is `nil`, the returned payload version is invalid, or options are invalid.
Callback error	`producerFunc` returns an error.
`ErrPayloadNotEncodable`	The returned payload cannot be encoded as JSON.
Partition creation error	The message’s covering partition cannot be created.
Database or context error	The transaction fails, retries are exhausted, or commit cannot be confirmed.

Produce a message — Options, idempotency, and result fields.
Register a producer — Defaults and slow-call logging.
Produce messages in a database transaction — A complete example that saves a user and requests a welcome email.