How to Add Pub/Sub to a chi App

// orders/orders.go
package orders

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"net/http"
	"time"

	"encore.dev/pubsub"
	"encore.dev/rlog"
	"encore.dev/storage/cache"
	"github.com/go-chi/chi/v5"
	"github.com/go-chi/chi/v5/middleware"
)

type OrderPlaced struct {
	OrderID    string `json:"order_id"`
	Email      string `json:"email"`
	TotalCents int64  `json:"total_cents"`
}

// In-memory locally, SNS + SQS on AWS or Google Pub/Sub on GCP when deployed.
var OrderEvents = pubsub.NewTopic[*OrderPlaced]("order-placed", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

//encore:service
type Service struct {
	router chi.Router
}

func initService() (*Service, error) {
	r := chi.NewRouter()
	r.Use(middleware.Recoverer)
	r.Post("/orders", placeOrder)
	return &Service{router: r}, nil
}

// Fallback sends every request that no Encore endpoint matches to chi.
//
//encore:api public raw path=/!fallback
func (s *Service) Fallback(w http.ResponseWriter, req *http.Request) {
	s.router.ServeHTTP(w, req)
}

func placeOrder(w http.ResponseWriter, r *http.Request) {
	var in struct {
		Email      string `json:"email"`
		TotalCents int64  `json:"total_cents"`
	}
	if err := json.NewDecoder(r.Body).Decode(&in); err != nil || in.Email == "" || in.TotalCents <= 0 {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "email and total_cents are required"})
		return
	}
	b := make([]byte, 8)
	rand.Read(b)
	ev := &OrderPlaced{OrderID: "ord_" + hex.EncodeToString(b), Email: in.Email, TotalCents: in.TotalCents}

	// The handler returns as soon as the event is stored; email sending
	// and anything else downstream happens in subscribers.
	if _, err := OrderEvents.Publish(r.Context(), ev); err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not queue order"})
		return
	}
	writeJSON(w, http.StatusAccepted, map[string]string{"order_id": ev.OrderID})
}

func writeJSON(w http.ResponseWriter, status int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	json.NewEncoder(w).Encode(v)
}

// Delivery is at least once, so the subscriber records which orders already
// got their email and skips duplicates.
var cluster = cache.NewCluster("orders", cache.ClusterConfig{
	EvictionPolicy: cache.NoEviction,
})

var confirmationSent = cache.NewStringKeyspace[string](cluster, cache.KeyspaceConfig{
	KeyPattern:    "confirmation-sent/:key",
	DefaultExpiry: cache.ExpireIn(7 * 24 * time.Hour),
})

var _ = pubsub.NewSubscription(OrderEvents, "send-confirmation", pubsub.SubscriptionConfig[*OrderPlaced]{
	Handler: sendConfirmation,
	RetryPolicy: &pubsub.RetryPolicy{
		MinBackoff: 10 * time.Second,
		MaxBackoff: 10 * time.Minute,
		MaxRetries: 8, // then the message moves to the dead-letter queue
	},
})

func sendConfirmation(ctx context.Context, ev *OrderPlaced) error {
	// Claim the order first. If the key exists, an earlier delivery sent it.
	err := confirmationSent.SetIfNotExists(ctx, ev.OrderID, "sending")
	if errors.Is(err, cache.KeyExists) {
		return nil
	} else if err != nil {
		return err
	}
	if err := sendEmail(ctx, ev.Email, "Order "+ev.OrderID+" confirmed"); err != nil {
		// Release the claim so the retry can send it.
		confirmationSent.Delete(ctx, ev.OrderID)
		return err
	}
	return nil
}

func sendEmail(ctx context.Context, to, subject string) error {
	rlog.Info("sending email", "to", to, "subject", subject)
	return nil // call your email provider here
}

package main

import (
	"context"
	"encoding/json"
	"log"
	"net/http"
	"os"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/sns"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	"github.com/go-chi/chi/v5"
)

type OrderPlaced struct {
	OrderID string `json:"order_id"`
	Email   string `json:"email"`
}

func main() {
	cfg, err := config.LoadDefaultConfig(context.Background())
	if err != nil {
		log.Fatal(err)
	}
	snsClient := sns.NewFromConfig(cfg)
	topicARN := os.Getenv("ORDER_TOPIC_ARN")

	r := chi.NewRouter()
	r.Post("/orders", func(w http.ResponseWriter, r *http.Request) {
		ev := OrderPlaced{OrderID: "ord_123", Email: "a@example.com"}
		body, _ := json.Marshal(ev)
		_, err := snsClient.Publish(r.Context(), &sns.PublishInput{
			TopicArn: aws.String(topicARN),
			Message:  aws.String(string(body)),
		})
		if err != nil {
			http.Error(w, "could not queue order", http.StatusInternalServerError)
			return
		}
		w.WriteHeader(http.StatusAccepted)
	})

	go consume(context.Background(), sqs.NewFromConfig(cfg), os.Getenv("CONFIRMATION_QUEUE_URL"))
	log.Fatal(http.ListenAndServe(":8080", r))
}

// consume long-polls the queue subscribed to the topic. With raw message
// delivery enabled on the subscription, the body is the event JSON.
func consume(ctx context.Context, client *sqs.Client, queueURL string) {
	for {
		out, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
			QueueUrl:            aws.String(queueURL),
			MaxNumberOfMessages: 10,
			WaitTimeSeconds:     20,
		})
		if err != nil {
			log.Print(err)
			continue
		}
		for _, m := range out.Messages {
			var ev OrderPlaced
			if err := json.Unmarshal([]byte(aws.ToString(m.Body)), &ev); err != nil {
				continue // left for the redrive policy to move to the DLQ
			}
			if err := sendConfirmation(ctx, ev); err != nil {
				continue // becomes visible again after the visibility timeout
			}
			client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
				QueueUrl:      aws.String(queueURL),
				ReceiptHandle: m.ReceiptHandle,
			})
		}
	}
}

func sendConfirmation(ctx context.Context, ev OrderPlaced) error { return nil }