package app import ( "bytes" "context" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "io" "log" "net/http" "strconv" "strings" "sync" "time" ) // Until now the app collected push tokens and never sent anything: with the App // in the background there is no socket, so a message simply arrived silently. // This is the other half — deciding who should be told, and telling them. // // The decision is deliberately conservative. A push goes out only to a // recipient who is not connected right now, has not muted the conversation, and // has not switched the notification off. Everyone else already sees the message // where they are looking. const ( pushTokenSafetyWindow = 5 * time.Minute pushRequestTimeout = 8 * time.Second pushPreviewRunes = 30 ) type pushTarget struct { UserID int64 Token string Platform string } type pushCredentials struct { AppID string AppKey string MasterSecret string BaseURL string } func (c pushCredentials) complete() bool { return c.AppID != "" && c.AppKey != "" && c.MasterSecret != "" } type pushAuth struct { mu sync.Mutex token string expires time.Time forKey string } func (a *App) pushSettings(ctx context.Context) (pushCredentials, bool) { if !a.configBool(ctx, "push.enabled", false) { return pushCredentials{}, false } credentials := pushCredentials{ AppID: strings.TrimSpace(a.configPlain(ctx, "push.app_id", "")), AppKey: strings.TrimSpace(a.configPlain(ctx, "push.app_key", "")), MasterSecret: strings.TrimSpace(a.configPlain(ctx, "push.master_secret", "")), BaseURL: strings.TrimRight(strings.TrimSpace(a.configPlain(ctx, "push.base_url", "https://restapi.getui.com")), "/"), } if credentials.BaseURL == "" { credentials.BaseURL = "https://restapi.getui.com" } return credentials, credentials.complete() } // token returns a cached auth token, refreshing it when it is close to expiry. // Getui allows 100 auth calls a minute, so caching is not optional. func (a *App) pushToken(ctx context.Context, credentials pushCredentials) (string, error) { a.push.mu.Lock() defer a.push.mu.Unlock() key := credentials.AppID + ":" + credentials.AppKey if a.push.forKey == key && a.push.token != "" && time.Now().Add(pushTokenSafetyWindow).Before(a.push.expires) { return a.push.token, nil } timestamp := strconv.FormatInt(time.Now().UnixMilli(), 10) digest := sha256.Sum256([]byte(credentials.AppKey + timestamp + credentials.MasterSecret)) body, err := a.pushPost(ctx, credentials, "/auth", "", map[string]string{ "sign": hex.EncodeToString(digest[:]), "timestamp": timestamp, "appkey": credentials.AppKey, }) if err != nil { return "", err } var payload struct { Code int `json:"code"` Msg string `json:"msg"` Data struct { Token string `json:"token"` ExpireTime string `json:"expire_time"` } `json:"data"` } if err = json.Unmarshal(body, &payload); err != nil { return "", fmt.Errorf("推送鉴权响应无法解析") } if payload.Code != 0 || payload.Data.Token == "" { return "", fmt.Errorf("推送鉴权失败:%s", payload.Msg) } expires := time.Now().Add(12 * time.Hour) if milliseconds, convErr := strconv.ParseInt(payload.Data.ExpireTime, 10, 64); convErr == nil && milliseconds > 0 { expires = time.UnixMilli(milliseconds) } a.push.token, a.push.expires, a.push.forKey = payload.Data.Token, expires, key return a.push.token, nil } func (a *App) pushPost(ctx context.Context, credentials pushCredentials, path, token string, payload any) ([]byte, error) { encoded, err := json.Marshal(payload) if err != nil { return nil, err } endpoint := credentials.BaseURL + "/v2/" + credentials.AppID + path request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(encoded)) if err != nil { return nil, err } request.Header.Set("Content-Type", "application/json;charset=utf-8") if token != "" { request.Header.Set("token", token) } response, err := (&http.Client{Timeout: pushRequestTimeout}).Do(request) if err != nil { return nil, fmt.Errorf("推送服务不可达") } defer response.Body.Close() body, err := io.ReadAll(io.LimitReader(response.Body, 1<<20)) if err != nil { return nil, err } if response.StatusCode < 200 || response.StatusCode >= 300 { return nil, fmt.Errorf("推送服务返回 HTTP %d", response.StatusCode) } return body, nil } // pushRecipients is the whole policy in one query: who, of these members, is // away from the app and still wants to hear about it. func (a *App) pushRecipients(ctx context.Context, conversationID, senderID int64, members []int64) []pushTarget { targets := []pushTarget{} for _, member := range members { if member == senderID || a.hub.online(member) { continue } var muted, allowed int if err := a.db.QueryRowContext(ctx, `SELECT m.muted,COALESCE(ns.im_enabled,1) FROM im_conversation_members m LEFT JOIN user_notification_settings ns ON ns.user_id=m.user_id WHERE m.conversation_id=? AND m.user_id=? AND m.status=1`, conversationID, member).Scan(&muted, &allowed); err != nil { continue } if muted == 1 || allowed != 1 { continue } rows, err := a.db.QueryContext(ctx, `SELECT push_token,platform FROM user_push_tokens WHERE user_id=? AND status=1 AND push_token<>''`, member) if err != nil { continue } for rows.Next() { target := pushTarget{UserID: member} if rows.Scan(&target.Token, &target.Platform) == nil { targets = append(targets, target) } } _ = rows.Close() } return targets } func pushPreview(messageType int, content any, showPreview bool) string { switch messageType { case messageTypeImage: return "[图片]" case messageTypeVoice: return "[语音]" } if !showPreview { return "给你发来一条消息" } data, _ := content.(map[string]any) text, _ := data["text"].(string) text = strings.TrimSpace(text) if text == "" { return "给你发来一条消息" } if runes := []rune(text); len(runes) > pushPreviewRunes { text = string(runes[:pushPreviewRunes]) + "…" } return text } // notifyOfflineMessage runs after the send has already succeeded, on its own // context: a slow or broken push service must never delay or fail a message // that is already delivered to everyone who is online. func (a *App) notifyOfflineMessage(conversationID, senderID int64, members []int64, item messageView) { credentials, ready := a.pushSettings(context.Background()) if !ready { return } go func() { ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) defer cancel() targets := a.pushRecipients(ctx, conversationID, senderID, members) if len(targets) == 0 { return } var sender string _ = a.db.QueryRowContext(ctx, `SELECT nickname FROM user_profiles WHERE user_id=?`, senderID).Scan(&sender) if strings.TrimSpace(sender) == "" { sender = "新消息" } body := pushPreview(item.Type, item.Content, a.configBool(ctx, "push.show_preview", true)) token, err := a.pushToken(ctx, credentials) if err != nil { log.Printf("push: 鉴权失败 conversation=%d: %v", conversationID, err) return } for _, target := range targets { if err := a.sendPush(ctx, credentials, token, target, sender, body, conversationID); err != nil { log.Printf("push: 下发失败 user=%d: %v", target.UserID, err) } } }() } func (a *App) sendPush(ctx context.Context, credentials pushCredentials, token string, target pushTarget, title, body string, conversationID int64) error { payload := map[string]any{ "request_id": randomToken()[:24], "settings": map[string]any{"ttl": 3600000}, "audience": map[string]any{"cid": []string{target.Token}}, "push_message": map[string]any{ "notification": map[string]any{ "title": title, "body": body, "click_type": "payload", // The client opens the conversation from this payload. "payload": fmt.Sprintf(`{"type":"conversation","conversationId":%d}`, conversationID), }, }, } raw, err := a.pushPost(ctx, credentials, "/push/single/cid", token, payload) if err != nil { return err } var result struct { Code int `json:"code"` Msg string `json:"msg"` } if err = json.Unmarshal(raw, &result); err != nil { return fmt.Errorf("推送响应无法解析") } // 10001/10002 mean the client id is unknown or retired; stop pushing to it. if result.Code == 10001 || result.Code == 10002 { _, _ = a.db.ExecContext(ctx, `UPDATE user_push_tokens SET status=0 WHERE user_id=? AND push_token=?`, target.UserID, target.Token) return nil } if result.Code != 0 { return fmt.Errorf("推送服务返回 %d:%s", result.Code, result.Msg) } return nil }