CORE-1120: completed implementation of new notifications service.

api/main.go

@@ -1,19 +1,23 @@
 package api
 
 import (
+	"database/sql"
 	"net/http"
 
 	v1 "github.com/cyverse-de/notifications/api/v1"
 	v2 "github.com/cyverse-de/notifications/api/v2"
 	"github.com/cyverse-de/notifications/common"
 	"github.com/cyverse-de/notifications/model"
 	"github.com/labstack/echo"
+	"gopkg.in/cyverse-de/messaging.v7"
 )
 
 // API defines the REST API of the notifications service
 type API struct {
 	Echo         *echo.Echo
 	AMQPSettings *common.AMQPSettings
+	AMQPClient   *messaging.Client
+	DB           *sql.DB
 	Service      string
 	Title        string
 	Version      string
@@ -36,8 +40,11 @@ func (a API) RegisterHandlers() {
 	// Register the group for API version 1.
 	v1Group := a.Echo.Group("/v1")
 	v1API := v1.API{
+		Echo:         a.Echo,
 		Group:        v1Group,
 		AMQPSettings: a.AMQPSettings,
+		AMQPClient:   a.AMQPClient,
+		DB:           a.DB,
 		Service:      a.Service,
 		Title:        a.Title,
 		Version:      a.Version,
@@ -47,8 +54,11 @@ func (a API) RegisterHandlers() {
 	// Register the group for API version 2.
 	v2Group := a.Echo.Group("/v2")
 	v2API := v2.API{
+		Echo:         a.Echo,
 		Group:        v2Group,
 		AMQPSettings: a.AMQPSettings,
+		AMQPClient:   a.AMQPClient,
+		DB:           a.DB,
 		Service:      a.Service,
 		Title:        a.Title,
 		Version:      a.Version,

api/main_docs.go

@@ -39,3 +39,7 @@ type errorResponseWrapper struct {
 	// in:body
 	Body model.ErrorResponse
 }
+
+// An empty response body.
+// swagger:response emptyResponse
+type emptyResponseWrapper struct{}

api/v1/listings.go

@@ -0,0 +1,193 @@
+package v1
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/cyverse-de/notifications/db"
+	"github.com/cyverse-de/notifications/model"
+	"github.com/cyverse-de/notifications/query"
+	"github.com/labstack/echo"
+)
+
+// GetMessagesHandler handles requests for listing notification messages for a user.
+func (a API) GetMessagesHandler(ctx echo.Context) error {
+	var err error
+
+	// Extract and validate the user query parameter.
+	user, err := query.ValidatedQueryParam(ctx, "user", "required")
+	if err != nil {
+		return ctx.JSON(http.StatusBadRequest, model.ErrorResponse{
+			Message: "missing required query parameter: user",
+		})
+	}
+
+	// Extract and validate the limit query parameter.
+	defaultLimit := uint64(0)
+	limit, err := query.ValidateUIntQueryParam(ctx, "limit", &defaultLimit)
+	if err != nil {
+		return ctx.JSON(http.StatusBadRequest, model.ErrorResponse{
+			Message: err.Error(),
+		})
+	}
+
+	// Extract and validate the offset query parameter.
+	defaultOffset := uint64(0)
+	offset, err := query.ValidateUIntQueryParam(ctx, "offset", &defaultOffset)
+	if err != nil {
+		return ctx.JSON(http.StatusBadRequest, model.ErrorResponse{
+			Message: err.Error(),
+		})
+	}
+
+	// Extract and validate the seen query parameter.
+	seen, err := query.ValidateBoolPQueryParam(ctx, "seen")
+	if err != nil {
+		return ctx.JSON(http.StatusBadRequest, model.ErrorResponse{
+			Message: err.Error(),
+		})
+	}
+
+	// Extract and validate the sort_dir query parameter.
+	defaultSortOrder := query.SortOrderDescending
+	sortOrderParam := query.NewSortOrderParam(&defaultSortOrder)
+	err = query.ValidateParseableParam(ctx, "sort_dir", sortOrderParam)
+	if err != nil {
+		return ctx.JSON(http.StatusBadRequest, model.ErrorResponse{
+			Message: err.Error(),
+		})
+	}
+
+	// Extract and validate the sort_field query parameter.
+	defaultSortField := query.V1ListingSortFieldTimestamp
+	sortFieldParam := query.NewV1ListingSortFieldParam(&defaultSortField)
+	err = query.ValidateParseableParam(ctx, "sort_field", sortFieldParam)
+	if err != nil {
+		return ctx.JSON(http.StatusBadRequest, model.ErrorResponse{
+			Message: err.Error(),
+		})
+	}
+
+	// Extract and reformat the filter.
+	var notificationType string
+	filter := strings.ReplaceAll(strings.ToLower(ctx.QueryParam("filter")), " ", "_")
+	if filter == "new" {
+		seen = new(bool)
+		*seen = false
+	} else if filter != "" {
+		notificationType = filter
+	}
+
+	// Start a transaction.
+	tx, err := a.DB.Begin()
+	if err != nil {
+		a.Echo.Logger.Error(err)
+		return err
+	}
+	defer tx.Rollback()
+
+	// Obtain the listing.
+	params := &db.V1NotificationListingParameters{
+		User:             user,
+		Limit:            limit,
+		Offset:           offset,
+		Seen:             seen,
+		SortOrder:        *(sortOrderParam.GetValue().(*query.SortOrder)),
+		SortField:        *(sortFieldParam.GetValue().(*query.V1ListingSortField)),
+		NotificationType: notificationType,
+	}
+	listing, err := db.V1ListNotifications(tx, params)
+	if err != nil {
+		a.Echo.Logger.Error(err)
+		return err
+	}
+
+	return ctx.JSON(http.StatusOK, listing)
+}
+
+// GetUnseenMessagesHandler handles requests for listing messages that haven't been marked as seen yet for a
+// user.
+func (a *API) GetUnseenMessagesHandler(ctx echo.Context) error {
+	var err error
+
+	// Extract and validate the user query parameter.
+	user, err := query.ValidatedQueryParam(ctx, "user", "required")
+	if err != nil {
+		return ctx.JSON(http.StatusBadRequest, model.ErrorResponse{
+			Message: "missing required query parameter: user",
+		})
+	}
+
+	// Start a transaction.
+	tx, err := a.DB.Begin()
+	if err != nil {
+		a.Echo.Logger.Error(err)
+		return err
+	}
+	defer tx.Rollback()
+
+	// Obtain the listing.
+	seen := false
+	params := &db.V1NotificationListingParameters{
+		User:             user,
+		Limit:            0,
+		Offset:           0,
+		Seen:             &seen,
+		SortOrder:        query.SortOrderAscending,
+		SortField:        query.V1ListingSortFieldTimestamp,
+		NotificationType: "",
+	}
+	listing, err := db.V1ListNotifications(tx, params)
+	if err != nil {
+		a.Echo.Logger.Error(err)
+		return err
+	}
+
+	return ctx.JSON(http.StatusOK, listing)
+}
+
+// CountMessagesHandler handles requests for counting notification messages.
+func (a *API) CountMessagesHandler(ctx echo.Context) error {
+	var err error
+
+	// Extract and validate the user query parameter.
+	user, err := query.ValidatedQueryParam(ctx, "user", "required")
+	if err != nil {
+		return ctx.JSON(http.StatusBadRequest, model.ErrorResponse{
+			Message: "missing required query parameter: user",
+		})
+	}
+
+	// Extract and validate the seen query parameter.
+	seen, err := query.ValidateBoolPQueryParam(ctx, "seen")
+	if err != nil {
+		return ctx.JSON(http.StatusBadRequest, model.ErrorResponse{
+			Message: err.Error(),
+		})
+	}
+
+	// Extract and reformat the notification type filter.
+	notificationType := strings.ReplaceAll(strings.ToLower(ctx.QueryParam("filter")), " ", "_")
+
+	// Start a transaction.
+	tx, err := a.DB.Begin()
+	if err != nil {
+		a.Echo.Logger.Error(err)
+		return err
+	}
+	defer tx.Rollback()
+
+	// Obtain the counts.
+	params := &db.V1NotificationCountingParameters{
+		User:             user,
+		Seen:             seen,
+		NotificationType: notificationType,
+	}
+	result, err := db.V1CountNotifications(tx, params)
+	if err != nil {
+		a.Echo.Logger.Error(err)
+		return err
+	}
+
+	return ctx.JSON(http.StatusOK, result)
+}

api/v1/main.go

@@ -1,23 +1,28 @@
 package v1
 
 import (
+	"database/sql"
 	"net/http"
 
 	"github.com/cyverse-de/notifications/common"
 	"github.com/cyverse-de/notifications/model"
 	"github.com/labstack/echo"
+	"gopkg.in/cyverse-de/messaging.v7"
 )
 
 // API defines version 1 of the REST API for the notifications service.
 type API struct {
+	Echo         *echo.Echo
 	Group        *echo.Group
 	AMQPSettings *common.AMQPSettings
+	AMQPClient   *messaging.Client
+	DB           *sql.DB
 	Service      string
 	Title        string
 	Version      string
 }
 
-// RootHandler handles GET requests to the /v1/ endpoint.
+// RootHandler handles GET requests to the /v1 endpoint.
 func (a API) RootHandler(ctx echo.Context) error {
 	resp := model.VersionRootResponse{
 		Service:    a.Service,
@@ -32,4 +37,12 @@ func (a API) RootHandler(ctx echo.Context) error {
 func (a API) RegisterHandlers() {
 	a.Group.GET("", a.RootHandler)
 	a.Group.GET("/", a.RootHandler)
+	a.Group.POST("/notification", a.NotificationRequestHandler)
+	a.Group.GET("/messages", a.GetMessagesHandler)
+	a.Group.GET("/unseen-messages", a.GetUnseenMessagesHandler)
+	a.Group.GET("/count-messages", a.CountMessagesHandler)
+	a.Group.POST("/seen", a.MarkMessagesAsSeen)
+	a.Group.POST("/mark-all-seen", a.MarkAllMessagesAsSeen)
+	a.Group.POST("/delete", a.DeleteMessages)
+	a.Group.DELETE("/delete-all", a.DeleteMatchingMessages)
 }