add more codes for kitchen-svc #3

cmd/barista/config.yml

@@ -6,8 +6,12 @@ http:
   host: '0.0.0.0'
   port: 5002
 
+postgres:
+  pool_max: 2
+  url: postgres://postgres:P@[email protected]:5432/postgres?sslmode=disable
+
 rabbit_mq:
-  url: amqp://guest:[email protected].255.101:5672/
+  url: amqp://guest:[email protected].240.102:5672/
 
 logger:
   log_level: 'debug'

cmd/barista/config/config.go

@@ -14,9 +14,15 @@ type (
 		configs.App  `yaml:"app"`
 		configs.HTTP `yaml:"http"`
 		configs.Log  `yaml:"logger"`
+		PG           `yaml:"postgres"`
 		RabbitMQ     `yaml:"rabbit_mq"`
 	}
 
+	PG struct {
+		PoolMax int    `env-required:"true" yaml:"pool_max" env:"PG_POOL_MAX"`
+		URL     string `env-required:"true" yaml:"url" env:"PG_URL"`
+	}
+
 	RabbitMQ struct {
 		URL string `env-required:"true" yaml:"url" env:"RABBITMQ_URL"`
 	}

cmd/counter/config.yml

@@ -8,10 +8,10 @@ http:
 
 postgres:
   pool_max: 2
-  url: postgres://postgres:P@[email protected].255.101:5432/postgres?sslmode=disable
+  url: postgres://postgres:P@[email protected].240.102:5432/postgres?sslmode=disable
 
 rabbit_mq:
-  url: amqp://guest:[email protected].255.101:5672/
+  url: amqp://guest:[email protected].240.102:5672/
 
 product_client:
   url: 0.0.0.0:5001

cmd/kitchen/config.yml

@@ -6,8 +6,12 @@ http:
   host: '0.0.0.0'
   port: 5004
 
+postgres:
+  pool_max: 2
+  url: postgres://postgres:P@[email protected]:5432/postgres?sslmode=disable
+
 rabbit_mq:
-  url: amqp://guest:[email protected].255.101:5672/
+  url: amqp://guest:[email protected].240.102:5672/
 
 logger:
   log_level: 'debug'

cmd/kitchen/config/config.go

@@ -14,9 +14,15 @@ type (
 		configs.App  `yaml:"app"`
 		configs.HTTP `yaml:"http"`
 		configs.Log  `yaml:"logger"`
+		PG           `yaml:"postgres"`
 		RabbitMQ     `yaml:"rabbit_mq"`
 	}
 
+	PG struct {
+		PoolMax int    `env-required:"true" yaml:"pool_max" env:"PG_POOL_MAX"`
+		URL     string `env-required:"true" yaml:"url" env:"PG_URL"`
+	}
+
 	RabbitMQ struct {
 		URL string `env-required:"true" yaml:"url" env:"RABBITMQ_URL"`
 	}

internal/barista/app/app.go

@@ -12,8 +12,10 @@ import (
 	"github.com/rabbitmq/amqp091-go"
 	"github.com/thangchung/go-coffeeshop/cmd/barista/config"
 	"github.com/thangchung/go-coffeeshop/internal/barista/features/orders/eventhandlers"
+	"github.com/thangchung/go-coffeeshop/internal/barista/features/orders/repo"
 	"github.com/thangchung/go-coffeeshop/pkg/event"
 	mylogger "github.com/thangchung/go-coffeeshop/pkg/logger"
+	"github.com/thangchung/go-coffeeshop/pkg/postgres"
 	"github.com/thangchung/go-coffeeshop/pkg/rabbitmq"
 	"github.com/thangchung/go-coffeeshop/pkg/rabbitmq/consumer"
 	"github.com/thangchung/go-coffeeshop/pkg/rabbitmq/publisher"
@@ -41,6 +43,18 @@ func (a *App) Run() error {
 
 	ctx, cancel := context.WithCancel(context.Background())
 
+	// PostgresDB
+	pg, err := postgres.NewPostgresDB(a.cfg.PG.URL, postgres.MaxPoolSize(a.cfg.PG.PoolMax))
+	if err != nil {
+		a.logger.Fatal("app - Run - postgres.NewPostgres: %s", err.Error())
+
+		cancel()
+
+		return err
+	}
+	defer pg.Close()
+
+	// rabbitmq
 	amqpConn, err := rabbitmq.NewRabbitMQConn(a.cfg.RabbitMQ.URL, a.logger)
 	if err != nil {
 		cancel()
@@ -65,8 +79,11 @@ func (a *App) Run() error {
 		return errors.Wrap(err, "publisher-Counter-NewOrderPublisher")
 	}
 
+	// repository
+	orderRepo := repo.NewOrderRepo(pg)
+
 	// event handlers.
-	a.handler = eventhandlers.NewBaristaOrderedEventHandler(counterOrderPub)
+	a.handler = eventhandlers.NewBaristaOrderedEventHandler(orderRepo, counterOrderPub)
 
 	// consumers
 	consumer, err := consumer.NewConsumer(

internal/barista/domain/interfaces.go

@@ -0,0 +1,11 @@
+package domain
+
+import (
+	"context"
+)
+
+type (
+	OrderRepo interface {
+		Create(context.Context, *BaristaOrder) error
+	}
+)

internal/barista/domain/order.go

@@ -0,0 +1,17 @@
+package domain
+
+import (
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/thangchung/go-coffeeshop/proto/gen"
+)
+
+type BaristaOrder struct {
+	ID       uuid.UUID    `json:"id" db:"id"`
+	ItemName string       `json:"itemName" db:"item_name"`
+	ItemType gen.ItemType `json:"itemType" db:"item_type"`
+	TimeUp   time.Time    `json:"timeUp" db:"time_up"`
+	Created  time.Time    `json:"created" db:"created"`
+	Updated  time.Time    `json:"updated" db:"updated"`
+}

internal/barista/features/orders/eventhandlers/barista_ordered.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	"github.com/thangchung/go-coffeeshop/internal/barista/domain"
 	"github.com/thangchung/go-coffeeshop/pkg/event"
 	"github.com/thangchung/go-coffeeshop/pkg/rabbitmq/publisher"
 	"github.com/thangchung/go-coffeeshop/proto/gen"
@@ -19,32 +20,47 @@ type BaristaOrderedEventHandler interface {
 var _ BaristaOrderedEventHandler = (*baristaOrderedEventHandler)(nil)
 
 type baristaOrderedEventHandler struct {
+	repo       domain.OrderRepo
 	counterPub *publisher.Publisher
 }
 
-func NewBaristaOrderedEventHandler(counterPub *publisher.Publisher) BaristaOrderedEventHandler {
+func NewBaristaOrderedEventHandler(repo domain.OrderRepo, counterPub *publisher.Publisher) BaristaOrderedEventHandler {
 	return &baristaOrderedEventHandler{
+		repo:       repo,
 		counterPub: counterPub,
 	}
 }
 
 func (h *baristaOrderedEventHandler) Handle(ctx context.Context, e *event.BaristaOrdered) error {
 	fmt.Println(e)
 
+	timeIn := time.Now()
+
 	delay := calculateDelay(e.ItemType)
 	time.Sleep(delay)
 
-	// todo: save to db
-	// ...
+	timeUp := time.Now().Add(delay)
+
+	err := h.repo.Create(ctx, &domain.BaristaOrder{
+		ID:       e.ItemLineID,
+		ItemType: e.ItemType,
+		ItemName: e.ItemType.String(),
+		TimeUp:   timeUp,
+		Created:  time.Now(),
+		Updated:  time.Now(),
+	})
+	if err != nil {
+		return errors.Wrap(err, "baristaOrderedEventHandler-h.repo.Create")
+	}
 
 	message := event.BaristaOrderUpdated{
 		OrderID:    e.OrderID,
 		ItemLineID: e.ItemLineID,
 		Name:       e.ItemType.String(),
 		ItemType:   e.ItemType,
 		MadeBy:     "teesee",
-		TimeIn:     time.Now(),
-		TimeUp:     time.Now().Add(5 * time.Minute),
+		TimeIn:     timeIn,
+		TimeUp:     timeUp,
 	}
 
 	eventBytes, err := json.Marshal(message)

internal/barista/features/orders/repo/orders_postgres.go

@@ -0,0 +1,50 @@
+package repo
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	"github.com/thangchung/go-coffeeshop/internal/barista/domain"
+	"github.com/thangchung/go-coffeeshop/pkg/postgres"
+)
+
+var _ domain.OrderRepo = (*orderRepo)(nil)
+
+type orderRepo struct {
+	pg *postgres.Postgres
+}
+
+func NewOrderRepo(pg *postgres.Postgres) domain.OrderRepo {
+	return &orderRepo{pg: pg}
+}
+
+func (d *orderRepo) Create(ctx context.Context, baristaOrder *domain.BaristaOrder) error {
+	tx, err := d.pg.Pool.Begin(ctx)
+	if err != nil {
+		return errors.Wrapf(err, "orderRepo-Create-d.pg.Pool.Begin(ctx)")
+	}
+
+	// insert order
+	sql, args, err := d.pg.Builder.
+		Insert(`"barista".barista_orders`).
+		Columns("id", "item_type", "item_name", "time_up", "created", "updated").
+		Values(
+			baristaOrder.ID,
+			baristaOrder.ItemType,
+			baristaOrder.ItemName,
+			baristaOrder.TimeUp,
+			baristaOrder.Created,
+			baristaOrder.Updated,
+		).
+		ToSql()
+	if err != nil {
+		return tx.Rollback(ctx)
+	}
+
+	_, err = d.pg.Pool.Exec(ctx, sql, args...)
+	if err != nil {
+		return tx.Rollback(ctx)
+	}
+
+	return tx.Commit(ctx)
+}

internal/counter/domain/order.go

@@ -155,7 +155,6 @@ func publishBaristaOrderEvent(
 ) error {
 	if isBarista {
 		// todo: refactor to event domain dispatcher
-		// ...
 		event := events.BaristaOrdered{
 			OrderID:    orderID,
 			ItemLineID: lineItemID,
@@ -175,7 +174,6 @@ func publishBaristaOrderEvent(
 		return nil
 	} else {
 		// todo: refactor to event domain dispatcher
-		// ...
 		event := events.KitchenOrdered{
 			OrderID:    orderID,
 			ItemLineID: lineItemID,

internal/counter/features/orders/repo/orders_postgres.go

@@ -117,6 +117,7 @@ func (d *orderRepo) GetAll(ctx context.Context) ([]*domain.Order, error) {
 				Price:          ol.Price,
 				ItemStatus:     ol.ItemStatus,
 				IsBaristaOrder: ol.IsBaristaOrder,
+				OrderID:        ol.OrderID,
 			})
 		}
 

internal/counter/grpc/counter_server.go

@@ -74,6 +74,7 @@ func (g *CounterServiceServerImpl) GetListOrderFulfillment(
 			LoyaltyMemberId: entity.LoyaltyMemberID.String(),
 			LineItems: lo.Map(entity.LineItems, func(item *domain.LineItem, _ int) *gen.LineItemDto {
 				return &gen.LineItemDto{
+					Id:             item.ID.String(),
 					ItemType:       item.ItemType,
 					Name:           item.Name,
 					Price:          float64(item.Price),

internal/kitchen/app/app.go

@@ -12,8 +12,10 @@ import (
 	"github.com/rabbitmq/amqp091-go"
 	"github.com/thangchung/go-coffeeshop/cmd/kitchen/config"
 	"github.com/thangchung/go-coffeeshop/internal/kitchen/features/orders/eventhandlers"
+	"github.com/thangchung/go-coffeeshop/internal/kitchen/features/orders/repo"
 	"github.com/thangchung/go-coffeeshop/pkg/event"
 	mylogger "github.com/thangchung/go-coffeeshop/pkg/logger"
+	"github.com/thangchung/go-coffeeshop/pkg/postgres"
 	"github.com/thangchung/go-coffeeshop/pkg/rabbitmq"
 	"github.com/thangchung/go-coffeeshop/pkg/rabbitmq/consumer"
 	"github.com/thangchung/go-coffeeshop/pkg/rabbitmq/publisher"
@@ -41,6 +43,18 @@ func (a *App) Run() error {
 
 	ctx, cancel := context.WithCancel(context.Background())
 
+	// PostgresDB
+	pg, err := postgres.NewPostgresDB(a.cfg.PG.URL, postgres.MaxPoolSize(a.cfg.PG.PoolMax))
+	if err != nil {
+		a.logger.Fatal("app - Run - postgres.NewPostgres: %s", err.Error())
+
+		cancel()
+
+		return err
+	}
+	defer pg.Close()
+
+	// rabbitmq
 	amqpConn, err := rabbitmq.NewRabbitMQConn(a.cfg.RabbitMQ.URL, a.logger)
 	if err != nil {
 		cancel()
@@ -65,8 +79,11 @@ func (a *App) Run() error {
 		return errors.Wrap(err, "publisher-Counter-NewOrderPublisher")
 	}
 
+	// repository
+	orderRepo := repo.NewOrderRepo(pg)
+
 	// event handlers.
-	a.handler = eventhandlers.NewKitchenOrderedEventHandler(counterOrderPub)
+	a.handler = eventhandlers.NewKitchenOrderedEventHandler(orderRepo, counterOrderPub)
 
 	// consumers
 	consumer, err := consumer.NewConsumer(

internal/kitchen/domain/interfaces.go

@@ -0,0 +1,11 @@
+package domain
+
+import (
+	"context"
+)
+
+type (
+	OrderRepo interface {
+		Create(context.Context, *KitchenOrder) error
+	}
+)

internal/kitchen/domain/order.go

@@ -0,0 +1,18 @@
+package domain
+
+import (
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/thangchung/go-coffeeshop/proto/gen"
+)
+
+type KitchenOrder struct {
+	ID       uuid.UUID    `json:"id" db:"id"`
+	OrderID  uuid.UUID    `json:"orderId" db:"order_id"`
+	ItemName string       `json:"itemName" db:"item_name"`
+	ItemType gen.ItemType `json:"itemType" db:"item_type"`
+	TimeUp   time.Time    `json:"timeUp" db:"time_up"`
+	Created  time.Time    `json:"created" db:"created"`
+	Updated  time.Time    `json:"updated" db:"updated"`
+}