Port from SQLite to Mongodb for cache layer

.github/workflows/build-and-test.yml

@@ -32,9 +32,14 @@ jobs:
 
       - name: Build
         run: go build ./cmd/oms
-  
-      - name: Test
+
+      - name: Unit Tests
         run: go test ./...
 
+      - name: Integration Tests
+        if: matrix.os == 'ubuntu-latest'
+        run: go test -tags=integration ./...
+
       - name: Coverage
-        run: make unit-test-coverage functional-test-coverage coverage-report
+        if: matrix.os == 'ubuntu-latest'
+        run: make unit-test-coverage integration-test-coverage coverage-report

Dockerfile

@@ -26,8 +26,6 @@ EXPOSE 8081
 EXPOSE 8082
 EXPOSE 8083
 EXPOSE 8084
-VOLUME /data
-ENV DATA_DIR=/data
 
 COPY --from=oms-builder /usr/local/bin/oms /usr/local/bin/oms
 

Makefile

@@ -1,21 +1,21 @@
 TEST_COVERAGE_OUTPUT_ROOT := $(CURDIR)/.coverage/unit
-FUNCTIONAL_COVERAGE_OUTPUT_ROOT := $(CURDIR)/.coverage/functional
+INTEGRATION_COVERAGE_OUTPUT_ROOT := $(CURDIR)/.coverage/integration
 COVERAGE_PROFILE := $(CURDIR)/.coverage/profile
 COVERAGE_REPORT := $(CURDIR)/.coverage/report.html
 
-test: unit-test functional-test
+test: unit-test integration-test
 
 unit-test:
 	go test ./app/{billing,order,shipment}
 
-functional-test:
-	go test ./app/test
-	
+integration-test:
+	go test -tags=integration ./app/test
+
 $(TEST_COVERAGE_OUTPUT_ROOT):
 	mkdir -p $(TEST_COVERAGE_OUTPUT_ROOT)
 
-$(FUNCTIONAL_COVERAGE_OUTPUT_ROOT):
-	mkdir -p $(FUNCTIONAL_COVERAGE_OUTPUT_ROOT)
+$(INTEGRATION_COVERAGE_OUTPUT_ROOT):
+	mkdir -p $(INTEGRATION_COVERAGE_OUTPUT_ROOT)
 
 $(SUMMARY_COVERAGE_OUTPUT_ROOT):
 	mkdir -p $(SUMMARY_COVERAGE_OUTPUT_ROOT)
@@ -26,12 +26,12 @@ unit-test-coverage: $(TEST_COVERAGE_OUTPUT_ROOT)
 	go test -cover ./app/order -args -test.gocoverdir=$(TEST_COVERAGE_OUTPUT_ROOT) 
 	go test -cover ./app/shipment -args -test.gocoverdir=$(TEST_COVERAGE_OUTPUT_ROOT) 
 
-functional-test-coverage: $(FUNCTIONAL_COVERAGE_OUTPUT_ROOT)
-	@echo Functional test coverage
-	go test -cover ./app/test -coverpkg ./... -args -test.gocoverdir=$(FUNCTIONAL_COVERAGE_OUTPUT_ROOT)
+integration-test-coverage: $(INTEGRATION_COVERAGE_OUTPUT_ROOT)
+	@echo Integration test coverage
+	go test -tags integration -cover ./app/test -coverpkg ./... -args -test.gocoverdir=$(INTEGRATION_COVERAGE_OUTPUT_ROOT)
 
-coverage-report: $(TEST_COVERAGE_OUTPUT_ROOT) $(FUNCTIONAL_COVERAGE_OUTPUT_ROOT) $(SUMMARY_COVERAGE_OUTPUT_ROOT)
+coverage-report: $(TEST_COVERAGE_OUTPUT_ROOT) $(INTEGRATION_COVERAGE_OUTPUT_ROOT) $(SUMMARY_COVERAGE_OUTPUT_ROOT)
 	@echo Summary coverage report
-	go tool covdata textfmt -i $(TEST_COVERAGE_OUTPUT_ROOT),$(FUNCTIONAL_COVERAGE_OUTPUT_ROOT) -o $(COVERAGE_PROFILE)
-	go tool covdata percent -i $(TEST_COVERAGE_OUTPUT_ROOT),$(FUNCTIONAL_COVERAGE_OUTPUT_ROOT)
-	go tool cover -html=$(COVERAGE_PROFILE) -o $(COVERAGE_REPORT)
+	go tool covdata textfmt -i $(TEST_COVERAGE_OUTPUT_ROOT),$(INTEGRATION_COVERAGE_OUTPUT_ROOT) -o $(COVERAGE_PROFILE)
+	go tool covdata percent -i $(TEST_COVERAGE_OUTPUT_ROOT),$(INTEGRATION_COVERAGE_OUTPUT_ROOT)
+	go tool cover -html=$(COVERAGE_PROFILE) -o $(COVERAGE_REPORT)

app/config/config.go

@@ -9,7 +9,7 @@ import (
 // AppConfig is a struct that holds the configuration for the Order/Shipment/Fraud/Billing system.
 type AppConfig struct {
 	BindOnIP     string
-	DataDir      string
+	MongoURL     string
 	BillingPort  int32
 	BillingURL   string
 	OrderPort    int32
@@ -44,7 +44,7 @@ func (c *AppConfig) ServiceHostPort(service string) (string, error) {
 func AppConfigFromEnv() (AppConfig, error) {
 	conf := AppConfig{
 		BindOnIP:     "127.0.0.1",
-		DataDir:      "./",
+		MongoURL:     "",
 		BillingPort:  8081,
 		BillingURL:   "http://127.0.0.1:8081",
 		OrderPort:    8082,
@@ -59,8 +59,8 @@ func AppConfigFromEnv() (AppConfig, error) {
 		conf.BindOnIP = ip
 	}
 
-	if p := os.Getenv("DATA_DIR"); p != "" {
-		conf.DataDir = p
+	if p := os.Getenv("MONGO_URL"); p != "" {
+		conf.MongoURL = p
 	}
 
 	if p := os.Getenv("BILLING_API_URL"); p != "" {

app/db/db.go

@@ -0,0 +1,211 @@
+package db
+
+import (
+	"context"
+	_ "embed"
+	"fmt"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+	"github.com/temporalio/reference-app-orders-go/app/config"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	mongodb "go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	_ "modernc.org/sqlite" // SQLite driver
+)
+
+// OrdersCollection is the name of the MongoDB collection to use for Orders.
+const OrdersCollection = "orders"
+
+// OrderStatus is a struct that represents the status of an Order
+type OrderStatus struct {
+	ID         string `db:"id" bson:"id"`
+	CustomerID string `db:"customer_id" bson:"customer_id"`
+	Status     string `db:"status" bson:"status"`
+
+	ReceivedAt time.Time `db:"received_at" bson:"received_at"`
+}
+
+// ShipmentStatus is a struct that represents the status of a Shipment
+type ShipmentStatus struct {
+	ID     string `db:"id" bson:"id"`
+	Status string `db:"status" bson:"status"`
+}
+
+// ShipmentCollection is the name of the MongoDB collection to use for Shipment data.
+const ShipmentCollection = "shipments"
+
+// DB is an interface that defines the methods that a database driver must implement
+type DB interface {
+	Connect(ctx context.Context) error
+	Setup() error
+	Close() error
+	InsertOrder(context.Context, *OrderStatus) error
+	UpdateOrderStatus(context.Context, string, string) error
+	GetOrders(context.Context, *[]OrderStatus) error
+	UpdateShipmentStatus(context.Context, string, string) error
+	GetShipments(context.Context, *[]ShipmentStatus) error
+}
+
+// CreateDB creates a new DB instance based on the configuration
+func CreateDB(config config.AppConfig) DB {
+	if config.MongoURL != "" {
+		return &MongoDB{uri: config.MongoURL}
+	}
+
+	return &SQLiteDB{path: "./api-store.db"}
+}
+
+// MongoDB is a struct that implements the DB interface for MongoDB
+type MongoDB struct {
+	uri    string
+	client *mongo.Client
+	db     *mongo.Database
+}
+
+// Connect connects to a MongoDB instance
+func (m *MongoDB) Connect(ctx context.Context) error {
+	client, err := mongo.Connect(ctx, options.Client().ApplyURI(m.uri))
+	if err != nil {
+		return err
+	}
+	m.client = client
+	m.db = client.Database("orders")
+	return nil
+}
+
+// Setup sets up the MongoDB instance
+func (m *MongoDB) Setup() error {
+	orders := m.db.Collection(OrdersCollection)
+	_, err := orders.Indexes().CreateOne(context.TODO(), mongodb.IndexModel{
+		Keys: map[string]interface{}{"received_at": 1},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create orders index: %w", err)
+	}
+
+	shipments := m.db.Collection(ShipmentCollection)
+	_, err = shipments.Indexes().CreateOne(context.TODO(), mongodb.IndexModel{
+		Keys: map[string]interface{}{"booked_at": 1},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create shipment index: %w", err)
+	}
+
+	return nil
+}
+
+// InsertOrder inserts an Order into the MongoDB instance
+func (m *MongoDB) InsertOrder(ctx context.Context, order *OrderStatus) error {
+	_, err := m.db.Collection(OrdersCollection).InsertOne(ctx, order)
+	return err
+}
+
+// UpdateOrderStatus updates an Order in the MongoDB instance
+func (m *MongoDB) UpdateOrderStatus(ctx context.Context, id string, status string) error {
+	_, err := m.db.Collection(OrdersCollection).UpdateOne(ctx, bson.M{"id": id}, bson.M{"$set": bson.M{"status": status}})
+	return err
+}
+
+// GetOrders returns a list of Orders from the MongoDB instance
+func (m *MongoDB) GetOrders(ctx context.Context, result *[]OrderStatus) error {
+	res, err := m.db.Collection(OrdersCollection).Find(ctx, bson.M{}, &options.FindOptions{
+		Sort: bson.M{"received_at": 1},
+	})
+	if err != nil {
+		return err
+	}
+
+	return res.All(ctx, result)
+}
+
+// UpdateShipmentStatus updates a Shipment in the MongoDB instance
+func (m *MongoDB) UpdateShipmentStatus(ctx context.Context, id string, status string) error {
+	_, err := m.db.Collection(ShipmentCollection).UpdateOne(
+		ctx,
+		bson.M{"id": id},
+		bson.M{
+			"$set":         bson.M{"status": status},
+			"$setOnInsert": bson.M{"booked_at": time.Now().UTC()},
+		},
+		options.Update().SetUpsert(true),
+	)
+	return err
+}
+
+// GetShipments returns a list of Shipments from the MongoDB instance
+func (m *MongoDB) GetShipments(ctx context.Context, result *[]ShipmentStatus) error {
+	res, err := m.db.Collection(ShipmentCollection).Find(ctx, bson.M{}, &options.FindOptions{
+		Sort: bson.M{"booked_at": 1},
+	})
+	if err != nil {
+		return err
+	}
+
+	return res.All(ctx, result)
+}
+
+// Close closes the connection to the MongoDB instance
+func (m *MongoDB) Close() error {
+	return m.client.Disconnect(context.Background())
+}
+
+// SQLiteDB is a struct that implements the DB interface for SQLite
+type SQLiteDB struct {
+	path string
+	db   *sqlx.DB
+}
+
+//go:embed schema.sql
+var sqliteSchema string
+
+// Connect connects to a SQLite instance
+func (s *SQLiteDB) Connect(_ context.Context) error {
+	db, err := sqlx.Connect("sqlite", s.path)
+	if err != nil {
+		return err
+	}
+	s.db = db
+	db.SetMaxOpenConns(1) // SQLite does not support concurrent writes
+	return nil
+}
+
+// Setup sets up the SQLite instance
+func (s *SQLiteDB) Setup() error {
+	_, err := s.db.Exec(sqliteSchema)
+	return err
+}
+
+// Close closes the connection to the SQLite instance
+func (s *SQLiteDB) Close() error {
+	return s.db.Close()
+}
+
+// InsertOrder inserts an Order into the SQLite instance
+func (s *SQLiteDB) InsertOrder(ctx context.Context, order *OrderStatus) error {
+	_, err := s.db.NamedExecContext(ctx, "INSERT OR IGNORE INTO orders (id, customer_id, received_at, status) VALUES (:id, :customer_id, :received_at, :status)", order)
+	return err
+}
+
+// UpdateOrderStatus updates an Order in the SQLite instance
+func (s *SQLiteDB) UpdateOrderStatus(ctx context.Context, id string, status string) error {
+	_, err := s.db.ExecContext(ctx, "UPDATE orders SET status = ? WHERE id = ?", status, id)
+	return err
+}
+
+// GetOrders returns a list of Orders from the SQLite instance
+func (s *SQLiteDB) GetOrders(ctx context.Context, result *[]OrderStatus) error {
+	return s.db.SelectContext(ctx, result, "SELECT id, status, received_at FROM orders ORDER BY received_at DESC")
+}
+
+// UpdateShipmentStatus updates a Shipment in the SQLite instance
+func (s *SQLiteDB) UpdateShipmentStatus(ctx context.Context, id string, status string) error {
+	_, err := s.db.ExecContext(ctx, "INSERT INTO shipments (id, booked_at, status) VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET status = ?", id, time.Now().UTC(), status, status)
+	return err
+}
+
+// GetShipments returns a list of Shipments from the SQLite instance
+func (s *SQLiteDB) GetShipments(ctx context.Context, result *[]ShipmentStatus) error {
+	return s.db.SelectContext(ctx, result, "SELECT id, status FROM shipments ORDER BY booked_at DESC")
+}

app/fraud/api_test.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"github.com/temporalio/reference-app-orders-go/app/fraud"
-	_ "modernc.org/sqlite"
 )
 
 func TestMaintenanceMode(t *testing.T) {