storagenode notification service and api added

Change-Id: I36898d7c43e1768e0cae0da8d83bb20b16f0cdde

storagenode/console/consolenotifications/notificationsapi.go

@@ -0,0 +1,141 @@
+// Copyright (C) 2019 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package consolenotifications
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/gorilla/mux"
+	"github.com/skyrings/skyring-common/tools/uuid"
+	"github.com/zeebo/errs"
+	"go.uber.org/zap"
+	"gopkg.in/spacemonkeygo/monkit.v2"
+
+	"storj.io/storj/storagenode/notifications"
+)
+
+const (
+	contentType = "Content-Type"
+
+	applicationJSON = "application/json"
+)
+
+var mon = monkit.Package()
+
+// Error is error type of storagenode web console.
+var Error = errs.Class("notifications console web error")
+
+// Notifications represents notification service.
+// architecture: Service
+type Notifications struct {
+	service *notifications.Service
+
+	log *zap.Logger
+}
+
+// jsonOutput defines json structure of api response data.
+type jsonOutput struct {
+	Data  interface{} `json:"data"`
+	Error string      `json:"error"`
+}
+
+// NewNotifications creates new instance of notification service.
+func NewNotifications(log *zap.Logger, service *notifications.Service) *Notifications {
+	return &Notifications{
+		log:     log,
+		service: service,
+	}
+}
+
+// ReadNotification updates specific notification in database as read.
+func (notification *Notifications) ReadNotification(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	defer mon.Task()(&ctx)(nil)
+	var err error
+
+	params := mux.Vars(r)
+	id, ok := params["id"]
+	if !ok {
+		notification.writeError(w, http.StatusInternalServerError, Error.Wrap(err))
+		return
+	}
+
+	notificationID, err := uuid.Parse(id)
+	if err != nil {
+		notification.writeError(w, http.StatusInternalServerError, Error.Wrap(err))
+		return
+	}
+
+	err = notification.service.Read(ctx, *notificationID)
+	if err != nil {
+		notification.writeError(w, http.StatusInternalServerError, Error.Wrap(err))
+		return
+	}
+}
+
+// ReadAllNotifications updates all notifications in database as read.
+func (notification *Notifications) ReadAllNotifications(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	defer mon.Task()(&ctx)(nil)
+
+	err := notification.service.ReadAll(ctx)
+	if err != nil {
+		notification.writeError(w, http.StatusInternalServerError, Error.Wrap(err))
+		return
+	}
+}
+
+// ListNotifications returns listed page of notifications from database.
+func (notification *Notifications) ListNotifications(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	defer mon.Task()(&ctx)(nil)
+	var err error
+
+	var request struct {
+		Cursor notifications.Cursor `json:"cursor"`
+	}
+
+	err = json.NewDecoder(r.Body).Decode(&request)
+	if err != nil {
+		notification.writeError(w, http.StatusInternalServerError, Error.Wrap(err))
+		return
+	}
+
+	page, err := notification.service.List(ctx, request.Cursor)
+	if err != nil {
+		notification.writeError(w, http.StatusInternalServerError, Error.Wrap(err))
+		return
+	}
+
+	notification.writeData(w, page)
+}
+
+// writeData is helper method to write JSON to http.ResponseWriter and log encoding error.
+func (notification *Notifications) writeData(w http.ResponseWriter, data interface{}) {
+	w.Header().Set(contentType, applicationJSON)
+	w.WriteHeader(http.StatusOK)
+
+	output := jsonOutput{Data: data}
+
+	if err := json.NewEncoder(w).Encode(output); err != nil {
+		notification.log.Error("json encoder error", zap.Error(err))
+	}
+}
+
+// writeError writes a JSON error payload to http.ResponseWriter log encoding error.
+func (notification *Notifications) writeError(w http.ResponseWriter, status int, err error) {
+	if status >= http.StatusInternalServerError {
+		notification.log.Error("api handler server error", zap.Int("status code", status), zap.Error(err))
+	}
+
+	w.Header().Set(contentType, applicationJSON)
+	w.WriteHeader(status)
+
+	output := jsonOutput{Error: err.Error()}
+
+	if err := json.NewEncoder(w).Encode(output); err != nil {
+		notification.log.Error("json encoder error", zap.Error(err))
+	}
+}

storagenode/console/consoleserver/server.go

@@ -17,6 +17,8 @@ import (
 
 	"storj.io/storj/pkg/storj"
 	"storj.io/storj/storagenode/console"
+	"storj.io/storj/storagenode/console/consolenotifications"
+	"storj.io/storj/storagenode/notifications"
 )
 
 const (
@@ -43,22 +45,26 @@ type Config struct {
 type Server struct {
 	log *zap.Logger
 
-	service  *console.Service
-	listener net.Listener
+	service       *console.Service
+	notifications *notifications.Service
+	listener      net.Listener
 
 	server http.Server
 }
 
 // NewServer creates new instance of storagenode console web server.
-func NewServer(logger *zap.Logger, assets http.FileSystem, service *console.Service, listener net.Listener) *Server {
+func NewServer(logger *zap.Logger, assets http.FileSystem, notifications *notifications.Service, service *console.Service, listener net.Listener) *Server {
 	server := Server{
-		log:      logger,
-		service:  service,
-		listener: listener,
+		log:           logger,
+		service:       service,
+		listener:      listener,
+		notifications: notifications,
 	}
 
 	router := mux.NewRouter()
 	apiRouter := router.PathPrefix("/api").Subrouter()
+	notificationRouter := router.PathPrefix("/api/notifications").Subrouter()
+	notificationController := consolenotifications.NewNotifications(server.log, server.notifications)
 
 	if assets != nil {
 		fs := http.FileServer(assets)
@@ -74,6 +80,9 @@ func NewServer(logger *zap.Logger, assets http.FileSystem, service *console.Serv
 	apiRouter.Handle("/dashboard", http.HandlerFunc(server.dashboardHandler)).Methods(http.MethodGet)
 	apiRouter.Handle("/satellites", http.HandlerFunc(server.satellitesHandler)).Methods(http.MethodGet)
 	apiRouter.Handle("/satellite/{id}", http.HandlerFunc(server.satelliteHandler)).Methods(http.MethodGet)
+	notificationRouter.Handle("/list", http.HandlerFunc(notificationController.ListNotifications)).Methods(http.MethodGet)
+	notificationRouter.Handle("/{id}/read", http.HandlerFunc(notificationController.ReadNotification)).Methods(http.MethodPost)
+	notificationRouter.Handle("/readall", http.HandlerFunc(notificationController.ReadAllNotifications)).Methods(http.MethodPost)
 
 	server.server = http.Server{
 		Handler: router,

storagenode/notifications/notifications.go

@@ -17,58 +17,58 @@ import (
 // architecture: Database
 type DB interface {
 	Insert(ctx context.Context, notification NewNotification) (Notification, error)
-	List(ctx context.Context, cursor NotificationCursor) (NotificationPage, error)
+	List(ctx context.Context, cursor Cursor) (Page, error)
 	Read(ctx context.Context, notificationID uuid.UUID) error
 	ReadAll(ctx context.Context) error
 }
 
-// NotificationType is a numeric value of specific notification type.
-type NotificationType int
+// Type is a numeric value of specific notification type.
+type Type int
 
 const (
-	// NotificationTypeCustom is a common notification type which doesn't describe node's core functionality.
+	// TypeCustom is a common notification type which doesn't describe node's core functionality.
 	// TODO: change type name when all notification types will be known
-	NotificationTypeCustom NotificationType = 0
-	// NotificationTypeAuditCheckFailure is a notification type which describes node's audit check failure.
-	NotificationTypeAuditCheckFailure NotificationType = 1
-	// NotificationTypeUptimeCheckFailure is a notification type which describes node's uptime check failure.
-	NotificationTypeUptimeCheckFailure NotificationType = 2
-	// NotificationTypeDisqualification is a notification type which describes node's disqualification status.
-	NotificationTypeDisqualification NotificationType = 3
+	TypeCustom Type = 0
+	// TypeAuditCheckFailure is a notification type which describes node's audit check failure.
+	TypeAuditCheckFailure Type = 1
+	// TypeUptimeCheckFailure is a notification type which describes node's uptime check failure.
+	TypeUptimeCheckFailure Type = 2
+	// TypeDisqualification is a notification type which describes node's disqualification status.
+	TypeDisqualification Type = 3
 )
 
 // NewNotification holds notification entity info which is being received from satellite or local client.
 type NewNotification struct {
 	SenderID storj.NodeID
-	Type     NotificationType
+	Type     Type
 	Title    string
 	Message  string
 }
 
 // Notification holds notification entity info which is being retrieved from database.
 type Notification struct {
-	ID        uuid.UUID
-	SenderID  storj.NodeID
-	Type      NotificationType
-	Title     string
-	Message   string
-	ReadAt    *time.Time
-	CreatedAt time.Time
+	ID        uuid.UUID    `json:"id"`
+	SenderID  storj.NodeID `json:"senderID"`
+	Type      Type         `json:"type"`
+	Title     string       `json:"title"`
+	Message   string       `json:"message"`
+	ReadAt    *time.Time   `json:"readAt"`
+	CreatedAt time.Time    `json:"createdAt"`
 }
 
-// NotificationCursor holds notification cursor entity which is used to create listed page from database.
-type NotificationCursor struct {
+// Cursor holds notification cursor entity which is used to create listed page from database.
+type Cursor struct {
 	Limit uint
 	Page  uint
 }
 
-// NotificationPage holds notification page entity which is used to show listed page of notifications on UI.
-type NotificationPage struct {
-	Notifications []Notification
+// Page holds notification page entity which is used to show listed page of notifications on UI.
+type Page struct {
+	Notifications []Notification `json:"notifications"`
 
-	Offset      uint64
-	Limit       uint
-	CurrentPage uint
-	PageCount   uint
-	TotalCount  uint64
+	Offset      uint64 `json:"offset"`
+	Limit       uint   `json:"limit"`
+	CurrentPage uint   `json:"currentPage"`
+	PageCount   uint   `json:"pageCount"`
+	TotalCount  uint64 `json:"totalCount"`
 }

storagenode/notifications/notifications_test.go

@@ -48,7 +48,7 @@ func TestNotificationsDB(t *testing.T) {
 			Message:  "testMessage2",
 		}
 
-		notificationCursor := notifications.NotificationCursor{
+		notificationCursor := notifications.Cursor{
 			Limit: 2,
 			Page:  1,
 		}
@@ -74,7 +74,7 @@ func TestNotificationsDB(t *testing.T) {
 		assert.Equal(t, expectedNotification2.Title, notificationFromDB2.Title)
 		assert.Equal(t, expectedNotification2.Message, notificationFromDB2.Message)
 
-		page := notifications.NotificationPage{}
+		page := notifications.Page{}
 
 		// test List method to return right form of page depending on cursor.
 		t.Run("test paged list", func(t *testing.T) {
@@ -90,7 +90,7 @@ func TestNotificationsDB(t *testing.T) {
 			assert.Equal(t, uint(1), page.CurrentPage)
 		})
 
-		notificationCursor = notifications.NotificationCursor{
+		notificationCursor = notifications.Cursor{
 			Limit: 5,
 			Page:  1,
 		}
@@ -135,7 +135,7 @@ func TestEmptyNotificationsDB(t *testing.T) {
 
 		notificationsdb := db.Notifications()
 
-		notificationCursor := notifications.NotificationCursor{
+		notificationCursor := notifications.Cursor{
 			Limit: 5,
 			Page:  1,
 		}

storagenode/notifications/service.go

@@ -0,0 +1,77 @@
+// Copyright (C) 2019 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package notifications
+
+import (
+	"context"
+
+	"github.com/skyrings/skyring-common/tools/uuid"
+	"go.uber.org/zap"
+	"gopkg.in/spacemonkeygo/monkit.v2"
+)
+
+var (
+	mon = monkit.Package()
+)
+
+// Service is the notification service between storage nodes and satellites.
+// architecture: Service
+type Service struct {
+	log *zap.Logger
+	db  DB
+}
+
+// NewService creates a new notification service.
+func NewService(log *zap.Logger, db DB) *Service {
+	return &Service{
+		log: log,
+		db:  db,
+	}
+}
+
+// Receive - receives notifications from satellite and Insert them into DB.
+func (service *Service) Receive(ctx context.Context, newNotification NewNotification) (Notification, error) {
+	notification, err := service.db.Insert(ctx, newNotification)
+	if err != nil {
+		return Notification{}, err
+	}
+
+	return notification, nil
+}
+
+// Read - change notification status to Read by ID.
+func (service *Service) Read(ctx context.Context, notificationID uuid.UUID) (err error) {
+	defer mon.Task()(&ctx)(&err)
+
+	err = service.db.Read(ctx, notificationID)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// ReadAll - change status of all user's notifications to Read.
+func (service *Service) ReadAll(ctx context.Context) (err error) {
+	defer mon.Task()(&ctx)(&err)
+
+	err = service.db.ReadAll(ctx)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// List - shows the list of paginated notifications.
+func (service *Service) List(ctx context.Context, cursor Cursor) (_ Page, err error) {
+	defer mon.Task()(&ctx)(&err)
+
+	notificationPage, err := service.db.List(ctx, cursor)
+	if err != nil {
+		return Page{}, err
+	}
+
+	return notificationPage, nil
+}

storagenode/peer.go

@@ -165,6 +165,10 @@ type Peer struct {
 		Chore    *gracefulexit.Chore
 	}
 
+	Notifications struct {
+		Service *notifications.Service
+	}
+
 	Bandwidth *bandwidth.Service
 }
 
@@ -209,6 +213,10 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB, revocationDB exten
 		}
 	}
 
+	{ // setup notification service.
+		peer.Notifications.Service = notifications.NewService(peer.Log, peer.DB.Notifications())
+	}
+
 	{ // setup contact service
 		c := config.Contact
 		if c.ExternalAddress == "" {
@@ -375,6 +383,7 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB, revocationDB exten
 		peer.Console.Endpoint = consoleserver.NewServer(
 			peer.Log.Named("console:endpoint"),
 			assets,
+			peer.Notifications.Service,
 			peer.Console.Service,
 			peer.Console.Listener,
 		)

storagenode/storagenodedb/notifications.go

@@ -65,7 +65,7 @@ func (db *notificationDB) Insert(ctx context.Context, notification notifications
 }
 
 // List returns listed page of notifications from database.
-func (db *notificationDB) List(ctx context.Context, cursor notifications.NotificationCursor) (_ notifications.NotificationPage, err error) {
+func (db *notificationDB) List(ctx context.Context, cursor notifications.Cursor) (_ notifications.Page, err error) {
 	defer mon.Task()(&ctx, cursor)(&err)
 
 	if cursor.Limit > 50 {
@@ -73,10 +73,10 @@ func (db *notificationDB) List(ctx context.Context, cursor notifications.Notific
 	}
 
 	if cursor.Page == 0 {
-		return notifications.NotificationPage{}, ErrNotificationsDB.Wrap(errs.New("page can not be 0"))
+		return notifications.Page{}, ErrNotificationsDB.Wrap(errs.New("page can not be 0"))
 	}
 
-	page := notifications.NotificationPage{
+	page := notifications.Page{
 		Limit:  cursor.Limit,
 		Offset: uint64((cursor.Page - 1) * cursor.Limit),
 	}
@@ -90,13 +90,13 @@ func (db *notificationDB) List(ctx context.Context, cursor notifications.Notific
 
 	err = db.QueryRowContext(ctx, countQuery).Scan(&page.TotalCount)
 	if err != nil {
-		return notifications.NotificationPage{}, ErrNotificationsDB.Wrap(err)
+		return notifications.Page{}, ErrNotificationsDB.Wrap(err)
 	}
 	if page.TotalCount == 0 {
 		return page, nil
 	}
 	if page.Offset > page.TotalCount-1 {
-		return notifications.NotificationPage{}, ErrNotificationsDB.Wrap(errs.New("page is out of range"))
+		return notifications.Page{}, ErrNotificationsDB.Wrap(errs.New("page is out of range"))
 	}
 
 	query := `
@@ -109,7 +109,7 @@ func (db *notificationDB) List(ctx context.Context, cursor notifications.Notific
 
 	rows, err := db.QueryContext(ctx, query, page.Limit, page.Offset)
 	if err != nil {
-		return notifications.NotificationPage{}, ErrNotificationsDB.Wrap(err)
+		return notifications.Page{}, ErrNotificationsDB.Wrap(err)
 	}
 
 	defer func() {
@@ -131,12 +131,12 @@ func (db *notificationDB) List(ctx context.Context, cursor notifications.Notific
 			&notification.CreatedAt,
 		)
 		if err = rows.Err(); err != nil {
-			return notifications.NotificationPage{}, ErrNotificationsDB.Wrap(err)
+			return notifications.Page{}, ErrNotificationsDB.Wrap(err)
 		}
 
 		notificationID, err = dbutil.BytesToUUID(notificationIDBytes)
 		if err != nil {
-			return notifications.NotificationPage{}, ErrNotificationsDB.Wrap(err)
+			return notifications.Page{}, ErrNotificationsDB.Wrap(err)
 		}
 
 		notification.ID = notificationID