// 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}}}}} }