refactor(): removed redundant files and small improvements
This commit is contained in:
@@ -1,22 +1,24 @@
|
||||
# Go parameters
|
||||
GOCMD=go
|
||||
GOBUILD=$(GOCMD) build -trimpath -ldflags="-s -w"
|
||||
GOTEST=$(GOCMD) test
|
||||
GODOWNLOAD=$(GOCMD) mod download
|
||||
BINARY_NAME=datarush
|
||||
MIGRATE_BINARY_NAME=datarush-migrate
|
||||
|
||||
BASE_BINARY_NAME=datarush
|
||||
GW_BINARY_NAME=$(BASE_BINARY_NAME)-gw
|
||||
MIGRATE_BINARY_NAME=$(BASE_BINARY_NAME)-migrate
|
||||
AUTH_BINARY_NAME=$(BASE_BINARY_NAME)-auth
|
||||
|
||||
BINARY_DIR=bin
|
||||
|
||||
# Protobuf parameters
|
||||
PROTOC=protoc
|
||||
PROTO_DIR=api/proto
|
||||
PROTO_FILE=$(PROTO_DIR)/auth.proto $(PROTO_DIR)/competition.proto $(PROTO_DIR)/results.proto $(PROTO_DIR)/review.proto $(PROTO_DIR)/submission.proto $(PROTO_DIR)/task.proto $(PROTO_DIR)/user.proto
|
||||
PROTO_FILE=${PROTO_DIR}/achievements.proto $(PROTO_DIR)/auth.proto $(PROTO_DIR)/competition.proto $(PROTO_DIR)/results.proto $(PROTO_DIR)/review.proto $(PROTO_DIR)/submission.proto $(PROTO_DIR)/task.proto $(PROTO_DIR)/user.proto
|
||||
PROTO_OUT=.
|
||||
|
||||
.PHONY: install i generate gen generate-gw test build run migrate lint fmt format clean help codegen examples
|
||||
|
||||
install:
|
||||
$(GODOWNLOAD)
|
||||
$(GOCMD) mod download
|
||||
$(GOCMD) mod tidy
|
||||
|
||||
i: install
|
||||
|
||||
@@ -35,16 +37,23 @@ generate-gw:
|
||||
test:
|
||||
$(GOTEST) ./...
|
||||
|
||||
build:
|
||||
$(GOBUILD) -o ./$(BINARY_DIR)/$(BINARY_NAME) ./cmd/server
|
||||
chmod +x ./$(BINARY_DIR)/$(BINARY_NAME)
|
||||
build-gw:
|
||||
$(GOBUILD) -o ./$(BINARY_DIR)/$(GW_BINARY_NAME) ./cmd/gw
|
||||
chmod +x ./$(BINARY_DIR)/$(GW_BINARY_NAME)
|
||||
|
||||
build-migrate:
|
||||
$(GOBUILD) -o ./$(BINARY_DIR)/$(MIGRATE_BINARY_NAME) ./cmd/migrate
|
||||
chmod +x ./$(BINARY_DIR)/$(MIGRATE_BINARY_NAME)
|
||||
|
||||
run: build
|
||||
./$(BINARY_DIR)/$(BINARY_NAME)
|
||||
build-auth:
|
||||
$(GOBUILD) -o ./$(BINARY_DIR)/$(AUTH_BINARY_NAME) ./cmd/auth
|
||||
chmod +x ./$(BINARY_DIR)/$(AUTH_BINARY_NAME)
|
||||
|
||||
build: build-gw build-migrate build-auth
|
||||
|
||||
run:
|
||||
./$(BINARY_DIR)/$(AUTH_BINARY_NAME) &
|
||||
./$(BINARY_DIR)/$(GW_BINARY_NAME)
|
||||
|
||||
migrate: build-migrate
|
||||
@cmd=$(word 2,$(MAKECMDGOALS)); \
|
||||
@@ -65,28 +74,15 @@ format: fmt
|
||||
clean:
|
||||
rm -rf bin/*
|
||||
|
||||
codegen:
|
||||
@if [ -z "$(SERVICE)" ]; then \
|
||||
echo "Usage: make codegen SERVICE=<service-name>"; \
|
||||
echo "Example: make codegen SERVICE=auth"; \
|
||||
exit 1; \
|
||||
fi
|
||||
$(GOBUILD) -o ./$(BINARY_DIR)/codegen ./cmd/codegen
|
||||
./$(BINARY_DIR)/codegen $(SERVICE) ./
|
||||
|
||||
|
||||
help:
|
||||
@echo "Available commands:"
|
||||
@echo "Help:"
|
||||
@echo " install - Install all deps using go mod download"
|
||||
@echo " i"
|
||||
@echo " generate - Generate gRPC code"
|
||||
@echo " gen"
|
||||
@echo " protoc"
|
||||
@echo " codegen - Generate service template"
|
||||
@echo " Usage: make codegen SERVICE=<name>"
|
||||
@echo " examples - Show integration examples"
|
||||
@echo " test - Run tests"
|
||||
@echo " build - Build the binary"
|
||||
@echo " build - Build all binaries"
|
||||
@echo " run - Run the application"
|
||||
@echo " lint - Run golangci-lint linter"
|
||||
@echo " format - Run golangci-lint formatter"
|
||||
|
||||
+1
-1
@@ -7,7 +7,7 @@ import (
|
||||
"flag"
|
||||
"log"
|
||||
|
||||
"datarush/internal/lms/config"
|
||||
"datarush/internal/migrate/config"
|
||||
|
||||
"github.com/golang-migrate/migrate/v4"
|
||||
"github.com/golang-migrate/migrate/v4/database/postgres"
|
||||
|
||||
@@ -1,6 +0,0 @@
|
||||
package main
|
||||
|
||||
|
||||
func main() {
|
||||
|
||||
}
|
||||
@@ -1,11 +0,0 @@
|
||||
GRPC_ENABLE_REFLECTION=true
|
||||
HTTP_HANDLER_ENABLE=true
|
||||
|
||||
POSTGRES_HOST=postgres
|
||||
POSTGRES_PORT=5432
|
||||
POSTGRES_USERNAME=postgres
|
||||
POSTGRES_PASSWORD=postgres
|
||||
POSTGRES_DATABASE=postgres
|
||||
|
||||
REDIS_URI=redis://redis:6379
|
||||
AUTH_GRPC_ADDR=auth:50052
|
||||
@@ -0,0 +1,19 @@
|
||||
SERVER_PORT=8080
|
||||
SERVER_HOST=0.0.0.0
|
||||
|
||||
AUTH_SERVICE_ADDR=auth:50052
|
||||
USER_SERVICE_ADDR=user:50052
|
||||
COMPETITION_SERVICE_ADDR=competition:50052
|
||||
TASK_SERVICE_ADDR=task:50052
|
||||
SUBMISSION_SERVICE_ADDR=submission:50052
|
||||
RESULTS_SERVICE_ADDR=results:50052
|
||||
REVIEW_SERVICE_ADDR=review:50052
|
||||
ACHIEVEMENTS_SERVICE_ADDR=achievements:50052
|
||||
|
||||
AWS_ACCESS_KEY_ID=your_access_key_here
|
||||
AWS_SECRET_ACCESS_KEY=your_secret_key_here
|
||||
AWS_REGION=us-east-1
|
||||
S3_BUCKET=datarush-submissions
|
||||
S3_ENDPOINT=
|
||||
|
||||
JWT_SECRET=your_jwt_secret_here
|
||||
@@ -1,9 +0,0 @@
|
||||
package domain
|
||||
|
||||
import (
|
||||
"errors"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrInvalidID = errors.New("invalid uuid")
|
||||
)
|
||||
@@ -1,42 +0,0 @@
|
||||
package domain
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/go-playground/validator/v10"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrOrderAlreadyExist = errors.New("order already exist")
|
||||
ErrOrderNotFound = errors.New("order not found")
|
||||
ErrInvalidOrderData = errors.New("invalid order data")
|
||||
)
|
||||
|
||||
type Order struct {
|
||||
ID uuid.UUID `db:"id" json:"id" validate:"required"`
|
||||
Item string `db:"item" json:"item" validate:"required"`
|
||||
Quantity int32 `db:"quantity" json:"quantity" validate:"required,gt=0"`
|
||||
}
|
||||
|
||||
func NewOrder(id uuid.UUID, item string, quantity int32) (*Order, error) {
|
||||
order := &Order{
|
||||
ID: id,
|
||||
Item: item,
|
||||
Quantity: quantity,
|
||||
}
|
||||
|
||||
err := order.Validate()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return order, nil
|
||||
}
|
||||
|
||||
func (o *Order) Validate() error {
|
||||
validate := validator.New()
|
||||
|
||||
return fmt.Errorf("%w: %w", ErrInvalidOrderData, validate.Struct(o))
|
||||
}
|
||||
@@ -1,51 +0,0 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
authPb "datarush/pkg/api/auth"
|
||||
|
||||
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
)
|
||||
|
||||
const (
|
||||
httpReadTimeout = 10 * time.Second
|
||||
httpWriteTimeout = 10 * time.Second
|
||||
httpIdleTimeout = 60 * time.Second
|
||||
)
|
||||
|
||||
func StartGateway(grpcPort, httpPort int, authGrpcAddr string) error {
|
||||
ctx := context.Background()
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
gwmux := runtime.NewServeMux()
|
||||
opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
|
||||
|
||||
// Register auth service
|
||||
if err := authPb.RegisterAuthServiceHandlerFromEndpoint(ctx, gwmux, authGrpcAddr, opts); err != nil {
|
||||
return fmt.Errorf("failed to register auth service: %w", err)
|
||||
}
|
||||
|
||||
srv := &http.Server{
|
||||
Addr: fmt.Sprintf(":%d", httpPort),
|
||||
Handler: gwmux,
|
||||
ReadTimeout: httpReadTimeout,
|
||||
WriteTimeout: httpWriteTimeout,
|
||||
IdleTimeout: httpIdleTimeout,
|
||||
}
|
||||
|
||||
log.Printf("starting gRPC-Gateway on port %d", httpPort)
|
||||
return srv.ListenAndServe()
|
||||
}
|
||||
|
||||
func GetGRPCListener(port int) (net.Listener, error) {
|
||||
return net.Listen("tcp", fmt.Sprintf(":%d", port))
|
||||
}
|
||||
@@ -1,26 +0,0 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"log"
|
||||
|
||||
"datarush/internal/lms/domain"
|
||||
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
func mapError(err error) error {
|
||||
if errors.Is(err, domain.ErrOrderNotFound) {
|
||||
return status.Error(codes.NotFound, err.Error())
|
||||
}
|
||||
if errors.Is(err, domain.ErrOrderAlreadyExist) {
|
||||
return status.Error(codes.AlreadyExists, err.Error())
|
||||
}
|
||||
if errors.Is(err, domain.ErrInvalidOrderData) || errors.Is(err, domain.ErrInvalidID) {
|
||||
return status.Error(codes.InvalidArgument, err.Error())
|
||||
}
|
||||
|
||||
log.Printf("internal server error: %v", err)
|
||||
return status.Error(codes.Internal, "internal server error")
|
||||
}
|
||||
@@ -1,63 +0,0 @@
|
||||
package http
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/jmoiron/sqlx"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
type HealthHandler struct {
|
||||
DB *sqlx.DB
|
||||
Redis *redis.Client
|
||||
}
|
||||
|
||||
func NewHealthHandler(db *sqlx.DB, redisDB *redis.Client) *HealthHandler {
|
||||
return &HealthHandler{
|
||||
DB: db,
|
||||
Redis: redisDB,
|
||||
}
|
||||
}
|
||||
|
||||
type HealthResponse struct {
|
||||
Status string `json:"status"`
|
||||
Details map[string]string `json:"details"`
|
||||
}
|
||||
|
||||
func (h *HealthHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
|
||||
defer cancel()
|
||||
|
||||
details := map[string]string{}
|
||||
|
||||
if err := h.DB.PingContext(ctx); err != nil {
|
||||
details["postgres"] = "unhealthy: " + err.Error()
|
||||
} else {
|
||||
details["postgres"] = "ok"
|
||||
}
|
||||
|
||||
if err := h.Redis.Ping(ctx).Err(); err != nil {
|
||||
details["redis"] = "unhealthy: " + err.Error()
|
||||
} else {
|
||||
details["redis"] = "ok"
|
||||
}
|
||||
|
||||
status := "ok"
|
||||
for _, v := range details {
|
||||
if v != "ok" {
|
||||
status = "unhealthy"
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
resp := HealthResponse{Status: status, Details: details}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if status != "ok" {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(resp)
|
||||
}
|
||||
@@ -1,102 +0,0 @@
|
||||
package inmemory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"datarush/internal/lms/domain"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type OrderRepository struct {
|
||||
mu sync.RWMutex
|
||||
orders map[string]*domain.Order
|
||||
}
|
||||
|
||||
func NewOrderRepository() *OrderRepository {
|
||||
return &OrderRepository{
|
||||
orders: make(map[string]*domain.Order),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *OrderRepository) Create(ctx context.Context, order *domain.Order) error {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
|
||||
if _, ok := r.orders[order.ID.String()]; ok {
|
||||
return domain.ErrOrderAlreadyExist
|
||||
}
|
||||
r.orders[order.ID.String()] = order
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *OrderRepository) Get(ctx context.Context, id uuid.UUID) (*domain.Order, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
|
||||
order, ok := r.orders[id.String()]
|
||||
if !ok {
|
||||
return nil, domain.ErrOrderNotFound
|
||||
}
|
||||
|
||||
return order, nil
|
||||
}
|
||||
|
||||
func (r *OrderRepository) Update(ctx context.Context, order *domain.Order) error {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
|
||||
if _, ok := r.orders[order.ID.String()]; !ok {
|
||||
return domain.ErrOrderNotFound
|
||||
}
|
||||
r.orders[order.ID.String()] = order
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *OrderRepository) Delete(ctx context.Context, id uuid.UUID) error {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
|
||||
if _, ok := r.orders[id.String()]; !ok {
|
||||
return domain.ErrOrderNotFound
|
||||
}
|
||||
|
||||
delete(r.orders, id.String())
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *OrderRepository) List(ctx context.Context) ([]*domain.Order, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
|
||||
orders := make([]*domain.Order, 0, len(r.orders))
|
||||
for _, order := range r.orders {
|
||||
orders = append(orders, order)
|
||||
}
|
||||
|
||||
return orders, nil
|
||||
}
|
||||
@@ -1,17 +0,0 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"datarush/internal/lms/domain"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type OrderRepository interface {
|
||||
Create(ctx context.Context, order *domain.Order) error
|
||||
Get(ctx context.Context, id uuid.UUID) (*domain.Order, error)
|
||||
Update(ctx context.Context, order *domain.Order) error
|
||||
Delete(ctx context.Context, id uuid.UUID) error
|
||||
List(ctx context.Context) ([]*domain.Order, error)
|
||||
}
|
||||
@@ -1,243 +0,0 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"datarush/internal/lms/domain"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jmoiron/sqlx"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
const (
|
||||
orderCachePrefix = "order:"
|
||||
cacheTTL = 5 * time.Minute
|
||||
)
|
||||
|
||||
type OrderRepository struct {
|
||||
db *sqlx.DB
|
||||
redisClient *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,
|
||||
redisClient: 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 uuid.UUID) (*domain.Order, error) {
|
||||
if r.cacheEnable {
|
||||
if order, err := r.getFromCache(ctx, id.String()); 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 uuid.UUID) 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.String())
|
||||
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.redisClient.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.redisClient.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())
|
||||
|
||||
err = r.redisClient.Set(ctx, key, data, cacheTTL).Err()
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *OrderRepository) invalidateCache(_ context.Context, id string) {
|
||||
if !r.cacheEnable {
|
||||
return
|
||||
}
|
||||
|
||||
go func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), r.redisClient.Options().ReadTimeout)
|
||||
defer cancel()
|
||||
|
||||
if err := r.redisClient.Del(ctx, r.cacheKey(id)).Err(); err != nil {
|
||||
log.Printf("warn: cache invalidation failed for order %s: %v", id, err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -1,182 +0,0 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"datarush/internal/lms/config"
|
||||
"datarush/internal/lms/interceptor"
|
||||
|
||||
httpHandlers "datarush/internal/lms/handler/http"
|
||||
|
||||
authPb "datarush/pkg/api/auth"
|
||||
|
||||
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
|
||||
"github.com/jmoiron/sqlx"
|
||||
_ "github.com/lib/pq" // postgres driver
|
||||
"github.com/redis/go-redis/v9"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
"google.golang.org/grpc/reflection"
|
||||
)
|
||||
|
||||
const (
|
||||
httpReadTimeout = 10 * time.Second
|
||||
httpWriteTimeout = 10 * time.Second
|
||||
httpIdleTimeout = 60 * time.Second
|
||||
redisRetryCount = 2
|
||||
redisMinRetryBackoff = 50 * time.Millisecond
|
||||
redisMaxRetryBackoff = 200 * time.Millisecond
|
||||
redisDialTimeout = 1 * time.Second
|
||||
redisDialerRetries = 3
|
||||
redisTimeout = 2 * time.Second
|
||||
)
|
||||
|
||||
type Server struct {
|
||||
grpcServer *grpc.Server
|
||||
config *config.Config
|
||||
db *sqlx.DB
|
||||
redisDB *redis.Client
|
||||
}
|
||||
|
||||
func New(cfg *config.Config) *Server {
|
||||
loggerInterceptor := interceptor.NewLoggerInterceptor()
|
||||
|
||||
grpcServer := grpc.NewServer(
|
||||
grpc.UnaryInterceptor(loggerInterceptor.Unary()),
|
||||
grpc.StreamInterceptor(loggerInterceptor.Stream()),
|
||||
)
|
||||
|
||||
return &Server{
|
||||
grpcServer: grpcServer,
|
||||
config: cfg,
|
||||
}
|
||||
}
|
||||
|
||||
func runHTTPHandler(s *Server, grpcServerEndpoint *string) error {
|
||||
ctx := context.Background()
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
gwmux := runtime.NewServeMux()
|
||||
opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
|
||||
|
||||
// Register auth service
|
||||
if err := registerAuthService(ctx, gwmux, s.config.AuthGRPCAddr, opts); err != nil {
|
||||
log.Printf("failed to register auth service: %v", err)
|
||||
}
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.Handle("/healthz", httpHandlers.NewHealthHandler(s.db, s.redisDB))
|
||||
mux.Handle("/", gwmux)
|
||||
|
||||
srv := &http.Server{
|
||||
Addr: fmt.Sprintf(":%d", s.config.HTTPPort),
|
||||
Handler: mux,
|
||||
ReadTimeout: httpReadTimeout,
|
||||
WriteTimeout: httpWriteTimeout,
|
||||
IdleTimeout: httpIdleTimeout,
|
||||
}
|
||||
|
||||
return srv.ListenAndServe()
|
||||
}
|
||||
|
||||
func registerAuthService(ctx context.Context, gwmux *runtime.ServeMux, authAddr string, opts []grpc.DialOption) error {
|
||||
if err := authPb.RegisterAuthServiceHandlerFromEndpoint(ctx, gwmux, authAddr, opts); err != nil {
|
||||
return fmt.Errorf("register auth service handler: %w", err)
|
||||
}
|
||||
log.Printf("registered auth service from %s", authAddr)
|
||||
return nil
|
||||
}
|
||||
|
||||
func registerAuthHandlerFromEndpoint(ctx context.Context, gwmux *runtime.ServeMux, endpoint string, opts []grpc.DialOption) error {
|
||||
// We'll import the proto and register it here
|
||||
// For now, this is a placeholder that will be called from gateway integration
|
||||
return nil
|
||||
}
|
||||
func getDatabase(cfg config.Config) (*sqlx.DB, error) {
|
||||
db, err := sqlx.Connect("postgres", cfg.BuildPostgresConnStr())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("connect to database: %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,
|
||||
MaxRetries: redisRetryCount,
|
||||
MinRetryBackoff: redisMinRetryBackoff,
|
||||
MaxRetryBackoff: redisMaxRetryBackoff,
|
||||
DialTimeout: redisDialTimeout,
|
||||
DialerRetries: redisDialerRetries,
|
||||
DialerRetryTimeout: redisDialTimeout,
|
||||
ReadTimeout: redisTimeout,
|
||||
WriteTimeout: redisTimeout,
|
||||
})
|
||||
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() {
|
||||
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
|
||||
|
||||
if s.config.GRPCEnableReflection {
|
||||
reflection.Register(s.grpcServer)
|
||||
log.Println("gRPC server will start with reflection")
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) Start() error {
|
||||
addr := fmt.Sprintf(":%d", s.config.GRPCPort)
|
||||
lis, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to listen: %w", err)
|
||||
}
|
||||
|
||||
if s.config.EnableHTTPHandler {
|
||||
go func() {
|
||||
log.Printf("starting HTTP gateway on port %d", s.config.HTTPPort)
|
||||
if err := runHTTPHandler(s, &addr); err != nil {
|
||||
log.Printf("HTTP gateway failed: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
log.Printf("starting gRPC server on port %d", s.config.GRPCPort)
|
||||
|
||||
if err := s.grpcServer.Serve(lis); err != nil {
|
||||
return fmt.Errorf("failed to serve: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) Stop() {
|
||||
s.grpcServer.GracefulStop()
|
||||
log.Println("gRPC server stopped gracefully")
|
||||
}
|
||||
@@ -1,56 +0,0 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"datarush/internal/lms/domain"
|
||||
"datarush/internal/lms/repository"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type OrderService struct {
|
||||
repo repository.OrderRepository
|
||||
}
|
||||
|
||||
func NewOrderService(repo repository.OrderRepository) *OrderService {
|
||||
return &OrderService{
|
||||
repo: repo,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *OrderService) Create(ctx context.Context, item string, quantity int32) (*domain.Order, error) {
|
||||
order, err := domain.NewOrder(uuid.New(), item, quantity)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := s.repo.Create(ctx, order); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return order, nil
|
||||
}
|
||||
|
||||
func (s *OrderService) Get(ctx context.Context, id uuid.UUID) (*domain.Order, error) {
|
||||
return s.repo.Get(ctx, id)
|
||||
}
|
||||
|
||||
func (s *OrderService) Update(ctx context.Context, id uuid.UUID, item string, quantity int32) (*domain.Order, error) {
|
||||
order, err := domain.NewOrder(id, item, quantity)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := s.repo.Update(ctx, order); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return order, nil
|
||||
}
|
||||
|
||||
func (s *OrderService) Delete(ctx context.Context, id uuid.UUID) error {
|
||||
return s.repo.Delete(ctx, id)
|
||||
}
|
||||
|
||||
func (s *OrderService) List(ctx context.Context) ([]*domain.Order, error) {
|
||||
return s.repo.List(ctx)
|
||||
}
|
||||
Reference in New Issue
Block a user