89 lines
3.1 KiB
Go
89 lines
3.1 KiB
Go
// Package notifications adapts the internal notification wire contract to product models.
|
|
package notifications
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"crypto/x509"
|
|
"os"
|
|
"time"
|
|
|
|
pb "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1"
|
|
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
)
|
|
|
|
type Config struct {
|
|
Address string `yaml:"address"`
|
|
CAFile string `yaml:"ca_file"`
|
|
ServerName string `yaml:"server_name"`
|
|
ActivationURL string `yaml:"activation_url"`
|
|
}
|
|
type Message struct {
|
|
Key, UserID, Email, Locale, Template string
|
|
Parameters map[string]string
|
|
ExpiresAt time.Time
|
|
}
|
|
type Sender interface {
|
|
Send(context.Context, Message) error
|
|
}
|
|
type Client struct {
|
|
conn *grpc.ClientConn
|
|
rpc pb.NotificationsServiceClient
|
|
}
|
|
|
|
func Open(c Config) (*Client, error) {
|
|
if c.Address == "" {
|
|
return nil, merrors.InvalidArgument("notification address is required")
|
|
}
|
|
tlsConfig := &tls.Config{MinVersion: tls.VersionTLS12, ServerName: c.ServerName}
|
|
if c.CAFile != "" {
|
|
pem, err := os.ReadFile(c.CAFile)
|
|
if err != nil {
|
|
return nil, merrors.InternalWrap(err, "read notification CA")
|
|
}
|
|
pool, err := x509.SystemCertPool()
|
|
if err != nil {
|
|
pool = x509.NewCertPool()
|
|
}
|
|
if !pool.AppendCertsFromPEM(pem) {
|
|
return nil, merrors.InvalidArgument("invalid notification CA")
|
|
}
|
|
tlsConfig.RootCAs = pool
|
|
}
|
|
var transport credentials.TransportCredentials = insecure.NewCredentials()
|
|
if c.CAFile != "" || c.ServerName != "" {
|
|
transport = credentials.NewTLS(tlsConfig)
|
|
}
|
|
conn, err := grpc.NewClient(c.Address, grpc.WithTransportCredentials(transport))
|
|
if err != nil {
|
|
return nil, merrors.InternalWrap(err, "create notification client")
|
|
}
|
|
return &Client{conn: conn, rpc: pb.NewNotificationsServiceClient(conn)}, nil
|
|
}
|
|
func (c *Client) Close() error { return c.conn.Close() }
|
|
func (c *Client) Send(ctx context.Context, m Message) error {
|
|
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
|
defer cancel()
|
|
res, err := c.rpc.Send(ctx, request(m))
|
|
if err != nil {
|
|
return merrors.ServiceUnavailableWrap(err, "notification queue unavailable")
|
|
}
|
|
if res.GetNotification().GetId() == "" || len(res.GetNotification().GetDeliveries()) != 1 {
|
|
return merrors.ServiceUnavailable("notification queue returned no receipt")
|
|
}
|
|
for _, d := range res.Notification.Deliveries {
|
|
switch d.State {
|
|
case pb.DeliveryState_DELIVERY_STATE_FAILED, pb.DeliveryState_DELIVERY_STATE_EXPIRED, pb.DeliveryState_DELIVERY_STATE_UNKNOWN:
|
|
return merrors.ServiceUnavailable("notification could not be delivered")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
func request(m Message) *pb.SendRequest {
|
|
return &pb.SendRequest{IdempotencyKey: m.Key, UserId: m.UserID, TemplateId: m.Template, TemplateVersion: "1", Locale: m.Locale, Parameters: m.Parameters, ExpiresAt: timestamppb.New(m.ExpiresAt), Targets: []*pb.Target{{Destination: &pb.Target_Email{Email: &pb.EmailTarget{Address: m.Email}}}}}
|
|
}
|