♻️ Use dealer postman instead of built-in feature to deliver email and notify

.idea/workspace.xml

@@ -5,11 +5,17 @@
   </component>
   <component name="ChangeListManager">
     <list default="true" id="3fefb2c4-b6f9-466b-a523-53352e8d6f95" name="更改" comment=":sparkles: Support stream controller event emit">
-      <change afterPath="$PROJECT_DIR$/pkg/internal/grpc/stream.go" afterDir="false" />
       <change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
       <change beforePath="$PROJECT_DIR$/go.mod" beforeDir="false" afterPath="$PROJECT_DIR$/go.mod" afterDir="false" />
       <change beforePath="$PROJECT_DIR$/go.sum" beforeDir="false" afterPath="$PROJECT_DIR$/go.sum" afterDir="false" />
-      <change beforePath="$PROJECT_DIR$/pkg/internal/grpc/server.go" beforeDir="false" afterPath="$PROJECT_DIR$/pkg/internal/grpc/server.go" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/pkg/internal/services/external_apns.go" beforeDir="false" />
+      <change beforePath="$PROJECT_DIR$/pkg/internal/services/external_firebase.go" beforeDir="false" />
+      <change beforePath="$PROJECT_DIR$/pkg/internal/services/factors.go" beforeDir="false" afterPath="$PROJECT_DIR$/pkg/internal/services/factors.go" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/pkg/internal/services/mailer.go" beforeDir="false" />
+      <change beforePath="$PROJECT_DIR$/pkg/internal/services/notifications.go" beforeDir="false" afterPath="$PROJECT_DIR$/pkg/internal/services/notifications.go" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/pkg/internal/services/tokens.go" beforeDir="false" afterPath="$PROJECT_DIR$/pkg/internal/services/tokens.go" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/pkg/main.go" beforeDir="false" afterPath="$PROJECT_DIR$/pkg/main.go" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/settings.toml" beforeDir="false" afterPath="$PROJECT_DIR$/settings.toml" afterDir="false" />
     </list>
     <option name="SHOW_DIALOG" value="false" />
     <option name="HIGHLIGHT_CONFLICTS" value="true" />
@@ -33,7 +39,7 @@
     <option name="RECENT_GIT_ROOT_PATH" value="$PROJECT_DIR$" />
   </component>
   <component name="ProblemsViewState">
-    <option name="selectedTabId" value="CurrentFile" />
+    <option name="selectedTabId" value="ProjectErrors" />
   </component>
   <component name="ProjectColorInfo">{
   &quot;customColor&quot;: &quot;&quot;,

go.mod

@@ -6,7 +6,7 @@ toolchain go1.22.1
 
 require (
 	firebase.google.com/go v3.13.0+incompatible
-	git.solsynth.dev/hydrogen/dealer v0.0.0-20240720114704-037fc8a18c60
+	git.solsynth.dev/hydrogen/dealer v0.0.0-20240721055146-d74cdddbaf49
 	git.solsynth.dev/hydrogen/paperclip v0.0.0-20240622051057-0f56dba45745
 	github.com/go-playground/validator/v10 v10.17.0
 	github.com/gofiber/fiber/v2 v2.52.4

go.sum

@@ -23,6 +23,8 @@ git.solsynth.dev/hydrogen/dealer v0.0.0-20240719153055-607eba001f65 h1:p9PIsp5Ry
 git.solsynth.dev/hydrogen/dealer v0.0.0-20240719153055-607eba001f65/go.mod h1:oPdUxLy6TFeRxiRC/BoNb3YUNcnSiOnJrzFTnCPSoCA=
 git.solsynth.dev/hydrogen/dealer v0.0.0-20240720114704-037fc8a18c60 h1:cy58ybsaMHX8lVoCa3bsWkwQGxD4sPmMAMsVYg42kqU=
 git.solsynth.dev/hydrogen/dealer v0.0.0-20240720114704-037fc8a18c60/go.mod h1:oPdUxLy6TFeRxiRC/BoNb3YUNcnSiOnJrzFTnCPSoCA=
+git.solsynth.dev/hydrogen/dealer v0.0.0-20240721055146-d74cdddbaf49 h1:DMmCBcnCO0qcER/p4EQ04CmWleb4YI3Br6QK5F8Q628=
+git.solsynth.dev/hydrogen/dealer v0.0.0-20240721055146-d74cdddbaf49/go.mod h1:IZd94qZZIj+MO9EqjGDqnAD9nWurlNPyhVPKemAY5lw=
 git.solsynth.dev/hydrogen/paperclip v0.0.0-20240622051057-0f56dba45745 h1:40BUsQMNXjqHyytkyF9py1HjTAWlRgO6R57YXUrHNy4=
 git.solsynth.dev/hydrogen/paperclip v0.0.0-20240622051057-0f56dba45745/go.mod h1:FsQGSLTl0gvo+9Jmbot02S72suyF9tFTrzDj70Xhifo=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=

pkg/internal/services/external_apns.go

@@ -1,25 +0,0 @@
-package services
-
-import (
-	"github.com/sideshow/apns2"
-	"github.com/sideshow/apns2/token"
-	"github.com/spf13/viper"
-)
-
-// ExtAPNS is Apple Notification Services client
-var ExtAPNS *apns2.Client
-
-func SetupAPNS() error {
-	authKey, err := token.AuthKeyFromFile(viper.GetString("apns_credentials"))
-	if err != nil {
-		return err
-	}
-
-	ExtAPNS = apns2.NewTokenClient(&token.Token{
-		AuthKey: authKey,
-		KeyID:   viper.GetString("apns_credentials_key"),
-		TeamID:  viper.GetString("apns_credentials_team"),
-	}).Production()
-
-	return nil
-}

pkg/internal/services/external_firebase.go

@@ -1,23 +0,0 @@
-package services
-
-import (
-	"context"
-	firebase "firebase.google.com/go"
-	"github.com/spf13/viper"
-	"google.golang.org/api/option"
-)
-
-// ExtFire is the firebase app client
-var ExtFire *firebase.App
-
-func SetupFirebase() error {
-	opt := option.WithCredentialsFile(viper.GetString("firebase_credentials"))
-	app, err := firebase.NewApp(context.Background(), nil, opt)
-	if err != nil {
-		return err
-	} else {
-		ExtFire = app
-	}
-
-	return nil
-}

pkg/internal/services/factors.go

@@ -1,10 +1,14 @@
 package services
 
 import (
+	"context"
 	"fmt"
+	"git.solsynth.dev/hydrogen/dealer/pkg/proto"
+	"git.solsynth.dev/hydrogen/passport/pkg/internal/gap"
 	"github.com/rs/zerolog/log"
 	"github.com/samber/lo"
 	"strings"
+	"time"
 
 	"git.solsynth.dev/hydrogen/passport/pkg/internal/database"
 	"git.solsynth.dev/hydrogen/passport/pkg/internal/models"
@@ -81,7 +85,17 @@ func GetFactorCode(factor models.AuthFactor) (bool, error) {
 
 		subject := fmt.Sprintf("[%s] Login verification code", viper.GetString("name"))
 		content := fmt.Sprintf(EmailPasswordTemplate, user.Name, factor.Secret, viper.GetString("maintainer"))
-		if err := SendMail(user.GetPrimaryEmail().Content, subject, content); err != nil {
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_, err := proto.NewPostmanClient(gap.H.GetDealerGrpcConn()).DeliverEmail(ctx, &proto.DeliverEmailRequest{
+			To: user.GetPrimaryEmail().Content,
+			Email: &proto.EmailRequest{
+				Subject:  subject,
+				TextBody: &content,
+			},
+		})
+		if err != nil {
 			log.Warn().Err(err).Uint("factor", factor.ID).Msg("Failed to delivery one-time-password via mail...")
 			return true, nil
 		}

pkg/internal/services/mailer.go

@@ -1,51 +0,0 @@
-package services
-
-import (
-	"crypto/tls"
-	"fmt"
-	"net/smtp"
-	"net/textproto"
-
-	"github.com/jordan-wright/email"
-	"github.com/spf13/viper"
-)
-
-func SendMail(target string, subject string, content string) error {
-	mail := &email.Email{
-		To:      []string{target},
-		From:    viper.GetString("mailer.name"),
-		Subject: subject,
-		Text:    []byte(content),
-		Headers: textproto.MIMEHeader{},
-	}
-	return mail.SendWithTLS(
-		fmt.Sprintf("%s:%d", viper.GetString("mailer.smtp_host"), viper.GetInt("mailer.smtp_port")),
-		smtp.PlainAuth(
-			"",
-			viper.GetString("mailer.username"),
-			viper.GetString("mailer.password"),
-			viper.GetString("mailer.smtp_host"),
-		),
-		&tls.Config{ServerName: viper.GetString("mailer.smtp_host")},
-	)
-}
-
-func SendMailHTML(target string, subject string, content string) error {
-	mail := &email.Email{
-		To:      []string{target},
-		From:    viper.GetString("mailer.name"),
-		Subject: subject,
-		HTML:    []byte(content),
-		Headers: textproto.MIMEHeader{},
-	}
-	return mail.SendWithTLS(
-		fmt.Sprintf("%s:%d", viper.GetString("mailer.smtp_host"), viper.GetInt("mailer.smtp_port")),
-		smtp.PlainAuth(
-			"",
-			viper.GetString("mailer.username"),
-			viper.GetString("mailer.password"),
-			viper.GetString("mailer.smtp_host"),
-		),
-		&tls.Config{ServerName: viper.GetString("mailer.smtp_host")},
-	)
-}

pkg/internal/services/notifications.go

@@ -3,20 +3,16 @@ package services
 import (
 	"context"
 	"fmt"
+	jsoniter "github.com/json-iterator/go"
+	"github.com/samber/lo"
 	"reflect"
-	"sync"
 	"time"
 
 	"git.solsynth.dev/hydrogen/dealer/pkg/proto"
 	"git.solsynth.dev/hydrogen/passport/pkg/internal/gap"
 
-	"firebase.google.com/go/messaging"
 	"git.solsynth.dev/hydrogen/passport/pkg/internal/database"
 	"git.solsynth.dev/hydrogen/passport/pkg/internal/models"
-	"github.com/rs/zerolog/log"
-	"github.com/sideshow/apns2"
-	payload2 "github.com/sideshow/apns2/payload"
-	"github.com/spf13/viper"
 )
 
 func AddNotifySubscriber(user models.Account, provider, id, tk, ua string) (models.NotificationSubscriber, error) {
@@ -54,7 +50,6 @@ func NewNotification(notification models.Notification) error {
 	if err := database.C.Save(&notification).Error; err != nil {
 		return err
 	}
-
 	if err := PushNotification(notification); err != nil {
 		return err
 	}
@@ -71,11 +66,6 @@ func NewNotificationBatch(notifications []models.Notification) error {
 	return nil
 }
 
-// PushNotification will push the notification whatever it exists record in the
-// database Recommend pushing another goroutine when you need to push a lot of
-// notifications And just use a block statement when you just push one
-// notification.
-// The time of creating a new subprocess is much more than push notification.
 func PushNotification(notification models.Notification) error {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
@@ -102,89 +92,85 @@ func PushNotification(notification models.Notification) error {
 		return err
 	}
 
+	var providers []string
+	var tokens []string
 	for _, subscriber := range subscribers {
-		switch subscriber.Provider {
+		providers = append(providers, subscriber.Provider)
-		case models.NotifySubscriberFirebase:
+		tokens = append(tokens, subscriber.DeviceToken)
-			if ExtFire != nil {
-				ctx := context.Background()
-				client, err := ExtFire.Messaging(ctx)
-				if err != nil {
-					log.Warn().Err(err).Msg("An error occurred when creating FCM client...")
-					break
-				}
-
-				var image string
-				if notification.Picture != nil {
-					image = *notification.Picture
-				}
-				message := &messaging.Message{
-					Notification: &messaging.Notification{
-						Title:    notification.Title,
-						Body:     notification.Body,
-						ImageURL: image,
-					},
-					Token: subscriber.DeviceToken,
-				}
-
-				if response, err := client.Send(ctx, message); err != nil {
-					log.Warn().Err(err).Msg("An error occurred when notify subscriber via FCM...")
-				} else {
-					log.Debug().
-						Str("response", response).
-						Int("subscriber", int(subscriber.ID)).
-						Msg("Notified subscriber via FCM.")
-				}
-			}
-		case models.NotifySubscriberAPNs:
-			if ExtAPNS != nil {
-				data := payload2.
-					NewPayload().
-					AlertTitle(notification.Title).
-					AlertBody(notification.Body).
-					Sound("default").
-					Category(notification.Topic).
-					MutableContent()
-				if notification.Avatar != nil {
-					data = data.Custom("avatar_url", *notification.Avatar)
-				}
-				if notification.Picture != nil {
-					data = data.Custom("picture_url", *notification.Picture)
-				}
-				rawData, err := data.MarshalJSON()
-				if err != nil {
-					log.Warn().Err(err).Msg("An error occurred when preparing to notify subscriber via APNs...")
-				}
-				payload := &apns2.Notification{
-					ApnsID:      subscriber.DeviceID,
-					DeviceToken: subscriber.DeviceToken,
-					Topic:       viper.GetString("apns_topic"),
-					Payload:     rawData,
-				}
-
-				if resp, err := ExtAPNS.Push(payload); err != nil {
-					log.Warn().Err(err).Msg("An error occurred when notify subscriber via APNs...")
-				} else {
-					log.Debug().
-						Str("reason", resp.Reason).
-						Int("status", resp.StatusCode).
-						Int("subscriber", int(subscriber.ID)).
-						Msg("Notified subscriber via APNs.")
-				}
-			}
-		}
 	}
 
-	return nil
+	metadata, _ := jsoniter.Marshal(notification.Metadata)
+
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	_, err = proto.NewPostmanClient(gap.H.GetDealerGrpcConn()).DeliverNotificationBatch(ctx, &proto.DeliverNotificationBatchRequest{
+		Providers:    providers,
+		DeviceTokens: tokens,
+		Notify: &proto.NotifyRequest{
+			Topic:       notification.Topic,
+			Title:       notification.Title,
+			Subtitle:    notification.Subtitle,
+			Body:        notification.Body,
+			Metadata:    metadata,
+			Avatar:      notification.Avatar,
+			Picture:     notification.Picture,
+			IsRealtime:  notification.IsRealtime,
+			IsForcePush: notification.IsForcePush,
+		},
+	})
+
+	return err
 }
 
 func PushNotificationBatch(notifications []models.Notification) {
-	var wg sync.WaitGroup
+	accountIdx := lo.Map(notifications, func(item models.Notification, index int) uint {
+		return item.AccountID
+	})
+	var subscribers []models.NotificationSubscriber
+	database.C.Where("account_id IN ?", accountIdx).Find(&subscribers)
+
+	stream := proto.NewStreamControllerClient(gap.H.GetDealerGrpcConn())
 	for _, notification := range notifications {
-		wg.Add(1)
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-		item := notification
+		_, _ = stream.PushStream(ctx, &proto.PushStreamRequest{
-		go func() {
+			UserId: uint64(notification.AccountID),
-			_ = PushNotification(item)
+			Body: models.UnifiedCommand{
-			wg.Done()
+				Action:  "notifications.new",
-		}()
+				Payload: notification,
+			}.Marshal(),
+		})
+		cancel()
+
+		// Skip push notification
+		if GetStatusDisturbable(notification.AccountID) != nil {
+			continue
+		}
+
+		var providers []string
+		var tokens []string
+		for _, subscriber := range subscribers {
+			providers = append(providers, subscriber.Provider)
+			tokens = append(tokens, subscriber.DeviceToken)
+		}
+
+		metadata, _ := jsoniter.Marshal(notification.Metadata)
+
+		ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+		_, _ = proto.NewPostmanClient(gap.H.GetDealerGrpcConn()).DeliverNotificationBatch(ctx, &proto.DeliverNotificationBatchRequest{
+			Providers:    providers,
+			DeviceTokens: tokens,
+			Notify: &proto.NotifyRequest{
+				Topic:       notification.Topic,
+				Title:       notification.Title,
+				Subtitle:    notification.Subtitle,
+				Body:        notification.Body,
+				Metadata:    metadata,
+				Avatar:      notification.Avatar,
+				Picture:     notification.Picture,
+				IsRealtime:  notification.IsRealtime,
+				IsForcePush: notification.IsForcePush,
+			},
+		})
+		cancel()
 	}
 }

pkg/internal/services/tokens.go

@@ -1,7 +1,10 @@
 package services
 
 import (
+	"context"
 	"fmt"
+	"git.solsynth.dev/hydrogen/dealer/pkg/proto"
+	"git.solsynth.dev/hydrogen/passport/pkg/internal/gap"
 	"strings"
 	"time"
 
@@ -113,5 +116,14 @@ func NotifyMagicToken(token models.MagicToken) error {
 		return fmt.Errorf("unsupported magic token type to notify")
 	}
 
-	return SendMail(user.GetPrimaryEmail().Content, subject, content)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	_, err := proto.NewPostmanClient(gap.H.GetDealerGrpcConn()).DeliverEmail(ctx, &proto.DeliverEmailRequest{
+		To: user.GetPrimaryEmail().Content,
+		Email: &proto.EmailRequest{
+			Subject:  subject,
+			TextBody: &content,
+		},
+	})
+	return err
 }

pkg/main.go

@@ -46,12 +46,6 @@ func main() {
 	if err := gap.RegisterService(); err != nil {
 		log.Error().Err(err).Msg("An error occurred when registering service to gateway...")
 	}
-	if err := services.SetupFirebase(); err != nil {
-		log.Error().Err(err).Msg("An error occurred when connecting Firebase...")
-	}
-	if err := services.SetupAPNS(); err != nil {
-		log.Error().Err(err).Msg("An error occurred when connecting APNs...")
-	}
 
 	// Server
 	go server.NewServer().Listen()

settings.toml

@@ -9,12 +9,6 @@ domain = "localhost"
 
 content_endpoint = "https://usercontent.solsynth.dev"
 
-apns_topic = "dev.solsynth.solian.Runner"
-apns_credentials = ""
-apns_credentials_team = "000000000"
-apns_credentials_key = "000000000"
-firebase_credentials = ""
-
 use_registration_magic_token = false
 
 [debug]
@@ -24,13 +18,6 @@ print_routes = false
 [dealer]
 addr = "127.0.0.1:7442"
 
-[mailer]
-name = "Alphabot <alphabot@smartsheep.studio>"
-smtp_host = "smtp.exmail.qq.com"
-smtp_port = 465
-username = "alphabot@smartsheep.studio"
-password = "gz937Zxxzfcd9SeH"
-
 [security]
 cookie_domain = "localhost"
 cookie_samesite = "Lax"