How do I publish messages from an Echo handler?

Do the database write first, then publish an event that carries the IDs a consumer needs and return the response without waiting for the consumer. Use c.Request().Context() for the publish call, and keep slow work such as sending email or calling third-party APIs in the subscriber.

How do I add Pub/Sub to an Echo app with Encore?

Declare a topic with pubsub.NewTopic[*Event]("name", pubsub.TopicConfig{DeliveryGuarantee: pubsub.AtLeastOnce}), call Publish from the Echo handler, and declare pubsub.NewSubscription in the consuming service with a RetryPolicy. Encore runs it locally with encore run and creates SNS and SQS with a dead-letter queue on AWS, or Google Pub/Sub on GCP.

How do I stop a subscriber from sending the same email twice?

Claim the work in the database inside a transaction, for example by inserting (user_id, kind) into a table with that pair as the primary key and ON CONFLICT DO NOTHING. Send only when the insert affected a row, and commit after the send succeeds so a failed send is retried.

How to Add Pub/Sub to an Echo App in Go (2026)

An Echo signup handler that writes the user row and then calls an email provider inline makes the user wait on that provider, and returns a 500 for a signup that already succeeded whenever the provider has a bad minute. Moving the email to a message queue fixes both, and brings its own setup: a topic, a queue per consumer, retries, a dead-letter queue and permissions for the publisher and the consumer in each environment.

This guide moves the welcome email of an Echo API onto Pub/Sub. The Echo handler publishes a UserSignedUp event, and a subscriber in a separate service sends the email exactly once per user even when the event is delivered twice. The code is shown with Encore, then with SNS and SQS wired up by hand.

Pub/Sub options for an Echo app at a glance

	Encore	SNS + SQS by hand
Topic and queue declared in	Go code (`pubsub.NewTopic`, `NewSubscription`)	Terraform or the AWS console
Local development	Topic and subscriber run under `encore run`	LocalStack or a shared dev account
Retries and dead-letter queue	`RetryPolicy` on the subscription	`redrive_policy` and a second queue
Consumer loop	Handler function called per event	Polling loop with `ReceiveMessage` and `DeleteMessage`
Permissions	Granted to the publishing and consuming services	IAM policies you write per role
Production	SNS and SQS on AWS, Pub/Sub on GCP, in your account	SNS and SQS you provision

How do you add Pub/Sub to Echo with Encore?

Encore works alongside Echo, and the topic, the subscription and the databases on both sides are declared in Go next to the code that uses them, then provisioned in your own AWS or GCP account on deploy. Echo keeps its router behind a raw fallback route in the accounts service, and the subscriber lives in an emails service with its own database.

Set up the app

Get started

brew install encoredev/tap/encore &&

encore app create

Copy

Run in your terminal to get started locally.

$ encore app create signups --example=hello-world$ cd signups && go get github.com/labstack/echo/v4@v4.16.3$ mkdir -p accounts/migrations emails/migrations

-- accounts/migrations/1_create_users.up.sql
CREATE TABLE users (
    id BIGSERIAL PRIMARY KEY,
    email TEXT NOT NULL UNIQUE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

The accounts service declares the topic and publishes to it from the Echo handler after the insert succeeds:

// accounts/accounts.go
package accounts

import (
	"net/http"
	"strings"

	"encore.dev/pubsub"
	"encore.dev/storage/sqldb"
	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
)

var db = sqldb.NewDatabase("accounts", sqldb.DatabaseConfig{
	Migrations: "./migrations",
})

type UserSignedUp struct {
	UserID int64  `json:"user_id"`
	Email  string `json:"email"`
}

// SNS and SQS on AWS, Google Pub/Sub on GCP, in-process with `encore run`.
var Signups = pubsub.NewTopic[*UserSignedUp]("user-signed-up", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

//encore:service
type Service struct {
	e *echo.Echo
}

func initService() (*Service, error) {
	e := echo.New()
	e.Use(middleware.Recover())
	e.POST("/signup", signup)
	return &Service{e: e}, nil
}

//encore:api public raw path=/!fallback
func (s *Service) Fallback(w http.ResponseWriter, req *http.Request) {
	s.e.ServeHTTP(w, req)
}

func signup(c echo.Context) error {
	var in struct {
		Email string `json:"email"`
	}
	if err := c.Bind(&in); err != nil || !strings.Contains(in.Email, "@") {
		return echo.NewHTTPError(http.StatusBadRequest, "a valid email is required")
	}
	ctx := c.Request().Context()

	var id int64
	err := db.QueryRow(ctx, `
		INSERT INTO users (email) VALUES ($1)
		ON CONFLICT (email) DO NOTHING
		RETURNING id
	`, in.Email).Scan(&id)
	if err == sqldb.ErrNoRows {
		return echo.NewHTTPError(http.StatusConflict, "email already registered")
	} else if err != nil {
		return err
	}

	// The response does not wait for the email provider.
	if _, err := Signups.Publish(ctx, &UserSignedUp{UserID: id, Email: in.Email}); err != nil {
		return err
	}
	return c.JSON(http.StatusCreated, map[string]int64{"id": id})
}

The emails service subscribes with a retry policy and keeps a sent_emails table whose primary key is (user_id, kind), which is what makes a second delivery of the same event harmless:

// emails/emails.go
package emails

import (
	"context"
	"time"

	"encore.app/accounts"
	"encore.dev/pubsub"
	"encore.dev/rlog"
	"encore.dev/storage/sqldb"
)

var db = sqldb.NewDatabase("emails", sqldb.DatabaseConfig{
	Migrations: "./migrations",
})

var _ = pubsub.NewSubscription(accounts.Signups, "send-welcome-email",
	pubsub.SubscriptionConfig[*accounts.UserSignedUp]{
		Handler:     sendWelcome,
		AckDeadline: 60 * time.Second,
		RetryPolicy: &pubsub.RetryPolicy{
			MinBackoff: 10 * time.Second,
			MaxBackoff: 10 * time.Minute,
			MaxRetries: 8,
		},
	},
)

// sendWelcome is idempotent: the sent_emails row is claimed before the
// email goes out, so a redelivered event does not send a second email.
func sendWelcome(ctx context.Context, ev *accounts.UserSignedUp) error {
	tx, err := db.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	res, err := tx.Exec(ctx, `
		INSERT INTO sent_emails (user_id, kind) VALUES ($1, 'welcome')
		ON CONFLICT DO NOTHING
	`, ev.UserID)
	if err != nil {
		return err
	}
	if res.RowsAffected() == 0 {
		return nil // already sent
	}
	if err := deliver(ctx, ev.Email, "Welcome aboard"); err != nil {
		return err // rolls back the claim, so the retry sends again
	}
	return tx.Commit()
}

func deliver(ctx context.Context, to, subject string) error {
	rlog.Info("sending email", "to", to, "subject", subject)
	return nil
}

The emails/migrations/1_create_sent.up.sql file creates sent_emails (user_id BIGINT, kind TEXT, sent_at TIMESTAMPTZ, PRIMARY KEY (user_id, kind)). encore run starts Postgres locally for both databases and runs the topic in-process, so curl -X POST localhost:4000/signup -d '{"email":"ada@example.com"}' returns at once and the welcome email shows up as a separate trace in the local development dashboard. A handler that returns an error is retried with backoff, and after MaxRetries the event moves to the subscription's dead-letter queue, as the Pub/Sub docs describe.

When the app deploys to AWS, Encore creates the SNS topic, an SQS queue and dead-letter queue for the subscription, and the permissions for accounts to publish and emails to consume, all inside your account and listed in the AWS infrastructure docs. On GCP the same code becomes a Pub/Sub topic and subscription. A coding agent that adds a second consumer, say a CRM sync, declares another NewSubscription on accounts.Signups in a new service and runs the whole flow locally before anyone reviews it, with no queue or policy to request first.

Best fit: Echo APIs that want to hand slow or flaky work to a subscriber with retries and a dead-letter queue, with the same code running locally and on SNS and SQS or Pub/Sub in their own cloud account.

The hand-built version publishes with the AWS SDK for Go v2 from the Echo handler and runs a polling worker as a separate process:

func runWorker(ctx context.Context, q *sqs.Client, queueURL string) {
	for ctx.Err() == nil {
		out, err := q.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
			QueueUrl:            &queueURL,
			MaxNumberOfMessages: 10,
			WaitTimeSeconds:     20,
		})
		if err != nil {
			time.Sleep(time.Second)
			continue
		}
		for _, m := range out.Messages {
			var ev UserSignedUp
			if json.Unmarshal([]byte(*m.Body), &ev) != nil || sendWelcome(ctx, &ev) != nil {
				continue // visibility timeout expires and SQS redelivers
			}
			q.DeleteMessage(ctx, &sqs.DeleteMessageInput{
				QueueUrl:      &queueURL,
				ReceiptHandle: m.ReceiptHandle,
			})
		}
	}
}

Around that loop sit an SNS topic, an SQS queue subscribed to it with raw message delivery, a dead-letter queue attached through a redrive policy, a queue policy that lets the topic send, and two IAM policies, one allowing sns:Publish for the API's task role and one allowing receive and delete for the worker's role. The worker needs its own ECS service and scaling settings, and local development means LocalStack or a shared AWS account where two developers drain each other's queue. The SQS developer guide covers the redrive settings.

Best fit: teams whose SNS topics and SQS queues are already defined in Terraform and shared with services outside the Echo app.

How to choose

Pick Encore to keep Echo and get a topic, a retrying subscriber and a dead-letter queue that run under encore run and are created in your own AWS or GCP account from the same Go code. Pick SNS and SQS by hand when the queues already exist in Terraform or other systems publish to them, and plan for the worker process, the policies and a local substitute.

How to Add Pub/Sub to an Echo App