d486a5e44f
continuous-integration/drone/push Build is passing
- Introduced a new `NotificationWorker` to promote due scheduled notifications from pending to sent, improving notification management. - Added `NotificationInterval` configuration to schedule the notification delivery worker, with a default value for flexibility. - Implemented `MarkDueScheduledAsSent` method in the notification repository to update the status of notifications based on their delivery time. - Updated the notification service to process due scheduled notifications during relevant operations, ensuring timely delivery. This update enhances the notification system by automating the delivery of scheduled notifications, improving user engagement and operational efficiency.
318 lines
8.8 KiB
Go
318 lines
8.8 KiB
Go
package notification
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"regexp"
|
|
"strings"
|
|
"time"
|
|
|
|
"tm/pkg/logger"
|
|
orm "tm/pkg/mongo"
|
|
"tm/pkg/response"
|
|
|
|
"go.mongodb.org/mongo-driver/v2/bson"
|
|
mongopkg "go.mongodb.org/mongo-driver/v2/mongo"
|
|
)
|
|
|
|
var (
|
|
ErrNotificationNotFound = errors.New("notification not found")
|
|
)
|
|
|
|
// Notification represents a stored notification in the database
|
|
type Notification struct {
|
|
orm.Model `bson:",inline"`
|
|
UserID string `bson:"user_id" json:"user_id"`
|
|
Title string `bson:"title" json:"title"`
|
|
Message string `bson:"message" json:"message"`
|
|
Link string `bson:"link,omitempty" json:"link,omitempty"`
|
|
Image string `bson:"image,omitempty" json:"image,omitempty"`
|
|
Priority string `bson:"priority" json:"priority"`
|
|
Type string `bson:"type" json:"type"`
|
|
EventType string `bson:"event_type,omitempty" json:"event_type,omitempty"`
|
|
Status string `bson:"status" json:"status"`
|
|
Methods map[string]string `bson:"methods,omitempty" json:"methods,omitempty"`
|
|
Metadata map[string]any `bson:"metadata,omitempty" json:"metadata,omitempty"`
|
|
ScheduleAt int64 `bson:"schedule_at,omitempty" json:"schedule_at,omitempty"`
|
|
Seen bool `bson:"seen" json:"seen"`
|
|
SeenAt int64 `bson:"seen_at,omitempty" json:"seen_at,omitempty"`
|
|
IsScheduled bool `bson:"is_scheduled" json:"is_scheduled"`
|
|
ScheduledAt int64 `bson:"scheduled_at,omitempty" json:"scheduled_at,omitempty"`
|
|
}
|
|
|
|
// SetID sets the notification ID (implements IDSetter interface)
|
|
func (n *Notification) SetID(id string) {
|
|
n.Model.SetID(id)
|
|
}
|
|
|
|
// GetID returns the notification ID (implements IDGetter interface)
|
|
func (n *Notification) GetID() string {
|
|
return n.Model.GetID()
|
|
}
|
|
|
|
// Repository defines methods for notification data access
|
|
type Repository interface {
|
|
Create(ctx context.Context, notification *Notification) error
|
|
GetByID(ctx context.Context, id string) (*Notification, error)
|
|
GetByUserID(ctx context.Context, userIDs []string, search, status string, seen *bool, priority, eventType, notificationType string, pagination *response.Pagination) (*orm.PaginatedResult[Notification], error)
|
|
Update(ctx context.Context, notification *Notification) error
|
|
MarkAsSeen(ctx context.Context, notificationID, userID string) error
|
|
MarkAllAsSeen(ctx context.Context, userID string) error
|
|
MarkDueScheduledAsSent(ctx context.Context, dueBefore int64) (int64, error)
|
|
}
|
|
|
|
type notificationRepository struct {
|
|
ormRepo orm.Repository[Notification]
|
|
collection *mongopkg.Collection
|
|
logger logger.Logger
|
|
}
|
|
|
|
// NewRepository creates a new notification repository
|
|
func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Repository {
|
|
// Create indexes
|
|
indexes := []orm.Index{
|
|
*orm.NewIndex("user_id_idx", bson.D{{Key: "user_id", Value: 1}}),
|
|
*orm.NewIndex("status_idx", bson.D{{Key: "status", Value: 1}}),
|
|
*orm.NewIndex("seen_idx", bson.D{{Key: "seen", Value: 1}}),
|
|
*orm.NewIndex("created_at_idx", bson.D{{Key: "created_at", Value: -1}}),
|
|
*orm.NewIndex("scheduled_delivery_idx", bson.D{
|
|
{Key: "status", Value: 1},
|
|
{Key: "is_scheduled", Value: 1},
|
|
{Key: "schedule_at", Value: 1},
|
|
}),
|
|
}
|
|
|
|
err := mongoManager.CreateIndexes("notifications", indexes)
|
|
if err != nil {
|
|
logger.Warn("Failed to create notification indexes", map[string]interface{}{
|
|
"error": err.Error(),
|
|
})
|
|
}
|
|
|
|
collection := mongoManager.GetCollection("notifications")
|
|
ormRepo := orm.NewRepository[Notification](collection, logger)
|
|
|
|
return ¬ificationRepository{
|
|
ormRepo: ormRepo,
|
|
collection: collection,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
// Create creates a new notification
|
|
func (r *notificationRepository) Create(ctx context.Context, notification *Notification) error {
|
|
now := time.Now().Unix()
|
|
notification.CreatedAt = now
|
|
notification.UpdatedAt = now
|
|
|
|
err := r.ormRepo.Create(ctx, notification)
|
|
if err != nil {
|
|
r.logger.Error("Failed to create notification", map[string]interface{}{
|
|
"error": err.Error(),
|
|
"user_id": notification.UserID,
|
|
})
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetByID retrieves a notification by ID
|
|
func (r *notificationRepository) GetByID(ctx context.Context, id string) (*Notification, error) {
|
|
return r.ormRepo.FindByID(ctx, id)
|
|
}
|
|
|
|
// GetByUserID retrieves notifications with optional user filters
|
|
func (r *notificationRepository) GetByUserID(ctx context.Context, userIDs []string, search, status string, seen *bool, priority, eventType, notificationType string, pagination *response.Pagination) (*orm.PaginatedResult[Notification], error) {
|
|
filter := bson.M{}
|
|
andConditions := bson.A{}
|
|
|
|
if q := strings.TrimSpace(search); q != "" {
|
|
pattern := regexp.QuoteMeta(q)
|
|
andConditions = append(andConditions, bson.M{
|
|
"$or": bson.A{
|
|
bson.M{"title": bson.M{"$regex": pattern, "$options": "i"}},
|
|
bson.M{"message": bson.M{"$regex": pattern, "$options": "i"}},
|
|
},
|
|
})
|
|
}
|
|
|
|
if len(userIDs) == 1 {
|
|
filter["user_id"] = userIDs[0]
|
|
} else if len(userIDs) > 1 {
|
|
filter["user_id"] = bson.M{"$in": userIDs}
|
|
}
|
|
|
|
if status != "" {
|
|
filter["status"] = status
|
|
}
|
|
|
|
if seen != nil {
|
|
filter["seen"] = *seen
|
|
}
|
|
|
|
if priority != "" {
|
|
filter["priority"] = priority
|
|
}
|
|
|
|
if eventType != "" {
|
|
if eventType == EventTypeInApp {
|
|
andConditions = append(andConditions, bson.M{
|
|
"$or": bson.A{
|
|
bson.M{"event_type": EventTypeInApp},
|
|
bson.M{"event_type": ""},
|
|
bson.M{"event_type": bson.M{"$exists": false}},
|
|
},
|
|
})
|
|
} else {
|
|
filter["event_type"] = eventType
|
|
}
|
|
}
|
|
|
|
if notificationType != "" {
|
|
filter["type"] = notificationType
|
|
}
|
|
|
|
if len(andConditions) == 1 {
|
|
for key, value := range andConditions[0].(bson.M) {
|
|
filter[key] = value
|
|
}
|
|
} else if len(andConditions) > 1 {
|
|
filter["$and"] = andConditions
|
|
}
|
|
|
|
mongoPagination, err := orm.BuildListPagination(
|
|
pagination.Limit,
|
|
pagination.Offset,
|
|
pagination.Cursor,
|
|
pagination.SortBy,
|
|
pagination.SortOrder,
|
|
filter,
|
|
orm.ListPaginationOptions{},
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
result, err := r.ormRepo.FindAll(ctx, filter, mongoPagination)
|
|
if err != nil {
|
|
r.logger.Error("Failed to find notifications", map[string]interface{}{
|
|
"error": err.Error(),
|
|
"user_ids": userIDs,
|
|
})
|
|
return nil, err
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
// Update updates a notification
|
|
func (r *notificationRepository) Update(ctx context.Context, notification *Notification) error {
|
|
notification.UpdatedAt = time.Now().Unix()
|
|
|
|
err := r.ormRepo.Update(ctx, notification)
|
|
if err != nil {
|
|
r.logger.Error("Failed to update notification", map[string]interface{}{
|
|
"error": err.Error(),
|
|
"user_id": notification.UserID,
|
|
})
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// MarkAsSeen marks a notification as seen
|
|
func (r *notificationRepository) MarkAsSeen(ctx context.Context, notificationID, userID string) error {
|
|
objectID, err := bson.ObjectIDFromHex(notificationID)
|
|
if err != nil {
|
|
return ErrNotificationNotFound
|
|
}
|
|
|
|
filter := bson.M{
|
|
"_id": objectID,
|
|
"user_id": userID,
|
|
}
|
|
|
|
update := bson.M{
|
|
"$set": bson.M{
|
|
"seen": true,
|
|
"seen_at": time.Now().Unix(),
|
|
"updated_at": time.Now().Unix(),
|
|
},
|
|
}
|
|
|
|
result, err := r.collection.UpdateOne(ctx, filter, update)
|
|
if err != nil {
|
|
r.logger.Error("Failed to mark notification as seen", map[string]interface{}{
|
|
"error": err.Error(),
|
|
"notification_id": notificationID,
|
|
"user_id": userID,
|
|
})
|
|
return err
|
|
}
|
|
|
|
if result.MatchedCount == 0 {
|
|
return ErrNotificationNotFound
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// MarkAllAsSeen marks all notifications for a user as seen
|
|
func (r *notificationRepository) MarkAllAsSeen(ctx context.Context, userID string) error {
|
|
filter := bson.M{
|
|
"user_id": userID,
|
|
"seen": false,
|
|
}
|
|
|
|
update := bson.M{
|
|
"$set": bson.M{
|
|
"seen": true,
|
|
"seen_at": time.Now().Unix(),
|
|
"updated_at": time.Now().Unix(),
|
|
},
|
|
}
|
|
|
|
_, err := r.collection.UpdateMany(ctx, filter, update)
|
|
if err != nil {
|
|
r.logger.Error("Failed to mark all notifications as seen", map[string]interface{}{
|
|
"error": err.Error(),
|
|
"user_id": userID,
|
|
})
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// MarkDueScheduledAsSent updates pending scheduled notifications whose delivery time has passed.
|
|
func (r *notificationRepository) MarkDueScheduledAsSent(ctx context.Context, dueBefore int64) (int64, error) {
|
|
filter := bson.M{
|
|
"status": string(DeliveryStatusPending),
|
|
"is_scheduled": true,
|
|
"schedule_at": bson.M{
|
|
"$gt": 0,
|
|
"$lte": dueBefore,
|
|
},
|
|
}
|
|
|
|
update := bson.M{
|
|
"$set": bson.M{
|
|
"status": string(DeliveryStatusSent),
|
|
"updated_at": time.Now().Unix(),
|
|
},
|
|
}
|
|
|
|
result, err := r.collection.UpdateMany(ctx, filter, update)
|
|
if err != nil {
|
|
r.logger.Error("Failed to mark due scheduled notifications as sent", map[string]interface{}{
|
|
"error": err.Error(),
|
|
"due_before": dueBefore,
|
|
})
|
|
return 0, err
|
|
}
|
|
|
|
return result.ModifiedCount, nil
|
|
}
|