im_server/server/im_business_events.go

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
}