Stage 11 - Real Storage: Postgres

# Local development dependencies. Postgres is added now; Redpanda joins in stage 12.
# Run `docker compose up -d`, then run the app on your host. the **Swap the engine** section shows a
# temporary `main.go`; stage 13 adds the `server` subcommand and shows how to add
# the app itself as a service.

services:
  db:
    image: postgres:16-alpine
    container_name: minimart-db
    environment:
      POSTGRES_USER: minimart
      POSTGRES_PASSWORD: minimart
      POSTGRES_DB: minimart
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U minimart"]
      interval: 5s
      timeout: 3s
      retries: 10

volumes:
  pgdata:

// Package conf reads all configuration from environment variables in one place.
// Defaults target the local docker-compose stack, so the service runs with no
// .env file in development. In production every value comes from the environment.
package conf

import (
	"os"
)

func env(key, def string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return def
}

// DatabaseURL is the primary (read-write) Postgres DSN. Writes go here.
func DatabaseURL() string {
	return env("DATABASE_URL", "postgres://minimart:minimart@localhost:5432/minimart?sslmode=disable")
}

// DatabaseReplicaURL is the read-replica DSN. Reads go here. It falls back to the
// primary when unset - the common case in local development, where there is only
// one database.
func DatabaseReplicaURL() string {
	if v := os.Getenv("DATABASE_REPLICA_URL"); v != "" {
		return v
	}
	return DatabaseURL()
}

// HTTPAddr is the address the HTTP server listens on.
func HTTPAddr() string {
	return env("HTTP_ADDR", ":8080")
}

// Env identifies the deployment environment for logs or environment-specific
// behavior at the edges.
func Env() string {
	return env("ENV", "development")
}

HTTP_ADDR=":8080"
ENV="development"

# Primary (read-write) Postgres. Reads use DATABASE_REPLICA_URL if set, else this.
DATABASE_URL="postgres://minimart:minimart@localhost:5432/minimart?sslmode=disable"
# DATABASE_REPLICA_URL="postgres://minimart:minimart@localhost:5433/minimart?sslmode=disable"

docker compose up -d

docker compose exec db psql -U minimart -c '\dt'

Did not find any relations.

-- migrate:up
CREATE TABLE products (
    id          TEXT        PRIMARY KEY,
    name        TEXT        NOT NULL,
    price_cents BIGINT      NOT NULL,
    stock       INT         NOT NULL DEFAULT 0,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- migrate:down
DROP TABLE IF EXISTS products;

-- migrate:up
CREATE TABLE orders (
    id          TEXT        PRIMARY KEY,
    user_id     TEXT        NOT NULL,
    total_cents BIGINT      NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE order_items (
    id               BIGSERIAL PRIMARY KEY,
    order_id         TEXT      NOT NULL REFERENCES orders (id),
    product_id       TEXT      NOT NULL,
    name             TEXT      NOT NULL,
    unit_price_cents BIGINT    NOT NULL,
    quantity         INT       NOT NULL,
    line_total_cents BIGINT    NOT NULL
);

CREATE INDEX order_items_order_id_idx ON order_items (order_id);

-- migrate:down
DROP TABLE IF EXISTS order_items;
DROP TABLE IF EXISTS orders;

// Package db applies schema migrations using dbmate as a library (not the CLI),
// so the same binary that serves traffic can also run `minimart migrate`.
package db

import (
	"fmt"
	"net/url"

	"github.com/amacneil/dbmate/v2/pkg/dbmate"
	_ "github.com/amacneil/dbmate/v2/pkg/driver/postgres" // registers the postgres driver

	"minimart/conf"
)

type Migrator struct {
	m *dbmate.DB
}

// New builds a Migrator pointed at the primary database. dbmate looks for
// migration files in ./db/migrations by default.
func New() (*Migrator, error) {
	u, err := url.Parse(conf.DatabaseURL())
	if err != nil {
		return nil, fmt.Errorf("parse DATABASE_URL: %w", err)
	}
	m := dbmate.New(u)
	m.AutoDumpSchema = false // don't write a schema.sql dump; migrations are the source of truth
	return &Migrator{m: m}, nil
}

// Migrate creates the database if it does not exist, then applies every pending
// migration in timestamp order. It is safe to run on every deploy.
func (mig *Migrator) Migrate() error {
	if err := mig.m.CreateAndMigrate(); err != nil {
		return fmt.Errorf("create and migrate: %w", err)
	}
	return nil
}

package cmd

import "minimart/db"

// RunMigrate applies all pending schema migrations, then exits. Deploys run this
// as a one-shot step before starting the server.
func RunMigrate() error {
	m, err := db.New()
	if err != nil {
		return err
	}
	return m.Migrate()
}

version: "2"

# Each domain gets a separate reader and writer package, generated from that
# domain's queries.sql. The schema source of truth is the dbmate migrations, so
# sqlc and the database can never drift. Regenerate with `sqlc generate`.
# Never hand-edit the generated *.sql.go / models.go / db.go files.
sql:
  - engine: "postgresql"
    schema: "db/migrations"
    queries: "catalog/infrastructure/postgres/reader/queries.sql"
    gen:
      go: &go
        package: "reader"
        out: "catalog/infrastructure/postgres/reader"
        sql_package: "pgx/v5"
        emit_pointers_for_null_types: true

  - engine: "postgresql"
    schema: "db/migrations"
    queries: "catalog/infrastructure/postgres/writer/queries.sql"
    gen:
      go:
        <<: *go
        package: "writer"
        out: "catalog/infrastructure/postgres/writer"

  - engine: "postgresql"
    schema: "db/migrations"
    queries: "order/infrastructure/postgres/reader/queries.sql"
    gen:
      go:
        <<: *go
        package: "reader"
        out: "order/infrastructure/postgres/reader"

  - engine: "postgresql"
    schema: "db/migrations"
    queries: "order/infrastructure/postgres/writer/queries.sql"
    gen:
      go:
        <<: *go
        package: "writer"
        out: "order/infrastructure/postgres/writer"

-- name: GetProduct :one
SELECT id, name, price_cents, stock, created_at, updated_at
FROM products
WHERE id = $1;

-- name: ListProducts :many
SELECT id, name, price_cents, stock, created_at, updated_at
FROM products
ORDER BY created_at;

-- name: SaveProduct :exec
INSERT INTO products (id, name, price_cents, stock)
VALUES ($1, $2, $3, $4)
ON CONFLICT (id) DO UPDATE
SET name        = EXCLUDED.name,
    price_cents = EXCLUDED.price_cents,
    stock       = EXCLUDED.stock,
    updated_at  = NOW();

-- name: GetOrder :one
SELECT id, user_id, total_cents, created_at
FROM orders
WHERE id = $1;

-- name: ListOrderItems :many
SELECT product_id, name, unit_price_cents, quantity, line_total_cents
FROM order_items
WHERE order_id = $1
ORDER BY id;

-- name: CreateOrder :exec
INSERT INTO orders (id, user_id, total_cents)
VALUES ($1, $2, $3);

-- name: CreateOrderItem :exec
INSERT INTO order_items (order_id, product_id, name, unit_price_cents, quantity, line_total_cents)
VALUES ($1, $2, $3, $4, $5, $6);

sqlc generate

ls catalog/infrastructure/postgres/reader

db.go  models.go  queries.sql.go

package postgres

import (
	"context"
	"errors"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"minimart/catalog/domain"
	"minimart/catalog/infrastructure/postgres/reader"
	"minimart/catalog/infrastructure/postgres/writer"
)

// CatalogRepository implements the catalog domain.Repository port over Postgres.
// It holds sqlc query objects bound to a read pool and a write pool.
type CatalogRepository struct {
	reader *reader.Queries
	writer *writer.Queries
}

// Compile-time proof the Postgres adapter satisfies the same port as the
// in-memory one. This is why swapping storage touches no business code.
var _ domain.Repository = (*CatalogRepository)(nil)

func NewCatalogRepository(readPool, writePool *pgxpool.Pool) *CatalogRepository {
	return &CatalogRepository{
		reader: reader.New(readPool),
		writer: writer.New(writePool),
	}
}

func (r *CatalogRepository) Save(ctx context.Context, p domain.Product) error {
	return r.writer.SaveProduct(ctx, writer.SaveProductParams{
		ID:         p.ID,
		Name:       p.Name,
		PriceCents: p.PriceCents,
		Stock:      int32(p.Stock),
	})
}

func (r *CatalogRepository) GetByID(ctx context.Context, id string) (domain.Product, error) {
	row, err := r.reader.GetProduct(ctx, id)
	if err != nil {
		// Translate the driver's "no rows" into the domain's not-found error, so
		// the application layer never learns which database we use.
		if errors.Is(err, pgx.ErrNoRows) {
			return domain.Product{}, domain.ErrProductNotFound
		}
		return domain.Product{}, err
	}
	return domain.Product{
		ID:         row.ID,
		Name:       row.Name,
		PriceCents: row.PriceCents,
		Stock:      int(row.Stock),
	}, nil
}

func (r *CatalogRepository) List(ctx context.Context) ([]domain.Product, error) {
	rows, err := r.reader.ListProducts(ctx)
	if err != nil {
		return nil, err
	}
	out := make([]domain.Product, 0, len(rows))
	for _, row := range rows {
		out = append(out, domain.Product{
			ID:         row.ID,
			Name:       row.Name,
			PriceCents: row.PriceCents,
			Stock:      int(row.Stock),
		})
	}
	return out, nil
}

package postgres

import (
	"context"
	"errors"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"minimart/order/domain"
	"minimart/order/infrastructure/postgres/reader"
	"minimart/order/infrastructure/postgres/writer"
)

type OrderRepository struct {
	writePool *pgxpool.Pool
	reader    *reader.Queries
	writer    *writer.Queries
}

var _ domain.Repository = (*OrderRepository)(nil)

func NewOrderRepository(readPool, writePool *pgxpool.Pool) *OrderRepository {
	return &OrderRepository{
		writePool: writePool,
		reader:    reader.New(readPool),
		writer:    writer.New(writePool),
	}
}

// Save writes the order and all its items in one transaction. If any item insert
// fails, the deferred Rollback undoes the whole thing.
func (r *OrderRepository) Save(ctx context.Context, o domain.Order) error {
	tx, err := r.writePool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx) // no-op after a successful Commit

	q := r.writer.WithTx(tx)
	if err := q.CreateOrder(ctx, writer.CreateOrderParams{
		ID:         o.ID,
		UserID:     o.UserID,
		TotalCents: o.TotalCents,
	}); err != nil {
		return err
	}
	for _, it := range o.Items {
		if err := q.CreateOrderItem(ctx, writer.CreateOrderItemParams{
			OrderID:        o.ID,
			ProductID:      it.ProductID,
			Name:           it.Name,
			UnitPriceCents: it.UnitPriceCents,
			Quantity:       int32(it.Quantity),
			LineTotalCents: it.LineTotalCents,
		}); err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

func (r *OrderRepository) GetByID(ctx context.Context, id string) (domain.Order, error) {
	row, err := r.reader.GetOrder(ctx, id)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return domain.Order{}, domain.ErrOrderNotFound
		}
		return domain.Order{}, err
	}
	items, err := r.reader.ListOrderItems(ctx, id)
	if err != nil {
		return domain.Order{}, err
	}
	domItems := make([]domain.Item, 0, len(items))
	for _, it := range items {
		domItems = append(domItems, domain.Item{
			ProductID:      it.ProductID,
			Name:           it.Name,
			UnitPriceCents: it.UnitPriceCents,
			Quantity:       int(it.Quantity),
			LineTotalCents: it.LineTotalCents,
		})
	}
	return domain.Order{
		ID:         row.ID,
		UserID:     row.UserID,
		TotalCents: row.TotalCents,
		Items:      domItems,
		CreatedAt:  row.CreatedAt,
	}, nil
}

// Package catalog wires the catalog domain into the modular monolith. It is the
// domain's composition root: the one place that names concrete adapters.
package catalog

import (
	"context"
	"net/http"

	"github.com/jackc/pgx/v5/pgxpool"

	"minimart/catalog/application"
	"minimart/catalog/infrastructure/postgres"
	cataloghttp "minimart/catalog/presentation/http"
	catalogmodule "minimart/catalog/presentation/module"
	"minimart/pkg/module"
	"minimart/pkg/module/contracts"
)

// Module is the catalog domain's self-wiring unit.
type Module struct {
	readPool  *pgxpool.Pool
	writePool *pgxpool.Pool
	svc       *application.Service
	handlers  *cataloghttp.Handlers
}

var _ module.Module = (*Module)(nil)

func NewModule(readPool, writePool *pgxpool.Pool) *Module {
	return &Module{readPool: readPool, writePool: writePool}
}

func (m *Module) Name() string { return "catalog" }

// Init chooses the concrete storage adapter. Swapping storage is this one line;
// no other layer is touched. (Earlier stages used memory.New() here.)
func (m *Module) Init(ctx context.Context) error {
	repo := postgres.NewCatalogRepository(m.readPool, m.writePool)
	m.svc = application.NewService(repo)
	m.handlers = cataloghttp.NewHandlers(m.svc)
	return nil
}

func (m *Module) RegisterContracts(ctx context.Context, c *contracts.Contracts) error {
	c.Catalog = catalogmodule.New(m.svc)
	return nil
}

func (m *Module) RegisterHTTP(ctx context.Context, mux *http.ServeMux) error {
	m.handlers.Register(mux)
	return nil
}

package order

import (
	"context"
	"net/http"

	"github.com/jackc/pgx/v5/pgxpool"

	"minimart/order/application"
	"minimart/order/infrastructure/postgres"
	orderhttp "minimart/order/presentation/http"
	"minimart/pkg/module"
	"minimart/pkg/module/contracts"
)

type Module struct {
	registry  *contracts.Contracts
	readPool  *pgxpool.Pool
	writePool *pgxpool.Pool
	svc       *application.Service
	handlers  *orderhttp.Handlers
}

var _ module.Module = (*Module)(nil)

func NewModule(registry *contracts.Contracts, readPool, writePool *pgxpool.Pool) *Module {
	return &Module{registry: registry, readPool: readPool, writePool: writePool}
}

func (m *Module) Name() string { return "order" }

func (m *Module) Init(ctx context.Context) error {
	repo := postgres.NewOrderRepository(m.readPool, m.writePool)
	m.svc = application.NewService(repo, m.registry)
	m.handlers = orderhttp.NewHandlers(m.svc)
	return nil
}

func (m *Module) RegisterContracts(ctx context.Context, c *contracts.Contracts) error { return nil }

func (m *Module) RegisterHTTP(ctx context.Context, mux *http.ServeMux) error {
	m.handlers.Register(mux)
	return nil
}

package cmd

import (
	"context"
	"fmt"

	"github.com/jackc/pgx/v5/pgxpool"

	"minimart/conf"
)

// newPools builds a read pool (replica) and a write pool (primary). Locally both
// DSNs resolve to the same database; in production they point at different hosts.
func newPools(ctx context.Context) (readPool, writePool *pgxpool.Pool, err error) {
	writePool, err = pgxpool.New(ctx, conf.DatabaseURL())
	if err != nil {
		return nil, nil, fmt.Errorf("connect primary: %w", err)
	}
	readPool, err = pgxpool.New(ctx, conf.DatabaseReplicaURL())
	if err != nil {
		writePool.Close()
		return nil, nil, fmt.Errorf("connect replica: %w", err)
	}
	return readPool, writePool, nil
}

package cmd

import (
	"context"
	"log"
	"net/http"

	"minimart/catalog"
	"minimart/conf"
	"minimart/order"
	"minimart/pkg/module"
	"minimart/pkg/module/contracts"
)

// StartServer runs the HTTP API. It builds the DB pools, then runs the four-phase
// module boot and serves.
func StartServer() error {
	ctx := context.Background()

	readPool, writePool, err := newPools(ctx)
	if err != nil {
		return err
	}
	defer writePool.Close()
	defer readPool.Close()

	mux := http.NewServeMux()
	mux.HandleFunc("GET /health", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"status":"ok"}`))
	})

	registry := &contracts.Contracts{}
	modules := []module.Module{
		catalog.NewModule(readPool, writePool),
		order.NewModule(registry, readPool, writePool),
	}

	for _, m := range modules {
		if err := m.Init(ctx); err != nil {
			return err
		}
	}
	for _, m := range modules {
		if err := m.RegisterContracts(ctx, registry); err != nil {
			return err
		}
	}
	if err := registry.Verify(); err != nil {
		return err
	}
	for _, m := range modules {
		if err := m.RegisterHTTP(ctx, mux); err != nil {
			return err
		}
	}

	log.Printf("minimart server listening on %s", conf.HTTPAddr())
	return http.ListenAndServe(conf.HTTPAddr(), mux)
}