pro/im_server
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: expire stale call pushes

Browse Source
This commit is contained in:
richjohnrambo
2026-09-14 12:11:43 +08:00
parent 4854e090ef
commit 789e662bdb
8 changed files with 164 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/calls.go
+3 -1
View File
@@ -37,6 +37,8 @@ const (
constCallEventHangUp = "hang-up"
// Message headers representing call states.
// Call is ringing.
constCallMsgStarted = "started"
// Call is established.
constCallMsgAccepted = "accepted"
// Previously establied call has successfully finished.
@@ -149,7 +151,7 @@ func normalizeCallInvite(head map[string]any, content any) (map[string]any, any,
for key, value := range head {
normalizedHead[key] = value
}
normalizedHead["webrtc"] = "started"
normalizedHead["webrtc"] = constCallMsgStarted
normalizedHead["x-im-type"] = "call"
if _, ok := normalizedHead["mime"]; !ok {
normalizedHead["mime"] = "application/json"
server/db/mysql/tenant_adapter.go
+5 -1
View File
@@ -2791,7 +2791,11 @@ func (a *adapter) TenantCallList(id t.TenantID, uid t.Uid, topic, cursor string,
func (a *adapter) finalizeStaleCalls(ctx context.Context) error {
now := t.TimeNow()
startedBefore := now.Add(-2 * time.Minute)
startedTimeout := imCallTimeoutSeconds()
if startedTimeout <= 0 {
startedTimeout = defaultCallEstablishmentTimeout
}
startedBefore := now.Add(-time.Duration(startedTimeout) * time.Second)
acceptedBefore := now.Add(-12 * time.Hour)
if _, err := a.db.ExecContext(ctx, `UPDATE im_call_records
SET status=?,end_reason=?,duration_ms=0,finished_at=?,updated_at=?
server/push.go
+8
View File
@@ -49,6 +49,14 @@ func (t *Topic) pushForData(fromUid types.Uid, data *MsgServerData, msgMarkedAsR
receipt.Payload.Webrtc = webrtc
audioOnly, _ := data.Head["aonly"].(bool)
receipt.Payload.AudioOnly = audioOnly
if webrtc == constCallMsgStarted {
timeout := imCallTimeoutSeconds()
if timeout <= 0 {
timeout = defaultCallEstablishmentTimeout
}
expiresAt := data.Timestamp.Add(time.Duration(timeout) * time.Second)
receipt.Payload.ExpiresAt = &expiresAt
}
}
if replace, found := data.Head["replace"].(string); found {
receipt.Payload.Replace = replace
server/push/fcm/payload.go
+24 -1
View File
@@ -36,6 +36,9 @@ func payloadToData(pl *push.Payload) (map[string]string, error) {
}
data["topic"] = pl.Topic
data["ts"] = pl.Timestamp.Format(time.RFC3339Nano)
if pl.ExpiresAt != nil && !pl.ExpiresAt.IsZero() {
data["expires_at"] = pl.ExpiresAt.UTC().Format(time.RFC3339Nano)
}
// Must use "xfrom" because "from" is a reserved word. Google did not bother to document it anywhere.
data["xfrom"] = pl.From
if pl.What == push.ActMsg {
@@ -89,6 +92,15 @@ func payloadToData(pl *push.Payload) (map[string]string, error) {
return data, nil
}
func pushExpired(data map[string]string) bool {
expiresAt := data["expires_at"]
if expiresAt == "" {
return false
}
parsed, err := time.Parse(time.RFC3339Nano, expiresAt)
return err == nil && !time.Now().UTC().Before(parsed)
}
func clonePayload(src map[string]string) map[string]string {
dst := make(map[string]string, len(src))
maps.Copy(dst, src)
@@ -103,6 +115,9 @@ func PrepareV1Notifications(rcpt *push.Receipt, config *configType) ([]*fcmv1.Me
logs.Warn.Println("fcm push: could not parse payload:", err)
return nil, nil
}
if pushExpired(data) {
return nil, nil
}
// Device IDs to send pushes to.
var devices map[t.Uid][]t.DeviceDef
@@ -322,6 +337,11 @@ func apnsNotificationConfig(what, topic string, data map[string]string, unread i
if config.TimeToLive > 0 {
expires = time.Now().UTC().Add(time.Duration(config.TimeToLive) * time.Second)
}
if expiresAt := data["expires_at"]; expiresAt != "" {
if parsed, err := time.Parse(time.RFC3339Nano, expiresAt); err == nil && parsed.Before(expires) {
expires = parsed
}
}
bundleId := config.ApnsBundleID
pushType := common.ApnsPushTypeAlert
priority := 10
@@ -335,7 +355,10 @@ func apnsNotificationConfig(what, topic string, data map[string]string, unread i
// a different adapter.
// pushType = common.ApnsPushTypeVoip
// bundleId += ".voip"
expires = time.Now().UTC().Add(time.Duration(voipTimeToLive) * time.Second)
callExpires := time.Now().UTC().Add(time.Duration(voipTimeToLive) * time.Second)
if expires.After(callExpires) {
expires = callExpires
}
} else if what == push.ActRead {
priority = 5
interruptionLevel = common.InterruptionLevelPassive
server/push/fcm/payload_test.go
+55
View File
@@ -0,0 +1,55 @@
package fcm
import (
"testing"
"time"
"github.com/tinode/chat/server/push"
"github.com/tinode/chat/server/store/types"
)
func TestPayloadToDataIncludesCallExpiresAt(t *testing.T) {
expiresAt := time.Date(2026, 9, 4, 10, 0, 30, 0, time.UTC)
data, err := payloadToData(&push.Payload{
What: push.ActMsg,
Topic: "p2pTopic",
From: "usrCaller",
SeqId: 30,
Webrtc: "started",
ExpiresAt: &expiresAt,
Content: map[string]any{
"txt": "call",
},
Timestamp: time.Date(2026, 9, 4, 10, 0, 0, 0, time.UTC),
})
if err != nil {
t.Fatalf("payload data: %v", err)
}
if data["expires_at"] != expiresAt.Format(time.RFC3339Nano) {
t.Fatalf("unexpected expires_at: %#v", data)
}
}
func TestPrepareV1NotificationsDropsExpiredCallInvite(t *testing.T) {
expiresAt := time.Now().Add(-time.Second)
messages, uids := PrepareV1Notifications(&push.Receipt{
TenantID: 7,
To: map[types.Uid]push.Recipient{
42: {},
},
Payload: push.Payload{
What: push.ActMsg,
Topic: "p2pTopic",
SeqId: 30,
Webrtc: "started",
ExpiresAt: &expiresAt,
Content: map[string]any{
"txt": "call",
},
Timestamp: time.Now().Add(-time.Minute),
},
}, &configType{})
if len(messages) != 0 || len(uids) != 0 {
t.Fatalf("expired call invite should be dropped: messages=%d uids=%#v", len(messages), uids)
}
}
server/push/jpush/push_jpush.go
+27 -1
View File
@@ -25,7 +25,7 @@ const (
defaultAPIDomain = "https://api.jpush.cn"
defaultPushPath = "/v3/push"
defaultTTL = 86400
defaultCallTTL = 60
defaultCallTTL = 30
)
var handler Handler
@@ -327,6 +327,9 @@ func preparePayloads(rcpt *push.Receipt, config *configType) ([]*jpushPayload, [
logs.Warn.Println("jpush: could not build payload:", err)
return nil, nil
}
if pushExpired(data) {
return nil, nil
}
useAlias := true
if config.UseAlias != nil {
useAlias = *config.UseAlias
@@ -399,6 +402,9 @@ func pushData(pl push.Payload) (map[string]string, error) {
}
if pl.Webrtc != "" {
data["webrtc"] = pl.Webrtc
if pl.ExpiresAt != nil && !pl.ExpiresAt.IsZero() {
data["expires_at"] = pl.ExpiresAt.UTC().Format(time.RFC3339Nano)
}
if callID := payloadContentString(pl.Content, "callId"); callID != "" {
data["call_id"] = callID
}
@@ -437,11 +443,31 @@ func pushData(pl push.Payload) (map[string]string, error) {
return data, nil
}
func pushExpired(data map[string]string) bool {
expiresAt := data["expires_at"]
if expiresAt == "" {
return false
}
parsed, err := time.Parse(time.RFC3339Nano, expiresAt)
return err == nil && !time.Now().UTC().Before(parsed)
}
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
if expiresAt := data["expires_at"]; expiresAt != "" {
if parsed, err := time.Parse(time.RFC3339Nano, expiresAt); err == nil {
remaining := int(time.Until(parsed).Seconds())
if remaining < ttl {
ttl = remaining
}
}
}
if ttl < 1 {
ttl = 1
}
}
payload := &jpushPayload{
Platform: "all",
server/push/jpush/push_jpush_test.go
+40 -10
View File
@@ -171,14 +171,16 @@ func TestPreparePayloadsDoesNotDropRecipientWithDeliveredSession(t *testing.T) {
}
func TestBuildPayloadUsesShortTTLForCalls(t *testing.T) {
expiresAt := time.Now().Add(30 * time.Second).UTC().Format(time.RFC3339Nano)
config := &configType{Title: "MeetingIM", DefaultTTLSeconds: 86400, CallTTLSeconds: 60}
payload := buildPayload(map[string][]string{"alias": {"usr1"}}, "p2pTopic", map[string]string{
"type": "call_invite",
"topic": "p2pTopic",
"seq": "30",
"call_type": "audio",
"type": "call_invite",
"topic": "p2pTopic",
"seq": "30",
"call_type": "audio",
"expires_at": expiresAt,
}, 0, config)
if payload.Options.TimeToLive != 60 {
if payload.Options.TimeToLive > 30 {
t.Fatalf("unexpected ttl: %d", payload.Options.TimeToLive)
}
if payload.Notification.Android.Title != "语音通话" {
@@ -187,12 +189,14 @@ func TestBuildPayloadUsesShortTTLForCalls(t *testing.T) {
}
func TestPushDataExtractsCallFields(t *testing.T) {
expiresAt := time.Date(2026, 9, 4, 10, 0, 30, 0, time.UTC)
data, err := pushData(push.Payload{
What: push.ActMsg,
Topic: "p2pTopic",
From: "usrCaller",
SeqId: 30,
Webrtc: "started",
What: push.ActMsg,
Topic: "p2pTopic",
From: "usrCaller",
SeqId: 30,
Webrtc: "started",
ExpiresAt: &expiresAt,
Content: map[string]any{
"callId": "call_1",
"callType": "audio",
@@ -205,6 +209,32 @@ func TestPushDataExtractsCallFields(t *testing.T) {
if data["type"] != "call_invite" || data["call_id"] != "call_1" || data["call_type"] != "audio" || data["caller_id"] != "usrCaller" {
t.Fatalf("unexpected call data: %#v", data)
}
if data["expires_at"] != expiresAt.Format(time.RFC3339Nano) {
t.Fatalf("unexpected expires_at: %#v", data)
}
}
func TestPreparePayloadsDropsExpiredCallInvite(t *testing.T) {
uid := types.Uid(42)
expiresAt := time.Now().Add(-time.Second)
config := &configType{Title: "MeetingIM", DefaultTTLSeconds: 86400, CallTTLSeconds: 30}
payloads, uids := preparePayloads(&push.Receipt{
TenantID: 7,
To: map[types.Uid]push.Recipient{
uid: {},
},
Payload: push.Payload{
What: push.ActMsg,
Topic: "p2pTopic",
SeqId: 30,
Webrtc: "started",
ExpiresAt: &expiresAt,
Timestamp: time.Now().Add(-time.Minute),
},
}, config)
if len(payloads) != 0 || len(uids) != 0 {
t.Fatalf("expired call invite should be dropped: payloads=%d uids=%#v", len(payloads), uids)
}
}
func TestSendPayloadUsesBasicAuth(t *testing.T) {
server/push/push.go
+2
View File
@@ -70,6 +70,8 @@ type Payload struct {
Topic string `json:"topic"`
// Timestamp of the action.
Timestamp time.Time `json:"ts"`
// Expiration time for short-lived payloads such as call invites.
ExpiresAt *time.Time `json:"expires_at,omitempty"`
// {data} notification.