Ecommerce Order Fulfillment Backend on AWS

res, err := tx.ExecContext(ctx, `
	UPDATE products SET stock = stock - $1
	WHERE sku = $2 AND stock >= $1
`, qty, sku)
if err != nil {
	return err
}
if res.RowsAffected() == 0 {
	return fmt.Errorf("insufficient stock")
}

-- orders/migrations/1_create_orders.up.sql
CREATE TABLE products (
    sku        TEXT PRIMARY KEY,
    name       TEXT NOT NULL,
    stock      INT NOT NULL CHECK (stock >= 0)
);

CREATE TABLE orders (
    id             BIGSERIAL PRIMARY KEY,
    sku            TEXT NOT NULL REFERENCES products(sku),
    qty            INT NOT NULL CHECK (qty > 0),
    status         TEXT NOT NULL,
    label_key      TEXT NOT NULL DEFAULT '',
    idempotency_key TEXT NOT NULL UNIQUE,
    reserved_until TIMESTAMPTZ,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE fulfillment_outbox (
    order_id   BIGINT PRIMARY KEY REFERENCES orders(id),
    sku        TEXT NOT NULL,
    qty        INT NOT NULL,
    status     TEXT NOT NULL DEFAULT 'pending'
                 CHECK (status IN ('pending', 'processing', 'published')),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

// orders/orders.go
package orders

import (
	"context"
	"errors"
	"fmt"
	"time"

	"encore.dev/beta/errs"
	"encore.dev/cron"
	"encore.dev/pubsub"
	"encore.dev/rlog"
	"encore.dev/storage/objects"
	"encore.dev/storage/sqldb"
)

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

// Packing labels: S3 on AWS, local files with `encore run`.
var Labels = objects.NewBucket("packing-labels", objects.BucketConfig{
	Public: false,
})

type FulfillmentJob struct {
	OrderID int64  `json:"order_id"`
	SKU     string `json:"sku"`
	Qty     int    `json:"qty"`
}

var Fulfillment = pubsub.NewTopic[*FulfillmentJob]("fulfillment", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

var _ = pubsub.NewSubscription(Fulfillment, "pack-order",
	pubsub.SubscriptionConfig[*FulfillmentJob]{
		Handler: packOrder,
		RetryPolicy: &pubsub.RetryPolicy{
			MaxRetries: 5,
			MinBackoff: 5 * time.Second,
			MaxBackoff: 10 * time.Minute,
		},
	},
)

func packOrder(ctx context.Context, job *FulfillmentJob) error {
	var status string
	if err := db.QueryRow(ctx, `SELECT status FROM orders WHERE id = $1`, job.OrderID).Scan(&status); err != nil {
		return err
	}
	if status != "reserved" {
		// Reservation was cancelled or already packed; acknowledge without side effects.
		return nil
	}

	key := fmt.Sprintf("labels/%d.pdf", job.OrderID)
	// In production, call the carrier API and write the real PDF bytes.
	body := []byte(fmt.Sprintf("PACK %s x%d order=%d
", job.SKU, job.Qty, job.OrderID))

	w := Labels.Upload(ctx, key)
	if _, err := w.Write(body); err != nil {
		w.Abort(err)
		return err
	}
	if err := w.Close(); err != nil {
		return err
	}

	res, err := db.Exec(ctx, `
		UPDATE orders SET status = 'packed', label_key = $2
		WHERE id = $1 AND status = 'reserved'
	`, job.OrderID, key)
	if err != nil {
		return err
	}
	if res.RowsAffected() == 0 {
		// Lost the race with cancellation; acknowledge the job.
		return nil
	}
	return nil
}

type PlaceOrderParams struct {
	SKU            string `json:"sku"`
	Qty            int    `json:"qty"`
	IdempotencyKey string `json:"idempotency_key"`
}

type PlaceOrderResponse struct {
	OrderID   int64  `json:"order_id"`
	MessageID string `json:"message_id"`
}

// PlaceOrder reserves stock and writes a fulfillment outbox row in the same transaction.
//
//encore:api public method=POST path=/orders
func PlaceOrder(ctx context.Context, p *PlaceOrderParams) (*PlaceOrderResponse, error) {
	if p.Qty < 1 {
		return nil, &errs.Error{Code: errs.InvalidArgument, Message: "qty must be >= 1"}
	}
	if p.IdempotencyKey == "" {
		return nil, &errs.Error{Code: errs.InvalidArgument, Message: "idempotency_key required"}
	}

	tx, err := db.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	var existing int64
	err = tx.QueryRow(ctx, `SELECT id FROM orders WHERE idempotency_key = $1`, p.IdempotencyKey).Scan(&existing)
	if err == nil {
		return &PlaceOrderResponse{OrderID: existing, MessageID: ""}, nil
	}
	if !errors.Is(err, sqldb.ErrNoRows) {
		return nil, err
	}

	res, err := tx.Exec(ctx, `
		UPDATE products SET stock = stock - $1
		WHERE sku = $2 AND stock >= $1
	`, p.Qty, p.SKU)
	if err != nil {
		return nil, err
	}
	if res.RowsAffected() == 0 {
		return nil, &errs.Error{Code: errs.FailedPrecondition, Message: "insufficient stock"}
	}

	var orderID int64
	if err := tx.QueryRow(ctx, `
		INSERT INTO orders (sku, qty, status, idempotency_key, reserved_until)
		VALUES ($1, $2, 'reserved', $3, now() + interval '15 minutes')
		RETURNING id
	`, p.SKU, p.Qty, p.IdempotencyKey).Scan(&orderID); err != nil {
		return nil, err
	}

	if _, err := tx.Exec(ctx, `
		INSERT INTO fulfillment_outbox (order_id, sku, qty, status) VALUES ($1, $2, $3, 'pending')
	`, orderID, p.SKU, p.Qty); err != nil {
		return nil, err
	}

	if err := tx.Commit(); err != nil {
		return nil, err
	}

	msgID, err := Fulfillment.Publish(ctx, &FulfillmentJob{
		OrderID: orderID,
		SKU:     p.SKU,
		Qty:     p.Qty,
	})
	if err != nil {
		// Order and outbox are committed; a worker can republish unpublished outbox rows.
		return &PlaceOrderResponse{OrderID: orderID, MessageID: ""}, nil
	}
	_, _ = db.Exec(ctx, `UPDATE fulfillment_outbox SET status = 'published', updated_at = now() WHERE order_id = $1`, orderID)
	return &PlaceOrderResponse{OrderID: orderID, MessageID: msgID}, nil
}

// PublishOutbox claims one pending (or stale processing) outbox row, publishes it, then marks published.
//
//encore:api private
func PublishOutbox(ctx context.Context) error {
	tx, err := db.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	var orderID int64
	var sku string
	var qty int
	err = tx.QueryRow(ctx, `
		SELECT o.order_id, o.sku, o.qty
		FROM fulfillment_outbox AS o
		JOIN orders AS ord ON ord.id = o.order_id
		WHERE ord.status = 'reserved'
		  AND (
			o.status = 'pending'
			OR (o.status = 'processing' AND o.updated_at < now() - interval '5 minutes')
		  )
		ORDER BY o.created_at
		FOR UPDATE OF o SKIP LOCKED
		LIMIT 1
	`).Scan(&orderID, &sku, &qty)
	if err != nil {
		if errors.Is(err, sqldb.ErrNoRows) {
			return nil
		}
		return err
	}
	if _, err := tx.Exec(ctx, `
		UPDATE fulfillment_outbox SET status = 'processing', updated_at = now()
		WHERE order_id = $1
	`, orderID); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return err
	}

	if _, err := Fulfillment.Publish(ctx, &FulfillmentJob{
		OrderID: orderID, SKU: sku, Qty: qty,
	}); err != nil {
		_, _ = db.Exec(ctx, `
			UPDATE fulfillment_outbox SET status = 'pending', updated_at = now()
			WHERE order_id = $1
		`, orderID)
		return err
	}
	_, err = db.Exec(ctx, `
		UPDATE fulfillment_outbox SET status = 'published', updated_at = now()
		WHERE order_id = $1
	`, orderID)
	return err
}

var _ = cron.NewJob("publish-fulfillment-outbox", cron.JobConfig{
	Title:    "Publish unpublished fulfillment jobs",
	Every:    1 * cron.Minute,
	Endpoint: PublishOutbox,
})

type OrderStatus struct {
	ID       int64  `json:"id"`
	Status   string `json:"status"`
	LabelKey string `json:"label_key"`
}

// Status lookup stays private to the owning checkout session: clients pass the
// idempotency key they used at place-order time instead of guessing numeric IDs.
//
//encore:api public method=GET path=/orders/by-key/:key
func GetOrderByKey(ctx context.Context, key string) (*OrderStatus, error) {
	var out OrderStatus
	err := db.QueryRow(ctx, `
		SELECT id, status, label_key FROM orders WHERE idempotency_key = $1
	`, key).Scan(&out.ID, &out.Status, &out.LabelKey)
	if err != nil {
		return nil, &errs.Error{Code: errs.NotFound, Message: "order not found"}
	}
	return &out, nil
}

// Release reservations that never moved to packed.
var _ = cron.NewJob("release-reservations", cron.JobConfig{
	Title:    "Release expired stock reservations",
	Every:    1 * cron.Hour,
	Endpoint: ReleaseExpired,
})

//encore:api private
func ReleaseExpired(ctx context.Context) error {
	rows, err := db.Query(ctx, `
		SELECT id, sku, qty FROM orders
		WHERE status = 'reserved' AND reserved_until < now()
	`)
	if err != nil {
		return err
	}
	defer rows.Close()

	for rows.Next() {
		var id int64
		var sku string
		var qty int
		if err := rows.Scan(&id, &sku, &qty); err != nil {
			return err
		}
		tx, err := db.Begin(ctx)
		if err != nil {
			return err
		}
		res, err := tx.Exec(ctx, `
			UPDATE orders SET status = 'cancelled'
			WHERE id = $1 AND status = 'reserved'
		`, id)
		if err != nil {
			tx.Rollback()
			return err
		}
		if res.RowsAffected() == 1 {
			if _, err := tx.Exec(ctx, `
				DELETE FROM fulfillment_outbox WHERE order_id = $1 AND status <> 'published' 
			`, id); err != nil {
				tx.Rollback()
				return err
			}
			if _, err := tx.Exec(ctx, `
				UPDATE products SET stock = stock + $1 WHERE sku = $2
			`, qty, sku); err != nil {
				tx.Rollback()
				return err
			}
		}
		if err := tx.Commit(); err != nil {
			return err
		}
		rlog.Info("released reservation", "order_id", id, "sku", sku)
	}
	return rows.Err()
}