310 lines
13 KiB
Go
310 lines
13 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/tinode/chat/server/logs"
|
|
immetrics "github.com/tinode/chat/server/metrics"
|
|
"github.com/tinode/chat/server/store"
|
|
"github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
const (
|
|
realtimeKindMessage = "message"
|
|
realtimeKindSessionControl = "session_control"
|
|
|
|
realtimeTargetTopic = "topic"
|
|
realtimeTargetUser = "user"
|
|
realtimeTargetDevice = "device"
|
|
realtimeTargetPlatf = "platform"
|
|
)
|
|
|
|
func startRedisRealtimeFanout(ctx context.Context) {
|
|
if err := store.StartRealtimeFanoutSubscriber(ctx, currentNodeID(), deliverRealtimeFanoutEnvelope); err != nil {
|
|
immetrics.IncRealtimeFanout("subscribe", realtimeKindMessage, realtimeTargetTopic, "error")
|
|
logs.Warn.Println("redis realtime fanout: subscriber disabled", err)
|
|
return
|
|
}
|
|
immetrics.IncRealtimeFanout("subscribe", realtimeKindMessage, realtimeTargetTopic, "ok")
|
|
logs.Info.Printf("redis realtime fanout: subscriber started node=%s", currentNodeID())
|
|
}
|
|
|
|
func publishRealtimeTopicMessage(topic string, msg *ServerComMessage) {
|
|
if msg == nil || !msg.TenantID.IsValid() || strings.TrimSpace(topic) == "" {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetTopic, "skip_invalid")
|
|
return
|
|
}
|
|
if globals.cluster != nil {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetTopic, "skip_native_cluster")
|
|
return
|
|
}
|
|
dedupKey := realtimeMessageDedupKey(topic, msg)
|
|
if err := store.PublishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindMessage, realtimeTargetTopic, topic, msg.SkipSid, dedupKey, msg); err != nil {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetTopic, "error")
|
|
logs.Warn.Printf("redis realtime fanout: publish topic=%s failed: %v", topic, err)
|
|
return
|
|
}
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetTopic, "ok")
|
|
}
|
|
|
|
func publishRealtimeUserMessage(uid types.Uid, msg *ServerComMessage) {
|
|
if msg == nil || !msg.TenantID.IsValid() || uid.IsZero() {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetUser, "skip_invalid")
|
|
return
|
|
}
|
|
dedupKey := realtimeMessageDedupKey(uid.UserId(), msg)
|
|
if err := store.PublishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindMessage, realtimeTargetUser, uid.UserId(), msg.SkipSid, dedupKey, msg); err != nil {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetUser, "error")
|
|
logs.Warn.Printf("redis realtime fanout: publish user=%s failed: %v", uid.UserId(), err)
|
|
return
|
|
}
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindMessage, realtimeTargetUser, "ok")
|
|
}
|
|
|
|
func publishRealtimeEvictUser(tenantID types.TenantID, uid types.Uid, reason string, skipSid string) {
|
|
if !tenantID.IsValid() || uid.IsZero() {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetUser, "skip_invalid")
|
|
return
|
|
}
|
|
msg := NoErrEvicted("", "", types.TimeNow())
|
|
msg.TenantID = tenantID
|
|
msg.AsUser = uid.UserId()
|
|
msg.SkipSid = skipSid
|
|
if reason != "" {
|
|
msg.Ctrl.Params = map[string]any{"reason": reason}
|
|
}
|
|
dedupKey := "evict:user:" + uid.UserId() + ":" + reason
|
|
if err := store.PublishRealtimeFanout(currentNodeID(), tenantID, realtimeKindSessionControl, realtimeTargetUser, uid.UserId(), skipSid, dedupKey, msg); err != nil {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetUser, "error")
|
|
logs.Warn.Printf("redis realtime fanout: publish evict user=%s failed: %v", uid.UserId(), err)
|
|
return
|
|
}
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetUser, "ok")
|
|
}
|
|
|
|
func publishRealtimeEvictDevice(tenantID types.TenantID, uid types.Uid, deviceID, reason string) {
|
|
if !tenantID.IsValid() || uid.IsZero() || strings.TrimSpace(deviceID) == "" {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetDevice, "skip_invalid")
|
|
return
|
|
}
|
|
msg := NoErrEvicted("", "", types.TimeNow())
|
|
msg.TenantID = tenantID
|
|
msg.AsUser = uid.UserId()
|
|
if reason != "" {
|
|
msg.Ctrl.Params = map[string]any{"reason": reason}
|
|
}
|
|
target := uid.UserId() + ":" + strings.TrimSpace(deviceID)
|
|
dedupKey := "evict:device:" + target + ":" + reason
|
|
if err := store.PublishRealtimeFanout(currentNodeID(), tenantID, realtimeKindSessionControl, realtimeTargetDevice, target, "", dedupKey, msg); err != nil {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetDevice, "error")
|
|
logs.Warn.Printf("redis realtime fanout: publish evict device=%s failed: %v", target, err)
|
|
return
|
|
}
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetDevice, "ok")
|
|
}
|
|
|
|
// publishRealtimeEvictUserPlatform evicts all sessions of the user on the given
|
|
// platform across cluster nodes. skipSid (the new login session) is part of the
|
|
// dedup key because repeated logins on the same platform within the dedup window
|
|
// must each reach the subscriber nodes.
|
|
func publishRealtimeEvictUserPlatform(tenantID types.TenantID, uid types.Uid, platf, reason, skipSid string) {
|
|
if !tenantID.IsValid() || uid.IsZero() || strings.TrimSpace(platf) == "" {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetPlatf, "skip_invalid")
|
|
return
|
|
}
|
|
msg := NoErrEvicted("", "", types.TimeNow())
|
|
msg.TenantID = tenantID
|
|
msg.AsUser = uid.UserId()
|
|
if reason != "" {
|
|
msg.Ctrl.Params = map[string]any{"reason": reason}
|
|
}
|
|
target := uid.UserId() + ":" + strings.TrimSpace(platf)
|
|
dedupKey := "evict:platform:" + target + ":" + skipSid
|
|
if err := store.PublishRealtimeFanout(currentNodeID(), tenantID, realtimeKindSessionControl, realtimeTargetPlatf, target, skipSid, dedupKey, msg); err != nil {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetPlatf, "error")
|
|
logs.Warn.Printf("redis realtime fanout: publish evict platform=%s failed: %v", target, err)
|
|
return
|
|
}
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindSessionControl, realtimeTargetPlatf, "ok")
|
|
}
|
|
|
|
// fanoutEvictReason extracts the machine-readable eviction reason from a
|
|
// session-control fanout message, if present.
|
|
func fanoutEvictReason(msg *ServerComMessage) string {
|
|
if msg == nil || msg.Ctrl == nil {
|
|
return ""
|
|
}
|
|
params, ok := msg.Ctrl.Params.(map[string]any)
|
|
if !ok {
|
|
return ""
|
|
}
|
|
reason, _ := params["reason"].(string)
|
|
return reason
|
|
}
|
|
|
|
func deliverRealtimeFanoutEnvelope(envelope *store.RealtimeFanoutEnvelope) {
|
|
if envelope == nil || !envelope.TenantID.IsValid() || len(envelope.Message) == 0 {
|
|
immetrics.IncRealtimeFanout("receive", realtimeKindMessage, realtimeTargetTopic, "skip_invalid")
|
|
return
|
|
}
|
|
immetrics.IncRealtimeFanout("receive", envelope.Kind, envelope.TargetType, "ok")
|
|
var msg ServerComMessage
|
|
if err := json.Unmarshal(envelope.Message, &msg); err != nil {
|
|
immetrics.IncRealtimeFanout("receive", envelope.Kind, envelope.TargetType, "malformed")
|
|
logs.Warn.Println("redis realtime fanout: malformed message", err)
|
|
return
|
|
}
|
|
msg.TenantID = envelope.TenantID
|
|
msg.RcptTo = envelope.Target
|
|
msg.SkipSid = envelope.SkipSid
|
|
msg.sess = nil
|
|
prepareRedisGroupFanoutMessage(&msg, envelope.Target)
|
|
|
|
switch envelope.Kind {
|
|
case realtimeKindSessionControl:
|
|
deliverRealtimeSessionControl(envelope, &msg)
|
|
default:
|
|
immetrics.ObserveRealtimeFanoutDelivery(envelope.Kind, envelope.TargetType, msg.Timestamp)
|
|
deliverRealtimeMessage(envelope, &msg)
|
|
}
|
|
}
|
|
|
|
func deliverRealtimeMessage(envelope *store.RealtimeFanoutEnvelope, msg *ServerComMessage) {
|
|
if globals.sessionStore == nil {
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_no_session_store")
|
|
return
|
|
}
|
|
switch envelope.TargetType {
|
|
case realtimeTargetUser:
|
|
uid := types.ParseUserId(envelope.Target)
|
|
if uid.IsZero() {
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_invalid_target")
|
|
return
|
|
}
|
|
delivered := globals.sessionStore.DeliverToUser(envelope.TenantID, uid, envelope.SkipSid, msg)
|
|
recordRealtimeFanoutDeliverResult(envelope.Kind, envelope.TargetType, delivered)
|
|
case realtimeTargetTopic, "":
|
|
deliverRealtimeTopicMessage(envelope, msg)
|
|
default:
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_unknown_target_type")
|
|
}
|
|
}
|
|
|
|
func deliverRealtimeTopicMessage(envelope *store.RealtimeFanoutEnvelope, msg *ServerComMessage) {
|
|
topic := strings.TrimSpace(envelope.Target)
|
|
if topic == "" {
|
|
topic = strings.TrimSpace(msg.RcptTo)
|
|
}
|
|
if topic == "" {
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, realtimeTargetTopic, "skip_invalid_target")
|
|
return
|
|
}
|
|
msg.RcptTo = topic
|
|
key := types.TopicKey{TenantID: envelope.TenantID, Topic: topic}
|
|
if globals.hub != nil && globals.hub.topicGet(key) != nil {
|
|
select {
|
|
case globals.hub.routeSrv <- msg:
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, realtimeTargetTopic, "hub_queued")
|
|
default:
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, realtimeTargetTopic, "hub_full")
|
|
logs.Err.Println("redis realtime fanout: hub routeSrv queue full", topic)
|
|
}
|
|
return
|
|
}
|
|
if types.GetTopicCat(topic) == types.TopicCatP2P {
|
|
delivered := globals.sessionStore.DeliverToP2PTopic(envelope.TenantID, topic, envelope.SkipSid, msg)
|
|
recordRealtimeFanoutDeliverResult(envelope.Kind, realtimeTargetTopic, delivered)
|
|
return
|
|
}
|
|
delivered := globals.sessionStore.DeliverToTopicSubscribers(envelope.TenantID, topic, envelope.SkipSid, msg)
|
|
recordRealtimeFanoutDeliverResult(envelope.Kind, realtimeTargetTopic, delivered)
|
|
}
|
|
|
|
// fanoutSessionControlStaleLimit caps how long after publication a
|
|
// session-control (evict) fanout message remains actionable. Delayed replays
|
|
// (consumer backlog, pending-entry redelivery after restart) carry the original
|
|
// skipSid, so a replayed evict would otherwise terminate freshly established
|
|
// sessions whose sid differs from the stale skipSid (新登录被旧消息误踢).
|
|
const fanoutSessionControlStaleLimit = 10 * time.Second
|
|
|
|
// fanoutSessionControlStale reports whether the session-control fanout message
|
|
// is too old to act on. Messages without a timestamp (unknown origin) are
|
|
// delivered to stay compatible with legacy publishers.
|
|
func fanoutSessionControlStale(msg *ServerComMessage) bool {
|
|
if msg == nil || msg.Ctrl == nil || msg.Ctrl.Timestamp.IsZero() {
|
|
return false
|
|
}
|
|
return time.Since(msg.Ctrl.Timestamp) > fanoutSessionControlStaleLimit
|
|
}
|
|
|
|
func deliverRealtimeSessionControl(envelope *store.RealtimeFanoutEnvelope, msg *ServerComMessage) {
|
|
if globals.sessionStore == nil {
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_no_session_store")
|
|
return
|
|
}
|
|
if fanoutSessionControlStale(msg) {
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_stale")
|
|
logs.Warn.Printf("redis realtime fanout: dropped stale session-control message target=%s older=%s", envelope.Target, fanoutSessionControlStaleLimit)
|
|
return
|
|
}
|
|
switch envelope.TargetType {
|
|
case realtimeTargetDevice:
|
|
uidText, deviceID, ok := strings.Cut(envelope.Target, ":")
|
|
if !ok {
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_invalid_target")
|
|
return
|
|
}
|
|
globals.sessionStore.EvictUserDevice(types.ParseUserId(uidText), deviceID, fanoutEvictReason(msg))
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted")
|
|
case realtimeTargetPlatf:
|
|
uidText, platf, ok := strings.Cut(envelope.Target, ":")
|
|
if !ok {
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_invalid_target")
|
|
return
|
|
}
|
|
globals.sessionStore.EvictUserPlatform(types.ParseUserId(uidText), platf, envelope.SkipSid, fanoutEvictReason(msg))
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted")
|
|
case realtimeTargetUser:
|
|
uid := types.ParseUserId(envelope.Target)
|
|
if uid.IsZero() {
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "skip_invalid_target")
|
|
return
|
|
}
|
|
globals.sessionStore.EvictUser(uid, envelope.SkipSid, fanoutEvictReason(msg))
|
|
immetrics.IncRealtimeFanout("deliver", envelope.Kind, envelope.TargetType, "evicted")
|
|
}
|
|
_ = msg
|
|
}
|
|
|
|
func recordRealtimeFanoutDeliverResult(kind, targetType string, delivered int) {
|
|
if delivered > 0 {
|
|
immetrics.IncRealtimeFanout("deliver", kind, targetType, "local_delivered")
|
|
return
|
|
}
|
|
immetrics.IncRealtimeFanout("deliver", kind, targetType, "no_local_session")
|
|
}
|
|
|
|
func realtimeMessageDedupKey(target string, msg *ServerComMessage) string {
|
|
if msg == nil {
|
|
return ""
|
|
}
|
|
if msg.Data != nil && msg.Data.SeqId > 0 {
|
|
topic := target
|
|
if msg.Data.Topic != "" {
|
|
topic = msg.Data.Topic
|
|
}
|
|
return "data:" + topic + ":" + strconv.FormatInt(msg.Data.SeqId, 10)
|
|
}
|
|
if msg.Info != nil {
|
|
return "info:" + target + ":" + msg.Info.What + ":" + msg.Info.From + ":" + msg.Info.Event + ":" + strconv.FormatInt(msg.Info.SeqId, 10)
|
|
}
|
|
if msg.Pres != nil {
|
|
return "pres:" + target + ":" + msg.Pres.What + ":" + strconv.FormatInt(msg.Pres.SeqId, 10)
|
|
}
|
|
return ""
|
|
}
|