package main import ( "encoding/json" "strings" "sync" "time" "github.com/tinode/chat/server/logs" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" ) const imExternalBusinessEventPollInterval = time.Second var imBusinessEventDispatchLocks sync.Map func stableIMEventID(parts ...any) string { return types.StableEventID(parts...) } func enqueueIMBusinessEvent(tenantID types.TenantID, eventID, targetType, target string, actor types.Uid, eventType string, payload any) error { if tenantID.IsZero() || eventID == "" || target == "" || eventType == "" { return types.ErrMalformed } encoded, err := json.Marshal(payload) if err != nil { return err } err = store.HTTPInfrastructure.CreateEvent(tenantID, &types.BusinessEvent{ EventID: eventID, TargetType: targetType, Target: target, ActorUserID: actor, EventType: eventType, Payload: encoded, }) if err == nil { dispatchIMBusinessEvents(tenantID) } return err } func createIMSystemNotification(tenantID types.TenantID, uid types.Uid, eventID, eventType, title, summary string, action types.KVMap) error { if uid.IsZero() { return types.ErrMalformed } return store.Security.CreateSystemNotification(tenantID, &types.SystemNotification{ TenantID: tenantID, UserID: uid, EventID: eventID, EventType: eventType, Title: title, Summary: summary, Action: action, }) } func publishIMMeetingEvent(tenantID types.TenantID, actor types.Uid, meeting *types.Meeting, eventType string) error { if meeting == nil { return types.ErrMalformed } for _, participant := range meeting.Participants { uid := participant.UserID if uid.IsZero() { continue } eventID := stableIMEventID("meeting", meeting.MeetingID, meeting.Version, eventType, uid.UserId()) if uid != actor { if err := createIMSystemNotification(tenantID, uid, stableIMEventID(eventID, "notification"), eventType, "会议通知", meeting.Title, types.KVMap{"meeting_id": meeting.MeetingID}); err != nil { return err } } if err := enqueueIMBusinessEvent(tenantID, eventID, "user", uid.UserId(), actor, eventType, map[string]any{"meeting_id": meeting.MeetingID, "version": meeting.Version, "state": meeting.State}); err != nil { return err } } return nil } func publishIMRelationEvent(tenantID types.TenantID, actor, target types.Uid, stableID, eventType, title, summary string) error { if target.IsZero() { return types.ErrMalformed } eventID := stableIMEventID("relation", stableID, eventType, target.UserId()) if target != actor && title != "" { if err := createIMSystemNotification(tenantID, target, stableIMEventID(eventID, "notification"), eventType, title, summary, types.KVMap{"user_id": actor.UserId()}); err != nil { return err } } return enqueueIMBusinessEvent(tenantID, eventID, "user", target.UserId(), actor, eventType, map[string]any{"user_id": actor.UserId()}) } func dispatchIMBusinessEvents(tenantID types.TenantID) { if tenantID.IsZero() || globals.hub == nil || store.HTTPInfrastructure == nil { return } lockValue, _ := imBusinessEventDispatchLocks.LoadOrStore(tenantID, &sync.Mutex{}) lock := lockValue.(*sync.Mutex) lock.Lock() defer lock.Unlock() now := time.Now().UTC().Round(time.Millisecond) events, err := store.HTTPInfrastructure.ListPendingEvents(tenantID, 100, now) if err != nil { logs.Warn.Printf("business outbox: list tenant=%d: %v", tenantID, err) return } for _, event := range events { if event.EventType == "registration.invite_used" { var payload struct { InviteCodeID int64 `json:"invite_code_id"` } if err = json.Unmarshal(event.Payload, &payload); err != nil || payload.InviteCodeID <= 0 { logs.Warn.Printf("business outbox: malformed invite usage event=%s: %v", event.EventID, err) return } if err = store.RegistrationInvites.SyncUsage(tenantID, payload.InviteCodeID); err != nil { logs.Warn.Printf("business outbox: sync invite usage event=%s: %v", event.EventID, err) return } if err = store.HTTPInfrastructure.MarkEventDelivered(tenantID, event.EventID, now); err != nil { logs.Warn.Printf("business outbox: mark delivered event=%s: %v", event.EventID, err) return } continue } if event.EventType == "security.password_reset" { var payload struct { UserID int64 `json:"user_id"` } if err = json.Unmarshal(event.Payload, &payload); err != nil || payload.UserID <= 0 { logs.Warn.Printf("business outbox: malformed password reset event=%s: %v", event.EventID, err) } else if globals.sessionStore != nil { globals.sessionStore.EvictUser(store.EncodeUid(payload.UserID), "") } if err = store.HTTPInfrastructure.MarkEventDelivered(tenantID, event.EventID, now); err != nil { logs.Warn.Printf("business outbox: mark delivered event=%s: %v", event.EventID, err) return } continue } from := "" if !event.ActorUserID.IsZero() { from = event.ActorUserID.UserId() } payload := json.RawMessage(event.Payload) if len(payload) == 0 { payload = json.RawMessage(`{}`) } message := &ServerComMessage{ TenantID: tenantID, RcptTo: event.Target, AsUser: from, Timestamp: event.CreatedAt, Info: &MsgServerInfo{Topic: event.Target, From: from, What: "im", Event: event.EventType, Payload: payload}, } if deliverIMBusinessEventToUserSessions(tenantID, event, message) { if err = store.HTTPInfrastructure.MarkEventDelivered(tenantID, event.EventID, now); err != nil { logs.Warn.Printf("business outbox: mark delivered event=%s: %v", event.EventID, err) return } continue } select { case globals.hub.routeSrv <- message: if err = store.HTTPInfrastructure.MarkEventDelivered(tenantID, event.EventID, now); err != nil { logs.Warn.Printf("business outbox: mark delivered event=%s: %v", event.EventID, err) return } default: logs.Warn.Printf("business outbox: hub queue full event=%s target=%s", event.EventID, strings.TrimSpace(event.Target)) return } } } // startIMExternalBusinessEventPoller bridges direct database writers until the // same event contract is transported by Kafka. func startIMExternalBusinessEventPoller() func() { stop := make(chan struct{}) go func() { ticker := time.NewTicker(imExternalBusinessEventPollInterval) defer ticker.Stop() for { select { case <-ticker.C: if globals.sessionStore == nil { continue } tenantIDs := make(map[types.TenantID]struct{}) globals.sessionStore.Range(func(_ string, session *Session) bool { if session != nil && !session.tenantID.IsZero() { tenantIDs[session.tenantID] = struct{}{} } return true }) for tenantID := range tenantIDs { dispatchIMBusinessEvents(tenantID) } case <-stop: return } } }() return func() { close(stop) } } func deliverIMBusinessEventToUserSessions(tenantID types.TenantID, event types.BusinessEvent, message *ServerComMessage) bool { if event.TargetType != "user" || globals.sessionStore == nil || message == nil { return false } uid := types.ParseUserId(event.Target) if uid.IsZero() { return false } delivered := false globals.sessionStore.Range(func(_ string, session *Session) bool { if session == nil || session.uid != uid || session.tenantID != tenantID { return true } if session.queueOut(message.copy()) { delivered = true } return true }) return delivered }