mirror of
https://github.com/3ybactuk/marketplace-go-service-project.git
synced 2025-10-30 14:03:45 +03:00
[hw-8] add: comment service
This commit is contained in:
3ybacTuK
190
comments/internal/app/app.go
Normal file
190
comments/internal/app/app.go
Normal file
@@ -0,0 +1,190 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"route256/comments/infra/config"
|
||||
"route256/comments/infra/db/postgres"
|
||||
"route256/comments/internal/app/server"
|
||||
"route256/comments/internal/domain/service"
|
||||
"time"
|
||||
|
||||
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
"github.com/rs/zerolog"
|
||||
"github.com/rs/zerolog/log"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
"google.golang.org/grpc/reflection"
|
||||
|
||||
mw "route256/comments/infra/grpc/middleware"
|
||||
repository "route256/comments/infra/repository/sqlc"
|
||||
|
||||
pb "route256/pkg/api/comments/v1"
|
||||
)
|
||||
|
||||
type App struct {
|
||||
config *config.Config
|
||||
controller *server.Server
|
||||
|
||||
grpcServer *grpc.Server
|
||||
httpServer *http.Server
|
||||
gwConn *grpc.ClientConn
|
||||
}
|
||||
|
||||
func NewApp(configPath string) (*App, error) {
|
||||
c, err := config.LoadConfig(configPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("unable to load config: %w", err)
|
||||
}
|
||||
|
||||
log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr})
|
||||
zerolog.SetGlobalLevel(zerolog.InfoLevel)
|
||||
|
||||
if c.Service.LogLevel != "" {
|
||||
level, logErr := zerolog.ParseLevel(c.Service.LogLevel)
|
||||
if logErr != nil {
|
||||
return nil, fmt.Errorf("unknown log level `%s` provided: %w", c.Service.LogLevel, logErr)
|
||||
}
|
||||
|
||||
zerolog.SetGlobalLevel(level)
|
||||
}
|
||||
|
||||
log.WithLevel(zerolog.GlobalLevel()).Msgf("using logging level=`%s`", zerolog.GlobalLevel().String())
|
||||
|
||||
shards, err := getPostgresPools(c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
repo := repository.NewCommentsRepository(shards[0], shards[1])
|
||||
|
||||
service := service.NewCommentsService(repo, c.App.EditInterval)
|
||||
controller := server.NewServer(service)
|
||||
|
||||
app := &App{
|
||||
config: c,
|
||||
controller: controller,
|
||||
}
|
||||
|
||||
return app, nil
|
||||
}
|
||||
|
||||
func (app *App) Shutdown(ctx context.Context) (err error) {
|
||||
if app.httpServer != nil {
|
||||
err = app.httpServer.Shutdown(ctx)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("failed http gateway server shutdown")
|
||||
}
|
||||
}
|
||||
|
||||
done := make(chan struct{})
|
||||
if app.grpcServer != nil {
|
||||
go func() {
|
||||
app.grpcServer.GracefulStop()
|
||||
|
||||
close(done)
|
||||
}()
|
||||
}
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-ctx.Done():
|
||||
if app.grpcServer != nil {
|
||||
app.grpcServer.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
if app.gwConn != nil {
|
||||
err2 := app.gwConn.Close()
|
||||
|
||||
if err2 != nil {
|
||||
err = err2
|
||||
|
||||
log.Error().Err(err).Msgf("failed gateway connection close")
|
||||
}
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (app *App) ListenAndServe(ctx context.Context) error {
|
||||
grpcAddr := fmt.Sprintf("%s:%s", app.config.Service.Host, app.config.Service.GRPCPort)
|
||||
l, err := net.Listen("tcp", grpcAddr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
app.grpcServer = grpc.NewServer(
|
||||
grpc.ChainUnaryInterceptor(
|
||||
mw.Logging,
|
||||
mw.Validate,
|
||||
),
|
||||
)
|
||||
reflection.Register(app.grpcServer)
|
||||
|
||||
pb.RegisterCommentsServer(app.grpcServer, app.controller)
|
||||
|
||||
go func() {
|
||||
if err = app.grpcServer.Serve(l); err != nil {
|
||||
log.Fatal().Err(err).Msg("failed to serve")
|
||||
}
|
||||
}()
|
||||
|
||||
log.Info().Msgf("Serving grpc loms at grpc://%s", l.Addr())
|
||||
|
||||
// Setup HTTP gateway
|
||||
|
||||
conn, err := grpc.NewClient(
|
||||
grpcAddr,
|
||||
grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("grpc.NewClient: %w", err)
|
||||
}
|
||||
app.gwConn = conn
|
||||
|
||||
gwmux := runtime.NewServeMux()
|
||||
if err = pb.RegisterCommentsHandler(ctx, gwmux, conn); err != nil {
|
||||
return fmt.Errorf("pb.RegisterLOMSHandler: %w", err)
|
||||
}
|
||||
|
||||
root := http.NewServeMux()
|
||||
root.Handle("/metrics", promhttp.Handler())
|
||||
root.Handle("/", gwmux)
|
||||
|
||||
app.httpServer = &http.Server{
|
||||
Addr: fmt.Sprintf("%s:%s", app.config.Service.Host, app.config.Service.HTTPPort),
|
||||
Handler: root,
|
||||
ReadTimeout: 10 * time.Second,
|
||||
}
|
||||
|
||||
log.Info().Msgf("Serving http loms at http://%s", app.httpServer.Addr)
|
||||
|
||||
return app.httpServer.ListenAndServe()
|
||||
}
|
||||
|
||||
func getPostgresPools(c *config.Config) ([]*pgxpool.Pool, error) {
|
||||
conns := make([]string, len(c.DbShards))
|
||||
for i, shard := range c.DbShards {
|
||||
conns[i] = fmt.Sprintf(
|
||||
"postgresql://%s:%s@%s:%s/%s?sslmode=disable",
|
||||
shard.User,
|
||||
shard.Password,
|
||||
shard.Host,
|
||||
shard.Port,
|
||||
shard.DBName,
|
||||
)
|
||||
}
|
||||
|
||||
pools, err := postgres.NewPools(context.Background(), conns...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return pools, nil
|
||||
}
|
||||
133
comments/internal/app/server/server.go
Normal file
133
comments/internal/app/server/server.go
Normal file
@@ -0,0 +1,133 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"route256/comments/internal/domain/entity"
|
||||
"route256/comments/internal/domain/model"
|
||||
|
||||
pb "route256/pkg/api/comments/v1"
|
||||
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/types/known/emptypb"
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
)
|
||||
|
||||
var _ pb.CommentsServer = (*Server)(nil)
|
||||
|
||||
type CommentsService interface {
|
||||
CommentGetByID(ctx context.Context, id int64) (*entity.Comment, error)
|
||||
CommentCreate(ctx context.Context, comment *entity.Comment) (int64, error)
|
||||
CommentListBySKU(ctx context.Context, sku int64) ([]*entity.Comment, error)
|
||||
CommentListByUser(ctx context.Context, userID int64) ([]*entity.Comment, error)
|
||||
CommentEdit(ctx context.Context, comment *entity.Comment) error
|
||||
}
|
||||
|
||||
type Server struct {
|
||||
pb.UnimplementedCommentsServer
|
||||
|
||||
service CommentsService
|
||||
}
|
||||
|
||||
func NewServer(commentsService CommentsService) *Server {
|
||||
return &Server{
|
||||
service: commentsService,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) CommentAdd(ctx context.Context, req *pb.CreateCommentRequest) (*pb.CreateCommentResponse, error) {
|
||||
id, err := s.service.CommentCreate(ctx, &entity.Comment{
|
||||
UserID: req.UserId,
|
||||
SKU: req.Sku,
|
||||
Text: req.Comment,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("service.InsertComment: %w", err)
|
||||
}
|
||||
|
||||
return &pb.CreateCommentResponse{
|
||||
Id: id,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *Server) CommentEdit(ctx context.Context, req *pb.EditCommentRequest) (*emptypb.Empty, error) {
|
||||
err := s.service.CommentEdit(ctx, &entity.Comment{
|
||||
ID: req.CommentId,
|
||||
UserID: req.UserId,
|
||||
Text: req.NewComment,
|
||||
})
|
||||
switch {
|
||||
case errors.Is(err, model.ErrCommentEditUserMismatch):
|
||||
return &emptypb.Empty{}, status.Error(codes.PermissionDenied, err.Error())
|
||||
case errors.Is(err, model.ErrCommentEditTimeout):
|
||||
return &emptypb.Empty{}, status.Error(codes.FailedPrecondition, err.Error())
|
||||
case err != nil:
|
||||
return &emptypb.Empty{}, status.Error(codes.Internal, err.Error())
|
||||
}
|
||||
|
||||
return &emptypb.Empty{}, nil
|
||||
}
|
||||
|
||||
func (s *Server) CommentGetByID(ctx context.Context, req *pb.GetCommentRequest) (*pb.GetCommentResponse, error) {
|
||||
comm, err := s.service.CommentGetByID(ctx, req.Id)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("service.GetCommentByID: %w", err)
|
||||
}
|
||||
|
||||
return &pb.GetCommentResponse{
|
||||
Comment: &pb.Comment{
|
||||
Id: comm.ID,
|
||||
UserId: comm.UserID,
|
||||
Sku: comm.SKU,
|
||||
Text: comm.Text,
|
||||
CreatedAt: timestamppb.New(comm.CreatedAt),
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *Server) CommentListBySKU(ctx context.Context, req *pb.ListBySkuRequest) (*pb.ListBySkuResponse, error) {
|
||||
comms, err := s.service.CommentListBySKU(ctx, req.Sku)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("service.ListCommentsBySku: %w", err)
|
||||
}
|
||||
|
||||
comments := make([]*pb.Comment, len(comms))
|
||||
for i, comm := range comms {
|
||||
comments[i] = &pb.Comment{
|
||||
Id: comm.ID,
|
||||
UserId: comm.UserID,
|
||||
Sku: comm.SKU,
|
||||
Text: comm.Text,
|
||||
CreatedAt: timestamppb.New(comm.CreatedAt),
|
||||
}
|
||||
}
|
||||
|
||||
return &pb.ListBySkuResponse{
|
||||
Comments: comments,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *Server) CommentListByUser(ctx context.Context, req *pb.ListByUserRequest) (*pb.ListByUserResponse, error) {
|
||||
comms, err := s.service.CommentListByUser(ctx, req.UserId)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("service.ListCommentsByUser: %w", err)
|
||||
}
|
||||
|
||||
comments := make([]*pb.Comment, len(comms))
|
||||
for i, comm := range comms {
|
||||
comments[i] = &pb.Comment{
|
||||
Id: comm.ID,
|
||||
UserId: comm.UserID,
|
||||
Sku: comm.SKU,
|
||||
Text: comm.Text,
|
||||
CreatedAt: timestamppb.New(comm.CreatedAt),
|
||||
}
|
||||
}
|
||||
|
||||
return &pb.ListByUserResponse{
|
||||
Comments: comments,
|
||||
}, nil
|
||||
}
|
||||
Reference in New Issue
Block a user