111 lines
3.4 KiB
Go
111 lines
3.4 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"net"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.sendico.io/tech/sxchnge/notifications/bus"
|
|
"git.sendico.io/tech/sxchnge/notifications/email"
|
|
pb "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1"
|
|
"git.sendico.io/tech/sxchnge/notifications/service"
|
|
"git.sendico.io/tech/sxchnge/notifications/store"
|
|
"git.sendico.io/tech/sxchnge/notifications/templates"
|
|
"git.sendico.io/tech/sxchnge/notifications/transport"
|
|
db "git.sendico.io/tech/sxchnge/shared/database"
|
|
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
|
mlogger "git.sendico.io/tech/sxchnge/shared/logging"
|
|
msgruntime "git.sendico.io/tech/sxchnge/shared/messaging/runtime"
|
|
"go.uber.org/zap"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials"
|
|
)
|
|
|
|
func Run(ctx context.Context, c Config, logger mlogger.Logger) error {
|
|
catalog, err := templates.Load(c.Templates)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
sender, err := email.New(c.Email)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
options, err := serverOptions(c)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
client, err := db.ConnectMongo(logger, &c.Database)
|
|
if err != nil {
|
|
return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "connect notifications database")
|
|
}
|
|
defer func() {
|
|
cleanup, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
_ = client.Disconnect(cleanup)
|
|
}()
|
|
repo, err := store.New(logger, client.Database())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
rt, err := msgruntime.New(logger, &c.Messaging)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rt.Close()
|
|
svc := service.New(repo, catalog, sender, string(c.Email.Driver), logger)
|
|
broker, err := bus.New(ctx, rt.Durable, rt.DurableAdmin, repo, svc, logger, c.Queue)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer broker.Close()
|
|
server := grpc.NewServer(append(options, grpc.MaxRecvMsgSize(128<<10), grpc.MaxConcurrentStreams(64))...)
|
|
pb.RegisterNotificationsServiceServer(server, transport.New(svc))
|
|
listener, err := net.Listen("tcp", c.Listen)
|
|
if err != nil {
|
|
return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "listen for notifications")
|
|
}
|
|
defer listener.Close()
|
|
runCtx, stop := context.WithCancel(ctx)
|
|
defer stop()
|
|
var wg sync.WaitGroup
|
|
for _, run := range []func(context.Context){broker.RunPublisher, broker.RunWorker} {
|
|
wg.Add(1)
|
|
go func() { defer wg.Done(); run(runCtx) }()
|
|
}
|
|
serveErr := make(chan error, 1)
|
|
go func() { serveErr <- server.Serve(listener) }()
|
|
logger.Info("Notifications service started", zap.String("listen", c.Listen), zap.String("email_driver", string(c.Email.Driver)))
|
|
select {
|
|
case <-ctx.Done():
|
|
case err = <-serveErr:
|
|
}
|
|
stop()
|
|
stopped := make(chan struct{})
|
|
go func() { server.GracefulStop(); close(stopped) }()
|
|
select {
|
|
case <-stopped:
|
|
case <-time.After(5 * time.Second):
|
|
server.Stop()
|
|
}
|
|
wg.Wait()
|
|
if err != nil {
|
|
return merrors.OpaqueWrap(merrors.ErrInternal, err, "notification server stopped unexpectedly")
|
|
}
|
|
logger.Info("Notifications service stopped")
|
|
return nil
|
|
}
|
|
|
|
// An omitted certificate pair uses plaintext gRPC on the private service network.
|
|
func serverOptions(c Config) ([]grpc.ServerOption, error) {
|
|
if c.TLSCert == "" && c.TLSKey == "" {
|
|
return nil, nil
|
|
}
|
|
cert, err := tls.LoadX509KeyPair(c.TLSCert, c.TLSKey)
|
|
if err != nil {
|
|
return nil, merrors.OpaqueWrap(merrors.ErrInvalidArg, err, "invalid gRPC TLS certificate")
|
|
}
|
|
return []grpc.ServerOption{grpc.Creds(credentials.NewTLS(&tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12}))}, nil
|
|
}
|