feat: 接入消息加密与实名认证能力

server/realtime_dispatcher.go

@@ -7,6 +7,7 @@ import (
 	"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"
@@ -16,6 +17,7 @@ import (
 const (
 	realtimeKindMessage        = "message"
 	realtimeKindSessionControl = "session_control"
+	realtimeKindCallEvent      = "call_event"
 
 	realtimeTargetTopic  = "topic"
 	realtimeTargetUser   = "user"
@@ -23,6 +25,16 @@ const (
 	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")
@@ -65,6 +77,35 @@ func publishRealtimeUserMessage(uid types.Uid, msg *ServerComMessage) {
 	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")
@@ -175,12 +216,79 @@ func deliverRealtimeFanoutEnvelope(envelope *store.RealtimeFanoutEnvelope) {
 	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")