notification system improvements

config/engine-api.toml

@@ -234,4 +234,4 @@ AUTHENTICATION_OIDC_FRONT_END_URL = "http://127.0.0.1:4200"
 AUTHENTICATION_OIDC_ENCRYPTION_KEY = "thisis24characterslongs."
 
 # NOTIFICATION_LIFETIME: The lifetime of a notification in the database.
-NOTIFICATION_LIFETIME = "1w"
+NOTIFICATION_LIFETIME = "168h" # 168h = 7 days, available units are "ns", "us" (or "µs"), "ms", "s", "m", "h"

internals/app/services.go

@@ -116,7 +116,7 @@ func initCoordinator() {
 
 	instanceName := viper.GetString("INSTANCE_NAME")
 	if err = coordinator.InitInstance(instanceName, models); err != nil {
-		zap.L().Fatal("Intialisation of coordinator master", zap.Error(err))
+		zap.L().Fatal("Initialization of coordinator master", zap.Error(err))
 	}
 	if viper.GetBool("ENABLE_CRONS_ON_START") {
 		for _, li := range coordinator.GetInstance().LogicalIndices {

internals/handlers/notification_handlers.go

@@ -1,8 +1,10 @@
 package handlers
 
 import (
+	"github.com/myrteametrics/myrtea-engine-api/v5/internals/export"
 	"net/http"
 	"strconv"
+	"time"
 
 	"github.com/go-chi/chi/v5"
 	"github.com/myrteametrics/myrtea-engine-api/v5/internals/dbutils"
@@ -67,6 +69,9 @@ func GetNotifications(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	notifications = append(notifications, notification.NewMockNotification(1, "level", "title", "subTitle", "description", time.Now(), []int64{1}, map[string]interface{}{"issueId": 1}))
+	notifications = append(notifications, notification.NewExportNotification(2, export.WrapperItem{Id: "test"}, 1))
+	notifications = append(notifications, notification.NewBaseNotification(3, false))
 	render.JSON(w, r, notifications)
 }
 

internals/handlers/notifier_handlers.go

@@ -1,17 +1,12 @@
 package handlers
 
 import (
-	"github.com/google/uuid"
-	"github.com/myrteametrics/myrtea-engine-api/v5/internals/export"
-	"github.com/myrteametrics/myrtea-engine-api/v5/internals/notifier/notification"
-	"net/http"
-	"time"
-
 	"github.com/myrteametrics/myrtea-engine-api/v5/internals/handlers/render"
 	"github.com/myrteametrics/myrtea-engine-api/v5/internals/models"
 	"github.com/myrteametrics/myrtea-engine-api/v5/internals/notifier"
 	"github.com/myrteametrics/myrtea-engine-api/v5/internals/security/users"
 	"go.uber.org/zap"
+	"net/http"
 )
 
 // NotificationsWSRegister godoc
@@ -46,20 +41,20 @@ func NotificationsWSRegister(w http.ResponseWriter, r *http.Request) {
 		zap.L().Error("Add new WS Client to manager", zap.Error(err))
 		return
 	}
-	go func(client *notifier.WebsocketClient) { // temporary for tests
-		zap.L().Info("starting notifier")
-		ticker := time.NewTicker(1 * time.Second)
-		after := time.After(30 * time.Second)
-		for {
-			select {
-			case <-ticker.C:
-				notifier.C().SendToUsers(notification.ExportNotification{Status: export.StatusPending, Export: export.WrapperItem{Id: uuid.New().String(), FileName: "test.bla"}}, []users.UserWithPermissions{user})
-				zap.L().Info("send notification")
-			case <-after:
-				return
-			}
-		}
-	}(client)
+	//go func(client *notifier.WebsocketClient) { // temporary for tests
+	//	zap.L().Info("starting notifier")
+	//	ticker := time.NewTicker(1 * time.Second)
+	//	after := time.After(30 * time.Second)
+	//	for {
+	//		select {
+	//		case <-ticker.C:
+	//			notifier.C().SendToUsers(notification.ExportNotification{Status: export.StatusPending, Export: export.WrapperItem{Id: uuid.New().String(), FileName: "test.bla"}}, []users.UserWithPermissions{user})
+	//			zap.L().Info("send notification")
+	//		case <-after:
+	//			return
+	//		}
+	//	}
+	//}(client)
 	go client.Write()
 
 	// go client.Read() // Disabled until proper usage

internals/handlers/processor_handlers.go

@@ -3,15 +3,13 @@ package handlers
 import (
 	"encoding/json"
 	"errors"
-	"github.com/myrteametrics/myrtea-engine-api/v5/internals/ingester"
-	"net/http"
-	"time"
-
 	"github.com/myrteametrics/myrtea-engine-api/v5/internals/handlers/render"
+	"github.com/myrteametrics/myrtea-engine-api/v5/internals/ingester"
 	"github.com/myrteametrics/myrtea-engine-api/v5/internals/processor"
 	"github.com/myrteametrics/myrtea-engine-api/v5/internals/scheduler"
 	"github.com/myrteametrics/myrtea-sdk/v4/models"
 	"go.uber.org/zap"
+	"net/http"
 )
 
 // ProcessorHandler is a basic struct allowing to set up a single aggregateIngester instance for all handlers
@@ -21,12 +19,8 @@ type ProcessorHandler struct {
 
 // NewProcessorHandler returns a pointer to an ProcessorHandler instance
 func NewProcessorHandler() *ProcessorHandler {
-	var aggregateIngester = ingester.NewAggregateIngester()
-	go aggregateIngester.Run()        // Start ingester
-	time.Sleep(10 * time.Millisecond) // goroutine warm-up
-
 	return &ProcessorHandler{
-		aggregateIngester: aggregateIngester,
+		aggregateIngester: ingester.NewAggregateIngester(),
 	}
 }
 

internals/ingester/aggregate.go

@@ -13,8 +13,9 @@ import (
 
 // AggregateIngester is a component which process scheduler.ExternalAggregate
 type AggregateIngester struct {
-	Data             chan []scheduler.ExternalAggregate
+	data             chan []scheduler.ExternalAggregate
 	metricQueueGauge *stdprometheus.Gauge
+	running          bool
 }
 
 var (
@@ -39,16 +40,17 @@ func _newRegisteredGauge() *stdprometheus.Gauge {
 // NewAggregateIngester returns a pointer to a new AggregateIngester instance
 func NewAggregateIngester() *AggregateIngester {
 	return &AggregateIngester{
-		Data:             make(chan []scheduler.ExternalAggregate, viper.GetInt("AGGREGATEINGESTER_QUEUE_BUFFER_SIZE")),
+		data:             make(chan []scheduler.ExternalAggregate, viper.GetInt("AGGREGATEINGESTER_QUEUE_BUFFER_SIZE")),
 		metricQueueGauge: _aggregateIngesterGauge,
+		running:          false,
 	}
 }
 
 // Run is the main routine of a TypeIngester instance
-func (ingester *AggregateIngester) Run() {
+func (ai *AggregateIngester) Run() {
 	zap.L().Info("Starting AggregateIngester")
 
-	for ir := range ingester.Data {
+	for ir := range ai.data {
 		zap.L().Debug("Received ExternalAggregate", zap.Int("ExternalAggregate items count", len(ir)))
 
 		err := HandleAggregates(ir)
@@ -57,25 +59,31 @@ func (ingester *AggregateIngester) Run() {
 		}
 
 		// Update queue gauge
-		(*ingester.metricQueueGauge).Set(float64(len(ingester.Data)))
+		(*ai.metricQueueGauge).Set(float64(len(ai.data)))
 	}
 
 }
 
 // Ingest process an array of scheduler.ExternalAggregate
-func (ingester *AggregateIngester) Ingest(aggregates []scheduler.ExternalAggregate) error {
-	dataLen := len(ingester.Data)
+func (ai *AggregateIngester) Ingest(aggregates []scheduler.ExternalAggregate) error {
+	dataLen := len(ai.data)
+
+	// Start ingester if not running
+	if !ai.running {
+		go ai.Run()
+		ai.running = true
+	}
 
 	zap.L().Debug("Ingesting data", zap.Any("aggregates", aggregates))
 
 	// Check for channel overloading
-	if dataLen+1 >= cap(ingester.Data) {
+	if dataLen+1 >= cap(ai.data) {
 		zap.L().Debug("Buffered channel would be overloaded with incoming bulkIngestRequest")
-		(*ingester.metricQueueGauge).Set(float64(dataLen))
+		(*ai.metricQueueGauge).Set(float64(dataLen))
 		return errors.New("channel overload")
 	}
 
-	ingester.Data <- aggregates
+	ai.data <- aggregates
 
 	return nil
 }

internals/notifier/notification/handler.go

@@ -45,6 +45,8 @@ func NewHandler(notificationLifetime time.Duration) *Handler {
 	// useless to start cleaner if lifetime is less than 0
 	if notificationLifetime > 0 {
 		go handler.startCleaner(context.Background())
+	} else {
+		zap.L().Info("Notification cleaner will not be started", zap.Duration("notificationLifetime", notificationLifetime))
 	}
 
 	return handler
@@ -68,7 +70,9 @@ func (h *Handler) RegisterNotificationTypes() {
 
 // startCleaner start a ticker to clean expired notifications in database every 24 hours
 func (h *Handler) startCleaner(context context.Context) {
-	ticker := time.NewTicker(time.Hour * 24)
+	cleanRate := time.Hour * 24
+	zap.L().Info("Starting notification cleaner", zap.Duration("cleanRate", cleanRate), zap.Duration("notificationLifetime", h.notificationLifetime))
+	ticker := time.NewTicker(cleanRate)
 	defer ticker.Stop()
 	for {
 		select {

internals/notifier/notification/notification.go

@@ -20,6 +20,15 @@ type BaseNotification struct {
 	Type         string `json:"type"`
 }
 
+// NewBaseNotification returns a new instance of a BaseNotification
+func NewBaseNotification(id int64, isRead bool) BaseNotification {
+	return BaseNotification{
+		Id:     id,
+		IsRead: isRead,
+		Type:   "BaseNotification",
+	}
+}
+
 // NewInstance returns a new instance of a BaseNotification
 func (n BaseNotification) NewInstance(id int64, data []byte, isRead bool) (Notification, error) {
 	var notification BaseNotification

internals/notifier/notification/notification_mock.go

@@ -14,6 +14,7 @@ type MockNotification struct {
 	Title        string                 `json:"title"`
 	SubTitle     string                 `json:"subtitle"`
 	Description  string                 `json:"description"`
+	Target       string                 `json:"target"`
 	Context      map[string]interface{} `json:"context,omitempty"`
 }