Files
im_server/server/push/jpush/push_jpush.go
T

623 lines
16 KiB
Go

// Package jpush implements push notification plugin for JPush.
package jpush
import (
"bytes"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"os"
"strconv"
"strings"
"time"
"github.com/tinode/chat/server/logs"
"github.com/tinode/chat/server/push"
"github.com/tinode/chat/server/store"
t "github.com/tinode/chat/server/store/types"
)
const (
defaultBufferSize = 1024
defaultAPIDomain = "https://api.jpush.cn"
defaultPushPath = "/v3/push"
defaultTTL = 86400
defaultCallTTL = 60
)
var handler Handler
type Handler struct {
input chan *push.Receipt
channel chan *push.ChannelReq
stop chan bool
client *http.Client
config configType
}
type configType struct {
Enabled bool `json:"enabled"`
DryRun bool `json:"dry_run"`
AppKey string `json:"app_key"`
MasterSecret string `json:"master_secret"`
APIDomain string `json:"api_domain"`
PushPath string `json:"push_path"`
Buffer int `json:"buffer"`
DefaultTTLSeconds int `json:"default_ttl_seconds"`
CallTTLSeconds int `json:"call_ttl_seconds"`
APNsProduction bool `json:"apns_production"`
UseAlias *bool `json:"use_alias"`
Classification int `json:"classification"`
Title string `json:"title"`
}
type jpushPayload struct {
Platform any `json:"platform"`
Audience map[string][]string `json:"audience,omitempty"`
Notification *jpushNotification `json:"notification,omitempty"`
Message *jpushMessage `json:"message,omitempty"`
Options jpushOptions `json:"options"`
CID string `json:"cid,omitempty"`
}
type jpushNotification struct {
Alert string `json:"alert,omitempty"`
Android *androidPayload `json:"android,omitempty"`
IOS *iosPayload `json:"ios,omitempty"`
HMOS *hmosPayload `json:"hmos,omitempty"`
}
type androidPayload struct {
Alert string `json:"alert"`
Title string `json:"title,omitempty"`
Extras map[string]string `json:"extras,omitempty"`
Priority int `json:"priority,omitempty"`
}
type iosPayload struct {
Alert iosAlert `json:"alert"`
Sound string `json:"sound,omitempty"`
Badge string `json:"badge,omitempty"`
Extras map[string]string `json:"extras,omitempty"`
}
type iosAlert struct {
Title string `json:"title,omitempty"`
Body string `json:"body,omitempty"`
}
type hmosPayload struct {
Alert string `json:"alert"`
Title string `json:"title,omitempty"`
Category string `json:"category,omitempty"`
Extras map[string]string `json:"extras,omitempty"`
}
type jpushMessage struct {
MsgContent string `json:"msg_content"`
Title string `json:"title,omitempty"`
Extras map[string]string `json:"extras,omitempty"`
}
type jpushOptions struct {
TimeToLive int `json:"time_to_live"`
APNsProduction bool `json:"apns_production"`
Classification int `json:"classification,omitempty"`
}
type jpushResponse struct {
SendNo any `json:"sendno"`
MsgID any `json:"msg_id"`
Error *struct {
Code int `json:"code"`
Message string `json:"message"`
} `json:"error,omitempty"`
}
func (Handler) Init(jsonconf json.RawMessage) (bool, error) {
if handler.input != nil {
return false, errors.New("already initialized")
}
var config configType
if err := json.Unmarshal(jsonconf, &config); err != nil {
return false, errors.New("failed to parse config: " + err.Error())
}
loadThirdPartyConfig(&config)
if v := strings.TrimSpace(os.Getenv("IM_JPUSH_APP_KEY")); v != "" {
config.AppKey = v
config.Enabled = true
}
if v := strings.TrimSpace(os.Getenv("IM_JPUSH_MASTER_SECRET")); v != "" {
config.MasterSecret = v
config.Enabled = true
}
if v := strings.TrimSpace(os.Getenv("IM_JPUSH_API_DOMAIN")); v != "" {
config.APIDomain = v
}
if !config.Enabled {
return false, nil
}
normalized, err := normalizeConfig(config)
if err != nil {
return false, err
}
handler.input = make(chan *push.Receipt, normalized.Buffer)
handler.channel = make(chan *push.ChannelReq, normalized.Buffer)
handler.stop = make(chan bool, 1)
handler.client = &http.Client{Timeout: 10 * time.Second}
handler.config = normalized
go func() {
for {
select {
case rcpt := <-handler.input:
go sendReceipt(rcpt, &handler.config, handler.client)
case req := <-handler.channel:
logs.Info.Println("jpush: channel subscription ignored", req.Uid.UserId(), req.Channel)
case <-handler.stop:
return
}
}
}()
return true, nil
}
func normalizeConfig(config configType) (configType, error) {
config.AppKey = strings.TrimSpace(config.AppKey)
config.MasterSecret = strings.TrimSpace(config.MasterSecret)
config.APIDomain = strings.TrimRight(strings.TrimSpace(config.APIDomain), "/")
config.PushPath = strings.TrimSpace(config.PushPath)
config.Title = strings.TrimSpace(config.Title)
if config.APIDomain == "" {
config.APIDomain = defaultAPIDomain
}
if config.PushPath == "" {
config.PushPath = defaultPushPath
}
if !strings.HasPrefix(config.PushPath, "/") {
config.PushPath = "/" + config.PushPath
}
if config.Buffer <= 0 {
config.Buffer = defaultBufferSize
}
if config.DefaultTTLSeconds <= 0 {
config.DefaultTTLSeconds = defaultTTL
}
if config.CallTTLSeconds <= 0 {
config.CallTTLSeconds = defaultCallTTL
}
if config.Title == "" {
config.Title = "MeetingIM"
}
if config.AppKey == "" || config.MasterSecret == "" {
return config, errors.New("jpush: missing app_key or master_secret")
}
return config, nil
}
func sendReceipt(rcpt *push.Receipt, config *configType, client *http.Client) {
payloads, uids := preparePayloads(rcpt, config)
for i := range payloads {
if config.DryRun {
logs.Info.Println("jpush dry run:", uids[i].UserId(), payloads[i].CID)
continue
}
if err := sendPayload(payloads[i], config, client); err != nil {
logs.Warn.Println("jpush push failed:", uids[i].UserId(), err)
return
}
}
}
func sendPayload(payload *jpushPayload, config *configType, client *http.Client) error {
body, err := json.Marshal(payload)
if err != nil {
return err
}
req, err := http.NewRequest(http.MethodPost, config.APIDomain+config.PushPath, bytes.NewReader(body))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(config.AppKey+":"+config.MasterSecret)))
resp, err := client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return fmt.Errorf("http %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
}
var parsed jpushResponse
if len(respBody) > 0 {
if err = json.Unmarshal(respBody, &parsed); err != nil {
return fmt.Errorf("decode response: %w", err)
}
}
if parsed.Error != nil {
return fmt.Errorf("jpush error %d: %s", parsed.Error.Code, parsed.Error.Message)
}
if parsed.MsgID != nil && logs.Info != nil {
logs.Info.Println("jpush sent:", parsed.MsgID)
}
return nil
}
func loadThirdPartyConfig(config *configType) {
if config == nil || store.ThirdPartyConfig == nil {
return
}
values, err := store.ThirdPartyConfig.Get("im_jpush",
"jpush_enabled",
"jpush_app_key",
"jpush_master_secret",
"jpush_api_domain",
"jpush_push_path",
"jpush_default_ttl_seconds",
"jpush_call_ttl_seconds",
"jpush_apns_production",
"jpush_use_alias",
"jpush_classification",
"jpush_title",
)
if err != nil || len(values) == 0 {
return
}
if v, ok := parseConfigBool(values["jpush_enabled"]); ok {
config.Enabled = v
}
if v := strings.TrimSpace(values["jpush_app_key"]); v != "" {
config.AppKey = v
}
if v := strings.TrimSpace(values["jpush_master_secret"]); v != "" {
config.MasterSecret = v
}
if v := strings.TrimSpace(values["jpush_api_domain"]); v != "" {
config.APIDomain = v
}
if v := strings.TrimSpace(values["jpush_push_path"]); v != "" {
config.PushPath = v
}
if v, ok := parseConfigInt(values["jpush_default_ttl_seconds"]); ok {
config.DefaultTTLSeconds = v
}
if v, ok := parseConfigInt(values["jpush_call_ttl_seconds"]); ok {
config.CallTTLSeconds = v
}
if v, ok := parseConfigBool(values["jpush_apns_production"]); ok {
config.APNsProduction = v
}
if v, ok := parseConfigBool(values["jpush_use_alias"]); ok {
config.UseAlias = &v
}
if v, ok := parseConfigInt(values["jpush_classification"]); ok {
config.Classification = v
}
if v := strings.TrimSpace(values["jpush_title"]); v != "" {
config.Title = v
}
}
func parseConfigBool(value string) (bool, bool) {
switch strings.ToLower(strings.TrimSpace(value)) {
case "1", "true", "yes", "on", "enabled":
return true, true
case "0", "false", "no", "off", "disabled":
return false, true
default:
return false, false
}
}
func parseConfigInt(value string) (int, bool) {
parsed, err := strconv.Atoi(strings.TrimSpace(value))
return parsed, err == nil
}
func preparePayloads(rcpt *push.Receipt, config *configType) ([]*jpushPayload, []t.Uid) {
if rcpt == nil || config == nil {
return nil, nil
}
data, err := pushData(rcpt.Payload)
if err != nil {
logs.Warn.Println("jpush: could not build payload:", err)
return nil, nil
}
useAlias := true
if config.UseAlias != nil {
useAlias = *config.UseAlias
}
uids := sortedRecipients(rcpt.To)
if len(uids) == 0 {
return nil, nil
}
var devices map[t.Uid][]t.DeviceDef
if !useAlias {
devices, _, err = providerDevicesForUsers(rcpt.TenantID, "jpush", uids...)
if err != nil {
logs.Warn.Println("jpush: db error", err)
return nil, nil
}
}
var payloads []*jpushPayload
var targets []t.Uid
for _, uid := range uids {
recipient := rcpt.To[uid]
if recipient.Delivered > 0 {
continue
}
userData := cloneData(data)
topic := rcpt.Payload.Topic
if t.GetTopicCat(topic) == t.TopicCatP2P {
if alias, err := t.P2PNameForUser(uid, topic); err == nil {
topic = alias
userData["topic"] = alias
}
}
var audience map[string][]string
if useAlias {
audience = map[string][]string{"alias": {aliasForUid(uid)}}
} else {
var registrationIDs []string
for _, device := range devices[uid] {
if device.DeviceId != "" && !containsString(recipient.Devices, device.DeviceId) {
registrationIDs = append(registrationIDs, device.DeviceId)
}
}
if len(registrationIDs) == 0 {
continue
}
audience = map[string][]string{"registration_id": registrationIDs}
}
payloads = append(payloads, buildPayload(audience, topic, userData, recipient.Unread, config))
targets = append(targets, uid)
}
return payloads, targets
}
func pushData(pl push.Payload) (map[string]string, error) {
data := map[string]string{
"what": pl.What,
"type": "im_message",
"topic": pl.Topic,
"ts": pl.Timestamp.Format(time.RFC3339Nano),
}
if pl.From != "" {
data["sender_id"] = pl.From
data["xfrom"] = pl.From
}
if pl.Silent {
data["silent"] = "true"
}
switch pl.What {
case push.ActMsg:
data["seq"] = strconv.FormatInt(pl.SeqId, 10)
if pl.ContentType != "" {
data["mime"] = pl.ContentType
data["message_type"] = messageTypeFromMime(pl.ContentType)
}
if text := payloadContentString(pl.Content, "text"); text != "" {
data["content"] = truncateRunes(text, push.MaxPayloadLength)
}
if pl.Webrtc != "" {
data["webrtc"] = pl.Webrtc
if callID := payloadContentString(pl.Content, "callId"); callID != "" {
data["call_id"] = callID
}
if callType := payloadContentString(pl.Content, "callType"); callType == "audio" || callType == "video" {
data["call_type"] = callType
}
data["type"] = "call_invite"
if pl.Webrtc != "started" {
data["type"] = "call_missed"
}
if pl.AudioOnly || data["call_type"] == "audio" {
data["call_type"] = "audio"
data["aonly"] = "true"
} else if data["call_type"] == "" {
data["call_type"] = "video"
}
if pl.From != "" {
data["caller_id"] = pl.From
}
}
if pl.Replace != "" {
data["replace"] = pl.Replace
data["silent"] = "true"
}
case push.ActSub:
data["type"] = "system_notification"
data["modeWant"] = pl.ModeWant.String()
data["modeGiven"] = pl.ModeGiven.String()
case push.ActRead:
data["type"] = "read"
data["seq"] = strconv.FormatInt(pl.SeqId, 10)
data["silent"] = "true"
default:
return nil, errors.New("unknown push type")
}
return data, nil
}
func buildPayload(audience map[string][]string, topic string, data map[string]string, unread int, config *configType) *jpushPayload {
alert, title := alertText(data, config)
ttl := config.DefaultTTLSeconds
if data["type"] == "call_invite" {
ttl = config.CallTTLSeconds
}
payload := &jpushPayload{
Platform: "all",
Audience: audience,
Options: jpushOptions{
TimeToLive: ttl,
APNsProduction: config.APNsProduction,
Classification: config.Classification,
},
}
if data["silent"] == "true" || data["type"] == "read" {
payload.Message = &jpushMessage{MsgContent: alert, Title: title, Extras: data}
return payload
}
badge := "+1"
if unread > 0 {
badge = strconv.Itoa(unread)
}
payload.Notification = &jpushNotification{
Alert: alert,
Android: &androidPayload{
Title: title,
Alert: alert,
Extras: data,
Priority: 1,
},
IOS: &iosPayload{
Alert: iosAlert{Title: title, Body: alert},
Sound: "default",
Badge: badge,
Extras: data,
},
HMOS: &hmosPayload{
Title: title,
Alert: alert,
Category: "IM",
Extras: data,
},
}
return payload
}
func alertText(data map[string]string, config *configType) (string, string) {
title := config.Title
switch data["type"] {
case "call_invite":
if data["call_type"] == "audio" {
return "收到一通语音通话", "语音通话"
}
return "收到一通视频通话", "视频通话"
case "call_missed":
return "你有一通未接来电", "未接来电"
case "system_notification":
return "你收到一条系统通知", title
default:
return "你收到一条新消息", title
}
}
func cidFor(topic string, data map[string]string) string {
parts := []string{data["type"], topic, data["seq"], data["replace"]}
return strings.Trim(strings.Join(parts, "_"), "_")
}
func messageTypeFromMime(mime string) string {
switch {
case strings.Contains(mime, "image"):
return "image"
case strings.Contains(mime, "audio"):
return "voice"
case strings.Contains(mime, "video"):
return "video"
default:
return "text"
}
}
func payloadContentString(content any, key string) string {
if content == nil {
return ""
}
if values, ok := content.(map[string]any); ok {
if value, ok := values[key].(string); ok {
return strings.TrimSpace(value)
}
}
if values, ok := content.(map[string]string); ok {
return strings.TrimSpace(values[key])
}
return ""
}
func truncateRunes(value string, limit int) string {
if limit <= 0 || len(value) <= limit {
return value
}
runes := []rune(value)
if len(runes) <= limit {
return value
}
return string(runes[:limit]) + "…"
}
func sortedRecipients(recipients map[t.Uid]push.Recipient) []t.Uid {
uids := make([]t.Uid, 0, len(recipients))
for uid := range recipients {
if !uid.IsZero() {
uids = append(uids, uid)
}
}
for i := 1; i < len(uids); i++ {
for j := i; j > 0 && uids[j].String() < uids[j-1].String(); j-- {
uids[j], uids[j-1] = uids[j-1], uids[j]
}
}
return uids
}
func cloneData(src map[string]string) map[string]string {
dst := make(map[string]string, len(src))
for k, v := range src {
dst[k] = v
}
return dst
}
func containsString(values []string, target string) bool {
for _, value := range values {
if value == target {
return true
}
}
return false
}
func aliasForUid(uid t.Uid) string {
// JPush alias does not accept '-' or '_'. Tinode user IDs are base64url-like,
// so map them to legal characters which never appear in the original alphabet.
return strings.NewReplacer("-", ".", "_", "|").Replace(uid.UserId())
}
func providerDevicesForUsers(tenantID t.TenantID, provider string, uids ...t.Uid) (map[t.Uid][]t.DeviceDef, int, error) {
if tokens, ok := store.DevicePushTokens.(store.DevicePushTokenProviderQueryInterface); ok {
return tokens.GetAllByProvider(tenantID, provider, uids...)
}
return map[t.Uid][]t.DeviceDef{}, 0, nil
}
func (Handler) IsReady() bool {
return handler.input != nil
}
func (Handler) Push() chan<- *push.Receipt {
return handler.input
}
func (Handler) Channel() chan<- *push.ChannelReq {
return handler.channel
}
func (Handler) Stop() {
if handler.stop != nil {
handler.stop <- true
}
handler.input = nil
handler.channel = nil
handler.stop = nil
}
func init() {
push.Register("jpush", &handler)
}