package app import ( "context" "encoding/json" "fmt" "io" "net/http" "net/http/httptest" "strings" "sync" "testing" "time" ) type capturedPush struct { mu sync.Mutex paths []string auths []map[string]any sends []map[string]any } func (c *capturedPush) record(path string, body map[string]any) { c.mu.Lock() defer c.mu.Unlock() c.paths = append(c.paths, path) if strings.HasSuffix(path, "/auth") { c.auths = append(c.auths, body) return } c.sends = append(c.sends, body) } func (c *capturedPush) snapshot() ([]string, []map[string]any, []map[string]any) { c.mu.Lock() defer c.mu.Unlock() return append([]string{}, c.paths...), append([]map[string]any{}, c.auths...), append([]map[string]any{}, c.sends...) } func pushFixture(t *testing.T, reply func(path string) string) (*App, *capturedPush) { t.Helper() db := isolatedIMDatabase(t) captured := &capturedPush{} server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { raw, _ := io.ReadAll(io.LimitReader(r.Body, 1<<20)) var body map[string]any _ = json.Unmarshal(raw, &body) if !strings.HasSuffix(r.URL.Path, "/auth") { body["__token_header"] = r.Header.Get("token") } captured.record(r.URL.Path, body) w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(reply(r.URL.Path))) })) t.Cleanup(server.Close) a := &App{db: db, hub: NewHub(), config: Config{Environment: "development", JWTSecret: "isolated-im-regression-secret-only"}} for _, statement := range []string{ `INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES ('push.enabled','true','boolean',''),('push.app_id','APP123','text',''),('push.app_key','KEY456','text',''), ('push.show_preview','true','boolean',''),('membership.free_unanswered_message_limit','0','integer','') ON DUPLICATE KEY UPDATE config_value=VALUES(config_value)`, } { if _, err := db.Exec(statement); err != nil { t.Fatal(err) } } secret, err := a.encryptSecret("SECRET789") if err != nil { t.Fatal(err) } if _, err := db.Exec(`INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES('push.master_secret',?,'secret','') ON DUPLICATE KEY UPDATE config_value=VALUES(config_value),value_type='secret'`, secret); err != nil { t.Fatal(err) } if _, err := db.Exec(`INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES('push.base_url',?,'text','') ON DUPLICATE KEY UPDATE config_value=VALUES(config_value)`, server.URL); err != nil { t.Fatal(err) } return a, captured } func okReply(path string) string { if strings.HasSuffix(path, "/auth") { return `{"code":0,"msg":"success","data":{"token":"auth-token-1","expire_time":"99999999999999"}}` } return `{"code":0,"msg":"success"}` } // The App collected push tokens for months and never sent anything: a message // arriving while the app was closed made no sound at all. func TestOfflineMessagePushMySQL(t *testing.T) { a, captured := pushFixture(t, okReply) db := a.db if _, err := db.Exec(`INSERT INTO user_push_tokens(user_id,device_id,provider,push_token,platform,status) VALUES(2,'device-2','unipush','cid-of-user-2','android',1)`); err != nil { t.Fatal(err) } var created struct { ID int64 `json:"id"` } if err := json.Unmarshal(imTestCall(t, a.directConversation, 1, "POST", "/api/v1/im/conversations", `{"userId":2}`, 200), &created); err != nil { t.Fatal(err) } settle := func() { t.Helper() deadline := time.Now().Add(3 * time.Second) for time.Now().Before(deadline) { if _, _, sends := captured.snapshot(); len(sends) > 0 { return } time.Sleep(30 * time.Millisecond) } } if _, _, err := a.persistMessageContext(context.Background(), created.ID, 1, "push-1", 1, map[string]any{"text": "在吗,明天有空吗"}); err != nil { t.Fatal(err) } settle() paths, auths, sends := captured.snapshot() if len(sends) != 1 { t.Fatalf("应当下发一条推送: paths=%v", paths) } if len(auths) != 1 || auths[0]["appkey"] != "KEY456" || auths[0]["sign"] == "" { t.Fatalf("鉴权请求不正确: %v", auths) } if !strings.Contains(paths[0], "/v2/APP123/auth") || !strings.Contains(paths[1], "/v2/APP123/push/single/cid") { t.Fatalf("接口路径不正确: %v", paths) } if sends[0]["__token_header"] != "auth-token-1" { t.Fatalf("推送请求必须带鉴权 token: %v", sends[0]) } audience := sends[0]["audience"].(map[string]any)["cid"].([]any) if len(audience) != 1 || audience[0] != "cid-of-user-2" { t.Fatalf("收件人不正确: %v", audience) } notification := sends[0]["push_message"].(map[string]any)["notification"].(map[string]any) if notification["title"] != "IM测试1" || notification["body"] != "在吗,明天有空吗" { t.Fatalf("通知内容: %v", notification) } // The auth token is reused: the provider rate-limits that endpoint hard. if _, _, err := a.persistMessageContext(context.Background(), created.ID, 1, "push-2", 1, map[string]any{"text": "再问一次"}); err != nil { t.Fatal(err) } deadline := time.Now().Add(3 * time.Second) for time.Now().Before(deadline) { if _, _, sends := captured.snapshot(); len(sends) > 1 { break } time.Sleep(30 * time.Millisecond) } if _, auths, sends := captured.snapshot(); len(auths) != 1 || len(sends) != 2 { t.Fatalf("第二次不应重新鉴权: auths=%d sends=%d", len(auths), len(sends)) } } func TestPushIsSkippedForEveryoneWhoDoesNotNeedItMySQL(t *testing.T) { a, captured := pushFixture(t, okReply) db := a.db for _, statement := range []string{ `INSERT INTO user_push_tokens(user_id,device_id,provider,push_token,platform,status) VALUES(2,'device-2','unipush','cid-2','android',1)`, `INSERT INTO user_push_tokens(user_id,device_id,provider,push_token,platform,status) VALUES(3,'device-3','unipush','cid-3','android',1)`, `INSERT INTO user_push_tokens(user_id,device_id,provider,push_token,platform,status) VALUES(1,'device-1','unipush','cid-1','android',1)`, } { if _, err := db.Exec(statement); err != nil { t.Fatal(err) } } var created struct { ID int64 `json:"id"` } if err := json.Unmarshal(imTestCall(t, a.directConversation, 1, "POST", "/api/v1/im/conversations", `{"userId":2}`, 200), &created); err != nil { t.Fatal(err) } // The client message id is ASCII only, so the case name and the id differ. attempt := 0 quiet := func(reason string) { t.Helper() attempt++ if _, _, err := a.persistMessageContext(context.Background(), created.ID, 1, fmt.Sprintf("quiet-%d", attempt), 1, map[string]any{"text": "在吗"}); err != nil { t.Fatal(err) } time.Sleep(400 * time.Millisecond) if _, _, sends := captured.snapshot(); len(sends) != 0 { t.Fatalf("%s: 不应下发推送,实际 %d 条", reason, len(sends)) } } // Muted conversation. if _, err := db.Exec(`UPDATE im_conversation_members SET muted=1 WHERE conversation_id=? AND user_id=2`, created.ID); err != nil { t.Fatal(err) } quiet("会话已免打扰") if _, err := db.Exec(`UPDATE im_conversation_members SET muted=0 WHERE conversation_id=? AND user_id=2`, created.ID); err != nil { t.Fatal(err) } // Notifications switched off by the recipient. if _, err := db.Exec(`INSERT INTO user_notification_settings(user_id,im_enabled) VALUES(2,0) ON DUPLICATE KEY UPDATE im_enabled=0`); err != nil { t.Fatal(err) } quiet("用户关闭了消息通知") if _, err := db.Exec(`UPDATE user_notification_settings SET im_enabled=1 WHERE user_id=2`); err != nil { t.Fatal(err) } // The switch in the admin console. if _, err := db.Exec(`UPDATE system_configs SET config_value='false' WHERE config_key='push.enabled'`); err != nil { t.Fatal(err) } quiet("总开关关闭") if _, err := db.Exec(`UPDATE system_configs SET config_value='true' WHERE config_key='push.enabled'`); err != nil { t.Fatal(err) } // Nobody pushes to themselves, and an online recipient already has it. if _, _, err := a.persistMessageContext(context.Background(), created.ID, 2, "own-message", 1, map[string]any{"text": "我自己发的"}); err != nil { t.Fatal(err) } time.Sleep(400 * time.Millisecond) _, _, sends := captured.snapshot() if len(sends) != 1 { t.Fatalf("只应给对方发 1 条,实际 %d", len(sends)) } audience := sends[0]["audience"].(map[string]any)["cid"].([]any) if len(audience) != 1 || audience[0] != "cid-1" { t.Fatalf("推送应当只发给对方,实际收件人 %v", audience) } } func TestPushPreviewRespectsTheContentSwitch(t *testing.T) { long := strings.Repeat("很长的一句话", 20) if got := pushPreview(1, map[string]any{"text": long}, true); len([]rune(got)) != pushPreviewRunes+1 { t.Fatalf("预览应当截断: %d", len([]rune(got))) } if got := pushPreview(1, map[string]any{"text": "晚上一起吃饭吗"}, true); got != "晚上一起吃饭吗" { t.Fatalf("got %q", got) } // Locked screens are public: with the switch off nothing of the text leaks. if got := pushPreview(1, map[string]any{"text": "银行卡密码是 1234"}, false); strings.Contains(got, "1234") { t.Fatalf("关闭预览后不得泄露正文: %q", got) } if got := pushPreview(messageTypeImage, nil, true); got != "[图片]" { t.Fatalf("got %q", got) } if got := pushPreview(messageTypeVoice, nil, true); got != "[语音]" { t.Fatalf("got %q", got) } } // A client id the provider no longer knows is retired, so it is not retried on // every single message from then on. func TestRetiredPushTokenIsDisabledMySQL(t *testing.T) { a, _ := pushFixture(t, func(path string) string { if strings.HasSuffix(path, "/auth") { return `{"code":0,"msg":"success","data":{"token":"auth-token-1","expire_time":"99999999999999"}}` } return `{"code":10001,"msg":"cid is invalid"}` }) db := a.db if _, err := db.Exec(`INSERT INTO user_push_tokens(user_id,device_id,provider,push_token,platform,status) VALUES(2,'device-2','unipush','stale-cid','android',1)`); err != nil { t.Fatal(err) } var created struct { ID int64 `json:"id"` } if err := json.Unmarshal(imTestCall(t, a.directConversation, 1, "POST", "/api/v1/im/conversations", `{"userId":2}`, 200), &created); err != nil { t.Fatal(err) } if _, _, err := a.persistMessageContext(context.Background(), created.ID, 1, "stale", 1, map[string]any{"text": "在吗"}); err != nil { t.Fatal(err) } deadline := time.Now().Add(3 * time.Second) for time.Now().Before(deadline) { var status int if db.QueryRow(`SELECT status FROM user_push_tokens WHERE push_token='stale-cid'`).Scan(&status) == nil && status == 0 { return } time.Sleep(30 * time.Millisecond) } t.Fatal("失效的 cid 应当被停用") }