egommerce/basket-service
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refactoring - added Functional Option pattern
Some checks failed
continuous-integration/drone/push Build is failing
Details

Browse Source
This commit is contained in:
Piotr Biernat 2023-04-04 00:57:11 +02:00
parent b6caa335b1
commit 751e1c70e6
18 changed files with 599 additions and 356 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
src/cmd/migrate/main.go
View File

@ -6,10 +6,13 @@ import (
"log" "log"
"os" "os"
"git.pbiernat.dev/egommerce/basket-service/internal/app/config"
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd"
"github.com/go-pg/migrations/v8" "github.com/go-pg/migrations/v8"
"github.com/go-pg/pg/v10" "github.com/go-pg/pg/v10"
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd"
"git.pbiernat.dev/egommerce/basket-service/internal/common"
cnf "git.pbiernat.dev/egommerce/basket-service/pkg/config"
) )
const ( const (
@ -32,16 +35,24 @@ Usage:
` `
func main() { func main() {
if config.ErrLoadingEnvs != nil { if cnf.ErrLoadingEnvs != nil {
log.Panicln("Error loading .env file", config.ErrLoadingEnvs) log.Panicln("Error loading .env file", cnf.ErrLoadingEnvs)
} }
// dbURL := config.GetEnv("DATABASE_URL", defDbURL) c := common.NewConfig()
loggerAddr := config.GetEnv("LOGGER_ADDR", defLoggerAddr)
mTblName := config.GetEnv("MIGRATIONS_TABLE_NAME", defMigrationsTableName)
logHost, logPort := fluentd.ParseAddr(loggerAddr) // dbURL := cnf.GetEnv("DATABASE_URL", defDbURL)
logger := fluentd.NewLogger(defAppName, logHost, logPort) mTblName := cnf.GetEnv("MIGRATIONS_TABLE_NAME", defMigrationsTableName)
logHost, logPort, err := fluentd.ParseAddr(c.LoggerAddr)
if err != nil {
log.Fatalf("Error parsing logger addr: %s. Err: %v", c.LoggerAddr, err)
}
logger, err := fluentd.NewLogger(c.GetAppFullName(), logHost, logPort) // @Refactor NewLogger return (logger, error)
if err != nil {
log.Fatalf("Error connecting to %s:%d. Err: %v", logHost, logPort, err)
}
defer logger.Close() defer logger.Close()
flag.Usage = usage flag.Usage = usage
@ -56,8 +67,8 @@ func main() {
mig := migrations.NewCollection() mig := migrations.NewCollection()
mig.SetTableName(mTblName) mig.SetTableName(mTblName)
err := mig.DiscoverSQLMigrations("./migrations")
if err != nil { if err := mig.DiscoverSQLMigrations("./migrations"); err != nil {
logger.Log("migration dicovery error: %#v", err) logger.Log("migration dicovery error: %#v", err)
} }

146
src/cmd/server/main.go
View File

@ -1,97 +1,83 @@
package main package main
import ( import (
"fmt"
"log" "log"
"os" "os"
"strconv"
"git.pbiernat.dev/egommerce/basket-service/internal/app/config" cnf "git.pbiernat.dev/egommerce/basket-service/pkg/config"
"git.pbiernat.dev/egommerce/basket-service/internal/app/database"
"git.pbiernat.dev/egommerce/basket-service/internal/app/server" "git.pbiernat.dev/egommerce/basket-service/internal/app/server"
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd" "git.pbiernat.dev/egommerce/basket-service/internal/common"
amqp "git.pbiernat.dev/egommerce/go-api-pkg/rabbitmq"
"github.com/go-redis/redis/v8"
)
const (
defAppName = "basket-svc"
defAppDomain = "basket-svc"
defPathPrefix = "/basket"
defNetAddr = ":80"
defLoggerAddr = "api-logger:24224"
defRegistryAddr = "api-registry:8500"
defDbURL = "postgres://postgres:12345678@postgres-db:5432/egommerce"
defCacheAddr = "api-cache:6379"
defCachePassword = "12345678"
defMongoDbURL = "mongodb://mongodb:12345678@mongo-db:27017"
defEventBusURL = "amqp://guest:guest@api-eventbus:5672"
ebEventsExchange = "api-events"
ebEventsQueue = "basket-svc"
defKVNmspc = "dev.egommerce/service/basket-svc"
) )
func main() { func main() {
if config.ErrLoadingEnvs != nil { if cnf.ErrLoadingEnvs != nil {
log.Panicln("Error loading .env file", config.ErrLoadingEnvs) log.Panicln("Error loading .env file", cnf.ErrLoadingEnvs)
} }
c := new(server.Config) c := common.NewConfig()
c.AppID, _ = os.Hostname() srv := server.New(
c.AppName = config.GetEnv("APP_NAME", defAppName) c,
c.AppDomain = config.GetEnv("APP_DOMAIN", defAppDomain) server.WithCache(c),
c.PathPrefix = config.GetEnv("APP_PATH_PREFIX", defPathPrefix) server.WithDatabase(c),
c.NetAddr = config.GetEnv("SERVER_ADDR", defNetAddr) server.WithEventbus(c),
c.Port, _ = strconv.Atoi(c.NetAddr[1:]) server.WithLogger(c),
c.LoggerAddr = config.GetEnv("LOGGER_ADDR", defLoggerAddr) server.WithRegistry(c),
c.RegistryAddr = config.GetEnv("REGISTRY_ADDR", defRegistryAddr) )
c.DbURL = config.GetEnv("DATABASE_URL", defDbURL)
c.CacheAddr = config.GetEnv("CACHE_ADDR", defCacheAddr)
c.CachePassword = config.GetEnv("CACHE_PASSWORD", defCachePassword)
c.EventBusURL = config.GetEnv("EVENTBUS_URL", defEventBusURL)
c.EventBusExchange = ebEventsExchange
c.KVNamespace = config.GetEnv("APP_KV_NAMESPACE", defKVNmspc)
logHost, logPort := fluentd.ParseAddr(c.LoggerAddr)
logger := fluentd.NewLogger(c.GetAppFullName(), logHost, logPort)
defer logger.Close()
// db conn
dbConn, err := database.Connect(c.DbURL)
if err != nil { // fixme: add wait-for-db...
logger.Log("Failed to connect to Database server: %v\n", err)
os.Exit(1)
}
defer dbConn.Close()
// redis conn
redis := redis.NewClient(&redis.Options{
Addr: c.CacheAddr,
Password: c.CachePassword,
DB: 0,
})
defer redis.Close()
// eventbus conn
ebConn, ebCh, err := amqp.Open(c.EventBusURL)
if err != nil {
logger.Log("Failed to connect to EventBus server: %v\n", err)
os.Exit(1)
}
defer ebCh.Close()
defer amqp.Close(ebConn)
err = amqp.NewExchange(ebCh, c.EventBusExchange)
if err != nil {
logger.Log("Failed to declare EventBus exchange: %v\n", err)
os.Exit(1)
}
// start server
srv := server.NewServer(c, logger, dbConn, redis, ebCh)
forever := make(chan struct{}) forever := make(chan struct{})
srv.StartWithGracefulShutdown(forever)
err := srv.Base.Start(forever, srv.Shutdown())
// server.SetupMiddleware(srv)
// server.SetupRouter(srv)
<-forever <-forever
// os.Exit(1) if err != nil {
log.Fatalf("Failed to start server. Reason: %v\n", err)
os.Exit(1)
}
fmt.Println("Done.")
os.Exit(0)
// c.AppDomain = cnf.GetEnv("APP_DOMAIN", defAppDomain)
// c.Port, _ = strconv.Atoi(c.NetAddr[1:])
// c.KVNamespace = cnf.GetEnv("APP_KV_NAMESPACE", defKVNmspc)
// logHost, logPort := fluentd.ParseAddr(c.LoggerAddr)
// logger := fluentd.NewLogger(c.GetAppFullName(), logHost, logPort)
// defer logger.Close()
// db conn
// dbConn, err := database.Connect(c.DbURL)
// if err != nil { // fixme: add wait-for-db...
// logger.Log("Failed to connect to Database server: %v\n", err)
// os.Exit(1)
// }
// defer dbConn.Close()
// redis conn
// redis := redis.NewClient(&redis.Options{
// Addr: c.CacheAddr,
// Password: c.CachePassword,
// DB: 0,
// })
// defer redis.Close()
// eventbus conn
// ebConn, ebCh, err := amqp.Open(c.EventBusURL)
// if err != nil {
// logger.Log("Failed to connect to EventBus server: %v\n", err)
// os.Exit(1)
// }
// defer ebCh.Close()
// defer amqp.Close(ebConn)
// err = amqp.NewExchange(ebCh, c.EventBusExchange)
// if err != nil {
// logger.Log("Failed to declare EventBus exchange: %v\n", err)
// os.Exit(1)
// }
} }

75
src/cmd/worker/main.go
View File

@ -1,9 +1,6 @@
package main package main
import ( import (
"bytes"
"encoding/json"
"errors"
"fmt" "fmt"
"log" "log"
"os" "os"
@ -12,17 +9,19 @@ import (
"syscall" "syscall"
"time" "time"
"git.pbiernat.dev/egommerce/basket-service/internal/app/config"
"git.pbiernat.dev/egommerce/basket-service/internal/app/database"
"git.pbiernat.dev/egommerce/basket-service/internal/app/event"
"git.pbiernat.dev/egommerce/basket-service/internal/app/server"
"git.pbiernat.dev/egommerce/basket-service/internal/app/service"
"git.pbiernat.dev/egommerce/basket-service/internal/app/ui"
discovery "git.pbiernat.dev/egommerce/go-api-pkg/consul"
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd" "git.pbiernat.dev/egommerce/go-api-pkg/fluentd"
"git.pbiernat.dev/egommerce/go-api-pkg/rabbitmq" "git.pbiernat.dev/egommerce/go-api-pkg/rabbitmq"
"github.com/go-redis/redis/v8" "github.com/go-redis/redis/v8"
"github.com/streadway/amqp" "github.com/streadway/amqp"
"git.pbiernat.dev/egommerce/go-api-pkg/consul"
"git.pbiernat.dev/egommerce/basket-service/internal/app/event"
"git.pbiernat.dev/egommerce/basket-service/internal/app/service"
"git.pbiernat.dev/egommerce/basket-service/internal/app/ui"
"git.pbiernat.dev/egommerce/basket-service/internal/common"
"git.pbiernat.dev/egommerce/basket-service/pkg/config"
"git.pbiernat.dev/egommerce/basket-service/pkg/database"
) )
const ( const (
@ -44,34 +43,30 @@ func main() {
log.Panicln("Error loading .env file", config.ErrLoadingEnvs) log.Panicln("Error loading .env file", config.ErrLoadingEnvs)
} }
c := new(server.Config) c := common.NewConfig()
c.AppID, _ = os.Hostname() logHost, logPort, err := fluentd.ParseAddr(c.LoggerAddr)
c.AppName = config.GetEnv("APP_NAME", defAppName) if err != nil {
c.LoggerAddr = config.GetEnv("LOGGER_ADDR", defLoggerAddr) log.Fatalf("Error parsing logger addr: %s. Err: %v", c.LoggerAddr, err)
c.RegistryAddr = config.GetEnv("REGISTRY_ADDR", defRegistryAddr) }
c.DbURL = config.GetEnv("DATABASE_URL", defDbURL)
c.CacheAddr = config.GetEnv("CACHE_ADDR", defCacheAddr)
c.CachePassword = config.GetEnv("CACHE_PASSWORD", defCachePassword)
c.EventBusURL = config.GetEnv("EVENTBUS_URL", defEventBusURL)
c.EventBusExchange = ebEventsExchange
c.EventBusQueue = ebEventsQueue
c.KVNamespace = config.GetEnv("APP_KV_NAMESPACE", defKVNmspc)
logHost, logPort := fluentd.ParseAddr(c.LoggerAddr) logger, err := fluentd.NewLogger(c.GetAppFullName(), logHost, logPort) // @Refactor NewLogger return (logger, error)
logger := fluentd.NewLogger(c.GetAppFullName(), logHost, logPort) if err != nil {
log.Fatalf("Error connecting to %s:%d. Err: %v", logHost, logPort, err)
}
defer logger.Close() defer logger.Close()
consul, err := discovery.NewService(c.RegistryAddr, c.AppID, c.AppName, c.AppID, "", "", 0) // consul, err := consul.NewService(c.RegistryAddr, c.Base.AppID, c.Base.AppName, c.Base.AppID, "", "", 0)
registry, err := consul.NewService(c.RegistryAddr, c.Base.AppID, c.Base.AppName, c.Base.AppID, "", "", 0)
if err != nil { if err != nil {
logger.Log("Error connecting to %s: %v", c.RegistryAddr, err) logger.Log("Error connecting to %s: %v", c.RegistryAddr, err)
} }
go func(consul *discovery.Service) { go func(consul *consul.Service) {
ticker := time.NewTicker(time.Second * 15) ticker := time.NewTicker(time.Second * 15)
for range ticker.C { for range ticker.C {
updateKVConfig(consul, c) // FIXME: duplicated in internal/app/server/server.go updateKVConfig(consul, c) // FIXME: duplicated in internal/app/server/server.go
} }
}(consul) }(registry)
// db conn // db conn
dbConn, err := database.Connect(c.DbURL) dbConn, err := database.Connect(c.DbURL)
@ -202,21 +197,21 @@ func main() {
<-forever <-forever
} }
func updateKVConfig(s *discovery.Service, oldCnf *server.Config) error { // FIXME: duplicated in internal/app/server/server.go func updateKVConfig(s *consul.Service, oldCnf *common.Config) error { // FIXME: duplicated in internal/app/server/server.go
data, _, err := s.KV().Get(oldCnf.KVNamespace, nil) // data, _, err := s.KV().Get(oldCnf.KVNamespace, nil)
if err != nil { // if err != nil {
return err // return err
} // }
if data == nil { // if data == nil {
return errors.New("empty KV config data") // return errors.New("empty KV config data")
} // }
buf := bytes.NewBuffer(data.Value) // buf := bytes.NewBuffer(data.Value)
decoder := json.NewDecoder(buf) // decoder := json.NewDecoder(buf)
if err := decoder.Decode(oldCnf); err != nil { // if err := decoder.Decode(oldCnf); err != nil {
return err // return err
} // }
return nil return nil
} }

2
src/go.mod
View File

@ -4,7 +4,7 @@ go 1.18
require ( require (
git.pbiernat.dev/egommerce/api-entities v0.0.26 git.pbiernat.dev/egommerce/api-entities v0.0.26
git.pbiernat.dev/egommerce/go-api-pkg v0.0.136 git.pbiernat.dev/egommerce/go-api-pkg v0.0.150
github.com/georgysavva/scany/v2 v2.0.0 github.com/georgysavva/scany/v2 v2.0.0
github.com/go-pg/migrations/v8 v8.1.0 github.com/go-pg/migrations/v8 v8.1.0
github.com/go-pg/pg/v10 v10.10.7 github.com/go-pg/pg/v10 v10.10.7

4
src/go.sum
View File

@ -3,6 +3,10 @@ git.pbiernat.dev/egommerce/api-entities v0.0.26 h1:Avz02GINwuYWOjw1fmZIJ3QgGEIz3
git.pbiernat.dev/egommerce/api-entities v0.0.26/go.mod h1:+BXvUcr6Cr6QNpJsW8BUfe1vVILdWDADNE0e3u0lNvU= git.pbiernat.dev/egommerce/api-entities v0.0.26/go.mod h1:+BXvUcr6Cr6QNpJsW8BUfe1vVILdWDADNE0e3u0lNvU=
git.pbiernat.dev/egommerce/go-api-pkg v0.0.136 h1:SzJRAkqJKdng/3d0V7o/R0yGh7QaZynPBn/P++on9RA= git.pbiernat.dev/egommerce/go-api-pkg v0.0.136 h1:SzJRAkqJKdng/3d0V7o/R0yGh7QaZynPBn/P++on9RA=
git.pbiernat.dev/egommerce/go-api-pkg v0.0.136/go.mod h1:w2N79aoumjrrcrGPJLkCwxAHtrLd7G4Uj8VOxvPooa0= git.pbiernat.dev/egommerce/go-api-pkg v0.0.136/go.mod h1:w2N79aoumjrrcrGPJLkCwxAHtrLd7G4Uj8VOxvPooa0=
git.pbiernat.dev/egommerce/go-api-pkg v0.0.149 h1:7K4z/XUMAPrnvOPcFfkeeNCpuc5IHoC1Pd68Ht7q9Ts=
git.pbiernat.dev/egommerce/go-api-pkg v0.0.149/go.mod h1:w2N79aoumjrrcrGPJLkCwxAHtrLd7G4Uj8VOxvPooa0=
git.pbiernat.dev/egommerce/go-api-pkg v0.0.150 h1:DMM3Kxb6HNw4BExA7Ss7P9ivs+TIeO9DxjfHPKeWrSg=
git.pbiernat.dev/egommerce/go-api-pkg v0.0.150/go.mod h1:w2N79aoumjrrcrGPJLkCwxAHtrLd7G4Uj8VOxvPooa0=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ= github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs= github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs=

20
src/internal/app/log.go
View File

@ -1,20 +0,0 @@
package app
import (
"log"
)
const AppName = "basket-svc"
func Panic(v ...any) {
log.Panicln(AppName+":", v)
}
func Panicf(format string, v ...any) {
log.Panicf(AppName+": "+format, v...)
}
func Panicln(v ...any) {
v = append([]any{AppName + ":"}, v...)
log.Panicln(v...)
}

31
src/internal/app/server/basket_handler.go
View File

@ -1,30 +1,31 @@
package server package server
// REFACTOR: APP DEDICATED CODE
import ( import (
"context" "context"
"time" "time"
def "git.pbiernat.dev/egommerce/api-entities/http" "git.pbiernat.dev/egommerce/api-entities/http"
"git.pbiernat.dev/egommerce/basket-service/internal/app/service" "git.pbiernat.dev/egommerce/basket-service/internal/app/service"
"git.pbiernat.dev/egommerce/basket-service/internal/app/ui" "git.pbiernat.dev/egommerce/basket-service/internal/app/ui"
"github.com/gofiber/fiber/v2" "github.com/gofiber/fiber/v2"
) )
func (s *Server) GetBasketHandler(c *fiber.Ctx) error { func (s *Server) GetBasketHandler(c *fiber.Ctx) error {
req := new(def.GetBasketRequest) req := new(http.GetBasketRequest)
if err := c.BodyParser(req); err != nil { if err := c.BodyParser(req); err != nil {
return s.Error400(c, err.Error()) return s.Base.Error400(c, err.Error())
} }
basketID := req.BasketID basketID := req.BasketID
basketSrv := service.NewBasketService(s.db, s.cache, s.ebCh, s.log) basketSrv := service.NewBasketService(s.Database, s.Cache, s.Eventbus, s.Logger)
ctx := context.Background() ctx := context.Background()
basket, err := basketSrv.FetchFromDB(ctx, basketID) basket, err := basketSrv.FetchFromDB(ctx, basketID)
if err != nil { if err != nil {
return s.Error400(c, "Failed to retrieve basket") return s.Base.Error400(c, "Failed to retrieve basket")
} }
res := &def.GetBasketResponse{ res := &http.GetBasketResponse{
ID: basket.ID, ID: basket.ID,
State: basket.State, State: basket.State,
CreatedAt: time.Duration(basket.CreatedAt.Time.Unix()), CreatedAt: time.Duration(basket.CreatedAt.Time.Unix()),
@ -38,16 +39,16 @@ func (s *Server) GetBasketHandler(c *fiber.Ctx) error {
func (s *Server) GetBasketItemsHandler(c *fiber.Ctx) error { func (s *Server) GetBasketItemsHandler(c *fiber.Ctx) error {
basketID := c.Params("basketId", "") basketID := c.Params("basketId", "")
basketSrv := service.NewBasketService(s.db, s.cache, s.ebCh, s.log) basketSrv := service.NewBasketService(s.Database, s.Cache, s.Eventbus, s.Logger)
ctx := context.Background() ctx := context.Background()
items, err := basketSrv.FetchItems(ctx, basketID) items, err := basketSrv.FetchItems(ctx, basketID)
if err != nil { if err != nil {
return s.Error400(c, "Failed to retrieve basket items") return s.Base.Error400(c, "Failed to retrieve basket items")
} }
var res []*def.GetBasketItemsResponse // FIXME var res []*http.GetBasketItemsResponse
for _, item := range items { for _, item := range items {
resItem := &def.GetBasketItemsResponse{ resItem := &http.GetBasketItemsResponse{
ID: item.ID, ID: item.ID,
BasketID: item.BasketID, BasketID: item.BasketID,
ProductID: item.ProductID, ProductID: item.ProductID,
@ -65,17 +66,17 @@ func (s *Server) GetBasketItemsHandler(c *fiber.Ctx) error {
} }
func (s *Server) CheckoutHandler(c *fiber.Ctx) error { func (s *Server) CheckoutHandler(c *fiber.Ctx) error {
reqID, _ := s.GetRequestID(c) reqID, _ := s.Base.GetRequestID(c)
req := new(def.BasketCheckoutRequest) req := new(http.BasketCheckoutRequest)
if err := c.BodyParser(req); err != nil { if err := c.BodyParser(req); err != nil {
return s.Error400(c, err.Error()) return s.Base.Error400(c, err.Error())
} }
basketID := req.BasketID basketID := req.BasketID
basketSrv := service.NewBasketService(s.db, s.cache, s.ebCh, s.log) basketSrv := service.NewBasketService(s.Database, s.Cache, s.Eventbus, s.Logger)
res, err := ui.CheckoutBasket(basketSrv, basketID, reqID) res, err := ui.CheckoutBasket(basketSrv, basketID, reqID)
if err != nil { if err != nil {
return s.Error400(c, "Failed to create order") return s.Base.Error400(c, "Failed to create order")
} }
return c.JSON(res) return c.JSON(res)

32
src/internal/app/server/config.go
View File

@ -1,32 +0,0 @@
package server
import "fmt"
type Config struct {
AppID string
AppName string
AppDomain string
PathPrefix string
NetAddr string
Port int
RegistryAddr string
KVNamespace string
LoggerAddr string `json:"logger_addr"`
DbURL string `json:"db_url"`
CacheAddr string `json:"cache_addr"`
CachePassword string `json:"cache_password"`
MongoDbUrl string `json:"mongodb_url"`
EventBusURL string `json:"eventbus_url"`
EventBusExchange string `json:"eventbus_exchange"`
EventBusQueue string `json:"eventbus_queue"`
HttpReadTimeout int `json:"http_read_timeout"`
HttpWriteTimeout int `json:"http_write_timeout"`
HttpIdleTimeout int `json:"http_idle_timeout"`
// Fields with json mapping are available trough ConsulKV
}
func (c *Config) GetAppFullName() string {
return fmt.Sprintf("%s_%s", c.AppName, c.AppID)
}

11
src/internal/app/server/health_handler.go
View File

@ -1,17 +1,18 @@
package server package server
// REFACTOR: UNIVERSAL SERVER CODE
import ( import (
"github.com/gofiber/fiber/v2" "github.com/gofiber/fiber/v2"
def "git.pbiernat.dev/egommerce/api-entities/http" "git.pbiernat.dev/egommerce/api-entities/http"
) )
func (s *Server) HealthHandler(c *fiber.Ctx) error { func (s *Server) HealthHandler(c *fiber.Ctx) error { // TODO add necessary logic
return c.JSON(&def.HealthResponse{ return c.JSON(&http.HealthResponse{
Status: "OK", Status: "OKa",
}) })
} }
func (s *Server) ConfigHandler(c *fiber.Ctx) error { func (s *Server) ConfigHandler(c *fiber.Ctx) error {
return c.JSON(s.conf) return c.JSON(s.Config)
} }

33
src/internal/app/server/middleware.go Normal file
View File

@ -0,0 +1,33 @@
package server
import (
"strings"
"github.com/gofiber/fiber/v2"
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd"
)
// "github.com/gofiber/fiber/v2"
// "github.com/gofiber/fiber/v2/middleware/cors"
func SetupMiddleware(s *Server) {
s.Base.Use(defaultCORS)
s.Base.Use(LoggingMiddleware(s.Logger))
}
func LoggingMiddleware(log *fluentd.Logger) func(c *fiber.Ctx) error {
return func(c *fiber.Ctx) error {
path := string(c.Request().URI().Path())
if strings.Contains(path, "/health") {
return c.Next()
}
log.Log("Request: %s, remote: %s, via: %s",
c.Request().URI().String(),
c.Context().RemoteIP().String(),
string(c.Context().UserAgent()))
return c.Next()
}
}

35
src/internal/app/server/router.go
View File

@ -1,10 +1,9 @@
package server package server
// REFACTOR: APP DEDICATED CODE
import ( import (
"net/http" "net/http"
"strings"
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd"
"github.com/gofiber/fiber/v2" "github.com/gofiber/fiber/v2"
"github.com/gofiber/fiber/v2/middleware/cors" "github.com/gofiber/fiber/v2/middleware/cors"
) )
@ -18,24 +17,19 @@ var (
}) })
) )
func SetupRoutes(s *Server) { func SetupRouter(s *Server) {
s.App.Options("*", defaultCORS) s.Base.Options("*", defaultCORS)
s.App.Get("/health", s.HealthHandler) s.Base.Get("/health", s.HealthHandler)
s.App.Get("/config", s.ConfigHandler) s.Base.Get("/config", s.ConfigHandler)
api := s.App.Group("/api") api := s.Base.Group("/api")
v1 := api.Group("/v1") v1 := api.Group("/v1")
v1.Get("/basket", s.GetBasketHandler) v1.Get("/basket", s.GetBasketHandler)
v1.Get("/basket/:basketId/items", s.GetBasketItemsHandler) v1.Get("/basket/:basketId/items", s.GetBasketItemsHandler)
v1.Post("/checkout", s.CheckoutHandler) v1.Post("/checkout", s.CheckoutHandler)
} }
func SetupMiddlewares(s *Server) {
s.App.Use(defaultCORS)
s.App.Use(LoggingMiddleware(s.log))
}
func CORSPreflightMiddleware(c *fiber.Ctx) error { func CORSPreflightMiddleware(c *fiber.Ctx) error {
if string(c.Request().Header.Method()) == http.MethodOptions { if string(c.Request().Header.Method()) == http.MethodOptions {
c.Response().SetStatusCode(http.StatusOK) c.Response().SetStatusCode(http.StatusOK)
@ -44,20 +38,3 @@ func CORSPreflightMiddleware(c *fiber.Ctx) error {
return c.Next() return c.Next()
} }
// Middlewares
func LoggingMiddleware(log *fluentd.Logger) func(c *fiber.Ctx) error {
return func(c *fiber.Ctx) error {
path := string(c.Request().URI().Path())
if strings.Contains(path, "/health") {
return c.Next()
}
log.Log("Request: %s, remote: %s, via: %s",
c.Request().URI().String(),
c.Context().RemoteIP().String(),
string(c.Context().UserAgent()))
return c.Next()
}
}

287
src/internal/app/server/server.go
View File

@ -4,73 +4,147 @@ import (
"bytes" "bytes"
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"log"
"os" "os"
"os/signal" "strconv"
"syscall"
"time" "time"
"github.com/go-redis/redis/v8" "github.com/go-redis/redis/v8"
"github.com/gofiber/fiber/v2"
"github.com/jackc/pgx/v5/pgxpool" "github.com/jackc/pgx/v5/pgxpool"
"github.com/streadway/amqp" "github.com/streadway/amqp"
def "git.pbiernat.dev/egommerce/api-entities/http" "git.pbiernat.dev/egommerce/go-api-pkg/consul"
discovery "git.pbiernat.dev/egommerce/go-api-pkg/consul"
"git.pbiernat.dev/egommerce/go-api-pkg/fluentd" "git.pbiernat.dev/egommerce/go-api-pkg/fluentd"
db "git.pbiernat.dev/egommerce/basket-service/pkg/database"
srv "git.pbiernat.dev/egommerce/basket-service/pkg/server"
"git.pbiernat.dev/egommerce/basket-service/internal/common"
) )
type Server struct { type (
*fiber.App Server struct {
conf *Config Base *srv.Server
log *fluentd.Logger Config *common.Config
db *pgxpool.Pool
cache *redis.Client Cache *redis.Client
ebCh *amqp.Channel Database *pgxpool.Pool
discovery *discovery.Service Eventbus *amqp.Channel
name string Logger *fluentd.Logger
addr string Registry *consul.Service
kvNmspc string kvNmspc string
} }
type Headers struct { OptionFn func(*Server) error
RequestID string `reqHeader:"x-request-id"` )
func New(c *common.Config, opts ...OptionFn) *Server {
svr := &Server{
Base: srv.New(c.Base),
} }
func NewServer(conf *Config, logger *fluentd.Logger, db *pgxpool.Pool, cache *redis.Client, ebCh *amqp.Channel) *Server { for _, opt := range opts {
consul, err := discovery.NewService(conf.RegistryAddr, conf.AppID, conf.AppName, conf.AppID, conf.AppDomain, conf.PathPrefix, conf.Port) if err := opt(svr); err != nil {
log.Fatalf("Failed to start HTTP Server. Err: %v\n", err)
}
}
SetupMiddleware(svr)
SetupRouter(svr)
return svr
}
func WithCache(c *common.Config) OptionFn {
redis := redis.NewClient(&redis.Options{
Addr: c.CacheAddr,
Password: c.CachePassword,
DB: 0,
})
return func(s *Server) error {
s.Cache = redis
return nil
}
// defer redis.Close()
}
func WithDatabase(c *common.Config) OptionFn {
dbConn, err := db.Connect(c.DbURL)
if err != nil { if err != nil {
logger.Log("Error connecting to %s: %v", conf.RegistryAddr, err) log.Fatalf("Failed to connect to the Database: %s. Err: %v\n", c.DbURL, err)
os.Exit(1)
} }
logger.Log("Registering service with name: %s, address: %s", consul.Name, consul.Address) return func(s *Server) error {
err = consul.Register() s.Database = dbConn
return nil
}
// defer dbConn.Close()
}
func WithEventbus(c *common.Config) OptionFn {
conn, err := amqp.Dial(c.EventBusURL)
if err != nil { if err != nil {
logger.Log("register error: %v", err) log.Fatalf("Failed to connect to the Eventbus: %s. Err: %v\n", c.EventBusURL, err)
} }
cnf := fiber.Config{ chn, err := conn.Channel()
AppName: conf.AppName, if err != nil {
ServerHeader: conf.AppName, log.Fatalf("Failed to open new Eventbus channel. Err: %v\n", err)
ReadTimeout: time.Millisecond * 50,
WriteTimeout: time.Millisecond * 50,
IdleTimeout: time.Millisecond * 50,
} }
s := &Server{
fiber.New(cnf), return func(s *Server) error {
conf, s.Eventbus = chn
logger,
db, return nil
cache,
ebCh,
consul,
conf.AppName,
conf.NetAddr,
conf.KVNamespace,
} }
// defer ebCh.Close()
// defer amqp.Close(ebConn)
}
func WithLogger(c *common.Config) OptionFn {
logHost, logPort, err := fluentd.ParseAddr(c.LoggerAddr)
if err != nil {
log.Fatalf("Failed to parse Fluentd address: %s. Err: %v", c.LoggerAddr, err)
}
logger, err := fluentd.NewLogger(c.GetAppFullName(), logHost, logPort)
if err != nil {
log.Fatalf("Failed to connect to the Fluentd on %s:%d. Err: %v", logHost, logPort, err)
}
return func(s *Server) error {
s.Logger = logger
return nil
}
// defer logger.Close()
}
func WithRegistry(c *common.Config) OptionFn {
// fmt.Printf("WithRegistry constructor: config: %v", c.Base)
port, _ := strconv.Atoi(c.Base.NetAddr[1:]) // FIXME: can be IP:PORT which now will cause error
registry, err := consul.NewService(c.RegistryAddr, c.Base.AppID, c.Base.AppName, c.Base.AppID, c.Base.AppName, c.Base.PathPrefix, port)
if err != nil {
log.Fatalf("Failed to connect to the Consul on: %s. Err: %v", c.RegistryAddr, err)
}
err = registry.Register()
if err != nil {
log.Fatalf("Failed to register in the Consul service. Err: %v", err)
}
return func(s *Server) error {
s.Registry = registry
go func(s *Server) { // Consul KV updater go func(s *Server) { // Consul KV updater
interval := time.Second * 15 ticker := time.NewTicker(time.Second * 15)
ticker := time.NewTicker(interval)
for range ticker.C { for range ticker.C {
s.updateKVConfig() s.updateKVConfig()
} }
@ -83,98 +157,105 @@ func NewServer(conf *Config, logger *fluentd.Logger, db *pgxpool.Pool, cache *re
} }
}(s) }(s)
SetupMiddlewares(s) return nil
SetupRoutes(s) }
return s
} }
// func (s *Server) Start() { // REFACTOR IN PROGRESS
// err := s.Listen(s.addr) // func (s *Server) Shutdown() error {
// s.log.Log("Starting error: %v", err) // s.Logger.Log("Server is going down... Unregistering service: %s", s.Discovery.GetID())
// s.Discovery.Unregister()
// s.clearMetadataCache()
// s.Cache.Close()
// s.Database.Close()
// s.Eventbus.Close()
// s.Logger.Close()
// return nil
// } // }
func (s *Server) StartWithGracefulShutdown(forever chan struct{}) { func (s *Server) Shutdown() srv.PurgeFn {
go func() { return func(srv *srv.Server) error {
sigint := make(chan os.Signal, 1) fmt.Printf("%v", s.Base)
signal.Notify(sigint, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) // s.Logger.Log("Server is going down... Unregistering service: %s", s.Base.AppID)
<-sigint s.Logger.Log("Server is going down... Unregistering service...")
if err := s.gracefulShutdown(); err != nil { s.Registry.Unregister()
s.log.Log("Server is not shutting down! Reason: %v", err) s.clearMetadataCache()
s.Eventbus.Close()
s.Database.Close()
s.Logger.Close()
return s.Base.Shutdown()
}
} }
close(forever) // END: REFACTOR IN PROGRESS
}()
if err := s.Listen(s.addr); err != nil { // @CHECK: merge s.Config and s.Base.Config to display all config as one array/map
s.log.Log("Server is not running! Reason: %v", err) func (s *Server) updateKVConfig() {
} config, _, err := s.Registry.KV().Get(s.kvNmspc, nil)
<-forever
}
func (s *Server) GetRequestID(c *fiber.Ctx) (string, error) {
var hdr = new(Headers)
if err := c.ReqHeaderParser(hdr); err != nil {
return "", err
}
return hdr.RequestID, nil
}
func (s *Server) Error400(c *fiber.Ctx, msg string) error {
return c.Status(fiber.StatusBadRequest).JSON(&def.ErrorResponse{Error: msg})
}
func (s *Server) Error404(c *fiber.Ctx, msg string) error {
return c.Status(fiber.StatusNotFound).JSON(&def.ErrorResponse{Error: msg})
}
func (s *Server) updateKVConfig() { // FIXME: duplicated in cmd/worker/main.go
config, _, err := s.discovery.KV().Get(s.kvNmspc, nil)
if err != nil || config == nil { if err != nil || config == nil {
return return
} }
kvCnf := bytes.NewBuffer(config.Value) kvCnf := bytes.NewBuffer(config.Value)
decoder := json.NewDecoder(kvCnf) decoder := json.NewDecoder(kvCnf)
if err := decoder.Decode(&s.conf); err != nil { if err := decoder.Decode(&s.Config); err != nil {
return return
} }
} }
func (s *Server) cacheMetadata() { func (s *Server) cacheMetadata() {
ctx := context.Background() ctx := context.TODO()
key, address := s.getMetadataIPsKey(), s.conf.AppID key, address := s.getMetadataIPsKey(), s.Base.Config.AppID
pos := s.cache.LPos(ctx, key, address, redis.LPosArgs{}).Val() pos := s.Cache.LPos(ctx, key, address, redis.LPosArgs{}).Val()
if pos >= 0 { if pos >= 0 {
s.cache.LRem(ctx, key, 0, address) s.Cache.LRem(ctx, key, 0, address)
} }
s.cache.LPush(ctx, key, address).Err() s.Cache.LPush(ctx, key, address).Err()
} }
func (s *Server) clearMetadataCache() { func (s *Server) clearMetadataCache() {
ctx := context.Background() ctx := context.TODO()
key, address := s.getMetadataIPsKey(), s.conf.AppID fmt.Printf("metadata: %v", s.Config.Base)
key, address := s.getMetadataIPsKey(), s.Config.Base.AppID
s.cache.LRem(ctx, key, 0, address) s.Cache.LRem(ctx, key, 0, address)
} }
func (s *Server) getMetadataIPsKey() string { func (s *Server) getMetadataIPsKey() string {
return "internal__" + s.conf.AppName + "__ips" return "internal__" + s.Base.Config.AppName + "__ips"
} }
func (s *Server) gracefulShutdown() error { //
s.log.Log("Server is going down... Unregistering service: %s", s.discovery.GetID()) //
s.discovery.Unregister() //
s.clearMetadataCache() //// OLD CODE TO BE REMOVED
// func NewServer(conf *Config, logger *fluentd.Logger, db *pgxpool.Pool, cache *redis.Client, ebCh *amqp.Channel) *Server {
s.ebCh.Close() // cnf := fiber.Config{
s.db.Close() // AppName: conf.AppName,
s.log.Close() // ServerHeader: conf.AppName,
// ReadTimeout: conf.ReadTimeout * time.Millisecond,
// WriteTimeout: conf.WriteTimeout * time.Millisecond,
// IdleTimeout: conf.IdleTimeout * time.Millisecond,
// }
// s := &Server{
// fiber.New(cnf),
// conf,
// logger,
// db,
// cache,
// ebCh,
// consul,
// conf.AppName,
// conf.NetAddr,
// conf.KVNamespace,
// }
return s.Shutdown() // return s
} // }

6
src/internal/app/service/basket.go
View File

@ -15,7 +15,7 @@ import (
) )
const ( const (
SERVICE_USER_AGENT = "basket-httpclient" ServiceUserAgent = "basket-httpclient"
) )
type BasketService struct { type BasketService struct {
@ -82,7 +82,7 @@ func (s *BasketService) FetchItem(ctx context.Context, basketID string, productI
func (s *BasketService) AddItem(ctx context.Context, itemID int, basketID string, qty int) error { func (s *BasketService) AddItem(ctx context.Context, itemID int, basketID string, qty int) error {
var price float64 = 0 var price float64 = 0
pricingAPI := api.NewPricingAPI(SERVICE_USER_AGENT, s.redis) pricingAPI := api.NewPricingAPI(ServiceUserAgent, s.redis)
productPrice, err := pricingAPI.GetProductPrice(itemID) productPrice, err := pricingAPI.GetProductPrice(itemID)
if err == nil { if err == nil {
@ -112,7 +112,7 @@ func (s *BasketService) RemoveItem(ctx context.Context, itemID int, basketID str
func (s *BasketService) UpdateItem(ctx context.Context, item *model.BasketItemModel, qty int) error { func (s *BasketService) UpdateItem(ctx context.Context, item *model.BasketItemModel, qty int) error {
var price float64 = 0 var price float64 = 0
pricingAPI := api.NewPricingAPI(SERVICE_USER_AGENT, s.redis) pricingAPI := api.NewPricingAPI(ServiceUserAgent, s.redis)
productPrice, err := pricingAPI.GetProductPrice(item.ProductID) productPrice, err := pricingAPI.GetProductPrice(item.ProductID)
if err == nil { if err == nil {

84
src/internal/common/config.go Normal file
View File

@ -0,0 +1,84 @@
package common
import (
"fmt"
"os"
cnf "git.pbiernat.dev/egommerce/basket-service/pkg/config"
srv "git.pbiernat.dev/egommerce/basket-service/pkg/server"
)
const (
// defAppDomain = "basket-svc"
defAppName = "basket-svc"
defCacheAddr = "api-cache:6379"
defCachePassword = "12345678"
defDbURL = "postgres://postgres:12345678@postgres-db:5432/egommerce"
defEventBusURL = "amqp://guest:guest@api-eventbus:56721"
defKVNmspc = "dev.egommerce/service/basket-svc"
defLoggerAddr = "api-logger:24224"
defNetAddr = ":80"
defMongoDbURL = "mongodb://mongodb:12345678@mongo-db:27017"
defPathPrefix = "/basket"
defRegistryAddr = "api-registry:8500"
defEbEventsExchange = "api-events"
defEbEventsQueue = "basket-svc"
)
type Config struct {
Base *srv.Config
// AppID string
// AppName string
// NetAddr string
// PathPrefix string
// IdleTimeout time.Duration `json:"idle_timeout"` // miliseconds
// ReadTimeout time.Duration `json:"read_timeout"` // miliseconds
// WriteTimeout time.Duration `json:"write_timeout"` // miliseconds
DbURL string `json:"db_url"`
CacheAddr string `json:"cache_addr"`
CachePassword string `json:"cache_password"`
EventBusExchange string `json:"eventbus_exchange"`
EventBusQueue string `json:"eventbus_queue"`
EventBusURL string `json:"eventbus_url"`
LoggerAddr string `json:"logger_addr"`
RegistryAddr string
// Fields with JSON mappings are available through Consul KV storage
// Port int
// KVNamespace string
// MongoDbUrl string `json:"mongodb_url"`
// HttpReadTimeout int `json:"http_read_timeout"`
// HttpWriteTimeout int `json:"http_write_timeout"`
// HttpIdleTimeout int `json:"http_idle_timeout"`
}
func NewConfig() *Config {
c := new(Config)
c.Base = new(srv.Config)
c.Base.AppID, _ = os.Hostname()
c.Base.AppName = cnf.GetEnv("APP_NAME", defAppName)
c.Base.NetAddr = cnf.GetEnv("SERVER_ADDR", defNetAddr)
c.Base.PathPrefix = cnf.GetEnv("APP_PATH_PREFIX", defPathPrefix)
c.CacheAddr = cnf.GetEnv("CACHE_ADDR", defCacheAddr)
c.CachePassword = cnf.GetEnv("CACHE_PASSWORD", defCachePassword)
c.DbURL = cnf.GetEnv("DATABASE_URL", defDbURL)
c.EventBusExchange = defEbEventsExchange
c.EventBusURL = cnf.GetEnv("EVENTBUS_URL", defEventBusURL)
c.LoggerAddr = cnf.GetEnv("LOGGER_ADDR", defLoggerAddr)
c.RegistryAddr = cnf.GetEnv("REGISTRY_ADDR", defRegistryAddr)
return c
}
func (c *Config) GetAppFullName() string {
return fmt.Sprintf("%s_%s", c.Base.AppName, c.Base.AppID) // @TODO check if Base prop can be private
}
func (c *Config) GetListenAddr() string {
return "" // @TODO: Implement me!
}

2
src/internal/app/config/env.go → src/pkg/config/config.go
View File

@ -12,7 +12,7 @@ func init() {
ErrLoadingEnvs = godotenv.Load() ErrLoadingEnvs = godotenv.Load()
} }
func GetEnv(name string, defVal string) string { // FIXME defVal and return types func GetEnv(name string, defVal string) string {
env := os.Getenv(name) env := os.Getenv(name)
if env == "" { if env == "" {
return defVal return defVal

0
src/internal/app/database/connect.go → src/pkg/database/connect.go
View File

21
src/pkg/server/config.go Normal file
View File

@ -0,0 +1,21 @@
package server
import (
"fmt"
"time"
)
type Config struct {
AppID string
AppName string
NetAddr string
PathPrefix string
IdleTimeout time.Duration // miliseconds
ReadTimeout time.Duration // miliseconds
WriteTimeout time.Duration // miliseconds
}
func (c *Config) GetAppFullName() string {
return fmt.Sprintf("%s_%s", c.AppName, c.AppID)
}

101
src/pkg/server/server.go Normal file
View File

@ -0,0 +1,101 @@
package server
import (
"fmt"
"log"
"os"
"os/signal"
"syscall"
"time"
"github.com/gofiber/fiber/v2"
"git.pbiernat.dev/egommerce/api-entities/http"
)
type (
HeaderRequestID struct {
RequestID string `reqHeader:"x-request-id"`
}
Server struct {
*fiber.App
*Config
addr string // e.g. "127.0.0.1:8080"
// name string // e.g. "awesome-rest-api"
// kvNmspc string
// cache *redis.Client
// db *pgxpool.Pool
// discovery *discovery.Service
// ebCh *amqp.Channel
// log *fluentd.Logger
}
PurgeFn func(*Server) error
)
func New(conf *Config) *Server {
return &Server{
App: fiber.New(fiber.Config{
AppName: conf.AppID,
ServerHeader: conf.AppName,
ReadTimeout: conf.ReadTimeout * time.Millisecond,
WriteTimeout: conf.WriteTimeout * time.Millisecond,
IdleTimeout: conf.IdleTimeout * time.Millisecond,
}),
addr: conf.NetAddr,
}
}
func (s *Server) GetRequestID(c *fiber.Ctx) (string, error) {
var hdr = new(HeaderRequestID)
if err := c.ReqHeaderParser(hdr); err != nil {
return "", err
}
return hdr.RequestID, nil
}
// @Refactor make single func with error message and optional http status code...
func (s *Server) Error400(c *fiber.Ctx, msg string) error {
return c.Status(fiber.StatusBadRequest).JSON(http.ErrorResponse{msg})
// test with &(reference) before http.ErrorMessage, but probably it's gonna be erroneous
}
func (s *Server) Error401(c *fiber.Ctx, msg string) error {
return c.Status(fiber.StatusUnauthorized).JSON(http.ErrorResponse{msg})
}
func (s *Server) Error403(c *fiber.Ctx, msg string) error {
return c.Status(fiber.StatusForbidden).JSON(http.ErrorResponse{msg})
}
func (s *Server) Error404(c *fiber.Ctx, msg string) error {
return c.Status(fiber.StatusNotFound).JSON(http.ErrorResponse{msg})
}
// @EndRefactor
func (s *Server) Start(forever chan struct{}, prgFn PurgeFn) error {
go func() {
fmt.Println("Starting...")
sigint := make(chan os.Signal, 1)
signal.Notify(sigint, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
<-sigint
fmt.Println("shutting down: after term signal.")
if err := prgFn(s); err != nil {
log.Fatalf("Failed to shutdown server. Reason: %v\n", err)
}
close(forever)
}()
err := s.Listen(s.addr)
<-forever
// if err := s.Listen(s.addr); err != nil {
// s.logger.Log("Failed to start server! Reason: %v\n", err)
// }
return err
}