connect/server/fcm.go

package server

import (
	"context"
	"encoding/json"
	"log"
	"time"

	firebase "firebase.google.com/go"
	"firebase.google.com/go/messaging"
	"github.com/robfig/cron/v3"
	"github.com/rotisserie/eris"
	"go.uber.org/zap"
	"google.golang.org/api/option"
	"sikey.com/websocket/models"
	"sikey.com/websocket/repositories"
)

const (
	iosFirebaseServiceAccountID = "firebase-adminsdk-uycdi@sikey-watch.iam.gserviceaccount.com"
	iosFirebaseCredentials      = `{
		"type": "service_account",
		"project_id": "sikey-watch",
		"private_key_id": "ab460af1d589188d0a8ab6c597f7daa283317838",
		"private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDKjxzFiEJkxMCw\nGhPaZDSFvOhBhYeiWpS1jKNf4lnkkQz/OlKQnK7wFBFGZJ8ZHJQGzKgkR35P9J8L\n7gpEkK3ar4yCtuZL4Vtr0CnvXeO8VJAoUwHkn0oDXTOvTWEWxmt7ROSz9qDNT1jn\n8m18+eJrP9bmqd/junm6RV0qErs0bqZKt2GN500jM67hFzhK7yrGevubr/qsQWS/\nFlpXaaGAvirOJ3tT/fRqVep0K1PUR66IOwsv510GZbvPWg6xG7NWvTOQBM5rvKKu\npHRXhe9N3HCI55wbFBTLsn/bIbxiLCvF4vpd3moFOzohstQKsAxZKcbk4NAxTxyU\nwnk1gNfzAgMBAAECggEABDoDQC3tMLwYRzRsGKEBpY7BC+j04RLhyn4CRLRs6sOK\nEWNOqLzUO6c3iw/7b+M0RYYMtiEVxiZVEUMozLB2KDuEg4+8c3Xi2feQgHGlmF7t\nWSIORo56VWKi+vPy5C5EvOA9o3Gwfkg7Ey/wT+zI6DL/UXXW+aIA+z2KSZ/1bw0c\nf2A4sUaS1OJ5yE9oTdPudO09mBGrBd4G2NigQJ9lgz+A9nh5+ECT/qZJs+Yc/gFT\nq6KhgJgVZydptbNaW9NgW0lKrtJPmxTZ3nO+u2iWpuGJGMb3tNGuR1P35OBK4UBk\nmwgrXK6LDPbuo1RMNVTBFeWf5yCAtekbgtxXXleZWQKBgQDxG3obkRwq02Wd+r/i\nbkScwGeEsnXbjIzeQsalb4FDOQX69J7GFnjAcG+23yjoiRPmA0gnGzXEsABocFoR\nJHgJGwiXN+sjFKbHOKKbLK1hGHSUZfCW+7KGx2nvUA8m4qXG3HP/XNFZQQaJrnb2\nLIE+w5GkAltR1uFVK3YCf19F2wKBgQDXEhaaaqG9+pHxTXRUmeJhruVooCs4w5eP\n9o9WUmX0w6UNgLe1AqLUFe0/zxOphD1MOUMs3FReWox6AdrRzdXlE/2gKR0NdGQG\nFiNSuhb58xnM6iMECg+im39HqCXRZRQKOeEZu7XJEPZU2+RvlM5ov00a+aif4Pfb\nEtarj9qtyQKBgQCazYeC95JcNMqDuiFFYoMPGcHdQy/EvOMdOUaNpaAa5xvd2v2u\nNXNmK0qu4W1Ej+6EugqzgRbuqAo7BBfv9bhUMFU9sht4tKO4Oba9ZtwTAT6ooSLt\n7cDcJGDx1DdGQjMqERUxGgkYA1YNREUBHeYFxE9YPGMhkpOuuW7Vf65ODQKBgCYb\nUl3x3s6mgw1aR+5lhbMBJiyvlHjuTwB3E1acKux/bdNCp0ovOWKSsALKUhWLFMFY\ntApSz7AYIyPLCFZ8PhXkwN+L6VXk9YQOkBusVT3cUQn5wlKI5dRN8PNlW64KVs6p\nCrVgiQkjNEI84/DRUPFGVAcfjT0mw0PRxq+HQZvhAoGBAIeRKHj/dceKR+pDw1vN\nTgSK61MfCAFMPSANLGYq+1lRoyDlc9ekFAZYaGXsSPy9ja3VSJkH/eu0rhUS6uYG\na7sVk/TwXoze/scHiR0JZNiKxoyoFaHJmblFdk0TC4+vwcpJFRSoQhh6gz0tZUL5\nAatOg4mA1vMm3Ll+xUY4ZzFe\n-----END PRIVATE KEY-----\n",
		"client_email": "firebase-adminsdk-uycdi@sikey-watch.iam.gserviceaccount.com",
		"client_id": "109500502903490303964",
		"auth_uri": "https://accounts.google.com/o/oauth2/auth",
		"token_uri": "https://oauth2.googleapis.com/token",
		"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
		"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/firebase-adminsdk-uycdi%40sikey-watch.iam.gserviceaccount.com",
		"universe_domain": "googleapis.com"
	  }`

	androidFirebaseServiceAccountID = "firebase-adminsdk-oasvo@sik-veryfit.iam.gserviceaccount.com"
	androidFirebaseCredentials      = `{
	   "type": "service_account",
	   "project_id": "sik-veryfit",
	   "private_key_id": "268ebd274a04ebdd14375440a193ddb1e7acec05",
	   "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQCZsf9lfcZ1/5hK\nq1CV55VVCi3q8Fi1r3Qw1UPyOAuHsHhIMCYs7wOas2Vimft43T6nMhEMAVnuZkg8\nOnM86HcECsK9wIFZGzGu7sKJQW2mJ3VEh/1PAo2L9Cakcr4K70RWKMtNV6jly5+o\nVYoxOyxNlWYGMTAc41XcjBYOPAwJFXnhbpFv/XbZMj/cFXUvB/ORpJXnDZqyVcBH\nlSxfTHl1XSXdXuPzIulIQeA5+1ZGtBBrb3fdTFdoMb8xcpMKwqILzVAlIDuwGMg4\n2MZ7pctOm9Qvqww9s/Vwhmog43MfGNslN8m5j9w7rBvA3wWbnIZ5Dl8f2VGhgnat\nEcWUC97LAgMBAAECggEAAo6FrkajkL6GlUrGK3aI+fFIP/Lmi+8vXWsfpKfGBR5A\nnO6ZYXM9Dhxj+gDg6K5VhQJLeDJJxM0sC0su4ybrQI6yvgCABKlz4nI/ZcotKU4c\n1nIQeTYASuZZzBuUib4HUEFKvEl60Db1k2KPonnP8RGdy+uFV8uTSmStIUAfGXLt\nnVoWKXvJmuT+hpKfvq2xVvYJFSoVrnrvbm9RVj9oUc2jbO3DXiNW/b73qLBgUh+4\nP+aLkspLQbLkmaz0oX6MaWHwnvOHPvVJUSqWHOssv60qeSoD+irM414YVBXaRXce\nQDEUq3Fdv5ORJzlru2iLZQoq71BHI7Ye3btV/DV7kQKBgQDPNW5p4yTMRrEK4GVE\ntCkag4cfKuBs9+wxl699jUlJJbQ7dqjHewN0zwFJmTCedph3SCBOM0JR3PpcwhAQ\n9HTuB3HxZ+vkV9w3UjDhlqOe13UvBrhjcqiQ1FAax59FfEA1vZVWAk4XcK3v+Sz8\nwtVl66haEa8+2puNmAUJtWYepQKBgQC94sVRM4220gMC9SPXr2diLt1DEMIAlbXO\noY7hAA89xRQGZYrcJw5bbUx4Dxkh24SPagmO102v+SZVUymo77bftYyB/+HWR2C2\ncIdJr304xcwxbcuPXYLm4cJOvOKVaZIYO1MaomMH4KhtasuqIc2DcjaLEoB0KrAG\ngY/aWsF8rwKBgQC/hgOom+tHZY//Hap89omHmEss49TM49vNewcaZZ13nwIYdHVM\n7MclELF/9biav8PtfR1fKsICKN7BHh77jPkHiponKIdBaHSELdRAr5xNFZma/fsw\n8KoeNCBWoGz3LQGsqq998GN+Bwi+5vJOL0hQDKJvnij9T9K37eu6LZFWgQKBgG+Z\nOKiJqQ86jISPeHpJ33Pn4SiKT8qyMkD32JZKR6rhCezcfuj2l0yKhzfEf9vTPSxg\ngK+PwFvdDXd2QfTEtfDyrVq1/Y/HMYLnzcsztV/JeEHQqqNerRFuu5k1D+IKQs+1\nBhmQOK2njrWH35zk8vj/BilkPVSIxh18xgPp9O+HAoGBAMvhSFbM5RxAtWTFsNRP\nOqY+SI0uUtcJsSWJLZGKXuYRjaT6DbAbiLTkGm4RflPXN35YydKJsiIcQhGfTkz/\nN3p0aTGz7mpmomHrKgQQ2vJaEFk4xbFctmnyepfAf1c/Odby5oYzeXSr/oSrqD3A\nJYfxCeHcusIh5jHG6OGGK5wT\n-----END PRIVATE KEY-----\n",
	   "client_email": "firebase-adminsdk-oasvo@sik-veryfit.iam.gserviceaccount.com",
	   "client_id": "108604649099796268804",
	   "auth_uri": "https://accounts.google.com/o/oauth2/auth",
	   "token_uri": "https://oauth2.googleapis.com/token",
	   "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
	   "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/firebase-adminsdk-oasvo%40sik-veryfit.iam.gserviceaccount.com",
	   "universe_domain": "googleapis.com"
	}`
)

const DefaultQueueStatus = -1
const DefaultRemainingRetries = 3

type FirebaseMessageServer struct {
	//iosApp    *firebase.App
	iosClient     *messaging.Client
	androidClient *messaging.Client
	repos         *repositories.Repositories
}

func newIosFirebaseMessageClient(ctx context.Context) *messaging.Client {
	conf := &firebase.Config{ServiceAccountID: iosFirebaseServiceAccountID}
	app, err := firebase.NewApp(ctx, conf,
		// option.WithHTTPClient(&http.Client{Timeout: 5 * time.Second}),
		option.WithCredentialsJSON([]byte(iosFirebaseCredentials)))
	if err != nil {
		log.Fatalln(err)
	}

	// firebase messaging client
	client, err := app.Messaging(ctx)
	if err != nil {
		panic(eris.Wrap(err, "ios: unable to create fcm client connect"))
	}
	return client
}

func androidFirebaseMessageClient(ctx context.Context) *messaging.Client {
	conf := &firebase.Config{ServiceAccountID: androidFirebaseServiceAccountID}
	app, err := firebase.NewApp(ctx, conf,
		// option.WithHTTPClient(&http.Client{Timeout: 5 * time.Second}),
		option.WithCredentialsJSON([]byte(androidFirebaseCredentials)))
	if err != nil {
		log.Fatalln(err)
	}

	// firebase messaging client
	client, err := app.Messaging(ctx)
	if err != nil {
		panic(eris.Wrap(err, "android: unable to create fcm client connect"))
	}
	return client
}

func NewFirebaseMessageServer(repos *repositories.Repositories) *FirebaseMessageServer {
	ctx := context.Background()
	firebaseMessageServer := &FirebaseMessageServer{
		iosClient:     newIosFirebaseMessageClient(ctx),
		androidClient: androidFirebaseMessageClient(ctx),
		repos:         repos,
	}

	c := cron.New()
	c.AddFunc("@every 5s", func() { firebaseMessageServer.queueRun(ctx) })
	c.Start()

	return firebaseMessageServer
}

func (f *FirebaseMessageServer) Send(ctx context.Context, uid string, message Message) error {
	// "newMsg": { "title": "News", "body": "click to view" },
	// "chatting": { "title": "Chat Message", "body": "click to view" },
	// "location": { "title": "Location Changed", "body": "click to view" },
	// "fence": { "title": "Electronic Fence", "body": "click to view" },
	// "notification": { "title": "New Notification","body": "click to view" },
	// "videoCall": { "title": "Call Notification", "body": "click to view" }
	var title string
	var body = "click to view"
	remainingRetries := 1
	messageType := message.MessageType()
	switch messageType {
	case MessageTypeDownChating, MessageTypeUpChating:
		title = "Chat Message"
	case MessageTypeLocation:
		title = "Location Changed"
		// 定位消息不参与 fcm
		zap.L().Info("[firebase] 定位消息不参与fcm推送", zap.String("user_id", uid))
		return nil
	case MessageTypeNotification:
		title = "New Notification"
	case MessageTypeVideoCall:
		remainingRetries = DefaultRemainingRetries
		title = "Call Notification"
		// 视频通话里的 payload 会有接通和挂断
		// 这里需要过滤掉挂断的通知消息
		if videoCall, ok := message.(*VideoCall); ok {
			if videoCall.Content.Dial == -1 {
				zap.L().Info("[firebase] 视频通话挂断通知不参与fcm推送", zap.String("user_id", uid))
				return nil
			}
		}
	default:
		title = "News"
	}

	// 加入消息到 fcm 队列
	queue := &models.FirebaseMessagingQueue{
		Title:            title,
		Body:             body,
		Receiver:         uid,
		Data:             string(serializeMessage(message)),
		RemainingRetries: remainingRetries,
	}
	if err := f.repos.FirebaseMessageQueueRepository.Create(ctx, queue); err != nil {
		zap.L().Info("[firebase] 加入队列错误", zap.Error(err), zap.String("user_id", uid))
		return eris.Wrap(err, "uanble to add fcm queue")
	}

	zap.L().Info("[firebase] 已加入推送队列", zap.String("user_id", uid), zap.Any("messageType", messageType), zap.Int("remainingRetries", remainingRetries))
	return nil
}

func (f *FirebaseMessageServer) queueRun(ctx context.Context) {
	//zap.L().Info("[firebase] queue running")
	// 每秒运行fcm队列
	queues, err := f.repos.FirebaseMessageQueueRepository.FindRetrieableQueue(ctx)
	if err != nil {
		zap.L().Error("[firebase] FindRetrieableQueue error", zap.Error(err))
		return
	}
	zap.L().Info("[firebase] queue running len ", zap.Int("queues", len(queues)))
	for _, queue := range queues {
		receiver := queue.Receiver
		firebaseToken, err := f.repos.FirebaseMessageRepository.GetFirebaseToken(ctx, receiver)
		if err != nil {
			_ = f.repos.FirebaseMessageQueueRepository.Delete(ctx, queue.ID)
			if eris.Is(err, models.ErrRecordNotFound) {
				zap.L().Info("[firebase] GetFirebaseToken NotFound 删除离线消息", zap.String("user_id", receiver), zap.Int("queue.ID", queue.ID))
				continue
			}
			zap.L().Error("[firebase] unable to find firebase message database", zap.Error(err), zap.String("user_id", receiver), zap.Int("queue.ID", queue.ID))
			return
		}

		msg := &messaging.Message{
			Token: firebaseToken.Token,
			Data: map[string]string{
				"message": queue.Data,
			},
			Android: &messaging.AndroidConfig{
				Notification: &messaging.AndroidNotification{
					Sound:    "default",
					Priority: messaging.PriorityHigh,
				},
			},
			APNS: &messaging.APNSConfig{
				Payload: &messaging.APNSPayload{
					Aps: &messaging.Aps{
						Sound: "default",
					},
				},
			},
			Notification: &messaging.Notification{
				Title: queue.Title,
				Body:  queue.Body,
			},
		}

		var client *messaging.Client
		switch firebaseToken.System {
		case "android":
			client = f.androidClient
			// 将JSON字符串转换为NatsPubMessage对象
			var message NatsPubMessage[NatsPubVideoCallMessage]
			err := json.Unmarshal([]byte(queue.Data), &message)
			if err != nil {
				zap.L().Error("[firebase] Error unmarshaling JSON: ", zap.Error(err), zap.String("user_id", receiver), zap.Int("queue.ID", queue.ID))
			} else {
				if message.Type == MessageTypeVideoCall {
					msg.Android.Notification.ChannelID = "video_call"
				}
			}
		default:
			client = f.iosClient
		}

		zap.L().Info("[firebase] Send", zap.Any("msg", msg), zap.String("user_id", receiver), zap.String("system", firebaseToken.System))
		var resultName string
		if resultName, err = client.Send(ctx, msg); err != nil {
			zap.L().Error("[firebase] 发送错误", zap.Error(err), zap.String("user_id", receiver), zap.Int("queue.ID", queue.ID))
			remainingRetries := queue.RemainingRetries - 1
			if remainingRetries == 0 {
				_ = f.repos.FirebaseMessageQueueRepository.Delete(ctx, queue.ID)
				return
			}
			queue.LastRetryTime = Now(time.Now().UTC())
			queue.RemainingRetries = remainingRetries
			_ = f.repos.FirebaseMessageQueueRepository.UpdateRetries(ctx, &queue)
			return
		}

		zap.L().Info("[firebase] Successfully",
			zap.Any("msg", msg),
			zap.String("user_id", receiver),
			zap.String("result", resultName))

		remainingRetries := queue.RemainingRetries - 1
		if remainingRetries == 0 {
			_ = f.repos.FirebaseMessageQueueRepository.Delete(ctx, queue.ID)
			return
		}
		queue.LastRetryTime = Now(time.Now().UTC())
		queue.RemainingRetries = remainingRetries
		_ = f.repos.FirebaseMessageQueueRepository.UpdateRetries(ctx, &queue)
	}
}

func Now(n time.Time) *time.Time {
	return &n
}

type NatsPubVideoCallMessage struct {
	AccountId string `json:"accountId"`
	Dial      int8   `json:"dial"` // -1: 挂断 1: 接通

	Receiver string `json:"receiver"`
	Sender   string `json:"sender"`
}

type NatsPubMessage[T NatsPubVideoCallMessage] struct {
	RequestId string `json:"requestId"`
	Type      int8   `json:"type"`
	Content   *T     `json:"content"`
}