diff --git a/server/calls.go b/server/calls.go index 1e8c6b6..7d0966b 100644 --- a/server/calls.go +++ b/server/calls.go @@ -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" diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 8a377eb..c427bce 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -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=? diff --git a/server/push.go b/server/push.go index ca35d10..6f46b27 100644 --- a/server/push.go +++ b/server/push.go @@ -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 diff --git a/server/push/fcm/payload.go b/server/push/fcm/payload.go index 41258ab..f45169a 100644 --- a/server/push/fcm/payload.go +++ b/server/push/fcm/payload.go @@ -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 diff --git a/server/push/fcm/payload_test.go b/server/push/fcm/payload_test.go new file mode 100644 index 0000000..48891b8 --- /dev/null +++ b/server/push/fcm/payload_test.go @@ -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) + } +} diff --git a/server/push/jpush/push_jpush.go b/server/push/jpush/push_jpush.go index 37e8a11..34af440 100644 --- a/server/push/jpush/push_jpush.go +++ b/server/push/jpush/push_jpush.go @@ -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", diff --git a/server/push/jpush/push_jpush_test.go b/server/push/jpush/push_jpush_test.go index c95debf..4d4a77e 100644 --- a/server/push/jpush/push_jpush_test.go +++ b/server/push/jpush/push_jpush_test.go @@ -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) { diff --git a/server/push/push.go b/server/push/push.go index f28bf7b..410e681 100644 --- a/server/push/push.go +++ b/server/push/push.go @@ -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.