The Order Repository

สิ่งที่จะสร้าง

migrations/order/0001_init.sql (ตาราง orders, order_items, และ outbox) และ services/order/internal/repo/orders.go — OrderRepo ที่มีเมธอด Create insert ตัวคำสั่งซื้อ รายการสินค้า และ outbox row ที่อธิบาย event ทั้งหมดไว้ใน pgx.Tx เดียว จากนั้น services/order/internal/server/server.go — Server ตัวจริงที่ implement ทั้งสามเมธอดของ OrderServiceServer แทนที่ placeholder จาก The gRPC Server → และในที่สุดก็นำ Catalog client ที่ inject ไว้มาใช้งานจริง: CreateOrder หาราคาปัจจุบันของแต่ละรายการจาก Catalog ก่อนที่จะแตะ repository เลย

ทำไม

OrderRepo.Create เขียนลงสามตาราง — orders, order_items, และ outbox — และทั้งสามการเขียนใช้ transaction เดียวกันโดยตั้งใจ ตัวคำสั่งซื้อ รายการสินค้าที่ประกอบขึ้นเป็นคำสั่งซื้อนั้น และข้อเท็จจริงว่า event นี้เกิดขึ้นแล้ว คือความจริงชุดเดียวที่ต้องเป็นอะตอมิก

ถ้า crash เกิดขึ้นระหว่างการ insert order กับการ insert outbox row คำสั่งซื้อจะมีอยู่ในฐานข้อมูลโดยไม่มีบันทึกที่ไหนเลยว่า order.created ควร publish ออกไป consumer ปลายทางทุกตัว รวมถึง Payment ใน Order Saga → Module 9 จะไม่มีวันรู้ว่าคำสั่งซื้อใบนี้มีอยู่จริง transaction เดียวทำให้ failure mode นี้เป็นไปไม่ได้ด้วยโครงสร้าง คือทั้งสามการเขียนสำเร็จพร้อมกัน หรือไม่สำเร็จเลยสักอย่าง

The Transactional Outbox → คือจุดที่ เหตุผล เบื้องหลังการเขียน event ลงตารางฐานข้อมูลแทนการ publish ตรงไป Kafka จะได้รับคำอธิบายเต็มรูปแบบ ส่วนบทนี้แค่เขียนโค้ดที่ทำให้การรับประกันแบบ transaction เดียวเป็นจริง

การตั้งราคาอยู่ในชั้น Server ไม่ใช่ repository ด้วยเหตุผลเดียวกับที่ validation อยู่ใน The Product API →: OrderRepo.Create รับรายการที่ตั้งราคาไว้แล้ว ([]PricedItem) และไม่รู้เลยด้วยซ้ำว่า Catalog มีอยู่ การตัดสินใจว่า จะหาราคาอย่างไร — เรียก gRPC สด ๆ ในกรณีของคอร์สนี้ — เป็นการตัดสินใจระดับชั้น API ที่ repository ไม่ควรต้องรู้

ข้อดีข้อเสีย

pgx.Tx เดียวที่ห่อการ insert order, การ insert แต่ละรายการ, และการ insert outbox

• Pros: เป็นอะตอมิกด้วยโครงสร้าง ไม่มี code path ไหนที่บันทึกคำสั่งซื้อโดยไม่มีรายการสินค้า หรือบันทึกโดยไม่มี event ไว้ประกาศในภายหลัง defer tx.Rollback(ctx) เพียงบรรทัดเดียวหลัง Begin ทำให้ error ที่จุดไหนก็ตามใน function นี้ ไม่ว่าจะ insert ผิดพลาดหรือ JSON marshal ล้มเหลว ย้อนกลับทุกอย่างที่เขียนไปใน call นี้ได้อย่างปลอดภัย
• Cons: transaction ถือ row lock คร่อมสามตารางตลอดช่วงชีวิตของตัวเอง แทนที่จะล็อกแค่ตารางเดียว ทำให้ critical section ยาวกว่าการ insert ตารางเดียว และความล้มเหลวของ JSON marshal สำหรับ outbox payload ซึ่งไม่น่าเกิดแต่เป็นไปได้ถ้าอนาคตมี field ที่ serialize ไม่ได้ จะทำให้การสร้างคำสั่งซื้อล้มเหลวทั้งหมด เท่ากับผูก reliability ของการ serialize event เข้ากับ reliability ของการสั่งซื้อไปเลย

codes.FailedPrecondition สำหรับสินค้าที่ Catalog รายงานว่าไม่พบ แทนที่จะเป็น codes.InvalidArgument

• Pros: ตัว request เองถูกต้องตามไวยากรณ์ — product_id เป็น string ที่มีรูปแบบถูกต้อง — ปัญหาคือ ข้อเท็จจริงเกี่ยวกับสถานะระบบปัจจุบัน (ไม่มีสินค้านั้นอยู่ตอนนี้) ซึ่งตรงกับความหมายของ codes.FailedPrecondition ตามแนวทาง status code ของ gRPC เอง แตกต่างจาก request ที่ผิดรูปแบบ client แยกความแตกต่างสองอย่างนี้ได้: “คุณส่งสิ่งที่ server ไม่มีวันรับได้” (InvalidArgument) กับ “สิ่งที่คุณส่งอาจถูกต้องขึ้นอยู่กับสถานะ” (FailedPrecondition)
• Cons: ความแตกต่างนี้ค่อนข้างละเอียดอ่อนและถูกใช้ไม่สม่ำเสมอใน API จริง — caller ต้องรู้ธรรมเนียม status code ของ gRPC จริง ๆ ถึงจะแยกสองแบบนี้ได้ และ client จำนวนมากก็แค่ปฏิบัติกับทั้งคู่เหมือนเป็น “bad request” แล้วไปต่อโดยไม่สนใจความแตกต่าง

ติดตั้ง

1. Migration

create extension if not exists pgcrypto;
create table orders ( id uuid primary key default gen_random_uuid(), customer_id text not null, status text not null default 'pending', total_cents bigint not null, created_at timestamptz not null default now() );
create table order_items ( id bigserial primary key, order_id uuid not null references orders(id) on delete cascade, product_id uuid not null, quantity int not null, unit_price_cents bigint not null );
create table outbox ( id uuid primary key default gen_random_uuid(), aggregate_id uuid not null, event_type text not null, payload jsonb not null, created_at timestamptz not null default now(), published_at timestamptz );
create index on outbox (published_at) where published_at is null;

บันทึกไฟล์นี้เป็น migrations/order/0001_init.sql แล้วรัน:

migrate -path migrations/order -database "$ORDER_DB_URL" up

มีการตัดสินใจเรื่องรูปร่างสองสามอย่างที่ควรพูดถึงตอนนี้ (ตาราง outbox จะได้รับคำอธิบายของตัวเองเต็ม ๆ ใน The Transactional Outbox →):

• status ถูกเก็บเป็น text ตัวพิมพ์เล็ก — 'pending', 'confirmed', 'cancelled' — ไม่ใช่ enum OrderStatus ของ proto ฐานข้อมูลไม่รู้จัก protobuf เลย toProtoStatus ของ Server (ด้านล่าง) คือจุดเดียวที่ map ระหว่างทั้งสองรูปแบบ
• order_items.order_id references orders(id) on delete cascade หมายความว่าการลบคำสั่งซื้อ ซึ่งคอร์สนี้ไม่เคยทำ แต่เครื่องมือ admin ในอนาคตอาจทำ จะลบรายการสินค้าของคำสั่งซื้อนั้นโดยอัตโนมัติ — ไม่มีแถวกำพร้าให้ต้องเก็บกวาดเอง
• outbox.aggregate_id คือ id ของแถวที่ event นี้พูดถึง — ทุก event ที่เกี่ยวกับคำสั่งซื้อในคอร์สนี้ชี้กลับไปที่ orders.id นั่นคือสิ่งที่ทำให้ consumer ในอนาคต (หรือ query สำหรับ debug) หา “ทุก event ของคำสั่งซื้อนี้” ได้ด้วย where aggregate_id = $1 เดียว

2. OrderRepo

// Package repo is the PostgreSQL-backed store for the Order service.
package repo

import (
  "context"
  "encoding/json"
  "fmt"
  "time"

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

// Item is the row shape of a single order_items row.
type Item struct {
  ProductID      string
  Quantity       int32
  UnitPriceCents int64
}

// Order is the row shape of the orders table, with its items attached.
type Order struct {
  ID         string
  CustomerID string
  Status     string
  TotalCents int64
  Items      []Item
  CreatedAt  time.Time
}

// PricedItem is a line item whose unit price has already been looked up
// (from the Catalog service) before Create is called.
type PricedItem struct {
  ProductID      string
  Quantity       int32
  UnitPriceCents int64
}

// OrderRepo is the PostgreSQL-backed store for orders.
type OrderRepo struct {
  db *pgxpool.Pool
}

// New returns an OrderRepo backed by db.
func New(db *pgxpool.Pool) *OrderRepo {
  return &OrderRepo{db: db}
}

type createdEventItem struct {
  ProductID      string `json:"product_id"`
  Quantity       int32  `json:"quantity"`
  UnitPriceCents int64  `json:"unit_price_cents"`
}

type createdEventPayload struct {
  OrderID    string             `json:"order_id"`
  CustomerID string             `json:"customer_id"`
  TotalCents int64              `json:"total_cents"`
  Items      []createdEventItem `json:"items"`
}

// Create inserts a new pending order, its line items, and an "order.created"
// outbox row in a single transaction — all three commit together or none do.
func (r *OrderRepo) Create(ctx context.Context, customerID string, items []PricedItem) (*Order, error) {
  var total int64
  for _, it := range items {
    total += int64(it.Quantity) * it.UnitPriceCents
  }

  tx, err := r.db.Begin(ctx)
  if err != nil {
    return nil, fmt.Errorf("repo: begin create order tx: %w", err)
  }
  defer tx.Rollback(ctx)

  var o Order
  err = tx.QueryRow(ctx, `
    insert into orders (customer_id, status, total_cents)
    values ($1, 'pending', $2)
    returning id, customer_id, status, total_cents, created_at`,
    customerID, total,
  ).Scan(&o.ID, &o.CustomerID, &o.Status, &o.TotalCents, &o.CreatedAt)
  if err != nil {
    return nil, fmt.Errorf("repo: insert order: %w", err)
  }

  eventItems := make([]createdEventItem, 0, len(items))
  for _, it := range items {
    if _, err := tx.Exec(ctx, `
      insert into order_items (order_id, product_id, quantity, unit_price_cents)
      values ($1, $2, $3, $4)`,
      o.ID, it.ProductID, it.Quantity, it.UnitPriceCents,
    ); err != nil {
      return nil, fmt.Errorf("repo: insert order item: %w", err)
    }
    o.Items = append(o.Items, Item{ProductID: it.ProductID, Quantity: it.Quantity, UnitPriceCents: it.UnitPriceCents})
    eventItems = append(eventItems, createdEventItem{ProductID: it.ProductID, Quantity: it.Quantity, UnitPriceCents: it.UnitPriceCents})
  }

  payload, err := json.Marshal(createdEventPayload{
    OrderID:    o.ID,
    CustomerID: o.CustomerID,
    TotalCents: o.TotalCents,
    Items:      eventItems,
  })
  if err != nil {
    return nil, fmt.Errorf("repo: marshal order.created payload: %w", err)
  }

  if _, err := tx.Exec(ctx, `
    insert into outbox (aggregate_id, event_type, payload)
    values ($1, 'order.created', $2)`,
    o.ID, payload,
  ); err != nil {
    return nil, fmt.Errorf("repo: insert outbox row: %w", err)
  }

  if err := tx.Commit(ctx); err != nil {
    return nil, fmt.Errorf("repo: commit create order tx: %w", err)
  }

  return &o, nil
}

// Get returns the order with the given id, items included. The returned
// error wraps pgx.ErrNoRows (checkable with errors.Is) when no such order
// exists.
func (r *OrderRepo) Get(ctx context.Context, id string) (*Order, error) {
  var o Order
  err := r.db.QueryRow(ctx, `
    select id, customer_id, status, total_cents, created_at
    from orders
    where id = $1`, id).Scan(&o.ID, &o.CustomerID, &o.Status, &o.TotalCents, &o.CreatedAt)
  if err != nil {
    return nil, fmt.Errorf("repo: get order %s: %w", id, err)
  }

  items, err := r.itemsForOrder(ctx, id)
  if err != nil {
    return nil, err
  }
  o.Items = items

  return &o, nil
}

// ListByCustomer returns every order placed by customerID, most recent
// first, items included.
func (r *OrderRepo) ListByCustomer(ctx context.Context, customerID string) ([]Order, error) {
  rows, err := r.db.Query(ctx, `
    select id, customer_id, status, total_cents, created_at
    from orders
    where customer_id = $1
    order by created_at desc`, customerID)
  if err != nil {
    return nil, fmt.Errorf("repo: list orders: %w", err)
  }
  defer rows.Close()

  var orders []Order
  for rows.Next() {
    var o Order
    if err := rows.Scan(&o.ID, &o.CustomerID, &o.Status, &o.TotalCents, &o.CreatedAt); err != nil {
      return nil, fmt.Errorf("repo: scan order: %w", err)
    }
    orders = append(orders, o)
  }
  if err := rows.Err(); err != nil {
    return nil, fmt.Errorf("repo: iterate orders: %w", err)
  }

  for i := range orders {
    items, err := r.itemsForOrder(ctx, orders[i].ID)
    if err != nil {
      return nil, err
    }
    orders[i].Items = items
  }

  return orders, nil
}

// itemsForOrder returns every order_items row for orderID, in insertion
// order. Shared by Get and ListByCustomer so both fetch items the same way.
func (r *OrderRepo) itemsForOrder(ctx context.Context, orderID string) ([]Item, error) {
  rows, err := r.db.Query(ctx, `
    select product_id, quantity, unit_price_cents
    from order_items
    where order_id = $1
    order by id`, orderID)
  if err != nil {
    return nil, fmt.Errorf("repo: list order items: %w", err)
  }
  defer rows.Close()

  var items []Item
  for rows.Next() {
    var it Item
    if err := rows.Scan(&it.ProductID, &it.Quantity, &it.UnitPriceCents); err != nil {
      return nil, fmt.Errorf("repo: scan order item: %w", err)
    }
    items = append(items, it)
  }
  if err := rows.Err(); err != nil {
    return nil, fmt.Errorf("repo: iterate order items: %w", err)
  }

  return items, nil
}

type statusEventPayload struct {
  OrderID string `json:"order_id"`
  Status  string `json:"status"`
}

// UpdateStatus moves an order to status ("confirmed" or "cancelled") and
// writes the matching "order.confirmed"/"order.cancelled" outbox row in the
// same transaction. The Order saga (Module 9) is what calls this once it
// learns the payment result.
func (r *OrderRepo) UpdateStatus(ctx context.Context, id, status string) error {
  eventType := map[string]string{
    "confirmed": "order.confirmed",
    "cancelled": "order.cancelled",
  }[status]
  if eventType == "" {
    return fmt.Errorf("repo: update order status: no outbox event for status %q", status)
  }

  tx, err := r.db.Begin(ctx)
  if err != nil {
    return fmt.Errorf("repo: begin update status tx: %w", err)
  }
  defer tx.Rollback(ctx)

  tag, err := tx.Exec(ctx, `update orders set status = $1 where id = $2`, status, id)
  if err != nil {
    return fmt.Errorf("repo: update order status: %w", err)
  }
  if tag.RowsAffected() == 0 {
    return fmt.Errorf("repo: update order status: no order with id %s", id)
  }

  payload, err := json.Marshal(statusEventPayload{OrderID: id, Status: status})
  if err != nil {
    return fmt.Errorf("repo: marshal %s payload: %w", eventType, err)
  }

  if _, err := tx.Exec(ctx, `
    insert into outbox (aggregate_id, event_type, payload)
    values ($1, $2, $3)`, id, eventType, payload,
  ); err != nil {
    return fmt.Errorf("repo: insert outbox row: %w", err)
  }

  return tx.Commit(ctx)
}

บันทึกไฟล์นี้เป็น services/order/internal/repo/orders.go มีรายละเอียดที่ควรพูดถึงนอกเหนือจาก transaction เดียวของ Create:

• Get และ ListByCustomer เรียก helper ส่วนตัว itemsForOrder เหมือนกัน แทนที่จะเขียน query select ... from order_items ซ้ำสองครั้ง ListByCustomer ยอมจ่ายด้วย query เพิ่มต่อคำสั่งซื้อหนึ่งครั้ง (N+1) — ยอมรับได้ในระดับของคอร์สนี้ และเป็นตัวเลือกที่จะทำเป็น query แบบ join เดียวได้ถ้าจุดนี้กลายเป็น hot path จริง ๆ ที่เป็น trade-off แบบเดียวกับที่ The Postgres Repository → พูดถึงไปแล้วสำหรับ repository ที่เรียบง่ายขนาดนี้
• UpdateStatus เช็ค eventType == "" ก่อนเปิด transaction เลย — สถานะที่ไม่ถูกต้อง (อะไรก็ตามที่ไม่ใช่ "confirmed" หรือ "cancelled") จะถูกปฏิเสธทันที ดังนั้น function นี้ไม่มีทางเริ่ม transaction ที่ต้องมา rollback อยู่ดี
• tag.RowsAffected() == 0 คือวิธีที่ UpdateStatus ตรวจจับ “ไม่มีคำสั่งซื้อนี้” UPDATE ไม่คืน pgx.ErrNoRows แบบที่ SELECT ซึ่งได้ศูนย์แถวคืนมา แต่จะสำเร็จพร้อม command tag ที่บอกว่าแตะไปกี่แถว และเลขศูนย์คือสัญญาณที่เราถือว่า “ไม่พบ” ในที่นี้
• ทุกค่าในทุก query ยังคงถูก bind เป็น $1, $2, … — วินัยป้องกัน SQL injection แบบเดียวกับ ProductRepo ไม่มีข้อยกเว้นแม้แต่กับ payload ที่ถูก marshal เป็น JSON แล้ว

3. CreateOrder RPC handler

// Package server implements orderv1.OrderServiceServer against an
// OrderRepo and a Catalog gRPC client.
package server

import (
  "context"
  "errors"
  "time"

  catalogv1 "github.com/avetavos/shopmicro/gen/shopmicro/catalog/v1"
  orderv1 "github.com/avetavos/shopmicro/gen/shopmicro/order/v1"
  "github.com/avetavos/shopmicro/services/order/internal/repo"
  "github.com/jackc/pgx/v5"
  "google.golang.org/grpc/codes"
  "google.golang.org/grpc/status"
)

// Server implements orderv1.OrderServiceServer.
type Server struct {
  orderv1.UnimplementedOrderServiceServer
  repo    *repo.OrderRepo
  catalog catalogv1.CatalogServiceClient
}

// New returns a Server backed by orderRepo, pricing items via catalog.
func New(orderRepo *repo.OrderRepo, catalog catalogv1.CatalogServiceClient) *Server {
  return &Server{repo: orderRepo, catalog: catalog}
}

// CreateOrder prices every requested item against the Catalog service, then
// persists the order.
func (s *Server) CreateOrder(ctx context.Context, req *orderv1.CreateOrderRequest) (*orderv1.Order, error) {
  if req.GetCustomerId() == "" {
    return nil, status.Error(codes.InvalidArgument, "customer_id is required")
  }
  if len(req.GetItems()) == 0 {
    return nil, status.Error(codes.InvalidArgument, "items must not be empty")
  }

  priced := make([]repo.PricedItem, 0, len(req.GetItems()))
  for _, item := range req.GetItems() {
    if item.GetQuantity() <= 0 {
      return nil, status.Errorf(codes.InvalidArgument, "quantity for product %s must be positive", item.GetProductId())
    }

    product, err := s.catalog.GetProduct(ctx, &catalogv1.GetProductRequest{Id: item.GetProductId()})
    if err != nil {
      if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
        return nil, status.Errorf(codes.FailedPrecondition, "product %s does not exist", item.GetProductId())
      }
      return nil, status.Errorf(codes.Internal, "price product %s: %v", item.GetProductId(), err)
    }

    priced = append(priced, repo.PricedItem{
      ProductID:      item.GetProductId(),
      Quantity:       item.GetQuantity(),
      UnitPriceCents: product.GetPriceCents(),
    })
  }

  o, err := s.repo.Create(ctx, req.GetCustomerId(), priced)
  if err != nil {
    return nil, status.Errorf(codes.Internal, "create order: %v", err)
  }
  return toProto(*o), nil
}

// GetOrder returns a single order by id.
func (s *Server) GetOrder(ctx context.Context, req *orderv1.GetOrderRequest) (*orderv1.Order, error) {
  if req.GetId() == "" {
    return nil, status.Error(codes.InvalidArgument, "id is required")
  }

  o, err := s.repo.Get(ctx, req.GetId())
  if err != nil {
    if errors.Is(err, pgx.ErrNoRows) {
      return nil, status.Error(codes.NotFound, "order not found")
    }
    return nil, status.Errorf(codes.Internal, "get order: %v", err)
  }
  return toProto(*o), nil
}

// ListOrders returns every order placed by a customer.
func (s *Server) ListOrders(ctx context.Context, req *orderv1.ListOrdersRequest) (*orderv1.ListOrdersResponse, error) {
  if req.GetCustomerId() == "" {
    return nil, status.Error(codes.InvalidArgument, "customer_id is required")
  }

  orders, err := s.repo.ListByCustomer(ctx, req.GetCustomerId())
  if err != nil {
    return nil, status.Errorf(codes.Internal, "list orders: %v", err)
  }

  resp := &orderv1.ListOrdersResponse{}
  for _, o := range orders {
    resp.Orders = append(resp.Orders, toProto(o))
  }
  return resp, nil
}

func toProto(o repo.Order) *orderv1.Order {
  items := make([]*orderv1.OrderItem, 0, len(o.Items))
  for _, it := range o.Items {
    items = append(items, &orderv1.OrderItem{
      ProductId:      it.ProductID,
      Quantity:       it.Quantity,
      UnitPriceCents: it.UnitPriceCents,
    })
  }
  return &orderv1.Order{
    Id:         o.ID,
    CustomerId: o.CustomerID,
    Status:     toProtoStatus(o.Status),
    TotalCents: o.TotalCents,
    Items:      items,
    CreatedAt:  o.CreatedAt.Format(time.RFC3339),
  }
}

func toProtoStatus(s string) orderv1.OrderStatus {
  switch s {
  case "pending":
    return orderv1.OrderStatus_ORDER_STATUS_PENDING
  case "confirmed":
    return orderv1.OrderStatus_ORDER_STATUS_CONFIRMED
  case "cancelled":
    return orderv1.OrderStatus_ORDER_STATUS_CANCELLED
  default:
    return orderv1.OrderStatus_ORDER_STATUS_UNSPECIFIED
  }
}

บันทึกไฟล์นี้เป็น services/order/internal/server/server.go มีรายละเอียดเรื่องการ map และ validation ที่ควรพูดถึง:

• CreateOrder ตั้งราคาทีละรายการ ใน loop ตามลำดับ แต่ละ call s.catalog.GetProduct คือ gRPC round trip หนึ่งครั้ง คำสั่งซื้อที่มีห้ารายการต่างกันจึงยิงเรียงกันห้าครั้งก่อนที่ transaction จะเริ่มด้วยซ้ำ นี่คือต้นทุน latency จริงที่ควรพูดตรง ๆ — เวอร์ชันโปรดักชันอาจตั้งราคาแบบขนานหรือรวมเป็น Catalog RPC เดียว แต่ทั้งสองอย่างไม่มีอยู่ใน .proto contract ของคอร์สนี้ ดังนั้น loop ตามลำดับคือสิ่งที่ถูกต้องจริง ๆ ในตอนนี้
• status.FromError(err) แกะ error ของ gRPC กลับมาเป็น codes.Code กับข้อความ นี่คือวิธีที่ CreateOrder แยก “Catalog บอกว่าสินค้านี้ไม่มีอยู่” (codes.NotFound จาก GetProduct ของ Catalog เอง) ออกจาก “การเรียก Catalog เองล้มเหลว” (network error, timeout) — มีแค่กรณีแรกเท่านั้นที่กลายเป็น codes.FailedPrecondition ส่วนกรณีที่สองกลายเป็น codes.Internal เพราะไม่ได้บอกข้อเท็จจริงอะไรเกี่ยวกับตัวสินค้าเลย
• errors.Is(err, pgx.ErrNoRows) เหมือนกับ ProductRepo/Server ใน Catalog ทุกประการ OrderRepo.Get ห่อสิ่งที่ Scan คืนกลับมาด้วย %w ดังนั้น GetOrder จึงยัง unwrap ห่วงโซ่เพื่อหา sentinel ข้างในและแปลงเป็น codes.NotFound ได้
• o.CreatedAt.Format(time.RFC3339) ของ toProto แปลง time.Time ของ repository เป็น string ธรรมดาที่ .proto contract ประกาศไว้สำหรับ Order.created_at — The Contracts → เลือก field string แทน google.protobuf.Timestamp สำหรับคอร์สนี้ ดังนั้นการ map จึงเป็นแค่ call Format เดียว ไม่ใช่การแปลงแบบ timestamppb

4. ต่อสาย Server เข้ากับ main.go

// Command order runs the Order gRPC server.
package main

import (
  "context"
  "log"
  "net"
  "os"
  "os/signal"
  "syscall"

  catalogv1 "github.com/avetavos/shopmicro/gen/shopmicro/catalog/v1"
  orderv1 "github.com/avetavos/shopmicro/gen/shopmicro/order/v1"
  "github.com/avetavos/shopmicro/pkg/config"
  "github.com/avetavos/shopmicro/pkg/pg"
  "github.com/avetavos/shopmicro/services/order/internal/repo"
  "github.com/avetavos/shopmicro/services/order/internal/server"
  "google.golang.org/grpc"
  "google.golang.org/grpc/credentials/insecure"
  "google.golang.org/grpc/reflection"
)

func main() {
  ctx := context.Background()

  dbURL := config.Get("ORDER_DB_URL", "postgres://shopmicro:shopmicro@localhost:5432/orders?sslmode=disable")
  grpcAddr := config.Get("ORDER_GRPC_ADDR", ":50052")
  catalogAddr := config.Get("CATALOG_GRPC_ADDR", ":50051")

  pool, err := pg.NewPool(ctx, dbURL)
  if err != nil {
    log.Fatalf("order: connect to postgres: %v", err)
  }
  defer pool.Close()

  catalogConn, err := grpc.NewClient(catalogAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
  if err != nil {
    log.Fatalf("order: dial catalog at %s: %v", catalogAddr, err)
  }
  defer catalogConn.Close()
  catalogClient := catalogv1.NewCatalogServiceClient(catalogConn)

  lis, err := net.Listen("tcp", grpcAddr)
  if err != nil {
    log.Fatalf("order: listen on %s: %v", grpcAddr, err)
  }

  orderRepo := repo.New(pool)
  orderServer := server.New(orderRepo, catalogClient)

  grpcServer := grpc.NewServer()
  orderv1.RegisterOrderServiceServer(grpcServer, orderServer)
  reflection.Register(grpcServer)

  go func() {
    log.Printf("order: gRPC server listening on %s", grpcAddr)
    if err := grpcServer.Serve(lis); err != nil {
      log.Fatalf("order: serve: %v", err)
    }
  }()

  stop := make(chan os.Signal, 1)
  signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
  <-stop

  log.Println("order: shutting down")
  grpcServer.GracefulStop()
}

บันทึกไฟล์นี้ทับ services/order/cmd/main.go การเปลี่ยนแปลงเดียวจากเวอร์ชันของ The gRPC Server →: orderRepo := repo.New(pool) กับ orderServer := server.New(orderRepo, catalogClient) สร้าง dependency chain ตัวจริง และ RegisterOrderServiceServer ลงทะเบียน orderServer แทน &placeholderServer{catalog: catalogClient} pool, listener, การ dial Catalog, การลงทะเบียน reflection, goroutine ที่ serve, และการจัดการ signal ทั้งหมดไม่เปลี่ยนแปลง — ไม่มีสิ่งไหนในนั้นเคยเกี่ยวข้องกับ placeholder เลยตั้งแต่แรก

ตรวจสอบผล

รัน Catalog แล้วรัน Order:

go run ./services/catalog/cmd
go run ./services/order/cmd

สร้างสินค้าใน Catalog ก่อน เพื่อให้ Order มีของจริงให้ตั้งราคา:

grpcurl -plaintext -d '{"name":"Coffee Mug","description":"350ml ceramic mug","price_cents":1299,"stock":50}' \
  localhost:50051 shopmicro.catalog.v1.CatalogService/CreateProduct

คัดลอก id จากผลลัพธ์ แล้วสั่งซื้อสองชิ้น:

grpcurl -plaintext -d '{"customer_id":"cust-1","items":[{"product_id":"8f14e45f-ceea-4c9d-b2a5-0c1e3f4a9b21","quantity":2}]}' \
  localhost:50052 shopmicro.order.v1.OrderService/CreateOrder

{
  "id": "3a7c9e21-1e4d-4b8a-9c6e-2f8b1d5a7c90",
  "customerId": "cust-1",
  "status": "ORDER_STATUS_PENDING",
  "totalCents": "2598",
  "items": [
    {
      "productId": "8f14e45f-ceea-4c9d-b2a5-0c1e3f4a9b21",
      "quantity": 2,
      "unitPriceCents": "1299"
    }
  ],
  "createdAt": "2026-07-14T09:12:03Z"
}

totalCents คือ 1299 × 2 = 2598 — ตั้งราคาสด ๆ จาก Catalog ไม่เคยมาจาก client เลย status กลับมาเป็น ORDER_STATUS_PENDING ตรงกับค่า default 'pending' ที่ migration เขียนไว้ ดึงกลับมาด้วย id:

grpcurl -plaintext -d '{"id":"3a7c9e21-1e4d-4b8a-9c6e-2f8b1d5a7c90"}' \
  localhost:50052 shopmicro.order.v1.OrderService/GetOrder

แสดงรายการคำสั่งซื้อทั้งหมดของลูกค้าคนนั้น:

grpcurl -plaintext -d '{"customer_id":"cust-1"}' \
  localhost:50052 shopmicro.order.v1.OrderService/ListOrders

ยืนยัน codes.FailedPrecondition สำหรับสินค้าที่ไม่มีอยู่จริง:

grpcurl -plaintext -d '{"customer_id":"cust-1","items":[{"product_id":"00000000-0000-0000-0000-000000000000","quantity":1}]}' \
  localhost:50052 shopmicro.order.v1.OrderService/CreateOrder

ERROR:
  Code: FailedPrecondition
  Message: product 00000000-0000-0000-0000-000000000000 does not exist

และยืนยัน codes.InvalidArgument สำหรับ items ที่ว่างเปล่า:

grpcurl -plaintext -d '{"customer_id":"cust-1","items":[]}' \
  localhost:50052 shopmicro.order.v1.OrderService/CreateOrder

ERROR:
  Code: InvalidArgument
  Message: items must not be empty

สุดท้าย ยืนยันว่าทั้ง module ยัง build ผ่าน:

go build ./...

ไม่มี output แปลว่าสำเร็จ

สรุป

migrations/order/0001_init.sql สร้าง orders, order_items, และ outbox — ตัวสุดท้ายมี partial index บนแถวที่ยังไม่ publish ซึ่ง The Transactional Outbox → จะอธิบายเต็ม ๆ OrderRepo.Create ใน services/order/internal/repo/orders.go insert คำสั่งซื้อ ทุกรายการสินค้า และ outbox row order.created อยู่ใน pgx.Tx เดียว ดังนั้นทั้งสามอย่างจึงสำเร็จพร้อมกันหรือไม่สำเร็จเลย; Get, ListByCustomer, และ UpdateStatus ทำให้ repository สมบูรณ์ ตัวสุดท้ายเขียน outbox row order.confirmed/order.cancelled ของตัวเองสำหรับ Order Saga → (Module 9) ที่จะเรียกใช้ในที่สุด Server ใน services/order/internal/server/server.go implement CreateOrder โดยตั้งราคาทุกรายการผ่าน Catalog client ที่ inject ไว้ — codes.FailedPrecondition สำหรับสินค้าที่ Catalog รายงานว่าไม่มี, codes.Internal สำหรับความล้มเหลวอื่น ๆ ของ Catalog — ก่อนที่จะเรียก repo.Create เลย; GetOrder และ ListOrders เชื่อมตรงไปที่ repository grpcurl ยืนยันเส้นทางทั้งหมดแบบ end-to-end: สร้างสินค้าใน Catalog สั่งซื้อใน Order โดยตั้งราคาจาก price_cents จริงของสินค้านั้น ดึงกลับมา แสดงรายการ และเห็นทั้ง codes.FailedPrecondition กับ codes.InvalidArgument ทำงานถูกต้อง นั่นคือ Order service ที่สร้างคำสั่งซื้อจริงพร้อมราคาที่ถูกต้อง ต่อไป The Transactional Outbox → จะอธิบายว่าทำไมการ insert outbox นั้น — ไม่ใช่การ publish ตรงไปที่ Kafka — คือสิ่งที่ทำให้ order.created เชื่อถือได้