feat: afro sms itegrated
This commit is contained in:
dawitel
parent
b690a1c933
commit
9ffcac096f
|
|
@ -59,7 +59,7 @@ func main() {
|
|||
betSvc := bet.NewService(store)
|
||||
|
||||
notificationRepo := repository.NewNotificationRepository(store)
|
||||
notificationSvc := notificationservice.New(notificationRepo, logger)
|
||||
notificationSvc := notificationservice.New(notificationRepo, logger, cfg)
|
||||
|
||||
app := httpserver.NewApp(cfg.Port, v, authSvc, logger, jwtutil.JwtConfig{
|
||||
JwtAccessKey: cfg.JwtKey,
|
||||
|
|
|
|||
5
go.mod
5
go.mod
|
|
@ -6,8 +6,8 @@ require (
|
|||
github.com/bytedance/sonic v1.13.2
|
||||
github.com/go-playground/validator/v10 v10.26.0
|
||||
github.com/gofiber/fiber/v2 v2.52.6
|
||||
github.com/golang-jwt/jwt/v5 v5.2.2
|
||||
github.com/gofiber/websocket/v2 v2.2.1
|
||||
github.com/golang-jwt/jwt/v5 v5.2.2
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/jackc/pgx/v5 v5.7.4
|
||||
github.com/joho/godotenv v1.5.1
|
||||
|
|
@ -18,9 +18,11 @@ require (
|
|||
|
||||
require (
|
||||
github.com/KyleBanks/depth v1.2.1 // indirect
|
||||
github.com/amanuelabay/afrosms-go v1.0.6 // indirect
|
||||
github.com/andybalholm/brotli v1.1.1 // indirect
|
||||
github.com/bytedance/sonic/loader v0.2.4 // indirect
|
||||
github.com/cloudwego/base64x v0.1.5 // indirect
|
||||
github.com/fasthttp/websocket v1.5.3 // indirect
|
||||
github.com/gabriel-vasile/mimetype v1.4.8 // indirect
|
||||
github.com/go-openapi/jsonpointer v0.21.1 // indirect
|
||||
github.com/go-openapi/jsonreference v0.21.0 // indirect
|
||||
|
|
@ -28,7 +30,6 @@ require (
|
|||
github.com/go-openapi/swag v0.23.1 // indirect
|
||||
github.com/go-playground/locales v0.14.1 // indirect
|
||||
github.com/go-playground/universal-translator v0.18.1 // indirect
|
||||
github.com/fasthttp/websocket v1.5.3 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
|
||||
github.com/jackc/puddle/v2 v2.2.2 // indirect
|
||||
|
|
|
|||
14
go.sum
14
go.sum
|
|
@ -4,6 +4,8 @@ github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6Xge
|
|||
github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
|
||||
github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
|
||||
github.com/agiledragon/gomonkey/v2 v2.3.1/go.mod h1:ap1AmDzcVOAz1YpeJ3TCzIgstoaWLA6jbbgxfB4w2iY=
|
||||
github.com/amanuelabay/afrosms-go v1.0.6 h1:/B9upckMqzr5/de7dbXPqIfmJm4utbQq0QUQePxmXtk=
|
||||
github.com/amanuelabay/afrosms-go v1.0.6/go.mod h1:5mzzZtWSCDdvQsA0OyYf5CtbdGpl9lXyrcpl8/DyBj0=
|
||||
github.com/andybalholm/brotli v1.0.4/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
|
||||
github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
|
||||
github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
|
||||
|
|
@ -20,6 +22,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
|
|||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/fasthttp/websocket v1.5.3 h1:TPpQuLwJYfd4LJPXvHDYPMFWbLjsT91n3GpWtCQtdek=
|
||||
github.com/fasthttp/websocket v1.5.3/go.mod h1:46gg/UBmTU1kUaTcwQXpUxtRwG2PvIZYeA8oL6vF3Fs=
|
||||
github.com/gabriel-vasile/mimetype v1.4.8 h1:FfZ3gj38NjllZIeJAmMhr+qKL8Wu+nOoI3GqacKw1NM=
|
||||
github.com/gabriel-vasile/mimetype v1.4.8/go.mod h1:ByKUIKGjh1ODkGM1asKUbQZOLGrPjydw3hYPU2YU9t8=
|
||||
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
|
||||
|
|
@ -46,14 +50,12 @@ github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91
|
|||
github.com/go-playground/validator/v10 v10.26.0 h1:SP05Nqhjcvz81uJaRfEV0YBSSSGMc/iMaVtFbr3Sw2k=
|
||||
github.com/go-playground/validator/v10 v10.26.0/go.mod h1:I5QpIEbmr8On7W0TktmJAumgzX4CA1XNl4ZmDuVHKKo=
|
||||
github.com/gofiber/fiber/v2 v2.32.0/go.mod h1:CMy5ZLiXkn6qwthrl03YMyW1NLfj0rhxz2LKl4t7ZTY=
|
||||
github.com/fasthttp/websocket v1.5.3 h1:TPpQuLwJYfd4LJPXvHDYPMFWbLjsT91n3GpWtCQtdek=
|
||||
github.com/fasthttp/websocket v1.5.3/go.mod h1:46gg/UBmTU1kUaTcwQXpUxtRwG2PvIZYeA8oL6vF3Fs=
|
||||
github.com/gofiber/fiber/v2 v2.52.6 h1:Rfp+ILPiYSvvVuIPvxrBns+HJp8qGLDnLJawAu27XVI=
|
||||
github.com/gofiber/fiber/v2 v2.52.6/go.mod h1:YEcBbO/FB+5M1IZNBP9FO3J9281zgPAreiI1oqg8nDw=
|
||||
github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8=
|
||||
github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk=
|
||||
github.com/gofiber/websocket/v2 v2.2.1 h1:C9cjxvloojayOp9AovmpQrk8VqvVnT8Oao3+IUygH7w=
|
||||
github.com/gofiber/websocket/v2 v2.2.1/go.mod h1:Ao/+nyNnX5u/hIFPuHl28a+NIkrqK7PRimyKaj4JxVU=
|
||||
github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8=
|
||||
github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk=
|
||||
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
|
||||
|
|
@ -110,11 +112,11 @@ github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUc
|
|||
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
|
||||
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
|
||||
github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
|
||||
github.com/savsgio/gotils v0.0.0-20230208104028-c358bd845dee h1:8Iv5m6xEo1NR1AvpV+7XmhI4r39LGNzwUL4YpMuL5vk=
|
||||
github.com/savsgio/gotils v0.0.0-20230208104028-c358bd845dee/go.mod h1:qwtSXrKuJh/zsFQ12yEE89xfCrGKK63Rr7ctU/uCo4g=
|
||||
github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
|
||||
github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
|
||||
github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
|
||||
github.com/savsgio/gotils v0.0.0-20230208104028-c358bd845dee h1:8Iv5m6xEo1NR1AvpV+7XmhI4r39LGNzwUL4YpMuL5vk=
|
||||
github.com/savsgio/gotils v0.0.0-20230208104028-c358bd845dee/go.mod h1:qwtSXrKuJh/zsFQ12yEE89xfCrGKK63Rr7ctU/uCo4g=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
|
||||
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ var (
|
|||
ErrLogLevel = errors.New("log level not set")
|
||||
ErrInvalidLevel = errors.New("invalid log level")
|
||||
ErrInvalidEnv = errors.New("env not set or invalid")
|
||||
ErrInvalidSMSAPIKey = errors.New("SMS API key is invalid")
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
|
|
@ -29,6 +30,10 @@ type Config struct {
|
|||
JwtKey string
|
||||
LogLevel slog.Level
|
||||
Env string
|
||||
AFRO_SMS_API_KEY string
|
||||
AFRO_SMS_SENDER_NAME string
|
||||
AFRO_SMS_RECEIVER_PHONE_NUMBER string
|
||||
ADRO_SMS_HOST_URL string
|
||||
}
|
||||
|
||||
func NewConfig() (*Config, error) {
|
||||
|
|
@ -38,20 +43,25 @@ func NewConfig() (*Config, error) {
|
|||
}
|
||||
return config, nil
|
||||
}
|
||||
|
||||
func (c *Config) loadEnv() error {
|
||||
err := godotenv.Load()
|
||||
if err != nil {
|
||||
return errors.New("failed to load env file")
|
||||
if err != nil && !os.IsNotExist(err) {
|
||||
return errors.New("failed to load env file: " + err.Error())
|
||||
}
|
||||
// env
|
||||
|
||||
env := os.Getenv("ENV")
|
||||
if env == "" {
|
||||
return ErrInvalidEnv
|
||||
}
|
||||
c.Env = env
|
||||
|
||||
portStr := os.Getenv("PORT")
|
||||
if portStr == "" {
|
||||
return ErrInvalidPort
|
||||
}
|
||||
port, err := strconv.Atoi(portStr)
|
||||
if err != nil {
|
||||
if err != nil || port < 1 || port > 65535 {
|
||||
return ErrInvalidPort
|
||||
}
|
||||
c.Port = port
|
||||
|
|
@ -61,24 +71,33 @@ func (c *Config) loadEnv() error {
|
|||
return ErrInvalidDbUrl
|
||||
}
|
||||
c.DbUrl = dbUrl
|
||||
|
||||
refreshExpiryStr := os.Getenv("REFRESH_EXPIRY")
|
||||
if refreshExpiryStr == "" {
|
||||
return ErrRefreshExpiry
|
||||
}
|
||||
refreshExpiry, err := strconv.Atoi(refreshExpiryStr)
|
||||
if err != nil {
|
||||
if err != nil || refreshExpiry <= 0 {
|
||||
return ErrRefreshExpiry
|
||||
}
|
||||
c.RefreshExpiry = refreshExpiry
|
||||
|
||||
jwtKey := os.Getenv("JWT_KEY")
|
||||
if jwtKey == "" {
|
||||
return ErrInvalidJwtKey
|
||||
}
|
||||
c.JwtKey = jwtKey
|
||||
|
||||
accessExpiryStr := os.Getenv("ACCESS_EXPIRY")
|
||||
if accessExpiryStr == "" {
|
||||
return ErrAccessExpiry
|
||||
}
|
||||
accessExpiry, err := strconv.Atoi(accessExpiryStr)
|
||||
if err != nil {
|
||||
if err != nil || accessExpiry <= 0 {
|
||||
return ErrAccessExpiry
|
||||
}
|
||||
c.AccessExpiry = accessExpiry
|
||||
// log level
|
||||
|
||||
logLevel := os.Getenv("LOG_LEVEL")
|
||||
if logLevel == "" {
|
||||
return ErrLogLevel
|
||||
|
|
@ -89,5 +108,23 @@ func (c *Config) loadEnv() error {
|
|||
return ErrInvalidLevel
|
||||
}
|
||||
c.LogLevel = lvl
|
||||
|
||||
c.AFRO_SMS_API_KEY = os.Getenv("AFRO_SMS_API_KEY")
|
||||
if c.AFRO_SMS_API_KEY == "" {
|
||||
return ErrInvalidSMSAPIKey
|
||||
}
|
||||
|
||||
c.AFRO_SMS_SENDER_NAME = os.Getenv("AFRO_SMS_SENDER_NAME")
|
||||
if c.AFRO_SMS_SENDER_NAME == "" {
|
||||
c.AFRO_SMS_SENDER_NAME = "FortuneBet"
|
||||
}
|
||||
|
||||
c.AFRO_SMS_RECEIVER_PHONE_NUMBER = os.Getenv("AFRO_SMS_RECEIVER_PHONE_NUMBER")
|
||||
|
||||
c.ADRO_SMS_HOST_URL = os.Getenv("ADRO_SMS_HOST_URL")
|
||||
if c.ADRO_SMS_HOST_URL == "" {
|
||||
c.ADRO_SMS_HOST_URL = "https://api.afrosms.com"
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,13 +2,16 @@ package notificationservice
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/SamuelTariku/FortuneBet-Backend/internal/config"
|
||||
"github.com/SamuelTariku/FortuneBet-Backend/internal/domain"
|
||||
"github.com/SamuelTariku/FortuneBet-Backend/internal/pkgs/helpers"
|
||||
"github.com/SamuelTariku/FortuneBet-Backend/internal/repository"
|
||||
afro "github.com/amanuelabay/afrosms-go"
|
||||
"github.com/gofiber/websocket/v2"
|
||||
)
|
||||
|
||||
|
|
@ -18,15 +21,17 @@ type Service struct {
|
|||
connections sync.Map
|
||||
notificationCh chan *domain.Notification
|
||||
stopCh chan struct{}
|
||||
config *config.Config
|
||||
}
|
||||
|
||||
func New(repo repository.NotificationRepository, logger *slog.Logger) NotificationStore {
|
||||
func New(repo repository.NotificationRepository, logger *slog.Logger, cfg *config.Config) NotificationStore {
|
||||
svc := &Service{
|
||||
repo: repo,
|
||||
logger: logger,
|
||||
connections: sync.Map{},
|
||||
notificationCh: make(chan *domain.Notification, 1000),
|
||||
stopCh: make(chan struct{}),
|
||||
config: cfg,
|
||||
}
|
||||
|
||||
go svc.startWorker()
|
||||
|
|
@ -37,12 +42,12 @@ func New(repo repository.NotificationRepository, logger *slog.Logger) Notificati
|
|||
|
||||
func (s *Service) addConnection(ctx context.Context, recipientID int64, c *websocket.Conn) {
|
||||
if c == nil {
|
||||
s.logger.Warn("Attempted to add nil WebSocket connection", "recipientID", recipientID)
|
||||
s.logger.Warn("[NotificationSvc.AddConnection] Attempted to add nil WebSocket connection", "recipientID", recipientID)
|
||||
return
|
||||
}
|
||||
|
||||
s.connections.Store(recipientID, c)
|
||||
s.logger.Info("Added WebSocket connection", "recipientID", recipientID)
|
||||
s.logger.Info("[NotificationSvc.AddConnection] Added WebSocket connection", "recipientID", recipientID)
|
||||
}
|
||||
|
||||
func (s *Service) SendNotification(ctx context.Context, notification *domain.Notification) error {
|
||||
|
|
@ -52,6 +57,7 @@ func (s *Service) SendNotification(ctx context.Context, notification *domain.Not
|
|||
|
||||
created, err := s.repo.CreateNotification(ctx, notification)
|
||||
if err != nil {
|
||||
s.logger.Error("[NotificationSvc.SendNotification] Failed to create notification", "id", notification.ID, "error", err)
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
@ -60,7 +66,7 @@ func (s *Service) SendNotification(ctx context.Context, notification *domain.Not
|
|||
select {
|
||||
case s.notificationCh <- notification:
|
||||
default:
|
||||
s.logger.Error("Notification channel full, dropping notification", "id", notification.ID)
|
||||
s.logger.Error("[NotificationSvc.SendNotification] Notification channel full, dropping notification", "id", notification.ID)
|
||||
}
|
||||
|
||||
return nil
|
||||
|
|
@ -69,17 +75,26 @@ func (s *Service) SendNotification(ctx context.Context, notification *domain.Not
|
|||
func (s *Service) MarkAsRead(ctx context.Context, notificationID string, recipientID int64) error {
|
||||
_, err := s.repo.UpdateNotificationStatus(ctx, notificationID, string(domain.DeliveryStatusSent), true, nil)
|
||||
if err != nil {
|
||||
s.logger.Error("[NotificationSvc.MarkAsRead] Failed to mark notification as read", "notificationID", notificationID, "recipientID", recipientID, "error", err)
|
||||
return err
|
||||
}
|
||||
s.logger.Info("[NotificationSvc.MarkAsRead] Notification marked as read", "notificationID", notificationID, "recipientID", recipientID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) ListNotifications(ctx context.Context, recipientID int64, limit, offset int) ([]domain.Notification, error) {
|
||||
return s.repo.ListNotifications(ctx, recipientID, limit, offset)
|
||||
notifications, err := s.repo.ListNotifications(ctx, recipientID, limit, offset)
|
||||
if err != nil {
|
||||
s.logger.Error("[NotificationSvc.ListNotifications] Failed to list notifications", "recipientID", recipientID, "limit", limit, "offset", offset, "error", err)
|
||||
return nil, err
|
||||
}
|
||||
s.logger.Info("[NotificationSvc.ListNotifications] Successfully listed notifications", "recipientID", recipientID, "count", len(notifications))
|
||||
return notifications, nil
|
||||
}
|
||||
|
||||
func (s *Service) ConnectWebSocket(ctx context.Context, recipientID int64, c *websocket.Conn) error {
|
||||
s.addConnection(ctx, recipientID, c)
|
||||
s.logger.Info("[NotificationSvc.ConnectWebSocket] WebSocket connection established", "recipientID", recipientID)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -87,17 +102,42 @@ func (s *Service) DisconnectWebSocket(recipientID int64) {
|
|||
s.connections.Delete(recipientID)
|
||||
if conn, loaded := s.connections.LoadAndDelete(recipientID); loaded {
|
||||
conn.(*websocket.Conn).Close()
|
||||
s.logger.Info("Disconnected WebSocket", "recipientID", recipientID)
|
||||
s.logger.Info("[NotificationSvc.DisconnectWebSocket] Disconnected WebSocket", "recipientID", recipientID)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) SendSMS(ctx context.Context, recipientID int64, message string) error {
|
||||
s.logger.Info("SMS notification requested", "recipientID", recipientID, "message", message)
|
||||
s.logger.Info("[NotificationSvc.SendSMS] SMS notification requested", "recipientID", recipientID, "message", message)
|
||||
|
||||
apiKey := s.config.AFRO_SMS_API_KEY
|
||||
senderName := s.config.AFRO_SMS_SENDER_NAME
|
||||
receiverPhone := s.config.AFRO_SMS_RECEIVER_PHONE_NUMBER
|
||||
hostURL := s.config.ADRO_SMS_HOST_URL
|
||||
endpoint := "/api/send"
|
||||
|
||||
request := afro.GetRequest(apiKey, endpoint, hostURL)
|
||||
request.Method = "GET"
|
||||
request.Sender(senderName)
|
||||
request.To(receiverPhone, message)
|
||||
|
||||
response, err := afro.MakeRequestWithContext(ctx, request)
|
||||
if err != nil {
|
||||
s.logger.Error("[NotificationSvc.SendSMS] Failed to send SMS", "recipientID", recipientID, "error", err)
|
||||
return err
|
||||
}
|
||||
|
||||
if response["acknowledge"] == "success" {
|
||||
s.logger.Info("[NotificationSvc.SendSMS] SMS sent successfully", "recipientID", recipientID)
|
||||
} else {
|
||||
s.logger.Error("[NotificationSvc.SendSMS] Failed to send SMS", "recipientID", recipientID, "response", response["response"])
|
||||
return errors.New("SMS delivery failed: " + response["response"].(string))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) SendEmail(ctx context.Context, recipientID int64, subject, message string) error {
|
||||
s.logger.Info("Email notification requested", "recipientID", recipientID, "subject", subject)
|
||||
s.logger.Info("[NotificationSvc.SendEmail] Email notification requested", "recipientID", recipientID, "subject", subject)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -107,6 +147,7 @@ func (s *Service) startWorker() {
|
|||
case notification := <-s.notificationCh:
|
||||
s.handleNotification(notification)
|
||||
case <-s.stopCh:
|
||||
s.logger.Info("[NotificationSvc.StartWorker] Worker stopped")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
|
@ -118,22 +159,22 @@ func (s *Service) handleNotification(notification *domain.Notification) {
|
|||
if conn, ok := s.connections.Load(notification.RecipientID); ok {
|
||||
data, err := notification.ToJSON()
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to serialize notification", "id", notification.ID, "error", err)
|
||||
s.logger.Error("[NotificationSvc.HandleNotification] Failed to serialize notification", "id", notification.ID, "error", err)
|
||||
return
|
||||
}
|
||||
if err := conn.(*websocket.Conn).WriteMessage(websocket.TextMessage, data); err != nil {
|
||||
s.logger.Error("Failed to send WebSocket message", "id", notification.ID, "error", err)
|
||||
s.logger.Error("[NotificationSvc.HandleNotification] Failed to send WebSocket message", "id", notification.ID, "error", err)
|
||||
notification.DeliveryStatus = domain.DeliveryStatusFailed
|
||||
} else {
|
||||
notification.DeliveryStatus = domain.DeliveryStatusSent
|
||||
}
|
||||
} else {
|
||||
s.logger.Warn("No WebSocket connection for recipient", "recipientID", notification.RecipientID)
|
||||
s.logger.Warn("[NotificationSvc.HandleNotification] No WebSocket connection for recipient", "recipientID", notification.RecipientID)
|
||||
notification.DeliveryStatus = domain.DeliveryStatusFailed
|
||||
}
|
||||
|
||||
if _, err := s.repo.UpdateNotificationStatus(ctx, notification.ID, string(notification.DeliveryStatus), notification.IsRead, notification.Metadata); err != nil {
|
||||
s.logger.Error("Failed to update notification status", "id", notification.ID, "error", err)
|
||||
s.logger.Error("[NotificationSvc.HandleNotification] Failed to update notification status", "id", notification.ID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -146,6 +187,7 @@ func (s *Service) startRetryWorker() {
|
|||
case <-ticker.C:
|
||||
s.retryFailedNotifications()
|
||||
case <-s.stopCh:
|
||||
s.logger.Info("[NotificationSvc.StartRetryWorker] Retry worker stopped")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
|
@ -155,7 +197,7 @@ func (s *Service) retryFailedNotifications() {
|
|||
ctx := context.Background()
|
||||
failedNotifications, err := s.repo.ListFailedNotifications(ctx, 100)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to list failed notifications", "error", err)
|
||||
s.logger.Error("[NotificationSvc.RetryFailedNotifications] Failed to list failed notifications", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -167,18 +209,20 @@ func (s *Service) retryFailedNotifications() {
|
|||
if conn, ok := s.connections.Load(notification.RecipientID); ok {
|
||||
data, err := notification.ToJSON()
|
||||
if err != nil {
|
||||
s.logger.Error("[NotificationSvc.RetryFailedNotifications] Failed to serialize notification for retry", "id", notification.ID, "error", err)
|
||||
continue
|
||||
}
|
||||
if err := conn.(*websocket.Conn).WriteMessage(websocket.TextMessage, data); err == nil {
|
||||
notification.DeliveryStatus = domain.DeliveryStatusSent
|
||||
if _, err := s.repo.UpdateNotificationStatus(ctx, notification.ID, string(notification.DeliveryStatus), notification.IsRead, notification.Metadata); err != nil {
|
||||
s.logger.Error("Failed to update after retry", "id", notification.ID, "error", err)
|
||||
s.logger.Error("[NotificationSvc.RetryFailedNotifications] Failed to update after retry", "id", notification.ID, "error", err)
|
||||
}
|
||||
s.logger.Info("[NotificationSvc.RetryFailedNotifications] Successfully retried notification", "id", notification.ID)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
s.logger.Error("Max retries reached for notification", "id", notification.ID)
|
||||
s.logger.Error("[NotificationSvc.RetryFailedNotifications] Max retries reached for notification", "id", notification.ID)
|
||||
}(notification)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user