ITQ/orderservice
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: added new repository implementation, small improvements

Browse Source
This commit is contained in:
ITQ
2025-11-11 21:35:06 +03:00
parent 9d58fe0b41
commit a2345bedac
13 changed files with 395 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/config/config.go
+18
View File
@@ -1,6 +1,7 @@
package config
import (
"fmt"
"log"
"os"
"strconv"
@@ -14,6 +15,12 @@ type Config struct {
EnableHTTPHandler bool
HTTPPort int
LogLevel string
DBHost string
DBPort int
DBUser string
DBPassword string
DBName string
RedisURI string
}
func Load() (*Config, error) {
@@ -25,6 +32,12 @@ func Load() (*Config, error) {
EnableHTTPHandler: mustGetBool("HTTP_HANDLER_ENABLE", false),
HTTPPort: mustGetInt("HTTP_PORT", 8080), //nolint:mnd // false-positive
LogLevel: getEnv("LOG_LEVEL", "info"),
DBHost: getEnv("POSTGRES_HOST", "localhost"),
DBPort: mustGetInt("POSTGRES_PORT", 5432),
DBUser: getEnv("POSTGRES_USERNAME", "postgres"),
DBPassword: getEnv("POSTGRES_PASSWORD", "postgres"),
DBName: getEnv("POSTGRES_DATABASE", "postgres"),
RedisURI: getEnv("REDIS_URI", "redis://localhost:6379"),
}, nil
}
@@ -52,3 +65,8 @@ func mustGetBool(key string, def bool) bool {
}
return b
}
func (c Config) BuildDsn() string {
return fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=disable",
c.DBHost, c.DBPort, c.DBUser, c.DBPassword, c.DBName)
}
internal/domain/order.go
+6 -4
View File
@@ -4,6 +4,8 @@ import (
"errors"
"fmt"
"strings"
"github.com/google/uuid"
)
var (
@@ -13,9 +15,9 @@ var (
)
type Order struct {
ID string
Item string
Quantity int32
ID uuid.UUID `db:"id" json:"id"`
Item string `db:"item" json:"item"`
Quantity int32 `db:"quantity" json:"quantity"`
}
func (o *Order) Validate() error {
@@ -25,7 +27,7 @@ func (o *Order) Validate() error {
if o.Quantity <= 0 {
return fmt.Errorf("%w: quantity must be positive", ErrInvalidOrderData)
}
if o.ID == "" {
if o.ID.String() == "" {
return fmt.Errorf("%w: ID cannot be empty", ErrInvalidOrderData)
}
return nil
internal/handler/grpc/errors.go
+2
View File
@@ -2,6 +2,7 @@ package handler
import (
"errors"
"log"
"orderservice/internal/domain"
@@ -20,5 +21,6 @@ func mapError(err error) error {
return status.Error(codes.InvalidArgument, err.Error())
}
log.Printf("internal server error: %v", err)
return status.Error(codes.Internal, "internal server error")
}
internal/handler/grpc/order.go
+2 -2
View File
@@ -22,7 +22,7 @@ func NewOrderHandler(service *service.OrderService) *OrderHandler {
func mapDomainStructToHandler(order *domain.Order) *pb.Order {
return &pb.Order{
Id: order.ID,
Id: order.ID.String(),
Item: order.Item,
Quantity: order.Quantity,
}
@@ -37,7 +37,7 @@ func (h *OrderHandler) CreateOrder(
return nil, mapError(err)
}
return &pb.CreateOrderResponse{Id: order.ID}, nil
return &pb.CreateOrderResponse{Id: order.ID.String()}, nil
}
func (h *OrderHandler) GetOrder(ctx context.Context, req *pb.GetOrderRequest) (*pb.GetOrderResponse, error) {
internal/repository/inmemory/order.go
+4 -4
View File
@@ -26,10 +26,10 @@ func (r *OrderRepository) Create(ctx context.Context, order *domain.Order) error
r.mu.Lock()
defer r.mu.Unlock()
if _, ok := r.orders[order.ID]; ok {
if _, ok := r.orders[order.ID.String()]; ok {
return domain.ErrOrderAlreadyExist
}
r.orders[order.ID] = order
r.orders[order.ID.String()] = order
return nil
}
@@ -58,10 +58,10 @@ func (r *OrderRepository) Update(ctx context.Context, order *domain.Order) error
r.mu.Lock()
defer r.mu.Unlock()
if _, ok := r.orders[order.ID]; !ok {
if _, ok := r.orders[order.ID.String()]; !ok {
return domain.ErrOrderNotFound
}
r.orders[order.ID] = order
r.orders[order.ID.String()] = order
return nil
}
internal/repository/postgres/order.go
+258
View File
@@ -0,0 +1,258 @@
package postgres
import (
"context"
"database/sql"
_ "embed"
"encoding/json"
"errors"
"fmt"
"log"
"time"
"orderservice/internal/domain"
"github.com/jmoiron/sqlx"
_ "github.com/lib/pq" // postgres driver
"github.com/redis/go-redis/v9"
)
//go:embed schema.sql
var Schema string
const (
orderCachePrefix = "order:"
cacheTTL = 5 * time.Minute
maxCacheRetries = 2
cacheRetryDelay = 100 * time.Millisecond
)
type OrderRepository struct {
db *sqlx.DB
cache *redis.Client
cacheEnable bool
}
type Config struct {
CacheEnable bool
}
func NewOrderRepository(db *sqlx.DB, redisClient *redis.Client, config *Config) *OrderRepository {
if config == nil {
config = &Config{
CacheEnable: true,
}
}
return &OrderRepository{
db: db,
cache: redisClient,
cacheEnable: config.CacheEnable,
}
}
func (r *OrderRepository) cacheKey(id string) string {
return orderCachePrefix + id
}
func (r *OrderRepository) Create(ctx context.Context, order *domain.Order) error {
tx, err := r.db.BeginTxx(ctx, nil)
if err != nil {
return fmt.Errorf("begin transaction: %w", err)
}
defer tx.Rollback()
query := `
insert into orders (id, item, quantity)
values (:id, :item, :quantity)
`
if _, err := tx.NamedExecContext(ctx, query, order); err != nil {
return fmt.Errorf("create order: %w", err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("commit transaction: %w", err)
}
if r.cacheEnable {
if err := r.setCacheWithRetry(ctx, order); err != nil {
log.Printf("WARN: cache set error for order %s: %v", order.ID, err)
}
}
return nil
}
func (r *OrderRepository) Get(ctx context.Context, id string) (*domain.Order, error) {
if r.cacheEnable {
if order, err := r.getFromCache(ctx, id); err == nil {
return order, nil
} else if !errors.Is(err, redis.Nil) {
log.Printf("WARN: cache get error for order %s: %v", id, err)
}
}
const query = `
select id, item, quantity
from orders
where id = $1
`
var order domain.Order
if err := r.db.GetContext(ctx, &order, query, id); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, domain.ErrOrderNotFound
}
return nil, fmt.Errorf("get order by id: %w", err)
}
if r.cacheEnable {
if err := r.setCacheWithRetry(ctx, &order); err != nil {
log.Printf("WARN: cache set error for order %s: %v", id, err)
}
}
return &order, nil
}
func (r *OrderRepository) Update(ctx context.Context, order *domain.Order) error {
tx, err := r.db.BeginTxx(ctx, nil)
if err != nil {
return fmt.Errorf("begin transaction: %w", err)
}
defer tx.Rollback()
query := `
update orders
set item = :item, quantity = :quantity
where id = :id
`
result, err := tx.NamedExecContext(ctx, query, order)
if err != nil {
return fmt.Errorf("update order: %w", err)
}
rowsAffected, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("get rows affected: %w", err)
}
if rowsAffected == 0 {
return domain.ErrOrderNotFound
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("commit transaction: %w", err)
}
if r.cacheEnable {
if err := r.setCacheWithRetry(ctx, order); err != nil {
log.Printf("WARN: cache set error for order %s: %v", order.ID, err)
}
}
return nil
}
func (r *OrderRepository) Delete(ctx context.Context, id string) error {
tx, err := r.db.BeginTxx(ctx, nil)
if err != nil {
return fmt.Errorf("begin transaction: %w", err)
}
defer tx.Rollback()
const query = `
delete from orders
where id = $1
`
result, err := tx.ExecContext(ctx, query, id)
if err != nil {
return fmt.Errorf("delete order: %w", err)
}
rowsAffected, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("get rows affected: %w", err)
}
if rowsAffected == 0 {
return domain.ErrOrderNotFound
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("commit transaction: %w", err)
}
r.invalidateCache(ctx, id)
return nil
}
func (r *OrderRepository) List(ctx context.Context) ([]*domain.Order, error) {
const query = `
select id, item, quantity
from orders
order by id
`
var orders []*domain.Order
if err := r.db.SelectContext(ctx, &orders, query); err != nil {
return nil, fmt.Errorf("list orders: %w", err)
}
return orders, nil
}
func (r *OrderRepository) getFromCache(ctx context.Context, id string) (*domain.Order, error) {
data, err := r.cache.Get(ctx, r.cacheKey(id)).Bytes()
if err != nil {
return nil, err
}
var order domain.Order
if err := json.Unmarshal(data, &order); err != nil {
r.cache.Del(ctx, r.cacheKey(id))
return nil, err
}
return &order, nil
}
func (r *OrderRepository) setCacheWithRetry(ctx context.Context, order *domain.Order) error {
data, err := json.Marshal(order)
if err != nil {
return err
}
key := r.cacheKey(order.ID.String())
for i := range maxCacheRetries {
err = r.cache.Set(ctx, key, data, cacheTTL).Err()
if err == nil {
return nil
}
if i < maxCacheRetries-1 {
time.Sleep(cacheRetryDelay)
}
}
return err
}
func (r *OrderRepository) invalidateCache(_ context.Context, id string) {
if !r.cacheEnable {
return
}
go func() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
if err := r.cache.Del(ctx, r.cacheKey(id)).Err(); err != nil {
log.Printf("WARN: cache invalidation failed for order %s: %v", id, err)
}
}()
}
internal/repository/postgres/schema.sql
+5
View File
@@ -0,0 +1,5 @@
create table if not exists orders (
id uuid primary key,
item varchar(500) not null,
quantity integer not null check (quantity > 0)
);
internal/server/server.go
+55 -8
View File
@@ -10,13 +10,15 @@ import (
"orderservice/internal/config"
"orderservice/internal/interceptor"
orderGrpcHandler "orderservice/internal/handler/grpc"
orderInMemory "orderservice/internal/repository/inmemory"
grpcHandlers "orderservice/internal/handler/grpc"
orderPostgresRepo "orderservice/internal/repository/postgres"
"orderservice/internal/service"
pb "orderservice/pkg/api/order"
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
"github.com/jmoiron/sqlx"
"github.com/redis/go-redis/v9"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/reflection"
@@ -25,6 +27,8 @@ import (
type Server struct {
grpcServer *grpc.Server
config *config.Config
db *sqlx.DB
redisDB *redis.Client
}
func New(cfg *config.Config) *Server {
@@ -46,21 +50,64 @@ func runHTTPHandler(s *Server, grpcServerEndpoint *string) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
mux := runtime.NewServeMux()
gwmux := runtime.NewServeMux()
opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
err := pb.RegisterOrderServiceHandlerFromEndpoint(ctx, mux, *grpcServerEndpoint, opts)
err := pb.RegisterOrderServiceHandlerFromEndpoint(ctx, gwmux, *grpcServerEndpoint, opts)
if err != nil {
return err
}
addr := fmt.Sprintf(":%d", s.config.HTTPPort)
return http.ListenAndServe(addr, mux)
return http.ListenAndServe(addr, gwmux)
}
func getDatabase(cfg config.Config) (*sqlx.DB, error) {
db, err := sqlx.Connect("postgres", cfg.BuildDsn())
if err != nil {
return nil, fmt.Errorf("connect to database: %w", err)
}
_, err = db.Exec(orderPostgresRepo.Schema)
if err != nil {
return nil, fmt.Errorf("run schema: %w", err)
}
return db, nil
}
func getRedis(cfg config.Config) (*redis.Client, error) {
conn, err := redis.ParseURL(cfg.RedisURI)
client := redis.NewClient(&redis.Options{
Addr: conn.Addr,
})
if err != nil {
return nil, fmt.Errorf("parse Redis URI: %w", err)
}
_, err = client.Ping(context.Background()).Result()
if err != nil {
return nil, fmt.Errorf("connect to Redis server: %w", err)
}
return client, nil
}
func (s *Server) RegisterServices() {
repo := orderInMemory.NewOrderRepository()
orderService := service.NewOrderService(repo)
orderHandler := orderGrpcHandler.NewOrderHandler(orderService)
db, err := getDatabase(*s.config)
if err != nil {
log.Print(err)
}
s.db = db
redisDB, err := getRedis(*s.config)
if err != nil {
log.Print(err)
}
s.redisDB = redisDB
orderRepo := orderPostgresRepo.NewOrderRepository(db, redisDB, &orderPostgresRepo.Config{CacheEnable: true})
orderService := service.NewOrderService(orderRepo)
orderHandler := grpcHandlers.NewOrderHandler(orderService)
pb.RegisterOrderServiceServer(s.grpcServer, orderHandler)
internal/service/order.go
+1 -1
View File
@@ -21,7 +21,7 @@ func NewOrderService(repo repository.OrderRepository) *OrderService {
func (s *OrderService) Create(ctx context.Context, item string, quantity int32) (*domain.Order, error) {
order := &domain.Order{
ID: uuid.NewString(),
ID: uuid.New(),
Item: item,
Quantity: quantity,
}