Event guarantee feature (#139)

.gitignore

@@ -7,4 +7,4 @@ vendor/
 minio_data/*
 logs/logs*.json
 coverage.out*
-./protobuf/gen/es/protobuf/node_modules
+protobuf/gen/es/protobuf/node_modules

cmd/server/server.go

@@ -22,7 +22,8 @@ import (
 	"github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/auth/v1/authv1connect"
 	"github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/chat/v1/chatv1connect"
 	"github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/events/v1/eventsv1connect"
-	"github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/message/messagev1connect"
+	"github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/message/v1/messagev1connect"
+
 	"github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/search/v1/searchv1connect"
 	"github.com/kavkaco/Kavka-Core/utils/hash"
 	"github.com/rs/cors"

config/config.go

@@ -21,13 +21,9 @@ type Env int
 const (
 	Development Env = iota
 	Production
-	Test
 )
 
-var (
-	CurrentEnv Env = Development
-	filename   string
-)
+var CurrentEnv Env = Development
 
 type (
 	Config struct {
@@ -108,25 +104,24 @@ func ConfigsDirPath() string {
 }
 
 func Read() *Config {
+	var fileName string
+
 	// Load KAVKA ENV
 	env := strings.ToLower(os.Getenv("KAVKA_ENV"))
 
 	if len(strings.TrimSpace(env)) == 0 || env == "development" {
 		CurrentEnv = Development
-		filename = "config.development.yml"
+		fileName = "config.development.yml"
 	} else if env == "production" {
 		CurrentEnv = Production
-		filename = "config.production.yml"
-	} else if env == "test" {
-		CurrentEnv = Test
-		filename = "config.test.yml"
+		fileName = "config.production.yml"
 	} else {
-		panic(errors.New("Invalid env value set for variable KAVKA_ENV: " + env))
+		log.Fatalln(errors.New("Invalid env value set for variable KAVKA_ENV: " + env))
 	}
 
 	// Load YAML configs
 	k := koanf.New(ConfigsDirPath())
-	if err := k.Load(file.Provider(fmt.Sprintf("%s/%s", ConfigsDirPath(), filename)), yaml.Parser()); err != nil {
+	if err := k.Load(file.Provider(fmt.Sprintf("%s/%s", ConfigsDirPath(), fileName)), yaml.Parser()); err != nil {
 		log.Fatalf("error loading config: %v", err)
 	}
 	config := &Config{}

config/config_test.go

@@ -1,7 +1,6 @@
 package config
 
 import (
-	"errors"
 	"os"
 	"testing"
 
@@ -20,22 +19,3 @@ func TestDevelopmentConfig(t *testing.T) {
 	require.NotEmpty(t, configs)
 	require.Equal(t, CurrentEnv, Development)
 }
-
-func TestTestConfig(t *testing.T) {
-	os.Setenv("KAVKA_ENV", "test")
-
-	Read()
-	require.Equal(t, CurrentEnv, Test)
-}
-
-func TestFunctionPanics(t *testing.T) {
-	os.Setenv("KAVKA_ENV", "panic")
-
-	defer func() {
-		if r := recover(); r == nil {
-			require.Error(t, errors.New("Expected panic but did not panic"))
-		}
-	}()
-
-	Read()
-}

delivery/grpc/handlers/auth_handler.go

@@ -9,7 +9,6 @@ import (
 	"github.com/kavkaco/Kavka-Core/internal/service/auth"
 	authv1 "github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/auth/v1"
 	"github.com/kavkaco/Kavka-Core/protobuf/proto_model_transformer"
-	"google.golang.org/genproto/googleapis/rpc/code"
 	"google.golang.org/protobuf/types/known/durationpb"
 )
 
@@ -24,7 +23,7 @@ func NewAuthGrpcHandler(authService auth.AuthService) AuthGrpcServer {
 func (a AuthGrpcServer) Login(ctx context.Context, req *connect.Request[authv1.LoginRequest]) (*connect.Response[authv1.LoginResponse], error) {
 	user, accessToken, refreshToken, varror := a.authService.Login(ctx, req.Msg.Email, req.Msg.Password)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_INTERNAL))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodePermissionDenied)
 	}
 
 	res := connect.NewResponse(&authv1.LoginResponse{
@@ -39,7 +38,7 @@ func (a AuthGrpcServer) Login(ctx context.Context, req *connect.Request[authv1.L
 func (a AuthGrpcServer) Register(ctx context.Context, req *connect.Request[authv1.RegisterRequest]) (*connect.Response[authv1.RegisterResponse], error) {
 	_, varror := a.authService.Register(ctx, req.Msg.Name, req.Msg.LastName, req.Msg.Username, req.Msg.Email, req.Msg.Password, req.Msg.VerifyEmailRedirectUrl)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_INVALID_ARGUMENT))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodeInvalidArgument)
 	}
 
 	res := connect.NewResponse(&authv1.RegisterResponse{})
@@ -50,7 +49,7 @@ func (a AuthGrpcServer) Register(ctx context.Context, req *connect.Request[authv
 func (a AuthGrpcServer) Authenticate(ctx context.Context, req *connect.Request[authv1.AuthenticateRequest]) (*connect.Response[authv1.AuthenticateResponse], error) {
 	user, varror := a.authService.Authenticate(ctx, req.Msg.AccessToken)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_PERMISSION_DENIED))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodePermissionDenied)
 	}
 
 	res := connect.NewResponse(&authv1.AuthenticateResponse{
@@ -63,7 +62,7 @@ func (a AuthGrpcServer) Authenticate(ctx context.Context, req *connect.Request[a
 func (a AuthGrpcServer) ChangePassword(ctx context.Context, req *connect.Request[authv1.ChangePasswordRequest]) (*connect.Response[authv1.ChangePasswordResponse], error) {
 	varror := a.authService.ChangePassword(ctx, req.Msg.AccessToken, req.Msg.OldPassword, req.Msg.NewPassword)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.CodeUnavailable)
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodePermissionDenied)
 	}
 
 	res := connect.NewResponse(&authv1.ChangePasswordResponse{})
@@ -73,7 +72,7 @@ func (a AuthGrpcServer) ChangePassword(ctx context.Context, req *connect.Request
 func (a AuthGrpcServer) RefreshToken(ctx context.Context, req *connect.Request[authv1.RefreshTokenRequest]) (*connect.Response[authv1.RefreshTokenResponse], error) {
 	newAccessToken, varror := a.authService.RefreshToken(ctx, req.Msg.UserId, req.Msg.RefreshToken)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_UNAVAILABLE))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodePermissionDenied)
 	}
 
 	res := connect.NewResponse(&authv1.RefreshTokenResponse{
@@ -86,7 +85,7 @@ func (a AuthGrpcServer) RefreshToken(ctx context.Context, req *connect.Request[a
 func (a AuthGrpcServer) SendResetPassword(ctx context.Context, req *connect.Request[authv1.SendResetPasswordRequest]) (*connect.Response[authv1.SendResetPasswordResponse], error) {
 	resetPasswordToken, timeout, varror := a.authService.SendResetPassword(ctx, req.Msg.Email, req.Msg.ResetPasswordRedirectUrl)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_UNAVAILABLE))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodePermissionDenied)
 	}
 
 	timeoutProto := durationpb.New(timeout)
@@ -101,7 +100,7 @@ func (a AuthGrpcServer) SendResetPassword(ctx context.Context, req *connect.Requ
 func (a AuthGrpcServer) SubmitResetPassword(ctx context.Context, req *connect.Request[authv1.SubmitResetPasswordRequest]) (*connect.Response[authv1.SubmitResetPasswordResponse], error) {
 	varror := a.authService.SubmitResetPassword(ctx, req.Msg.ResetPasswordToken, req.Msg.NewPassword)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_UNAVAILABLE))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodePermissionDenied)
 	}
 
 	res := connect.NewResponse(&authv1.SubmitResetPasswordResponse{})
@@ -111,7 +110,7 @@ func (a AuthGrpcServer) SubmitResetPassword(ctx context.Context, req *connect.Re
 func (a AuthGrpcServer) VerifyEmail(ctx context.Context, req *connect.Request[authv1.VerifyEmailRequest]) (*connect.Response[authv1.VerifyEmailResponse], error) {
 	varror := a.authService.VerifyEmail(ctx, req.Msg.VerifyEmailToken)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_UNAVAILABLE))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodePermissionDenied)
 	}
 
 	res := connect.NewResponse(&authv1.VerifyEmailResponse{})

delivery/grpc/handlers/chat_handler.go

@@ -14,7 +14,6 @@ import (
 	chatv1 "github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/chat/v1"
 	"github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/chat/v1/chatv1connect"
 	"github.com/kavkaco/Kavka-Core/protobuf/proto_model_transformer"
-	"google.golang.org/genproto/googleapis/rpc/code"
 )
 
 type chatHandler struct {
@@ -34,7 +33,7 @@ func (h chatHandler) CreateChannel(ctx context.Context, req *connect.Request[cha
 
 	chat, varror := h.chatService.CreateChannel(ctx, userID, req.Msg.Title, req.Msg.Username, req.Msg.Description)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_UNAVAILABLE))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodeUnavailable)
 	}
 
 	chatProto, err := proto_model_transformer.ChatToProto(*chat)
@@ -69,7 +68,7 @@ func (h chatHandler) CreateGroup(ctx context.Context, req *connect.Request[chatv
 
 	chat, varror := h.chatService.CreateGroup(ctx, userID, req.Msg.Title, req.Msg.Username, req.Msg.Description)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_INTERNAL))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodeUnavailable)
 	}
 
 	chatProto, err := proto_model_transformer.ChatToProto(*chat)
@@ -148,7 +147,7 @@ func (h chatHandler) JoinChat(ctx context.Context, req *connect.Request[chatv1.J
 
 	joinResult, varror := h.chatService.JoinChat(ctx, chatID, userID)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_INTERNAL))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodeInternal)
 	}
 
 	if !joinResult.Joined {
@@ -157,7 +156,7 @@ func (h chatHandler) JoinChat(ctx context.Context, req *connect.Request[chatv1.J
 
 	protoChat, err := proto_model_transformer.ChatToProto(*joinResult.UpdatedChat)
 	if err != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_INTERNAL))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodeInternal)
 	}
 
 	res := &connect.Response[chatv1.JoinChatResponse]{Msg: &chatv1.JoinChatResponse{

delivery/grpc/handlers/events_handler.go

@@ -2,8 +2,10 @@ package grpc_handlers
 
 import (
 	"context"
+	"time"
 
 	"connectrpc.com/connect"
+	"github.com/kavkaco/Kavka-Core/config"
 	"github.com/kavkaco/Kavka-Core/delivery/grpc/interceptor"
 	"github.com/kavkaco/Kavka-Core/infra/stream"
 	"github.com/kavkaco/Kavka-Core/internal/model"
@@ -21,24 +23,24 @@ func NewEventsGrpcHandler(logger *log.SubLogger, streamer stream.StreamSubscribe
 	return &eventsHandler{logger, streamer}
 }
 
-func (e *eventsHandler) SubscribeEventsStream(ctx context.Context, req *connect.Request[eventsv1.SubscribeEventsStreamRequest], str *connect.ServerStream[eventsv1.SubscribeEventsStreamResponse]) error {
+func (e *eventsHandler) SubscribeEventsStream(ctx context.Context, req *connect.Request[eventsv1.SubscribeEventsStreamRequest], stream *connect.ServerStream[eventsv1.SubscribeEventsStreamResponse]) error {
 	userID := ctx.Value(interceptor.UserID{}).(model.UserID)
 
 	done := ctx.Done()
 	userCh := make(chan *eventsv1.SubscribeEventsStreamResponse)
 	e.streamer.UserSubscribe(userID, userCh)
 
-	e.logger.Debug("user stream established")
+	e.logger.Trace("user stream established")
 
 	for {
-		if str == nil {
+		if stream == nil {
 			e.logger.Error("user stream is closed")
 			return nil
 		}
 
 		select {
 		case <-done:
-			e.logger.Debug("user disconnected!")
+			e.logger.Trace("user disconnected!")
 			e.streamer.UserUnsubscribe(userID)
 			return nil
 		case event, ok := <-userCh:
@@ -47,9 +49,11 @@ func (e *eventsHandler) SubscribeEventsStream(ctx context.Context, req *connect.
 				continue
 			}
 
-			e.logger.Debug("events-handler", "event-name", event.Name)
+			if config.CurrentEnv == config.Development {
+				time.Sleep(500 * time.Millisecond)
+			}
 
-			err := str.Send(event)
+			err := stream.Send(event)
 			if err != nil {
 				log.Error("unable to send message with grpc: " + err.Error())
 				continue

delivery/grpc/handlers/message_handler.go

@@ -9,19 +9,20 @@ import (
 	"github.com/kavkaco/Kavka-Core/internal/model"
 	"github.com/kavkaco/Kavka-Core/internal/service/message"
 	"github.com/kavkaco/Kavka-Core/log"
-	messagev1 "github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/message"
-	"github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/message/messagev1connect"
+
+	messagev1 "github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/message/v1"
+	"github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/message/v1/messagev1connect"
 	"github.com/kavkaco/Kavka-Core/protobuf/proto_model_transformer"
 	"github.com/kavkaco/Kavka-Core/utils/vali"
 	"google.golang.org/genproto/googleapis/rpc/code"
 )
 
 type MessageGrpcServer struct {
 	logger         *log.SubLogger
-	messageService message.MessageService
+	messageService *message.MessageService
 }
 
-func NewMessageGrpcHandler(logger *log.SubLogger, messageService message.MessageService) messagev1connect.MessageServiceHandler {
+func NewMessageGrpcHandler(logger *log.SubLogger, messageService *message.MessageService) messagev1connect.MessageServiceHandler {
 	return MessageGrpcServer{logger, messageService}
 }
 

delivery/grpc/handlers/search_handler.go

@@ -10,7 +10,6 @@ import (
 	searchv1 "github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/search/v1"
 	"github.com/kavkaco/Kavka-Core/protobuf/gen/go/protobuf/search/v1/searchv1connect"
 	"github.com/kavkaco/Kavka-Core/protobuf/proto_model_transformer"
-	"google.golang.org/genproto/googleapis/rpc/code"
 )
 
 type searchHandler struct {
@@ -25,7 +24,7 @@ func NewSearchGrpcHandler(logger *log.SubLogger, searchService search.SearchServ
 func (s *searchHandler) Search(ctx context.Context, req *connect.Request[searchv1.SearchRequest]) (*connect.Response[searchv1.SearchResponse], error) {
 	result, varror := s.searchService.Search(ctx, req.Msg.Input)
 	if varror != nil {
-		return nil, grpc_helpers.GrpcVarror(varror, connect.Code(code.Code_INTERNAL))
+		return nil, grpc_helpers.GrpcVarror(varror, connect.CodeUnavailable)
 	}
 
 	chats, err := proto_model_transformer.ChatsToProto(result.Chats)

delivery/grpc/helpers/errdetail.go → delivery/grpc/helpers/error_detail.go

@@ -23,7 +23,10 @@ func VarrorAsGrpcErrDetails(varror *vali.Varror) (*connect.ErrorDetail, error) {
 
 func GrpcVarror(varror *vali.Varror, code connect.Code) error {
 	connectErr := connect.NewError(code, varror.Error)
-	varrorDetail, _ := VarrorAsGrpcErrDetails(varror)
+	varrorDetail, err := VarrorAsGrpcErrDetails(varror)
+	if err != nil {
+		return err
+	}
 
 	if len(varror.ValidationErrors) > 0 {
 		connectErr.AddDetail(varrorDetail)

docker-compose.yml

@@ -51,6 +51,7 @@ services:
     image: nats
     ports:
       - "4222:4222"
+    restart: always
 
 networks:
   default_network: