426 lines
17 KiB
Go
426 lines
17 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/tinode/chat/server/auth"
|
|
"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"
|
|
realtimeKindCallEvent = "call_event"
|
|
|
|
realtimeTargetTopic = "topic"
|
|
realtimeTargetUser = "user"
|
|
realtimeTargetDevice = "device"
|
|
realtimeTargetPlatf = "platform"
|
|
)
|
|
|
|
type realtimeCallEventMessage struct {
|
|
Topic string `json:"topic"`
|
|
From string `json:"from"`
|
|
Sid string `json:"sid,omitempty"`
|
|
Event string `json:"event"`
|
|
Seq int64 `json:"seq"`
|
|
Payload json.RawMessage `json:"payload,omitempty"`
|
|
Timestamp time.Time `json:"ts,omitempty"`
|
|
}
|
|
|
|
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 publishRealtimeCallEvent(msg *ClientComMessage) {
|
|
if msg == nil || msg.Note == nil || !msg.TenantID.IsValid() ||
|
|
msg.RcptTo == "" || msg.AsUser == "" || msg.Note.What != "call" ||
|
|
!isDetachedCallEvent(msg.Note.Event) || globals.cluster != nil {
|
|
return
|
|
}
|
|
sid := ""
|
|
if msg.sess != nil {
|
|
sid = msg.sess.sid
|
|
}
|
|
event := realtimeCallEventMessage{
|
|
Topic: msg.RcptTo,
|
|
From: msg.AsUser,
|
|
Sid: sid,
|
|
Event: normalizeCallEvent(msg.Note.Event),
|
|
Seq: msg.Note.SeqId,
|
|
Payload: msg.Note.Payload,
|
|
Timestamp: msg.Timestamp,
|
|
}
|
|
dedupKey := "call:" + msg.RcptTo + ":" + msg.AsUser + ":" + event.Event + ":" + strconv.FormatInt(msg.Note.SeqId, 10)
|
|
if err := store.PublishRealtimeFanout(currentNodeID(), msg.TenantID, realtimeKindCallEvent, realtimeTargetTopic, msg.RcptTo, sid, dedupKey, event); err != nil {
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindCallEvent, realtimeTargetTopic, "error")
|
|
logs.Warn.Printf("redis realtime fanout: publish call event topic=%s from=%s event=%s seq=%d failed: %v",
|
|
msg.RcptTo, msg.AsUser, event.Event, msg.Note.SeqId, err)
|
|
return
|
|
}
|
|
immetrics.IncRealtimeFanout("publish", realtimeKindCallEvent, realtimeTargetTopic, "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
|
|
if envelope.SkipTopic != "" {
|
|
if msg.Pres != nil {
|
|
msg.Pres.SkipTopic = envelope.SkipTopic
|
|
}
|
|
if msg.Info != nil {
|
|
msg.Info.SkipTopic = envelope.SkipTopic
|
|
}
|
|
}
|
|
msg.sess = nil
|
|
prepareRedisGroupFanoutMessage(&msg, envelope.Target)
|
|
|
|
switch envelope.Kind {
|
|
case realtimeKindSessionControl:
|
|
deliverRealtimeSessionControl(envelope, &msg)
|
|
case realtimeKindCallEvent:
|
|
deliverRealtimeCallEvent(envelope)
|
|
default:
|
|
immetrics.ObserveRealtimeFanoutDelivery(envelope.Kind, envelope.TargetType, msg.Timestamp)
|
|
deliverRealtimeMessage(envelope, &msg)
|
|
}
|
|
}
|
|
|
|
func deliverRealtimeCallEvent(envelope *store.RealtimeFanoutEnvelope) {
|
|
var event realtimeCallEventMessage
|
|
if err := json.Unmarshal(envelope.Message, &event); err != nil {
|
|
immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "malformed")
|
|
logs.Warn.Println("redis realtime fanout: malformed call event", err)
|
|
return
|
|
}
|
|
topic := strings.TrimSpace(event.Topic)
|
|
if topic == "" {
|
|
topic = strings.TrimSpace(envelope.Target)
|
|
}
|
|
if topic == "" || event.From == "" || event.Seq <= 0 || !isDetachedCallEvent(event.Event) {
|
|
immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "skip_invalid")
|
|
return
|
|
}
|
|
key := types.TopicKey{TenantID: envelope.TenantID, Topic: topic}
|
|
if globals.hub == nil || globals.hub.topicGet(key) == nil {
|
|
immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "no_local_topic")
|
|
return
|
|
}
|
|
uid := types.ParseUserId(event.From)
|
|
if uid.IsZero() {
|
|
immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "skip_invalid_user")
|
|
return
|
|
}
|
|
ts := event.Timestamp
|
|
if ts.IsZero() {
|
|
ts = types.TimeNow()
|
|
}
|
|
msg := &ClientComMessage{
|
|
Note: &MsgClientNote{
|
|
Topic: topic,
|
|
What: "call",
|
|
Event: normalizeCallEvent(event.Event),
|
|
SeqId: event.Seq,
|
|
Payload: event.Payload,
|
|
},
|
|
Original: topic,
|
|
RcptTo: topic,
|
|
AsUser: event.From,
|
|
AuthLvl: int(auth.LevelAuth),
|
|
Timestamp: ts,
|
|
TenantID: envelope.TenantID,
|
|
sess: &Session{
|
|
proto: NONE,
|
|
sid: "redis-call:" + strings.TrimSpace(event.Sid),
|
|
uid: uid,
|
|
tenantID: envelope.TenantID,
|
|
authLvl: auth.LevelAuth,
|
|
lastAction: ts.UnixNano(),
|
|
},
|
|
init: true,
|
|
}
|
|
if msg.sess.sid == "redis-call:" {
|
|
msg.sess.sid = "redis-call:" + event.From
|
|
}
|
|
select {
|
|
case globals.hub.routeCli <- msg:
|
|
immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "hub_queued")
|
|
default:
|
|
immetrics.IncRealtimeFanout("deliver", realtimeKindCallEvent, realtimeTargetTopic, "hub_full")
|
|
logs.Err.Println("redis realtime fanout: hub routeCli queue full for call event", topic)
|
|
}
|
|
}
|
|
|
|
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 ""
|
|
}
|