feat: complete registration invitation and auth redesign

server/im_business_events.go

@@ -3,6 +3,7 @@ package main
 import (
 	"encoding/json"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/tinode/chat/server/logs"
@@ -10,6 +11,10 @@ import (
 	"github.com/tinode/chat/server/store/types"
 )
 
+const imExternalBusinessEventPollInterval = time.Second
+
+var imBusinessEventDispatchLocks sync.Map
+
 func stableIMEventID(parts ...any) string {
 	return types.StableEventID(parts...)
 }
@@ -87,6 +92,11 @@ 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 {
@@ -94,6 +104,39 @@ func dispatchIMBusinessEvents(tenantID types.TenantID) {
 		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()
@@ -118,3 +161,34 @@ func dispatchIMBusinessEvents(tenantID types.TenantID) {
 		}
 	}
 }
+
+// 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) }
+}