Shipment Tracking Event Pipeline on AWS and GCP

// Hand-rolled SNS publish after validating the scan payload.
out, err := snsClient.Publish(ctx, &sns.PublishInput{
	TopicArn: aws.String(statusTopicArn),
	Message:  aws.String(string(payloadJSON)),
})

-- shipments/migrations/1_create_shipments.up.sql
CREATE TABLE shipments (
    id            BIGSERIAL PRIMARY KEY,
    tracking_code TEXT NOT NULL UNIQUE,
    status        TEXT NOT NULL,
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE shipment_events (
    id            BIGSERIAL PRIMARY KEY,
    event_id      TEXT NOT NULL UNIQUE,
    shipment_id   BIGINT NOT NULL REFERENCES shipments(id),
    status        TEXT NOT NULL,
    location      TEXT NOT NULL DEFAULT '',
    recorded_at   TIMESTAMPTZ NOT NULL,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE webhook_deliveries (
    event_id    TEXT PRIMARY KEY,
    delivered_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

// shipments/shipments.go
package shipments

import (
	"context"
	"errors"
	"time"

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

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

var secrets struct {
	CarrierToken string // shared secret for handheld/carrier scanners
}

//encore:authhandler
func AuthHandler(ctx context.Context, token string) (auth.UID, error) {
	if token == "" || token != secrets.CarrierToken {
		return "", &errs.Error{Code: errs.Unauthenticated, Message: "invalid token"}
	}
	return auth.UID("carrier"), nil
}

// StatusEvent is one scan or status change. event_id makes handlers idempotent.
type StatusEvent struct {
	EventID      string    `json:"event_id"`
	TrackingCode string    `json:"tracking_code"`
	Status       string    `json:"status"`
	Location     string    `json:"location"`
	RecordedAt   time.Time `json:"recorded_at"`
}

var StatusEvents = pubsub.NewTopic[*StatusEvent]("shipment-status", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

// Projector: update the tracking store. Safe to run twice for the same event_id.
var _ = pubsub.NewSubscription(StatusEvents, "project-tracking",
	pubsub.SubscriptionConfig[*StatusEvent]{
		Handler: applyStatus,
		RetryPolicy: &pubsub.RetryPolicy{
			MaxRetries: 8,
			MinBackoff: 2 * time.Second,
			MaxBackoff: 5 * time.Minute,
		},
	},
)

func applyStatus(ctx context.Context, ev *StatusEvent) error {
	var shipmentID int64
	err := db.QueryRow(ctx, `
		INSERT INTO shipments (tracking_code, status, updated_at)
		VALUES ($1, $2, $3)
		ON CONFLICT (tracking_code) DO UPDATE
		  SET status = CASE
				WHEN EXCLUDED.updated_at >= shipments.updated_at THEN EXCLUDED.status
				ELSE shipments.status
			  END,
			  updated_at = GREATEST(shipments.updated_at, EXCLUDED.updated_at)
		RETURNING id
	`, ev.TrackingCode, ev.Status, ev.RecordedAt).Scan(&shipmentID)
	if err != nil {
		return err
	}

	_, err = db.Exec(ctx, `
		INSERT INTO shipment_events (event_id, shipment_id, status, location, recorded_at)
		VALUES ($1, $2, $3, $4, $5)
		ON CONFLICT (event_id) DO NOTHING
	`, ev.EventID, shipmentID, ev.Status, ev.Location, ev.RecordedAt)
	return err
}

// Webhook fan-out: separate subscription so a slow partner does not block the store.
var _ = pubsub.NewSubscription(StatusEvents, "notify-webhooks",
	pubsub.SubscriptionConfig[*StatusEvent]{
		Handler: notifyWebhook,
		RetryPolicy: &pubsub.RetryPolicy{
			MaxRetries: 10,
			MinBackoff: 5 * time.Second,
			MaxBackoff: 15 * time.Minute,
		},
	},
)

func notifyWebhook(ctx context.Context, ev *StatusEvent) error {
	var already string
	err := db.QueryRow(ctx, `
		SELECT event_id FROM webhook_deliveries WHERE event_id = $1
	`, ev.EventID).Scan(&already)
	if err == nil {
		// Already delivered for this event_id.
		return nil
	}
	if !errors.Is(err, sqldb.ErrNoRows) {
		return err
	}
	// POST to the shop webhook URL here. Return an error to retry before marking delivered.
	rlog.Info("webhook sent", "tracking", ev.TrackingCode, "status", ev.Status)
	_, err = db.Exec(ctx, `
		INSERT INTO webhook_deliveries (event_id) VALUES ($1)
		ON CONFLICT (event_id) DO NOTHING
	`, ev.EventID)
	return err
}

type RecordScanParams struct {
	EventID      string `json:"event_id"`
	TrackingCode string `json:"tracking_code"`
	Status       string `json:"status"`
	Location     string `json:"location"`
}

type RecordScanResponse struct {
	MessageID string `json:"message_id"`
}

// RecordScan accepts a scan from a handheld or carrier feed and enqueues it.
// Auth-gated so only verified carrier or warehouse credentials can rewrite status.
//
//encore:api auth method=POST path=/scans
func RecordScan(ctx context.Context, p *RecordScanParams) (*RecordScanResponse, error) {
	id, err := StatusEvents.Publish(ctx, &StatusEvent{
		EventID:      p.EventID,
		TrackingCode: p.TrackingCode,
		Status:       p.Status,
		Location:     p.Location,
		RecordedAt:   time.Now().UTC(),
	})
	if err != nil {
		return nil, err
	}
	return &RecordScanResponse{MessageID: id}, nil
}

type TrackingResponse struct {
	TrackingCode string `json:"tracking_code"`
	Status       string `json:"status"`
	UpdatedAt    time.Time `json:"updated_at"`
}

//encore:api public method=GET path=/track/:code
func Track(ctx context.Context, code string) (*TrackingResponse, error) {
	var out TrackingResponse
	err := db.QueryRow(ctx, `
		SELECT tracking_code, status, updated_at FROM shipments WHERE tracking_code = $1
	`, code).Scan(&out.TrackingCode, &out.Status, &out.UpdatedAt)
	if err != nil {
		return nil, err
	}
	return &out, nil
}

// Hourly sweep for shipments stuck in "in_transit" longer than 48 hours.
// Cron jobs run in deployed environments, not under `encore run`.
var _ = cron.NewJob("reconcile-stuck", cron.JobConfig{
	Title:    "Reconcile stuck shipments",
	Every:    1 * cron.Hour,
	Endpoint: ReconcileStuck,
})

//encore:api private
func ReconcileStuck(ctx context.Context) error {
	rows, err := db.Query(ctx, `
		SELECT tracking_code FROM shipments
		WHERE status = 'in_transit' AND updated_at < now() - interval '48 hours'
	`)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var code string
		if err := rows.Scan(&code); err != nil {
			return err
		}
		rlog.Warn("stuck shipment", "tracking", code)
		// Publish an exception event or open a support ticket.
	}
	return rows.Err()
}