[BACK-3478] Add email notifications work system processors to email users on specific events.

Author: lostlevels

clinics/service.go

@@ -28,6 +28,7 @@ type Client interface {
 	ListEHREnabledClinics(ctx context.Context) ([]clinic.Clinic, error)
 	SyncEHRData(ctx context.Context, clinicID string) error
 	GetPatients(ctx context.Context, clinicId string, userToken string, params *clinic.ListPatientsParams, injectedParams url.Values) ([]clinic.Patient, error)
+	GetPatient(ctx context.Context, clinicID, patientID string) (*clinic.Patient, error)
 }
 
 type config struct {
@@ -155,7 +156,7 @@ func (d *defaultClient) SharePatientAccount(ctx context.Context, clinicID, patie
 	}
 	if response.StatusCode() == http.StatusConflict {
 		// User is already shared with the clinic
-		return d.getPatient(ctx, clinicID, patientID)
+		return d.GetPatient(ctx, clinicID, patientID)
 	}
 	if response.StatusCode() != http.StatusOK {
 		err = errors.Preparedf(ErrorCodeClinicClientFailure,
@@ -182,7 +183,7 @@ func (d *defaultClient) SyncEHRData(ctx context.Context, clinicID string) error
 	return nil
 }
 
-func (d *defaultClient) getPatient(ctx context.Context, clinicID, patientID string) (*clinic.Patient, error) {
+func (d *defaultClient) GetPatient(ctx context.Context, clinicID, patientID string) (*clinic.Patient, error) {
 	response, err := d.httpClient.GetPatientWithResponse(ctx, clinic.ClinicId(clinicID), clinic.PatientId(patientID))
 	if err != nil {
 		return nil, err

data/service/api/v1/email_notifications.go

@@ -0,0 +1,86 @@
+package v1
+
+import (
+	"net/http"
+	"time"
+
+	dataService "github.com/tidepool-org/platform/data/service"
+	"github.com/tidepool-org/platform/request"
+	serviceApi "github.com/tidepool-org/platform/service/api"
+
+	"github.com/tidepool-org/platform/work/service/emailnotificationsprocessor"
+)
+
+func EmailNotificationRoutes() []dataService.Route {
+	return []dataService.Route{
+		dataService.Post("/v1/notifications/account/claims", queueClaimAccountNotification, serviceApi.RequireServer),
+		dataService.Post("/v1/notifications/account/connections", queueConnectAccountNotification, serviceApi.RequireServer),
+		dataService.Post("/v1/notifications/device/connections/issues", sendDeviceIssuesNotification, serviceApi.RequireServer),
+	}
+}
+
+func queueClaimAccountNotification(dataServiceContext dataService.Context) {
+	res := dataServiceContext.Response()
+	req := dataServiceContext.Request()
+
+	responder := request.MustNewResponder(res, req)
+
+	var data emailnotificationsprocessor.ClaimAccountReminderData
+	if err := request.DecodeRequestBody(req.Request, &data); err != nil {
+		request.MustNewResponder(res, req).Error(http.StatusBadRequest, err)
+		return
+	}
+
+	createDetails := emailnotificationsprocessor.NewClaimAccountWorkCreate(time.Now().Add(time.Hour*24*7), data)
+	_, err := dataServiceContext.WorkClient().Create(req.Context(), createDetails)
+	if err != nil {
+		responder.Error(http.StatusInternalServerError, err)
+		return
+	}
+
+	responder.Empty(http.StatusCreated)
+}
+
+func queueConnectAccountNotification(dataServiceContext dataService.Context) {
+	res := dataServiceContext.Response()
+	req := dataServiceContext.Request()
+
+	responder := request.MustNewResponder(res, req)
+
+	var data emailnotificationsprocessor.ConnectAccountReminderData
+	if err := request.DecodeRequestBody(req.Request, &data); err != nil {
+		request.MustNewResponder(res, req).Error(http.StatusBadRequest, err)
+		return
+	}
+
+	createDetails := emailnotificationsprocessor.NewConnectAccountWorkCreate(time.Now().Add(time.Hour*24*7), data)
+	_, err := dataServiceContext.WorkClient().Create(req.Context(), createDetails)
+	if err != nil {
+		responder.Error(http.StatusInternalServerError, err)
+		return
+	}
+
+	responder.Empty(http.StatusCreated)
+}
+
+func sendDeviceIssuesNotification(dataServiceContext dataService.Context) {
+	res := dataServiceContext.Response()
+	req := dataServiceContext.Request()
+
+	responder := request.MustNewResponder(res, req)
+
+	var data emailnotificationsprocessor.DeviceConnectionIssuesData
+	if err := request.DecodeRequestBody(req.Request, &data); err != nil {
+		request.MustNewResponder(res, req).Error(http.StatusBadRequest, err)
+		return
+	}
+
+	createDetails := emailnotificationsprocessor.NewDeviceConnectionIssuesWorkCreate(data)
+	_, err := dataServiceContext.WorkClient().Create(req.Context(), createDetails)
+	if err != nil {
+		responder.Error(http.StatusInternalServerError, err)
+		return
+	}
+
+	responder.Empty(http.StatusCreated)
+}

data/service/api/v1/v1.go

@@ -33,6 +33,7 @@ func Routes() []service.Route {
 	routes = append(routes, SourcesRoutes()...)
 	routes = append(routes, SummaryRoutes()...)
 	routes = append(routes, AlertsRoutes()...)
+	routes = append(routes, EmailNotificationRoutes()...)
 	routes = append(routes, abbottServiceApiV1.Routes()...)
 
 	return routes

data/service/service/standard.go

@@ -3,17 +3,21 @@ package service
 import (
 	"context"
 	"log"
+	"net/http"
 	"os"
 
 	"github.com/IBM/sarama"
+	"github.com/kelseyhightower/envconfig"
 
 	eventsCommon "github.com/tidepool-org/go-common/events"
 
 	abbottClient "github.com/tidepool-org/platform-plugin-abbott/abbott/client"
 	abbottProvider "github.com/tidepool-org/platform-plugin-abbott/abbott/provider"
 	abbottWork "github.com/tidepool-org/platform-plugin-abbott/abbott/work"
 
+	confirmationClient "github.com/tidepool-org/hydrophone/client"
 	"github.com/tidepool-org/platform/application"
+	"github.com/tidepool-org/platform/auth"
 	"github.com/tidepool-org/platform/clinics"
 	dataDeduplicatorDeduplicator "github.com/tidepool-org/platform/data/deduplicator/deduplicator"
 	dataDeduplicatorFactory "github.com/tidepool-org/platform/data/deduplicator/factory"
@@ -29,6 +33,7 @@ import (
 	"github.com/tidepool-org/platform/errors"
 	"github.com/tidepool-org/platform/events"
 	logInternal "github.com/tidepool-org/platform/log"
+	"github.com/tidepool-org/platform/mailer"
 	metricClient "github.com/tidepool-org/platform/metric/client"
 	oauthProvider "github.com/tidepool-org/platform/oauth/provider"
 	"github.com/tidepool-org/platform/permission"
@@ -41,8 +46,12 @@ import (
 	summaryClient "github.com/tidepool-org/platform/summary/client"
 	syncTaskMongo "github.com/tidepool-org/platform/synctask/store/mongo"
 	"github.com/tidepool-org/platform/twiist"
+	"github.com/tidepool-org/platform/user"
+	userClient "github.com/tidepool-org/platform/user/client"
 	workService "github.com/tidepool-org/platform/work/service"
 	workStoreStructuredMongo "github.com/tidepool-org/platform/work/store/structured/mongo"
+
+	"github.com/tidepool-org/platform/work/service/emailnotificationsprocessor"
 )
 
 type Standard struct {
@@ -55,13 +64,16 @@ type Standard struct {
 	syncTaskStore                  *syncTaskMongo.Store
 	workStructuredStore            *workStoreStructuredMongo.Store
 	dataDeduplicatorFactory        *dataDeduplicatorFactory.Factory
-	clinicsClient                  *clinics.Client
+	clinicsClient                  clinics.Client
 	dataClient                     *Client
 	dataRawClient                  *dataRawService.Client
 	dataSourceClient               *dataSourceServiceClient.Client
+	mailerClient                   mailer.Mailer
 	summaryClient                  *summaryClient.Client
 	workClient                     *workService.Client
 	abbottClient                   *abbottClient.Client
+	userClient                     user.Client
+	confirmationClient             confirmationClient.ClientWithResponsesInterface
 	workCoordinator                *workService.Coordinator
 	userEventsHandler              events.Runner
 	twiistServiceAccountAuthorizer *twiist.ServiceAccountAuthorizer
@@ -116,9 +128,18 @@ func (s *Standard) Initialize(provider application.Provider) error {
 	if err := s.initializeDataSourceClient(); err != nil {
 		return err
 	}
+	if err := s.initializeMailerClient(); err != nil {
+		return err
+	}
+	if err := s.initializeUserClient(); err != nil {
+		return err
+	}
 	if err := s.initializeSummaryClient(); err != nil {
 		return err
 	}
+	if err := s.initializeConfirmationClient(); err != nil {
+		return err
+	}
 	if err := s.initializeWorkClient(); err != nil {
 		return err
 	}
@@ -470,7 +491,7 @@ func (s *Standard) initializeClinicsClient() error {
 	if err != nil {
 		return errors.Wrap(err, "unable to create clinics client")
 	}
-	s.clinicsClient = &clnt
+	s.clinicsClient = clnt
 
 	return nil
 }
@@ -511,6 +532,65 @@ func (s *Standard) initializeDataSourceClient() error {
 	return nil
 }
 
+func (s *Standard) initializeMailerClient() error {
+	s.Logger().Debug("Initializing mailer client")
+	client, err := mailer.Client()
+	if err != nil {
+		return errors.Wrap(err, "unable to create mailer client")
+	}
+	s.mailerClient = client
+	return nil
+}
+
+func (s *Standard) initializeUserClient() error {
+	s.Logger().Debug("Initializing user client")
+	client, err := userClient.NewDefaultClient(userClient.Params{
+		ConfigReporter: s.ConfigReporter(),
+		Logger:         s.Logger(),
+		UserAgent:      s.UserAgent(),
+	})
+	if err != nil {
+		return errors.Wrap(err, "unable to create user client")
+	}
+	s.userClient = client
+	return nil
+}
+
+type confirmationClientConfig struct {
+	ServiceAddress string `envconfig:"TIDEPOOL_CONFIRMATION_CLIENT_ADDRESS"`
+}
+
+func (c *confirmationClientConfig) Load() error {
+	return envconfig.Process("", c)
+}
+
+func (s *Standard) initializeConfirmationClient() error {
+	s.Logger().Debug("Initializing confirmation client")
+
+	cfg := &confirmationClientConfig{}
+	if err := cfg.Load(); err != nil {
+		return errors.Wrap(err, "unable to load confirmations client config")
+	}
+
+	opts := confirmationClient.WithRequestEditorFn(func(ctx context.Context, req *http.Request) error {
+		token, err := s.AuthClient().ServerSessionToken()
+		if err != nil {
+			return err
+		}
+
+		req.Header.Add(auth.TidepoolSessionTokenHeaderKey, token)
+		return nil
+	})
+
+	client, err := confirmationClient.NewClientWithResponses(cfg.ServiceAddress, opts)
+	if err != nil {
+		return errors.Wrap(err, "unable to create confirmations client")
+	}
+	s.confirmationClient = client
+
+	return nil
+}
+
 func (s *Standard) initializeSummaryClient() error {
 	s.Logger().Debug("Creating summarizer registry")
 
@@ -618,6 +698,23 @@ func (s *Standard) initializeWorkCoordinator() error {
 		return errors.Wrap(err, "unable to register abbott processors")
 	}
 
+	emailDependencies := emailnotificationsprocessor.Dependencies{
+		DataSources:   s.dataSourceStructuredStore.NewDataSourcesRepository(),
+		Mailer:        s.mailerClient,
+		Auth:          s.AuthClient(),
+		Users:         s.userClient,
+		Clinics:       s.clinicsClient,
+		Confirmations: s.confirmationClient,
+	}
+	emailNotifProcessors, err := emailnotificationsprocessor.NewProcessors(emailDependencies)
+	if err != nil {
+		return errors.Wrap(err, "unable to create email notifications processor")
+	}
+
+	if err = s.workCoordinator.RegisterProcessors(emailNotifProcessors); err != nil {
+		return errors.Wrap(err, "unable to register email notifications processor")
+	}
+
 	s.Logger().Debug("Starting work coordinator")
 
 	s.workCoordinator.Start()

work/service/coordinator.go

@@ -45,7 +45,6 @@ type Coordinator struct {
 	managerContext             context.Context
 	managerCancelFunc          context.CancelFunc
 	managerWaitGroup           sync.WaitGroup
-	timer                      *time.Timer
 }
 
 func NewCoordinator(logger log.Logger, serverSessionTokenProvider ServerSessionTokenProvider, workClient WorkClient) (*Coordinator, error) {
@@ -156,9 +155,6 @@ func (c *Coordinator) startManager() {
 	go func() {
 		defer c.managerWaitGroup.Done()
 
-		c.startTimer()
-		defer c.stopTimer()
-
 		for {
 			select {
 			case <-c.managerContext.Done(): // Drain and complete any interrupted tasks
@@ -167,13 +163,10 @@ func (c *Coordinator) startManager() {
 				}
 				return
 			case completion := <-c.workersCompletionChannel:
-				c.stopTimer()
 				c.completeWork(completion)
 				c.requestAndDispatchWork()
-				c.startTimer()
-			case <-c.timer.C:
+			case <-c.tick():
 				c.requestAndDispatchWork()
-				c.startTimer()
 			}
 		}
 	}()
@@ -331,23 +324,10 @@ func (c *Coordinator) completeWork(completion *coordinatorProcessingCompletion)
 	}
 }
 
-func (c *Coordinator) startTimer() {
+func (c *Coordinator) tick() <-chan time.Time {
 	jitter := int64(float64(c.frequency) * CoordinatorDelayJitter)
 	frequencyWithJitter := c.frequency + time.Duration(rand.Int63n(jitter*2+1)-jitter)
-	if c.timer == nil {
-		c.timer = time.NewTimer(frequencyWithJitter)
-	} else {
-		c.timer.Reset(frequencyWithJitter)
-	}
-}
-
-func (c *Coordinator) stopTimer() {
-	if c.timer != nil {
-		if !c.timer.Stop() {
-			<-c.timer.C
-		}
-		c.timer = nil
-	}
+	return time.After(frequencyWithJitter)
 }
 
 type coordinatorProcessingIdentifier struct {