fix: align group management APIs

server/db/adapter.go

@@ -259,9 +259,6 @@ type TenantHTTPInfrastructureAdapter interface {
 	TenantHTTPIdempotencyBegin(t.TenantID, *t.HTTPIdempotencyRecord) (*t.HTTPIdempotencyRecord, bool, error)
 	TenantHTTPIdempotencyComplete(t.TenantID, *t.HTTPIdempotencyRecord) error
 	TenantHTTPIdempotencyAbort(t.TenantID, *t.HTTPIdempotencyRecord) error
-	TenantBusinessEventCreate(t.TenantID, *t.BusinessEvent) error
-	TenantBusinessEventListPending(t.TenantID, int, time.Time) ([]t.BusinessEvent, error)
-	TenantBusinessEventMarkDelivered(t.TenantID, string, time.Time) error
 	TenantKafkaConsumeBegin(t.TenantID, *t.KafkaConsumeRecord, time.Time) (*t.KafkaConsumeRecord, bool, error)
 	TenantKafkaConsumeSucceed(t.TenantID, string, string, time.Time) error
 	TenantKafkaConsumeFail(t.TenantID, string, string, string, string, time.Time) (*t.KafkaConsumeRecord, error)
@@ -375,6 +372,7 @@ type TenantBusinessAdapter interface {
 	TenantGroupSettingsCreate(t.TenantID, *t.GroupSettings) (*t.GroupSettings, error)
 	TenantGroupSettingsGet(t.TenantID, string) (*t.GroupSettings, error)
 	TenantGroupSettingsUpdate(t.TenantID, string, t.GroupSettingsPatch) (*t.GroupSettings, error)
+	TenantGroupDissolve(t.TenantID, string, t.Uid, string, time.Time) (*t.GroupSettings, error)
 	TenantGroupMemberUpsert(t.TenantID, *t.GroupMember) (*t.GroupMember, error)
 	TenantGroupMemberGet(t.TenantID, string, t.Uid) (*t.GroupMember, error)
 	TenantGroupMemberList(t.TenantID, string, t.GroupMemberListOptions) ([]t.GroupMember, string, error)
@@ -400,8 +398,6 @@ type TenantBusinessAdapter interface {
 	TenantGroupRoleUpdate(t.TenantID, string, t.Uid, t.Uid, string, t.AccessMode, time.Time) error
 	TenantGroupOwnerTransfer(t.TenantID, string, t.Uid, t.Uid, int, time.Time) error
 	TenantGroupCanPost(t.TenantID, string, t.Uid, bool, time.Time) (bool, error)
-	TenantGroupEventListPending(t.TenantID, int, time.Time) ([]t.GroupEvent, error)
-	TenantGroupEventMarkDelivered(t.TenantID, string, time.Time) error
 
 	TenantWorkbenchModules(t.TenantID, t.Uid) ([]t.WorkbenchModule, int, error)
 	TenantOrganizationDepartments(t.TenantID, t.Uid, string, string, int) ([]t.OrganizationDepartment, string, error)

server/db/mysql/infrastructure_schema.go

@@ -17,18 +17,6 @@ func ensureIMHTTPInfrastructureTables(exec sqlExecutor) error {
 			UNIQUE INDEX uk_im_http_idempotency_scope(user_id, method, route, idempotency_key),
 			INDEX idx_im_http_idempotency_expiry(expires_at)
 		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
-		`CREATE TABLE IF NOT EXISTS im_business_events_outbox (
-			id BIGINT NOT NULL AUTO_INCREMENT, event_id VARCHAR(64) NOT NULL,
-			target_type VARCHAR(16) NOT NULL, target VARCHAR(128) NOT NULL,
-			actor_user_id BIGINT NOT NULL DEFAULT 0, event_type VARCHAR(64) NOT NULL,
-			payload JSON, status VARCHAR(16) NOT NULL DEFAULT 'pending', attempts INT NOT NULL DEFAULT 0,
-			next_retry_at DATETIME(3), delivered_at DATETIME(3),
-			created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
-			updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
-			PRIMARY KEY(id), UNIQUE INDEX uk_im_business_events_outbox_event(event_id),
-			INDEX idx_im_business_events_outbox_pending(status, next_retry_at, id),
-			INDEX idx_im_business_events_outbox_target(target_type, target, id)
-		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
 		`CREATE TABLE IF NOT EXISTS im_kafka_consume_log (
 			id BIGINT NOT NULL AUTO_INCREMENT, event_id VARCHAR(96) NOT NULL,
 			consumer_group VARCHAR(128) NOT NULL, source_topic VARCHAR(128) NOT NULL,

server/db/mysql/infrastructure_schema_test.go

@@ -12,13 +12,12 @@ func TestEnsureIMHTTPInfrastructureTablesUsesSelectedTenantDatabase(t *testing.T
 	if err := ensureIMHTTPInfrastructureTables(exec); err != nil {
 		t.Fatal(err)
 	}
-	if len(exec.statements) != 7 {
-		t.Fatalf("created %d infrastructure tables, want 7", len(exec.statements))
+	if len(exec.statements) != 6 {
+		t.Fatalf("created %d infrastructure tables, want 6", len(exec.statements))
 	}
 	joined := strings.ToLower(strings.Join(exec.statements, "\n"))
 	for _, table := range []string{
 		"im_http_idempotency",
-		"im_business_events_outbox",
 		"im_kafka_consume_log",
 		"im_kafka_dead_letters",
 		"im_realname_appeals",
@@ -35,4 +34,7 @@ func TestEnsureIMHTTPInfrastructureTablesUsesSelectedTenantDatabase(t *testing.T
 	if strings.Contains(joined, "im_kafka_outbox") {
 		t.Fatalf("kafka outbox table must not be created: %s", joined)
 	}
+	if strings.Contains(joined, "im_business_events_outbox") {
+		t.Fatalf("business event outbox table must not be created: %s", joined)
+	}
 }

server/db/mysql/registration_invites.go

@@ -424,14 +424,10 @@ func (a *adapter) TenantRegistrationInviteRecordUse(id t.TenantID, value string,
 			}
 		}
 	}
-	eventID := t.StableEventID("registration-invite-use", id, row.ID, store.DecodeUid(uid))
-	if err = insertBusinessEventTx(tx, eventID, "user", uid.UserId(), uid, "registration.invite_used", map[string]any{
-		"tenant_id": id, "invite_code_id": row.ID, "user_id": store.DecodeUid(uid), "bound_user_id": row.BoundUserID,
-		"request_id": requestID, "used_at": now,
-	}); err != nil {
+	if err = tx.Commit(); err != nil {
 		return err
 	}
-	return tx.Commit()
+	return a.TenantRegistrationInviteSyncUsage(id, row.ID)
 }
 
 func upsertFriendshipPairTx(tx *sqlx.Tx, uid1, uid2 t.Uid, topic string, now time.Time) error {

server/db/mysql/schema.sql

@@ -13,7 +13,7 @@
 #
 # - Default tenant business database: tinode
 #   - im_users, im_topics, im_subscriptions, im_messages and extension tables
-#   - im_http_idempotency and im_business_events_outbox for reliable HTTP writes
+#   - im_http_idempotency for HTTP write idempotency
 #   - im_file_review_tasks for asynchronous content safety and malware checks
 #   - business tables do not store tenant_id in independent-database mode
 #

server/db/mysql/tenant_adapter.go

@@ -2173,10 +2173,6 @@ func (a *adapter) TenantMessageRecall(id t.TenantID, topic string, seq int64, ac
 	if err != nil {
 		return nil, err
 	}
-	if err = insertBusinessEventTx(tx, t.StableEventID("message", topic, seq, expectedVersion+1, "recalled"),
-		"topic", topic, actor, "message.recalled", map[string]any{"topic": topic, "seq": seq, "version": expectedVersion + 1}); err != nil {
-		return nil, err
-	}
 	if err = tx.Commit(); err != nil {
 		return nil, err
 	}
@@ -2223,10 +2219,6 @@ func (a *adapter) TenantGroupMessageRecall(id t.TenantID, topic string, seq int6
 	if err != nil {
 		return nil, err
 	}
-	if err = insertBusinessEventTx(tx, t.StableEventID("message", topic, seq, expectedVersion+1, "recalled"),
-		"topic", topic, actor, "message.recalled", map[string]any{"topic": topic, "seq": seq, "version": expectedVersion + 1}); err != nil {
-		return nil, err
-	}
 	if err = tx.Commit(); err != nil {
 		return nil, err
 	}
@@ -2280,10 +2272,6 @@ func (a *adapter) TenantMessageEdit(id t.TenantID, topic string, seq int64, acto
 	if err != nil {
 		return nil, err
 	}
-	if err = insertBusinessEventTx(tx, t.StableEventID("message", topic, seq, expectedVersion+1, "edited"),
-		"topic", topic, actor, "message.edited", map[string]any{"topic": topic, "seq": seq, "version": expectedVersion + 1}); err != nil {
-		return nil, err
-	}
 	if err = tx.Commit(); err != nil {
 		return nil, err
 	}
@@ -2336,10 +2324,6 @@ func (a *adapter) TenantGroupMessageEdit(id t.TenantID, topic string, seq int64,
 	if err != nil {
 		return nil, err
 	}
-	if err = insertBusinessEventTx(tx, t.StableEventID("message", topic, seq, expectedVersion+1, "edited"),
-		"topic", topic, actor, "message.edited", map[string]any{"topic": topic, "seq": seq, "version": expectedVersion + 1}); err != nil {
-		return nil, err
-	}
 	if err = tx.Commit(); err != nil {
 		return nil, err
 	}
@@ -3959,10 +3943,6 @@ func insertMeetingEventsTx(tx *sqlx.Tx, event principalMeetingEvent, participant
 				return err
 			}
 		}
-		if err := insertBusinessEventTx(tx, eventID, "user", participant.UserId(), event.Actor, event.EventType,
-			map[string]any{"meeting_id": event.MeetingID, "version": event.Version, "state": event.State}); err != nil {
-			return err
-		}
 	}
 	return nil
 }
@@ -8964,10 +8944,6 @@ func (a *adapter) TenantFriendRequestCreate(id t.TenantID, req *t.FriendRequest)
 		"新的好友申请", req.Message, map[string]any{"type": "friend_request", "request_id": req.RequestID, "user_id": req.ApplicantID.UserId()}); err != nil {
 		return nil, err
 	}
-	if err = insertBusinessEventTx(tx, eventID, "user", req.TargetID.UserId(), req.ApplicantID, eventType,
-		map[string]any{"request_id": req.RequestID, "user_id": req.ApplicantID.UserId()}); err != nil {
-		return nil, err
-	}
 	if err = tx.Commit(); err != nil {
 		return nil, err
 	}
@@ -9319,14 +9295,6 @@ func (a *adapter) TenantFriendRequestAccept(id t.TenantID, requestID string, act
 		"好友申请已处理", t.FriendRequestAccepted, map[string]any{"type": "friend_request", "request_id": row.RequestID, "user_id": actor.UserId(), "topic": topic}); err != nil {
 		return nil, err
 	}
-	if err = insertBusinessEventTx(tx, applicantEventID, "user", applicant.UserId(), actor, eventType,
-		map[string]any{"request_id": row.RequestID, "user_id": actor.UserId(), "topic": topic}); err != nil {
-		return nil, err
-	}
-	if err = insertBusinessEventTx(tx, t.StableEventID("relation", row.RequestID+":"+t.FriendRequestAccepted+":self", eventType, actor.UserId()),
-		"user", actor.UserId(), actor, eventType, map[string]any{"request_id": row.RequestID, "user_id": actor.UserId(), "topic": topic}); err != nil {
-		return nil, err
-	}
 	if err = tx.Commit(); err != nil {
 		return nil, err
 	}
@@ -9376,14 +9344,6 @@ func (a *adapter) TenantFriendRequestResolve(id t.TenantID, requestID string, ac
 		"好友申请已处理", status, map[string]any{"type": "friend_request", "request_id": req.RequestID, "user_id": actor.UserId()}); err != nil {
 		return nil, err
 	}
-	if err = insertBusinessEventTx(tx, eventID, "user", req.ApplicantID.UserId(), actor, eventType,
-		map[string]any{"request_id": req.RequestID, "user_id": actor.UserId()}); err != nil {
-		return nil, err
-	}
-	if err = insertBusinessEventTx(tx, t.StableEventID("relation", req.RequestID+":"+status+":self", eventType, actor.UserId()),
-		"user", actor.UserId(), actor, eventType, map[string]any{"request_id": req.RequestID, "user_id": actor.UserId()}); err != nil {
-		return nil, err
-	}
 	if err = tx.Commit(); err != nil {
 		return nil, err
 	}

server/db/mysql/tenant_infrastructure.go

@@ -140,94 +140,6 @@ func (a *adapter) httpIdempotencyGet(uid t.Uid, method, route, key string) (*t.H
 		ResponseBody: row.ResponseBody, CreatedAt: row.CreatedAt, UpdatedAt: row.UpdatedAt, ExpiresAt: row.ExpiresAt}, nil
 }
 
-func (a *adapter) TenantBusinessEventCreate(id t.TenantID, event *t.BusinessEvent) error {
-	tenantAdp, err := a.infrastructureAdapter(id)
-	if err != nil {
-		return err
-	}
-	if event == nil || event.EventID == "" || event.Target == "" || event.EventType == "" ||
-		(event.TargetType != "user" && event.TargetType != "topic") {
-		return t.ErrMalformed
-	}
-	if tenantAdp != a {
-		return tenantAdp.TenantBusinessEventCreate(id, event)
-	}
-	now := t.TimeNow()
-	_, err = a.db.Exec(`INSERT INTO im_business_events_outbox
-		(event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at)
-		VALUES(?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, event.EventID,
-		event.TargetType, event.Target, decodeUID(event.ActorUserID), event.EventType, event.Payload,
-		t.BusinessEventPending, now, now)
-	return err
-}
-
-func (a *adapter) TenantBusinessEventListPending(id t.TenantID, limit int, now time.Time) ([]t.BusinessEvent, error) {
-	tenantAdp, err := a.infrastructureAdapter(id)
-	if err != nil {
-		return nil, err
-	}
-	if tenantAdp != a {
-		return tenantAdp.TenantBusinessEventListPending(id, limit, now)
-	}
-	if limit <= 0 || limit > 500 {
-		limit = 100
-	}
-	rows, err := a.db.Queryx(`SELECT * FROM im_business_events_outbox
-		WHERE status=? AND (next_retry_at IS NULL OR next_retry_at<=?) ORDER BY id LIMIT ?`, t.BusinessEventPending, now, limit)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-	events := make([]t.BusinessEvent, 0, limit)
-	for rows.Next() {
-		var row struct {
-			ID          int64
-			EventID     string `db:"event_id"`
-			TargetType  string `db:"target_type"`
-			Target      string
-			ActorUserID int64  `db:"actor_user_id"`
-			EventType   string `db:"event_type"`
-			Payload     []byte
-			Status      string
-			Attempts    int
-			NextRetryAt *time.Time `db:"next_retry_at"`
-			DeliveredAt *time.Time `db:"delivered_at"`
-			CreatedAt   time.Time  `db:"created_at"`
-			UpdatedAt   time.Time  `db:"updated_at"`
-		}
-		if err = rows.StructScan(&row); err != nil {
-			return nil, err
-		}
-		events = append(events, t.BusinessEvent{ID: row.ID, EventID: row.EventID, TargetType: row.TargetType,
-			Target: row.Target, ActorUserID: store.EncodeUid(row.ActorUserID), EventType: row.EventType,
-			Payload: row.Payload, Status: row.Status, Attempts: row.Attempts, NextRetryAt: row.NextRetryAt,
-			DeliveredAt: row.DeliveredAt, CreatedAt: row.CreatedAt, UpdatedAt: row.UpdatedAt})
-	}
-	return events, rows.Err()
-}
-
-func (a *adapter) TenantBusinessEventMarkDelivered(id t.TenantID, eventID string, now time.Time) error {
-	tenantAdp, err := a.infrastructureAdapter(id)
-	if err != nil {
-		return err
-	}
-	if eventID == "" {
-		return t.ErrMalformed
-	}
-	if tenantAdp != a {
-		return tenantAdp.TenantBusinessEventMarkDelivered(id, eventID, now)
-	}
-	res, err := a.db.Exec(`UPDATE im_business_events_outbox SET status=?,delivered_at=?,attempts=attempts+1,updated_at=?
-		WHERE event_id=? AND status=?`, t.BusinessEventDelivered, now, now, eventID, t.BusinessEventPending)
-	if err != nil {
-		return err
-	}
-	if count, _ := res.RowsAffected(); count == 0 {
-		return t.ErrNotFound
-	}
-	return nil
-}
-
 func (a *adapter) TenantKafkaConsumeBegin(id t.TenantID, record *t.KafkaConsumeRecord, deadline time.Time) (*t.KafkaConsumeRecord, bool, error) {
 	tenantAdp, err := a.infrastructureAdapter(id)
 	if err != nil {
@@ -600,19 +512,6 @@ func truncateKafkaString(value string, limit int) string {
 	return string(runes[:limit])
 }
 
-func insertBusinessEventTx(tx *sqlx.Tx, eventID, targetType, target string, actor t.Uid, eventType string, payload any) error {
-	encoded, err := json.Marshal(payload)
-	if err != nil {
-		return err
-	}
-	now := t.TimeNow()
-	_, err = tx.Exec(`INSERT INTO im_business_events_outbox
-		(event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at)
-		VALUES(?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, eventID, targetType,
-		target, decodeUID(actor), eventType, encoded, t.BusinessEventPending, now, now)
-	return err
-}
-
 func insertSystemNotificationTx(tx *sqlx.Tx, uid t.Uid, eventID, eventType, title, summary string, action any) error {
 	encoded, err := json.Marshal(action)
 	if err != nil {

server/db/mysql/tests/mysql_test.go

@@ -61,7 +61,7 @@ func TestGroupBusinessTablesCreated(t *testing.T) {
 	tables := []string{
 		"im_group_settings", "im_group_members", "im_group_invitations", "im_group_join_requests",
 		"im_group_announcements", "im_group_announcement_reads", "im_group_mutes", "im_group_qrcodes",
-		"im_group_files", "im_group_admin_logs", "im_group_idempotency", "im_group_events_outbox",
+		"im_group_files", "im_group_admin_logs", "im_group_idempotency",
 	}
 	args := make([]any, len(tables))
 	for i, table := range tables {
@@ -90,7 +90,7 @@ func TestGroupBusinessTablesCreated(t *testing.T) {
 }
 
 func TestHTTPInfrastructureTablesCreated(t *testing.T) {
-	tables := []string{"im_http_idempotency", "im_business_events_outbox"}
+	tables := []string{"im_http_idempotency"}
 	args := make([]any, len(tables))
 	for i, table := range tables {
 		args[i] = table

server/im_business_events.go

@@ -3,7 +3,6 @@ package main
 import (
 	"encoding/json"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/tinode/chat/server/logs"
@@ -11,10 +10,6 @@ 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...)
 }
@@ -28,14 +23,44 @@ func enqueueIMBusinessEvent(tenantID types.TenantID, eventID, targetType, target
 	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 publishIMBusinessRealtimeEvent(tenantID, targetType, target, actor, eventType, encoded)
+}
+
+func publishIMBusinessRealtimeEvent(tenantID types.TenantID, targetType, target string, actor types.Uid, eventType string, payload json.RawMessage) error {
+	from := ""
+	if !actor.IsZero() {
+		from = actor.UserId()
 	}
-	return err
+	if len(payload) == 0 {
+		payload = json.RawMessage(`{}`)
+	}
+	message := &ServerComMessage{
+		TenantID: tenantID, RcptTo: target, AsUser: from, Timestamp: time.Now().UTC().Round(time.Millisecond),
+		Info: &MsgServerInfo{Topic: target, From: from, What: "im", Event: eventType, Payload: payload},
+	}
+	switch targetType {
+	case "user":
+		uid := types.ParseUserId(target)
+		if uid.IsZero() {
+			return types.ErrMalformed
+		}
+		if globals.sessionStore != nil {
+			globals.sessionStore.DeliverToUser(tenantID, uid, "", message)
+		}
+		publishRealtimeUserMessage(uid, message)
+	case "topic":
+		if globals.hub != nil {
+			select {
+			case globals.hub.routeSrv <- message:
+			default:
+				logs.Warn.Printf("business realtime: hub queue full target=%s event=%s", strings.TrimSpace(target), eventType)
+			}
+		}
+		publishRealtimeTopicMessage(target, message)
+	default:
+		return types.ErrMalformed
+	}
+	return nil
 }
 
 func createIMSystemNotification(tenantID types.TenantID, uid types.Uid, eventID, eventType,
@@ -87,136 +112,3 @@ func publishIMRelationEvent(tenantID types.TenantID, actor, target types.Uid, st
 	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
-}

server/im_business_events_test.go

@@ -1,120 +0,0 @@
-package main
-
-import (
-	"testing"
-	"time"
-
-	"github.com/tinode/chat/server/store"
-	"github.com/tinode/chat/server/store/types"
-)
-
-type passwordResetEventStore struct {
-	testHTTPInfrastructureStore
-	delivered []string
-}
-
-func (s *passwordResetEventStore) MarkEventDelivered(_ types.TenantID, eventID string, _ time.Time) error {
-	s.delivered = append(s.delivered, eventID)
-	return nil
-}
-
-func TestDispatchPasswordResetEvent(t *testing.T) {
-	oldInfrastructure := store.HTTPInfrastructure
-	oldHub, oldSessions := globals.hub, globals.sessionStore
-	defer func() {
-		store.HTTPInfrastructure = oldInfrastructure
-		globals.hub, globals.sessionStore = oldHub, oldSessions
-	}()
-
-	infrastructure := &passwordResetEventStore{}
-	infrastructure.events = []types.BusinessEvent{{
-		EventID: "password-reset-1", EventType: "security.password_reset",
-		TargetType: "user", Target: "42", Payload: []byte(`{"user_id":42}`),
-	}}
-	store.HTTPInfrastructure = infrastructure
-	globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 1)}
-	globals.sessionStore = nil
-
-	dispatchIMBusinessEvents(types.TenantID(1))
-
-	if len(infrastructure.delivered) != 1 || infrastructure.delivered[0] != "password-reset-1" {
-		t.Fatalf("password reset event was not marked delivered: %v", infrastructure.delivered)
-	}
-	if len(globals.hub.routeSrv) != 0 {
-		t.Fatal("password reset event was routed as a regular client event")
-	}
-}
-
-func TestDispatchRegistrationInviteUsageEvent(t *testing.T) {
-	oldInfrastructure, oldInvites := store.HTTPInfrastructure, store.RegistrationInvites
-	oldHub := globals.hub
-	defer func() {
-		store.HTTPInfrastructure, store.RegistrationInvites = oldInfrastructure, oldInvites
-		globals.hub = oldHub
-	}()
-
-	infrastructure := &passwordResetEventStore{}
-	infrastructure.events = []types.BusinessEvent{{
-		EventID: "invite-use-1", EventType: "registration.invite_used",
-		TargetType: "user", Target: "42", Payload: []byte(`{"invite_code_id":99}`),
-	}}
-	invites := &testRegistrationInviteStore{}
-	store.HTTPInfrastructure = infrastructure
-	store.RegistrationInvites = invites
-	globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 1)}
-
-	dispatchIMBusinessEvents(types.TenantID(1))
-
-	if invites.syncedCode != 99 {
-		t.Fatalf("invite usage projection code=%d, want 99", invites.syncedCode)
-	}
-	if len(infrastructure.delivered) != 1 || infrastructure.delivered[0] != "invite-use-1" {
-		t.Fatalf("invite usage event was not marked delivered: %v", infrastructure.delivered)
-	}
-	if len(globals.hub.routeSrv) != 0 {
-		t.Fatal("invite usage event was routed as a regular client event")
-	}
-}
-
-func TestDispatchUserBusinessEventToOnlineSession(t *testing.T) {
-	oldInfrastructure := store.HTTPInfrastructure
-	oldHub, oldSessions := globals.hub, globals.sessionStore
-	defer func() {
-		store.HTTPInfrastructure = oldInfrastructure
-		globals.hub, globals.sessionStore = oldHub, oldSessions
-	}()
-
-	uid := types.ParseUserId("usrv97xq86jw3c")
-	infrastructure := &passwordResetEventStore{}
-	infrastructure.events = []types.BusinessEvent{{
-		EventID: "friend-request-1", EventType: "friend.request.created",
-		TargetType: "user", Target: uid.UserId(), Payload: []byte(`{"request_id":"frq_1","user_id":"usrSrc"}`),
-		CreatedAt: time.Now().UTC(),
-	}}
-	store.HTTPInfrastructure = infrastructure
-	globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 1)}
-	globals.sessionStore = NewSessionStore(time.Minute)
-	session := &Session{sid: "s1", proto: WEBSOCK, uid: uid, tenantID: types.TenantID(1), send: make(chan any, 1)}
-	globals.sessionStore.sessCache[session.sid] = session
-
-	dispatchIMBusinessEvents(types.TenantID(1))
-
-	if len(infrastructure.delivered) != 1 || infrastructure.delivered[0] != "friend-request-1" {
-		t.Fatalf("business event was not marked delivered: %v", infrastructure.delivered)
-	}
-	if len(globals.hub.routeSrv) != 0 {
-		t.Fatal("online user event should be delivered directly to session without topic routing")
-	}
-	select {
-	case raw := <-session.send:
-		msg, ok := raw.(*ServerComMessage)
-		if !ok || msg.Info == nil {
-			t.Fatalf("unexpected outbound message: %#v", raw)
-		}
-		if msg.Info.What != "im" || msg.Info.Event != "friend.request.created" || msg.Info.Topic != uid.UserId() {
-			t.Fatalf("unexpected info event: %#v", msg.Info)
-		}
-	default:
-		t.Fatal("online user session did not receive business info event")
-	}
-}

server/im_group_events.go

@@ -2,80 +2,57 @@ package main
 
 import (
 	"encoding/json"
-	"strconv"
+	"strings"
 	"time"
 
 	"github.com/tinode/chat/server/logs"
-	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
 )
 
-// dispatchIMGroupEvents delivers committed outbox events to online topic
-// sessions. Events remain pending when the hub is unavailable or its queue is
-// full, so the next successful group mutation retries them.
-func dispatchIMGroupEvents(tenantID types.TenantID) {
-	if tenantID.IsZero() || globals.hub == nil || store.GroupBusiness == nil {
+func publishIMGroupRealtimeEvent(tenantID types.TenantID, topic string, actor types.Uid, eventType string, payload any) {
+	topic = strings.TrimSpace(topic)
+	eventType = strings.TrimSpace(eventType)
+	if tenantID.IsZero() || topic == "" || eventType == "" {
 		return
 	}
-	now := time.Now().UTC().Round(time.Millisecond)
-	events, err := store.GroupBusiness.ListPendingEvents(tenantID, 100, now)
+	encoded, err := json.Marshal(payload)
 	if err != nil {
-		logs.Warn.Printf("group outbox: list tenant=%d: %v", tenantID, err)
-		return
+		logs.Warn.Printf("group realtime: encode event topic=%s event=%s: %v", topic, eventType, err)
+		encoded = []byte(`{}`)
 	}
-	for _, event := range events {
-		members, memberErr := store.Topics.GetUsers(tenantID, event.Topic, &types.QueryOpt{Limit: 1000})
-		if memberErr != nil {
-			logs.Warn.Printf("group outbox: list members event=%s: %v", event.EventID, memberErr)
-			return
-		}
-		for _, member := range members {
-			uid := types.ParseUid(member.User)
-			if uid.IsZero() {
-				if decoded, parseErr := strconv.ParseInt(member.User, 10, 64); parseErr == nil && decoded > 0 {
-					uid = store.EncodeUid(decoded)
-				}
-			}
-			if uid.IsZero() || uid == event.ActorUserID {
-				continue
-			}
-			if err = createIMSystemNotification(tenantID, uid,
-				stableIMEventID(event.EventID, uid.UserId(), "notification"), event.EventType,
-				"群聊动态", event.EventType, types.KVMap{"topic": event.Topic}); err != nil {
-				logs.Warn.Printf("group outbox: create notification event=%s user=%s: %v", event.EventID, uid.UserId(), err)
-				return
-			}
-		}
-		payload := json.RawMessage(event.Payload)
-		if len(payload) == 0 {
-			payload = json.RawMessage(`{}`)
-		}
-		from := ""
-		if !event.ActorUserID.IsZero() {
-			from = event.ActorUserID.UserId()
-		}
-		message := &ServerComMessage{
-			TenantID:  tenantID,
-			RcptTo:    event.Topic,
-			AsUser:    from,
-			Timestamp: event.CreatedAt,
-			Info: &MsgServerInfo{
-				Topic:   event.Topic,
-				From:    from,
-				What:    "group",
-				Event:   event.EventType,
-				Payload: payload,
-			},
-		}
+	from := ""
+	if !actor.IsZero() {
+		from = actor.UserId()
+	}
+	message := &ServerComMessage{
+		TenantID:  tenantID,
+		RcptTo:    topic,
+		AsUser:    from,
+		Timestamp: time.Now().UTC().Round(time.Millisecond),
+		Info: &MsgServerInfo{
+			Topic:   topic,
+			From:    from,
+			What:    "group",
+			Event:   eventType,
+			Payload: json.RawMessage(encoded),
+		},
+	}
+	if globals.hub != nil {
 		select {
 		case globals.hub.routeSrv <- message:
-			if err = store.GroupBusiness.MarkEventDelivered(tenantID, event.EventID, now); err != nil {
-				logs.Warn.Printf("group outbox: mark delivered event=%s: %v", event.EventID, err)
-				return
-			}
 		default:
-			logs.Warn.Printf("group outbox: hub queue full event=%s", event.EventID)
-			return
+			logs.Warn.Printf("group realtime: hub queue full topic=%s event=%s", topic, eventType)
 		}
 	}
+	publishRealtimeTopicMessage(topic, message)
+}
+
+func userIDs(uids []types.Uid) []string {
+	ids := make([]string, 0, len(uids))
+	for _, uid := range uids {
+		if !uid.IsZero() {
+			ids = append(ids, uid.UserId())
+		}
+	}
+	return ids
 }

server/internal_admin_query_resources.go

@@ -115,18 +115,7 @@ var internalTenantResourceSpecs = map[string]*internalQuerySpec{
 	},
 }
 
-var internalOutboxResourceSpecs = map[string]*internalQuerySpec{
-	"business-events": {
-		Select: "SELECT id,event_id,target_type,target,actor_user_id,event_type,payload,status,attempts,next_retry_at,delivered_at,created_at,updated_at",
-		From:   "FROM im_business_events_outbox", DefaultSort: "ORDER BY id DESC", DetailKey: "id",
-		Filters: []internalQueryFilter{{"event_type", "event_type", "=", "string"}, {"status", "status", "=", "string"}, {"target_type", "target_type", "=", "string"}, {"target", "target", "=", "string"}, {"created_at_start", "created_at", ">=", "time"}, {"created_at_end", "created_at", "<=", "time"}},
-	},
-	"group-events": {
-		Select: "SELECT id,event_id,topic,actor_user_id,event_type,payload,status,attempts,next_retry_at,delivered_at,created_at,updated_at",
-		From:   "FROM im_group_events_outbox", DefaultSort: "ORDER BY id DESC", DetailKey: "id",
-		Filters: []internalQueryFilter{{"event_type", "event_type", "=", "string"}, {"status", "status", "=", "string"}, {"topic", "topic", "=", "string"}, {"created_at_start", "created_at", ">=", "time"}, {"created_at_end", "created_at", "<=", "time"}},
-	},
-}
+var internalOutboxResourceSpecs = map[string]*internalQuerySpec{}
 
 var internalOrgResourceSpecs = map[string]*internalQuerySpec{
 	"departments": {
@@ -184,8 +173,6 @@ func serveInternalDashboardSummary(wrt http.ResponseWriter, req *http.Request, t
 	summary := map[string]any{}
 	queries := map[string]string{
 		"dead_kafka":        "SELECT COUNT(*) AS count FROM im_kafka_dead_letters WHERE status='dead'",
-		"pending_business":  "SELECT COUNT(*) AS count FROM im_business_events_outbox WHERE status='pending'",
-		"pending_group":     "SELECT COUNT(*) AS count FROM im_group_events_outbox WHERE status='pending'",
 		"pending_realname":  "SELECT COUNT(*) AS count FROM im_realname_verifications WHERE status='pending'",
 		"received_abuse":    "SELECT COUNT(*) AS count FROM im_abuse_reports WHERE status='received'",
 		"received_feedback": "SELECT COUNT(*) AS count FROM im_feedback_submissions WHERE status='received'",

server/internal_api.go

@@ -98,7 +98,7 @@ func serveInternalAdminQuery(wrt http.ResponseWriter, req *http.Request, prefix
 	if !ok {
 		return
 	}
-	if req.Method == http.MethodPost && !(len(parts) == 3 && parts[2] == "users") {
+	if req.Method == http.MethodPost && !isInternalAdminPostRoute(parts) {
 		writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
 		return
 	}
@@ -114,6 +114,8 @@ func serveInternalAdminQuery(wrt http.ResponseWriter, req *http.Request, prefix
 		serveInternalAdminResourceList(wrt, req, tenant.ID, "group-messages", "")
 	case req.Method == http.MethodPost && len(parts) == 3 && parts[2] == "users":
 		serveInternalUserCreate(wrt, req, tenant.ID)
+	case req.Method == http.MethodPost && len(parts) == 5 && parts[2] == "users" && parts[4] == "action":
+		serveInternalUserAction(wrt, req, tenant.ID, parts[3])
 	case req.Method == http.MethodGet && len(parts) == 3 && parts[2] == "users":
 		serveInternalAdminResourceList(wrt, req, tenant.ID, "users", "")
 	case len(parts) == 4 && parts[2] == "users":
@@ -163,6 +165,11 @@ func serveInternalAdminQuery(wrt http.ResponseWriter, req *http.Request, prefix
 	}
 }
 
+func isInternalAdminPostRoute(parts []string) bool {
+	return (len(parts) == 3 && parts[2] == "users") ||
+		(len(parts) == 5 && parts[2] == "users" && parts[4] == "action")
+}
+
 type internalUserCreateRequest struct {
 	Username string `json:"username"`
 	Nickname string `json:"nickname"`
@@ -170,6 +177,15 @@ type internalUserCreateRequest struct {
 	Password string `json:"password"`
 }
 
+type internalUserActionRequest struct {
+	EventType    string `json:"event_type"`
+	ActorUserID  string `json:"actor_user_id"`
+	DeviceID     string `json:"device_id"`
+	PasswordHash string `json:"password_hash"`
+	Reason       string `json:"reason"`
+	Hard         bool   `json:"hard"`
+}
+
 func serveInternalUserCreate(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) {
 	var body internalUserCreateRequest
 	if err := json.NewDecoder(io.LimitReader(req.Body, 1<<20)).Decode(&body); err != nil {
@@ -244,6 +260,78 @@ func serveInternalUserCreate(wrt http.ResponseWriter, req *http.Request, tenantI
 	})
 }
 
+func serveInternalUserAction(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, rawUserID string) {
+	var body internalUserActionRequest
+	if err := json.NewDecoder(io.LimitReader(req.Body, 1<<20)).Decode(&body); err != nil {
+		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
+		return
+	}
+	eventType := strings.TrimSpace(body.EventType)
+	if !isInternalUserActionSupported(eventType) {
+		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "unsupported user action", map[string]any{"event_type": eventType})
+		return
+	}
+	target := parseKafkaCommandUser(rawUserID)
+	if target.IsZero() {
+		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "用户ID不能为空", nil)
+		return
+	}
+	payload := adminCommandPayload{
+		CommandID:    firstNonEmpty(strings.TrimSpace(req.Header.Get("X-Request-ID")), "internal-"+strconv.FormatInt(time.Now().UTC().UnixNano(), 10)),
+		ActorUserID:  strings.TrimSpace(body.ActorUserID),
+		UserID:       strconv.FormatInt(internalAdminUIDRaw(target), 10),
+		TargetID:     strconv.FormatInt(internalAdminUIDRaw(target), 10),
+		DeviceID:     strings.TrimSpace(body.DeviceID),
+		PasswordHash: strings.TrimSpace(body.PasswordHash),
+		Reason:       strings.TrimSpace(body.Reason),
+		Hard:         body.Hard,
+	}
+	envelope := kafkaEnvelope{
+		EventID:    payload.CommandID,
+		TraceID:    strings.TrimSpace(req.Header.Get("X-Request-ID")),
+		EventType:  eventType,
+		BizType:    "user",
+		BizID:      strconv.FormatInt(internalAdminUIDRaw(target), 10),
+		OccurredAt: time.Now().UTC(),
+	}
+	var (
+		result map[string]any
+		err    error
+	)
+	if eventType == adminCommandUserPasswordReset {
+		result, err = executeKafkaUserPasswordResetCommand(tenantID, payload)
+	} else {
+		result, err = executeKafkaUserGovernanceCommand(tenantID, envelope, payload, parseKafkaCommandUser(payload.ActorUserID))
+	}
+	if err != nil {
+		writeInternalQueryError(wrt, req, err)
+		return
+	}
+	if err := enqueueKafkaAuditEvent(req.Context(), tenantID, "admin", parseKafkaCommandUser(payload.ActorUserID),
+		eventType, "internal_user_action", payload.CommandID, "succeeded", "", result); err != nil {
+		logs.Warn.Printf("internal user action audit enqueue failed tenant:%d user:%d event:%s err:%v", tenantID, internalAdminUIDRaw(target), eventType, err)
+	}
+	writeIMSuccess(wrt, req, map[string]any{
+		"user_id":    internalAdminUIDRaw(target),
+		"event_type": eventType,
+		"status":     "succeeded",
+		"result":     result,
+	})
+}
+
+func isInternalUserActionSupported(eventType string) bool {
+	switch strings.TrimSpace(eventType) {
+	case adminCommandUserBan, adminCommandUserUnban,
+		adminCommandUserFreeze, adminCommandUserUnfreeze,
+		adminCommandUserDeviceKick, adminCommandUserLogoutAll,
+		adminCommandUserCancel, adminCommandUserDelete,
+		adminCommandUserPasswordReset:
+		return true
+	default:
+		return false
+	}
+}
+
 func authorizeInternalAPI(wrt http.ResponseWriter, req *http.Request, capability string) bool {
 	if !internalAPI.config.Enabled {
 		writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)

server/kafka_admin_commands.go

@@ -24,6 +24,7 @@ const (
 	adminCommandGroupMemberUnmute  = "group.member.unmute"
 	adminCommandGroupMemberRemove  = "group.member.remove"
 	adminCommandGroupOwnerTransfer = "group.owner.transfer"
+	adminCommandGroupDissolve      = "group.dissolve"
 	adminCommandUserBan            = "user.ban"
 	adminCommandUserUnban          = "user.unban"
 	adminCommandUserFreeze         = "user.freeze"
@@ -75,6 +76,8 @@ type groupGovernancePatch struct {
 	MemberUploadFiles *bool `json:"member_upload_files"`
 	MemberPrivateChat *bool `json:"member_private_chat"`
 	MemberInvite      *bool `json:"member_invite"`
+	MaxMemberCount    *int  `json:"max_member_count"`
+	MaxAdminCount     *int  `json:"max_admin_count"`
 }
 
 type adminCommandResultPayload struct {
@@ -221,7 +224,8 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
 		if err != nil {
 			return nil, err
 		}
-		dispatchIMGroupEvents(tenantID)
+		publishIMGroupRealtimeEvent(tenantID, topic, actor, "group.settings.updated",
+			map[string]any{"topic": topic, "version": settings.Version})
 		return map[string]any{"topic": topic, "version": settings.Version, "updated_at": settings.UpdatedAt}, nil
 
 	case adminCommandGroupGovernUpdate:
@@ -231,12 +235,15 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
 			MemberUploadFiles: payload.Governance.MemberUploadFiles,
 			MemberPrivateChat: payload.Governance.MemberPrivateChat,
 			MemberInvite:      payload.Governance.MemberInvite,
+			MaxMemberCount:    payload.Governance.MaxMemberCount,
+			MaxAdminCount:     payload.Governance.MaxAdminCount,
 			Version:           payload.Version,
 		})
 		if err != nil {
 			return nil, err
 		}
-		dispatchIMGroupEvents(tenantID)
+		publishIMGroupRealtimeEvent(tenantID, topic, actor, "group.settings.updated",
+			map[string]any{"topic": topic, "version": settings.Version})
 		return map[string]any{"topic": topic, "version": settings.Version, "updated_at": settings.UpdatedAt}, nil
 
 	case adminCommandGroupAdminSet:
@@ -259,7 +266,8 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
 			}
 			succeeded = append(succeeded, map[string]any{"user_id": internalAdminUIDRaw(target), "role": types.GroupMemberRoleAdmin})
 		}
-		dispatchIMGroupEvents(tenantID)
+		publishIMGroupRealtimeEvent(tenantID, topic, actor, "group.role.updated",
+			map[string]any{"topic": topic, "succeeded": succeeded})
 		return map[string]any{"topic": topic, "succeeded": succeeded}, nil
 
 	case adminCommandGroupAdminRemove:
@@ -271,7 +279,8 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
 			types.GroupMemberRoleMember, types.ModeCPublic, now); err != nil {
 			return nil, err
 		}
-		dispatchIMGroupEvents(tenantID)
+		publishIMGroupRealtimeEvent(tenantID, topic, actor, "group.role.updated",
+			map[string]any{"user_id": internalAdminUIDRaw(target), "role": types.GroupMemberRoleMember})
 		return map[string]any{"topic": topic, "user_id": internalAdminUIDRaw(target), "role": types.GroupMemberRoleMember}, nil
 
 	case adminCommandGroupMemberMute:
@@ -291,7 +300,8 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
 		if err != nil {
 			return nil, err
 		}
-		dispatchIMGroupEvents(tenantID)
+		publishIMGroupRealtimeEvent(tenantID, topic, actor, "group.mute.updated",
+			map[string]any{"user_id": internalAdminUIDRaw(target), "muted": true, "version": mute.Version})
 		return map[string]any{"topic": topic, "user_id": internalAdminUIDRaw(target), "muted": true, "version": mute.Version}, nil
 
 	case adminCommandGroupMemberUnmute:
@@ -302,7 +312,8 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
 		if err := store.GroupBusiness.DeleteMute(tenantID, topic, target, actor, now); err != nil {
 			return nil, err
 		}
-		dispatchIMGroupEvents(tenantID)
+		publishIMGroupRealtimeEvent(tenantID, topic, actor, "group.mute.updated",
+			map[string]any{"user_id": internalAdminUIDRaw(target), "muted": false})
 		return map[string]any{"topic": topic, "user_id": internalAdminUIDRaw(target), "muted": false}, nil
 
 	case adminCommandGroupMemberRemove:
@@ -313,7 +324,8 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
 		if err := store.GroupBusiness.Leave(tenantID, topic, target, now); err != nil {
 			return nil, err
 		}
-		dispatchIMGroupEvents(tenantID)
+		publishIMGroupRealtimeEvent(tenantID, topic, actor, "group.member.left",
+			map[string]any{"user_id": internalAdminUIDRaw(target)})
 		return map[string]any{"topic": topic, "user_id": internalAdminUIDRaw(target), "status": "left"}, nil
 
 	case adminCommandGroupOwnerTransfer:
@@ -324,12 +336,32 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
 		if err := store.GroupBusiness.TransferOwner(tenantID, topic, actor, target, payload.Version, now); err != nil {
 			return nil, err
 		}
-		dispatchIMGroupEvents(tenantID)
+		publishIMGroupRealtimeEvent(tenantID, topic, actor, "group.owner.transferred",
+			map[string]any{"owner": internalAdminUIDRaw(target), "previous_owner": internalAdminUIDRaw(actor)})
 		return map[string]any{"topic": topic, "owner": internalAdminUIDRaw(target), "previous_owner": internalAdminUIDRaw(actor)}, nil
+
+	case adminCommandGroupDissolve:
+		if err := dissolveGroupByAdminCommand(tenantID, topic, actor, strings.TrimSpace(payload.Reason), now); err != nil {
+			return nil, err
+		}
+		publishIMGroupRealtimeEvent(tenantID, topic, actor, "group.dissolved",
+			map[string]any{"topic": topic, "reason": strings.TrimSpace(payload.Reason)})
+		return map[string]any{"topic": topic, "status": "dissolved"}, nil
 	}
 	return nil, errors.New("unsupported admin command: " + envelope.EventType)
 }
 
+func dissolveGroupByAdminCommand(tenantID types.TenantID, topic string, actor types.Uid, reason string, now time.Time) error {
+	if strings.TrimSpace(topic) == "" || actor.IsZero() {
+		return types.ErrMalformed
+	}
+	if store.GroupBusiness == nil {
+		return errors.New("group business store is not initialized")
+	}
+	_, err := store.GroupBusiness.Dissolve(tenantID, topic, actor, reason, now)
+	return err
+}
+
 func executeKafkaMessageAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, payload adminCommandPayload) (map[string]any, error) {
 	conversationType := strings.TrimSpace(payload.ConversationType)
 	topic := strings.TrimSpace(payload.Topic)
@@ -386,7 +418,7 @@ func executeKafkaUserGovernanceCommand(tenantID types.TenantID, envelope kafkaEn
 		if err := store.Users.UpdateState(tenantID, target, types.StateSuspended); err != nil {
 			return nil, err
 		}
-		evictUserSessions(target)
+		evictUserSessions(tenantID, target)
 		notifyUserStateChanged(tenantID, target, types.StateSuspended)
 		result["state"] = types.StateSuspended.String()
 		result["evicted"] = true
@@ -407,6 +439,7 @@ func executeKafkaUserGovernanceCommand(tenantID types.TenantID, envelope kafkaEn
 		if globals.sessionStore != nil {
 			evicted = globals.sessionStore.EvictUserDevice(target, deviceID)
 		}
+		publishRealtimeEvictDevice(tenantID, target, deviceID, "admin_device_kick")
 		if store.Devices != nil {
 			if err := store.Devices.Delete(tenantID, target, deviceID); err != nil && !errors.Is(err, types.ErrNotFound) {
 				return nil, err
@@ -416,7 +449,7 @@ func executeKafkaUserGovernanceCommand(tenantID types.TenantID, envelope kafkaEn
 		result["evicted_sessions"] = evicted
 		return result, nil
 	case adminCommandUserLogoutAll:
-		evictUserSessions(target)
+		evictUserSessions(tenantID, target)
 		result["evicted"] = true
 		return result, nil
 	case adminCommandUserCancel, adminCommandUserDelete:
@@ -456,7 +489,7 @@ func executeKafkaUserPasswordResetCommand(tenantID types.TenantID, payload admin
 		}
 		authVersion = version
 	}
-	evictUserSessions(target)
+	evictUserSessions(tenantID, target)
 	return map[string]any{
 		"user_id":      internalAdminUIDRaw(target),
 		"action":       adminCommandUserPasswordReset,
@@ -465,10 +498,11 @@ func executeKafkaUserPasswordResetCommand(tenantID types.TenantID, payload admin
 	}, nil
 }
 
-func evictUserSessions(uid types.Uid) {
+func evictUserSessions(tenantID types.TenantID, uid types.Uid) {
 	if globals.sessionStore != nil {
 		globals.sessionStore.EvictUser(uid, "")
 	}
+	publishRealtimeEvictUser(tenantID, uid, "admin_user_evict", "")
 }
 
 func notifyUserStateChanged(tenantID types.TenantID, uid types.Uid, state types.ObjState) {
@@ -479,7 +513,7 @@ func notifyUserStateChanged(tenantID types.TenantID, uid types.Uid, state types.
 }
 
 func deleteUserByAdminCommand(tenantID types.TenantID, uid types.Uid, hard bool, _ time.Time) error {
-	evictUserSessions(uid)
+	evictUserSessions(tenantID, uid)
 	usersRemoveUser(tenantID, uid)
 	if globals.hub != nil && globals.hub.unreg != nil {
 		done := make(chan bool)

server/kafka_workbench_audit_test.go

@@ -128,9 +128,7 @@ func TestEnqueueKafkaWorkbenchEventRunsSynchronously(t *testing.T) {
 	}()
 
 	kafkaBus = &kafkaRuntime{}
-	infrastructure := &testHTTPInfrastructureStore{}
 	security := &testSecurityStore{}
-	store.HTTPInfrastructure = infrastructure
 	store.Security = security
 
 	err := enqueueKafkaWorkbenchEvent(context.Background(), 7, types.Uid(42), "meeting", "meeting.created", "meeting-1",
@@ -144,7 +142,4 @@ func TestEnqueueKafkaWorkbenchEventRunsSynchronously(t *testing.T) {
 	if security.notifications[0].UserID != types.Uid(43) || security.notifications[0].EventType != "meeting.created" {
 		t.Fatalf("unexpected notification: %#v", security.notifications[0])
 	}
-	if len(infrastructure.events) != 2 {
-		t.Fatalf("expected business events for both targets, got %#v", infrastructure.events)
-	}
 }

server/main.go

@@ -146,6 +146,8 @@ var globals struct {
 	sessionStore *SessionStore
 	// Cluster data.
 	cluster *Cluster
+	// Unique node ID used by Redis realtime fanout when native clustering is disabled.
+	nodeID string
 	// gRPC server.
 	grpcServer *grpc.Server
 	// Plugins.
@@ -346,6 +348,8 @@ type configType struct {
 	ExpvarPath string `json:"expvar"`
 	// URL path for internal server status. Disabled if the path is blank.
 	ServerStatusPath string `json:"server_status"`
+	// Unique IM node ID for Redis realtime fanout. Defaults to IM_NODE_ID or HOSTNAME.
+	NodeID string `json:"node_id"`
 	// Take IP address of the client from HTTP header 'X-Forwarded-For'.
 	// Useful when tinode is behind a proxy. If missing, fallback to default RemoteAddr.
 	UseXForwardedFor bool `json:"use_x_forwarded_for"`
@@ -466,6 +470,7 @@ func main() {
 	// Initialize cluster and receive calculated workerId.
 	// Cluster won't be started here yet.
 	workerId := clusterInit(config.Cluster, clusterSelf)
+	globals.nodeID = resolveIMNodeID(config.NodeID)
 
 	if *pprofFile != "" {
 		*pprofFile = toAbsolutePath(curwd, *pprofFile)
@@ -740,14 +745,12 @@ func main() {
 	globals.sessionStore = NewSessionStore(idleSessionTimeout + 15*time.Second)
 	// The hub (the main message router)
 	globals.hub = newHub()
-	startRedisGroupFanout(context.Background())
+	startRedisRealtimeFanout(context.Background())
 	startKafkaAdminCommandConsumer()
 	startKafkaConfigEventConsumer()
 	startKafkaCacheInvalidateConsumer()
 	startKafkaMessagePersistConsumer()
 	startKafkaDeadLetterConsumer()
-	stopExternalBusinessEvents := startIMExternalBusinessEventPoller()
-	defer stopExternalBusinessEvents()
 
 	// Start accepting cluster traffic.
 	if globals.cluster != nil {

server/meetingim.conf

@@ -15,6 +15,7 @@
   "permanent_accounts": false,
   "expvar": "/debug/vars",
   "server_status": "-",
+  "node_id": "",
   "use_x_forwarded_for": true,
   "x_frame_options": "SAMEORIGIN",
   "default_country_code": "CN",

server/metrics/metrics.go

@@ -28,6 +28,17 @@ var (
 		Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5},
 	}, []string{"driver", "operation", "status"})
 
+	RealtimeFanoutEventsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "realtime_fanout_events_total",
+		Help: "Total number of realtime fanout events by stage, kind, target type, and status.",
+	}, []string{"stage", "kind", "target_type", "status"})
+
+	RealtimeFanoutDeliveryDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "realtime_fanout_delivery_duration_seconds",
+		Help:    "Realtime fanout delivery age measured from message timestamp to local dispatch.",
+		Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
+	}, []string{"kind", "target_type"})
+
 	WSConnectionsActive = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
 		Name: "ws_connections_active",
 		Help: "Current active WebSocket connections.",
@@ -37,7 +48,8 @@ var (
 )
 
 func init() {
-	Registry.MustRegister(WSConnectionsActive, HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration)
+	Registry.MustRegister(WSConnectionsActive, HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration,
+		RealtimeFanoutEventsTotal, RealtimeFanoutDeliveryDuration)
 }
 
 func ObserveDB(driverName, operation string, started time.Time, err error) {
@@ -48,6 +60,26 @@ func ObserveDB(driverName, operation string, started time.Time, err error) {
 	DBQueryDuration.WithLabelValues(driverName, operation, status).Observe(time.Since(started).Seconds())
 }
 
+func IncRealtimeFanout(stage, kind, targetType, status string) {
+	RealtimeFanoutEventsTotal.WithLabelValues(normalizeLabel(stage), normalizeLabel(kind),
+		normalizeLabel(targetType), normalizeLabel(status)).Inc()
+}
+
+func ObserveRealtimeFanoutDelivery(kind, targetType string, timestamp time.Time) {
+	if timestamp.IsZero() {
+		return
+	}
+	RealtimeFanoutDeliveryDuration.WithLabelValues(normalizeLabel(kind), normalizeLabel(targetType)).
+		Observe(time.Since(timestamp).Seconds())
+}
+
+func normalizeLabel(value string) string {
+	if value == "" {
+		return "unknown"
+	}
+	return value
+}
+
 func expvarIntValue(name string) int64 {
 	value := expvar.Get(name)
 	if value == nil {

server/realtime_dispatcher.go

@@ -0,0 +1,238 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"strconv"
+	"strings"
+
+	"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"
+
+	realtimeTargetTopic  = "topic"
+	realtimeTargetUser   = "user"
+	realtimeTargetDevice = "device"
+)
+
+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 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")
+}
+
+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
+	msg.sess = nil
+	prepareRedisGroupFanoutMessage(&msg, envelope.Target)
+
+	switch envelope.Kind {
+	case realtimeKindSessionControl:
+		deliverRealtimeSessionControl(envelope, &msg)
+	default:
+		immetrics.ObserveRealtimeFanoutDelivery(envelope.Kind, envelope.TargetType, msg.Timestamp)
+		deliverRealtimeMessage(envelope, &msg)
+	}
+}
+
+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)
+}
+
+func deliverRealtimeSessionControl(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 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)
+		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)
+		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.Event + ":" + strconv.FormatInt(msg.Info.SeqId, 10)
+	}
+	if msg.Pres != nil {
+		return "pres:" + target + ":" + msg.Pres.What + ":" + strconv.FormatInt(msg.Pres.SeqId, 10)
+	}
+	return ""
+}

server/realtime_dispatcher_test.go

@@ -0,0 +1,126 @@
+package main
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/tinode/chat/server/store"
+	"github.com/tinode/chat/server/store/types"
+)
+
+func TestRealtimeEnvelopeDeliversToUserSessions(t *testing.T) {
+	oldStore := globals.sessionStore
+	defer func() { globals.sessionStore = oldStore }()
+
+	tenantID := types.TenantID(1)
+	uid := types.Uid(1001)
+	globals.sessionStore = realtimeTestSessionStore(
+		realtimeTestSession("sid-1", tenantID, uid, ""),
+		realtimeTestSession("sid-2", tenantID, uid, ""),
+		realtimeTestSession("sid-other", tenantID, types.Uid(1002), ""),
+	)
+	msg := &ServerComMessage{
+		Data: &MsgServerData{Topic: uid.UserId(), From: types.Uid(1002).UserId(), SeqId: 12, Timestamp: time.Now()},
+	}
+	raw, err := json.Marshal(msg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	deliverRealtimeFanoutEnvelope(&store.RealtimeFanoutEnvelope{
+		TenantID:   tenantID,
+		Kind:       realtimeKindMessage,
+		TargetType: realtimeTargetUser,
+		Target:     uid.UserId(),
+		Message:    raw,
+	})
+
+	for _, sid := range []string{"sid-1", "sid-2"} {
+		sess := globals.sessionStore.Get(sid)
+		select {
+		case got := <-sess.send:
+			if got.(*ServerComMessage).TenantID != tenantID {
+				t.Fatalf("tenant not restored for %s", sid)
+			}
+		default:
+			t.Fatalf("expected delivery to %s", sid)
+		}
+	}
+	if len(globals.sessionStore.Get("sid-other").send) != 0 {
+		t.Fatal("unexpected delivery to other user")
+	}
+}
+
+func TestRealtimeP2PTopicFallbackRewritesVisibleTopic(t *testing.T) {
+	oldStore := globals.sessionStore
+	oldHub := globals.hub
+	defer func() {
+		globals.sessionStore = oldStore
+		globals.hub = oldHub
+	}()
+
+	tenantID := types.TenantID(1)
+	uid1 := types.Uid(1001)
+	uid2 := types.Uid(1002)
+	topic := uid1.P2PName(uid2)
+	if topic == "" {
+		t.Fatal("failed to create p2p topic")
+	}
+	globals.hub = nil
+	globals.sessionStore = realtimeTestSessionStore(
+		realtimeTestSession("sid-u1", tenantID, uid1, uid2.UserId()),
+		realtimeTestSession("sid-u2", tenantID, uid2, uid1.UserId()),
+	)
+	msg := &ServerComMessage{
+		Data: &MsgServerData{Topic: topic, From: uid1.UserId(), SeqId: 13, Timestamp: time.Now()},
+	}
+	raw, err := json.Marshal(msg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	deliverRealtimeFanoutEnvelope(&store.RealtimeFanoutEnvelope{
+		TenantID:   tenantID,
+		Kind:       realtimeKindMessage,
+		TargetType: realtimeTargetTopic,
+		Target:     topic,
+		Message:    raw,
+	})
+
+	expectTopic := map[string]string{"sid-u1": uid2.UserId(), "sid-u2": uid1.UserId()}
+	for sid, visibleTopic := range expectTopic {
+		sess := globals.sessionStore.Get(sid)
+		select {
+		case got := <-sess.send:
+			data := got.(*ServerComMessage).Data
+			if data.Topic != visibleTopic {
+				t.Fatalf("%s got topic %q, want %q", sid, data.Topic, visibleTopic)
+			}
+		default:
+			t.Fatalf("expected p2p delivery to %s", sid)
+		}
+	}
+}
+
+func realtimeTestSessionStore(sessions ...*Session) *SessionStore {
+	ss := NewSessionStore(time.Minute)
+	for _, sess := range sessions {
+		ss.sessCache[sess.sid] = sess
+	}
+	return ss
+}
+
+func realtimeTestSession(sid string, tenantID types.TenantID, uid types.Uid, sub string) *Session {
+	sess := &Session{
+		sid:      sid,
+		tenantID: tenantID,
+		uid:      uid,
+		send:     make(chan any, 4),
+		subs:     make(map[string]*Subscription),
+	}
+	if sub != "" {
+		sess.subs[sub] = &Subscription{}
+	}
+	return sess
+}

server/session.go

@@ -14,6 +14,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"os"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -1375,6 +1376,22 @@ func currentNodeID() string {
 	if globals.cluster != nil && globals.cluster.thisNodeName != "" {
 		return globals.cluster.thisNodeName
 	}
+	if globals.nodeID != "" {
+		return globals.nodeID
+	}
+	return resolveIMNodeID("")
+}
+
+func resolveIMNodeID(configured string) string {
+	if configured = strings.TrimSpace(configured); configured != "" {
+		return configured
+	}
+	if value := strings.TrimSpace(os.Getenv("IM_NODE_ID")); value != "" {
+		return value
+	}
+	if value := strings.TrimSpace(os.Getenv("HOSTNAME")); value != "" {
+		return value
+	}
 	return "local"
 }
 

server/sessionstore.go

@@ -200,6 +200,90 @@ func (ss *SessionStore) Range(f func(sid string, s *Session) bool) {
 	ss.lock.Unlock()
 }
 
+func (ss *SessionStore) DeliverToUser(tenantID types.TenantID, uid types.Uid, skipSid string, msg *ServerComMessage) int {
+	if ss == nil || !tenantID.IsValid() || uid.IsZero() || msg == nil {
+		return 0
+	}
+	delivered := 0
+	ss.Range(func(_ string, sess *Session) bool {
+		if sess == nil || sess.isCluster() || sess.isProxy() || sess.tenantID != tenantID || sess.uid != uid || sess.sid == skipSid {
+			return true
+		}
+		if sess.queueOut(msg.copy()) {
+			delivered++
+		}
+		return true
+	})
+	return delivered
+}
+
+func (ss *SessionStore) DeliverToTopicSubscribers(tenantID types.TenantID, topic, skipSid string, msg *ServerComMessage) int {
+	if ss == nil || !tenantID.IsValid() || topic == "" || msg == nil {
+		return 0
+	}
+	delivered := 0
+	ss.Range(func(_ string, sess *Session) bool {
+		if sess == nil || sess.isCluster() || sess.isProxy() || sess.tenantID != tenantID || sess.sid == skipSid || !sess.hasSub(topic) {
+			return true
+		}
+		msgCopy := msg.copy()
+		if msgCopy.RcptTo == "" {
+			msgCopy.RcptTo = topic
+		}
+		if msgCopy.Data != nil && msgCopy.Data.Topic == "" {
+			msgCopy.Data.Topic = topic
+		}
+		if sess.queueOut(msgCopy) {
+			delivered++
+		}
+		return true
+	})
+	return delivered
+}
+
+func (ss *SessionStore) DeliverToP2PTopic(tenantID types.TenantID, topic, skipSid string, msg *ServerComMessage) int {
+	if ss == nil || !tenantID.IsValid() || topic == "" || msg == nil {
+		return 0
+	}
+	uid1, uid2, err := types.ParseP2P(topic)
+	if err != nil {
+		return ss.DeliverToTopicSubscribers(tenantID, topic, skipSid, msg)
+	}
+	participants := map[types.Uid]struct{}{uid1: {}, uid2: {}}
+	delivered := 0
+	ss.Range(func(_ string, sess *Session) bool {
+		if sess == nil || sess.isCluster() || sess.isProxy() || sess.tenantID != tenantID || sess.sid == skipSid {
+			return true
+		}
+		if _, ok := participants[sess.uid]; !ok {
+			return true
+		}
+		viewTopic, err := types.P2PNameForUser(sess.uid, topic)
+		if err != nil || viewTopic == "" {
+			return true
+		}
+		if !sess.hasSub(topic) && !sess.hasSub(viewTopic) {
+			return true
+		}
+		msgCopy := msg.copy()
+		msgCopy.RcptTo = sess.uid.UserId()
+		if msgCopy.Data != nil {
+			msgCopy.Data.Topic = viewTopic
+		}
+		if msgCopy.Info != nil {
+			msgCopy.Info.Topic = viewTopic
+		}
+		if msgCopy.Pres != nil {
+			msgCopy.Pres.Topic = viewTopic
+		}
+		if sess.queueOut(msgCopy) {
+			delivered++
+		}
+		return true
+	})
+	return delivered
+}
+
 // Shutdown terminates sessionStore. No need to clean up.
 // Don't send to clustered sessions, their servers are not being shut down.
 func (ss *SessionStore) Shutdown() {

server/store/group_message_runtime.go

@@ -51,6 +51,17 @@ type groupFanoutEnvelope struct {
 	Message      json.RawMessage `json:"message"`
 }
 
+type RealtimeFanoutEnvelope struct {
+	OriginNodeID string          `json:"origin_node_id"`
+	TenantID     types.TenantID  `json:"tenant_id"`
+	Kind         string          `json:"kind"`
+	TargetType   string          `json:"target_type"`
+	Target       string          `json:"target"`
+	SkipSid      string          `json:"skip_sid,omitempty"`
+	DedupKey     string          `json:"dedup_key,omitempty"`
+	Message      json.RawMessage `json:"message"`
+}
+
 type groupMessageRuntime interface {
 	Close() error
 	Claim(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) (*groupMessageUniqueRecord, bool, error)
@@ -71,6 +82,8 @@ type groupMessageRuntime interface {
 	GetTopicNodeOnlineMembers(tenantID types.TenantID, topic, nodeID string) ([]types.Uid, error)
 	PublishGroupFanout(originNodeID string, message any) error
 	StartGroupFanoutSubscriber(ctx context.Context, nodeID string, handler func([]byte)) error
+	PublishRealtimeFanout(originNodeID string, tenantID types.TenantID, kind, targetType, target, skipSid, dedupKey string, message any) error
+	StartRealtimeFanoutSubscriber(ctx context.Context, nodeID string, handler func(*RealtimeFanoutEnvelope)) error
 	CacheObject(tenantID types.TenantID, namespace, id string, value any) error
 	GetCachedObject(tenantID types.TenantID, namespace, id string, value any) (bool, error)
 	DeleteCachedObject(tenantID types.TenantID, namespace, id string) error
@@ -237,6 +250,20 @@ func StartGroupFanoutSubscriber(ctx context.Context, nodeID string, handler func
 	return groupMsgRuntime.StartGroupFanoutSubscriber(ctx, nodeID, handler)
 }
 
+func PublishRealtimeFanout(originNodeID string, tenantID types.TenantID, kind, targetType, target, skipSid, dedupKey string, message any) error {
+	if groupMsgRuntime == nil || originNodeID == "" || !tenantID.IsValid() || targetType == "" || target == "" || message == nil {
+		return nil
+	}
+	return groupMsgRuntime.PublishRealtimeFanout(originNodeID, tenantID, kind, targetType, target, skipSid, dedupKey, message)
+}
+
+func StartRealtimeFanoutSubscriber(ctx context.Context, nodeID string, handler func(*RealtimeFanoutEnvelope)) error {
+	if groupMsgRuntime == nil || nodeID == "" || handler == nil {
+		return nil
+	}
+	return groupMsgRuntime.StartRealtimeFanoutSubscriber(ctx, nodeID, handler)
+}
+
 func CacheObject(tenantID types.TenantID, namespace, id string, value any) error {
 	if groupMsgRuntime == nil || !tenantID.IsValid() || namespace == "" || id == "" || value == nil {
 		return nil
@@ -761,28 +788,49 @@ func (r *redisGroupMessageRuntime) GetTopicNodeOnlineMembers(tenantID types.Tena
 }
 
 func (r *redisGroupMessageRuntime) PublishGroupFanout(originNodeID string, message any) error {
+	return r.PublishRealtimeFanout(originNodeID, types.ZeroTenantID, "message", "topic", "", "", "", message)
+}
+
+func (r *redisGroupMessageRuntime) PublishRealtimeFanout(originNodeID string, tenantID types.TenantID, kind, targetType, target, skipSid, dedupKey string, message any) error {
 	if !r.cfg.GroupFanout {
 		return nil
 	}
+	if originNodeID == "" || message == nil {
+		return nil
+	}
 	payload, err := json.Marshal(message)
 	if err != nil {
 		return err
 	}
-	envelope, err := json.Marshal(groupFanoutEnvelope{
+	envelope, err := json.Marshal(RealtimeFanoutEnvelope{
 		OriginNodeID: originNodeID,
+		TenantID:     tenantID,
+		Kind:         kind,
+		TargetType:   targetType,
+		Target:       target,
+		SkipSid:      skipSid,
+		DedupKey:     dedupKey,
 		Message:      payload,
 	})
 	if err != nil {
 		return err
 	}
-	return r.client.Publish(context.Background(), r.groupFanoutChannel(), envelope).Err()
+	return r.client.Publish(context.Background(), r.realtimeFanoutChannel(tenantID), envelope).Err()
 }
 
 func (r *redisGroupMessageRuntime) StartGroupFanoutSubscriber(ctx context.Context, nodeID string, handler func([]byte)) error {
+	return r.StartRealtimeFanoutSubscriber(ctx, nodeID, func(envelope *RealtimeFanoutEnvelope) {
+		if envelope != nil {
+			handler(envelope.Message)
+		}
+	})
+}
+
+func (r *redisGroupMessageRuntime) StartRealtimeFanoutSubscriber(ctx context.Context, nodeID string, handler func(*RealtimeFanoutEnvelope)) error {
 	if !r.cfg.GroupFanout {
 		return nil
 	}
-	pubsub := r.client.Subscribe(ctx, r.groupFanoutChannel())
+	pubsub := r.client.PSubscribe(ctx, r.realtimeFanoutPattern())
 	if _, err := pubsub.Receive(ctx); err != nil {
 		_ = pubsub.Close()
 		return err
@@ -798,11 +846,27 @@ func (r *redisGroupMessageRuntime) StartGroupFanoutSubscriber(ctx context.Contex
 				if !ok {
 					return
 				}
-				var envelope groupFanoutEnvelope
+				var envelope RealtimeFanoutEnvelope
 				if err := json.Unmarshal([]byte(msg.Payload), &envelope); err != nil || envelope.OriginNodeID == nodeID || len(envelope.Message) == 0 {
 					continue
 				}
-				handler(envelope.Message)
+				if envelope.TenantID.IsValid() && envelope.DedupKey != "" {
+					id := fmt.Sprintf("%d:%s", envelope.TenantID, envelope.DedupKey)
+					claimed, err := r.ClaimShortState("rt_dedup", id, 5*time.Minute)
+					if err != nil || !claimed {
+						continue
+					}
+				}
+				if envelope.TargetType == "" {
+					envelope.TargetType = "topic"
+				}
+				if envelope.Kind == "" {
+					envelope.Kind = "message"
+				}
+				if !envelope.TenantID.IsValid() && envelope.Target == "" {
+					continue
+				}
+				handler(&envelope)
 			}
 		}
 	}()
@@ -1035,6 +1099,17 @@ func (r *redisGroupMessageRuntime) groupFanoutChannel() string {
 	return r.key(types.ZeroTenantID, "group_fanout")
 }
 
+func (r *redisGroupMessageRuntime) realtimeFanoutChannel(tenantID types.TenantID) string {
+	if tenantID.IsValid() {
+		return r.key(tenantID, "rt", "fanout")
+	}
+	return r.groupFanoutChannel()
+}
+
+func (r *redisGroupMessageRuntime) realtimeFanoutPattern() string {
+	return strings.TrimSuffix(r.prefix, ":") + ":*:rt:fanout"
+}
+
 func (r *redisGroupMessageRuntime) objectCacheKey(tenantID types.TenantID, namespace, id string) string {
 	return r.key(tenantID, "cache", namespace, id)
 }

server/store/group_message_runtime_test.go

@@ -113,6 +113,14 @@ func (r *testGroupMessageRuntime) StartGroupFanoutSubscriber(context.Context, st
 	return nil
 }
 
+func (r *testGroupMessageRuntime) PublishRealtimeFanout(string, types.TenantID, string, string, string, string, string, any) error {
+	return nil
+}
+
+func (r *testGroupMessageRuntime) StartRealtimeFanoutSubscriber(context.Context, string, func(*RealtimeFanoutEnvelope)) error {
+	return nil
+}
+
 func (r *testGroupMessageRuntime) CacheObject(types.TenantID, string, string, any) error {
 	return nil
 }

server/store/infrastructure.go

@@ -11,9 +11,6 @@ type HTTPInfrastructurePersistenceInterface interface {
 	BeginIdempotency(types.TenantID, *types.HTTPIdempotencyRecord) (*types.HTTPIdempotencyRecord, bool, error)
 	CompleteIdempotency(types.TenantID, *types.HTTPIdempotencyRecord) error
 	AbortIdempotency(types.TenantID, *types.HTTPIdempotencyRecord) error
-	CreateEvent(types.TenantID, *types.BusinessEvent) error
-	ListPendingEvents(types.TenantID, int, time.Time) ([]types.BusinessEvent, error)
-	MarkEventDelivered(types.TenantID, string, time.Time) error
 	BeginKafkaConsume(types.TenantID, *types.KafkaConsumeRecord, time.Time) (*types.KafkaConsumeRecord, bool, error)
 	MarkKafkaConsumeSucceeded(types.TenantID, string, string, time.Time) error
 	MarkKafkaConsumeFailed(types.TenantID, string, string, string, string, time.Time) (*types.KafkaConsumeRecord, error)
@@ -64,30 +61,6 @@ func (httpInfrastructureMapper) AbortIdempotency(tenantID types.TenantID, record
 	return infrastructure.TenantHTTPIdempotencyAbort(tenantID, record)
 }
 
-func (httpInfrastructureMapper) CreateEvent(tenantID types.TenantID, event *types.BusinessEvent) error {
-	infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
-	if err != nil {
-		return err
-	}
-	return infrastructure.TenantBusinessEventCreate(tenantID, event)
-}
-
-func (httpInfrastructureMapper) ListPendingEvents(tenantID types.TenantID, limit int, now time.Time) ([]types.BusinessEvent, error) {
-	infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
-	if err != nil {
-		return nil, err
-	}
-	return infrastructure.TenantBusinessEventListPending(tenantID, limit, now)
-}
-
-func (httpInfrastructureMapper) MarkEventDelivered(tenantID types.TenantID, eventID string, now time.Time) error {
-	infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
-	if err != nil {
-		return err
-	}
-	return infrastructure.TenantBusinessEventMarkDelivered(tenantID, eventID, now)
-}
-
 func (httpInfrastructureMapper) BeginKafkaConsume(tenantID types.TenantID, record *types.KafkaConsumeRecord, deadline time.Time) (*types.KafkaConsumeRecord, bool, error) {
 	infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
 	if err != nil {

server/store/store.go

@@ -1620,6 +1620,7 @@ type GroupBusinessPersistenceInterface interface {
 	CreateSettings(types.TenantID, *types.GroupSettings) (*types.GroupSettings, error)
 	GetSettings(types.TenantID, string) (*types.GroupSettings, error)
 	UpdateSettings(types.TenantID, string, types.GroupSettingsPatch) (*types.GroupSettings, error)
+	Dissolve(types.TenantID, string, types.Uid, string, time.Time) (*types.GroupSettings, error)
 	UpsertMember(types.TenantID, *types.GroupMember) (*types.GroupMember, error)
 	GetMember(types.TenantID, string, types.Uid) (*types.GroupMember, error)
 	ListMembers(types.TenantID, string, types.GroupMemberListOptions) ([]types.GroupMember, string, error)
@@ -1644,8 +1645,6 @@ type GroupBusinessPersistenceInterface interface {
 	UpdateRole(types.TenantID, string, types.Uid, types.Uid, string, types.AccessMode, time.Time) error
 	TransferOwner(types.TenantID, string, types.Uid, types.Uid, int, time.Time) error
 	CanPost(types.TenantID, string, types.Uid, bool, time.Time) (bool, error)
-	ListPendingEvents(types.TenantID, int, time.Time) ([]types.GroupEvent, error)
-	MarkEventDelivered(types.TenantID, string, time.Time) error
 }
 
 type groupBusinessMapper struct{}
@@ -1721,6 +1720,25 @@ func (groupBusinessMapper) UpdateSettings(tenantID types.TenantID, topic string,
 	return settings, err
 }
 
+func (groupBusinessMapper) Dissolve(tenantID types.TenantID, topic string, actor types.Uid, reason string, now time.Time) (*types.GroupSettings, error) {
+	if strings.TrimSpace(topic) == "" || actor.IsZero() {
+		return nil, types.ErrMalformed
+	}
+	tenantAdp, err := tenantBusinessAdapter(tenantID)
+	if err != nil {
+		return nil, err
+	}
+	settings, err := tenantAdp.TenantGroupDissolve(tenantID, strings.TrimSpace(topic), actor, reason, now)
+	if err == nil {
+		if settings != nil {
+			_ = CacheObject(tenantID, cacheNamespaceGroupSettings, strings.TrimSpace(topic), settings)
+		} else {
+			_ = DeleteCachedObject(tenantID, cacheNamespaceGroupSettings, strings.TrimSpace(topic))
+		}
+	}
+	return settings, err
+}
+
 func (groupBusinessMapper) UpsertMember(tenantID types.TenantID, member *types.GroupMember) (*types.GroupMember, error) {
 	if member == nil || member.Topic == "" || member.UserID.IsZero() {
 		return nil, types.ErrMalformed
@@ -2096,28 +2114,6 @@ func (groupBusinessMapper) CanPost(tenantID types.TenantID, topic string, uid ty
 	return tenantAdp.TenantGroupCanPost(tenantID, strings.TrimSpace(topic), uid, admin, now)
 }
 
-func (groupBusinessMapper) ListPendingEvents(tenantID types.TenantID, limit int, now time.Time) ([]types.GroupEvent, error) {
-	if limit <= 0 || limit > 100 {
-		limit = 50
-	}
-	tenantAdp, err := tenantBusinessAdapter(tenantID)
-	if err != nil {
-		return nil, err
-	}
-	return tenantAdp.TenantGroupEventListPending(tenantID, limit, now)
-}
-
-func (groupBusinessMapper) MarkEventDelivered(tenantID types.TenantID, eventID string, now time.Time) error {
-	if strings.TrimSpace(eventID) == "" || now.IsZero() {
-		return types.ErrMalformed
-	}
-	tenantAdp, err := tenantBusinessAdapter(tenantID)
-	if err != nil {
-		return err
-	}
-	return tenantAdp.TenantGroupEventMarkDelivered(tenantID, strings.TrimSpace(eventID), now)
-}
-
 func groupMemberCacheID(topic string, uid types.Uid) string {
 	return strings.TrimSpace(topic) + ":" + uid.String()
 }

server/store/types/groups.go

@@ -63,6 +63,8 @@ type GroupSettingsPatch struct {
 	MemberUploadFiles *bool
 	MemberPrivateChat *bool
 	MemberInvite      *bool
+	MaxMemberCount    *int
+	MaxAdminCount     *int
 	MemberCountDelta  int
 	Version           int
 }
@@ -209,19 +211,3 @@ type GroupFile struct {
 	UploaderRaw    int64      `db:"uploader_id"`
 	DeletedByRaw   *int64     `db:"deleted_by"`
 }
-
-type GroupEvent struct {
-	ID          int64      `db:"id"`
-	EventID     string     `db:"event_id"`
-	Topic       string     `db:"topic"`
-	ActorUserID Uid        `db:"-"`
-	EventType   string     `db:"event_type"`
-	Payload     []byte     `db:"payload"`
-	Status      string     `db:"status"`
-	Attempts    int        `db:"attempts"`
-	NextRetryAt *time.Time `db:"next_retry_at"`
-	DeliveredAt *time.Time `db:"delivered_at"`
-	CreatedAt   time.Time  `db:"created_at"`
-	UpdatedAt   time.Time  `db:"updated_at"`
-	ActorRaw    int64      `db:"actor_user_id"`
-}

server/store/types/infrastructure.go

@@ -10,9 +10,6 @@ const (
 	HTTPIdempotencyProcessing = "processing"
 	HTTPIdempotencyCompleted  = "completed"
 
-	BusinessEventPending   = "pending"
-	BusinessEventDelivered = "delivered"
-
 	KafkaConsumeProcessing = "processing"
 	KafkaConsumeSucceeded  = "succeeded"
 	KafkaConsumeFailed     = "failed"
@@ -40,23 +37,6 @@ type HTTPIdempotencyRecord struct {
 	ExpiresAt      time.Time
 }
 
-// BusinessEvent is a durable event destined for an online user or topic.
-type BusinessEvent struct {
-	ID          int64
-	EventID     string
-	TargetType  string
-	Target      string
-	ActorUserID Uid
-	EventType   string
-	Payload     []byte
-	Status      string
-	Attempts    int
-	NextRetryAt *time.Time
-	DeliveredAt *time.Time
-	CreatedAt   time.Time
-	UpdatedAt   time.Time
-}
-
 // KafkaConsumeRecord stores idempotent consumer state for one event and group.
 type KafkaConsumeRecord struct {
 	EventID            string

server/tinode.conf

@@ -15,6 +15,7 @@
   "permanent_accounts": false,
   "expvar": "/debug/vars",
   "server_status": "-",
+  "node_id": "",
   "use_x_forwarded_for": true,
   "x_frame_options": "SAMEORIGIN",
   "default_country_code": "CN",

server/topic.go

@@ -1177,11 +1177,7 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid,
 	pluginMessage(t.tenantID, data.Data, plgActCreate)
 
 	t.broadcastToSessions(data)
-	if t.cat == types.TopicCatGrp {
-		if err := store.PublishGroupFanout(currentNodeID(), data); err != nil {
-			logs.Warn.Printf("topic[%s]: failed to publish redis group fanout: %v", t.name, err)
-		}
-	}
+	publishRealtimeTopicMessage(t.name, data)
 
 	// sendPush will update unread message count and send push notification.
 	if pushRcpt := t.pushForData(asUid, data.Data, markedReadBySender); pushRcpt != nil {
@@ -1260,10 +1256,7 @@ func (t *Topic) handlePubBroadcast(msg *ClientComMessage) {
 	}
 
 	// Save to DB at master topic.
-	var attachments []string
-	if msg.Extra != nil && len(msg.Extra.Attachments) > 0 {
-		attachments = msg.Extra.Attachments
-	}
+	attachments := messageAttachmentURLs(msg)
 
 	if err := t.saveAndBroadcastMessage(msg, asUid, msg.Pub.NoEcho, attachments, msg.Pub.Head, msg.Pub.Content); err != nil {
 		logs.Err.Printf("topic[%s]: failed to save messagge - %s", t.name, err)
@@ -1275,6 +1268,105 @@ func (t *Topic) handlePubBroadcast(msg *ClientComMessage) {
 	}
 }
 
+func messageAttachmentURLs(msg *ClientComMessage) []string {
+	if msg == nil || msg.Pub == nil {
+		return nil
+	}
+	attachments := make([]string, 0)
+	seen := make(map[string]bool)
+	add := func(raw string) {
+		url, key := normalizeAttachmentURL(raw)
+		if url == "" || seen[key] {
+			return
+		}
+		seen[key] = true
+		attachments = append(attachments, url)
+	}
+	if msg.Extra != nil {
+		for _, attachment := range msg.Extra.Attachments {
+			add(attachment)
+		}
+	}
+	for _, attachment := range inferAttachmentURLsFromContent(msg.Pub.Content) {
+		add(attachment)
+	}
+	return attachments
+}
+
+func inferAttachmentURLsFromContent(content any) []string {
+	var attachments []string
+	seen := make(map[string]bool)
+	addFileID := func(raw string) {
+		url, key := normalizeAttachmentURL(raw)
+		if url != "" && !seen[key] {
+			seen[key] = true
+			attachments = append(attachments, url)
+		}
+	}
+
+	var walk func(any, int)
+	walk = func(value any, depth int) {
+		if value == nil || depth <= 0 {
+			return
+		}
+		switch typed := value.(type) {
+		case map[string]any:
+			if fileID, ok := typed["file_id"].(string); ok {
+				addFileID(fileID)
+			}
+			if ref, ok := typed["ref"].(string); ok {
+				addFileID(ref)
+			}
+			for _, child := range typed {
+				walk(child, depth-1)
+			}
+		case types.KVMap:
+			walk(map[string]any(typed), depth)
+		case []any:
+			for _, child := range typed {
+				walk(child, depth-1)
+			}
+		}
+	}
+	walk(content, 8)
+	return attachments
+}
+
+func normalizeAttachmentURL(raw string) (string, string) {
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return "", ""
+	}
+	if fid := parseAttachmentFileID(raw); !fid.IsZero() {
+		return "/v0/file/s/" + fid.String(), fid.String()
+	}
+	return raw, raw
+}
+
+func parseAttachmentFileID(raw string) types.Uid {
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return types.ZeroUid
+	}
+	if fid := types.ParseUid(raw); !fid.IsZero() {
+		return fid
+	}
+	if idx := strings.LastIndex(raw, "/v0/file/s/"); idx >= 0 {
+		tail := raw[idx+len("/v0/file/s/"):]
+		if cut := strings.IndexAny(tail, "?#/"); cut >= 0 {
+			tail = tail[:cut]
+		}
+		if dot := strings.IndexByte(tail, '.'); dot >= 0 {
+			tail = tail[:dot]
+		}
+		return types.ParseUid(tail)
+	}
+	if dot := strings.IndexByte(raw, '.'); dot >= 0 {
+		return types.ParseUid(raw[:dot])
+	}
+	return types.ZeroUid
+}
+
 func (t *Topic) validateMessageReplacement(asUid types.Uid, head map[string]any, now time.Time) error {
 	if head == nil || head["replace"] == nil {
 		return nil

server/topic_test.go

@@ -3540,6 +3540,61 @@ func TestHandleBroadcastDataWithAttachments(t *testing.T) {
 	}
 }
 
+func TestHandleBroadcastDataInfersAttachmentFromContentFileID(t *testing.T) {
+	numUsers := 2
+	helper := TopicTestHelper{}
+	helper.setUp(t, numUsers, types.TopicCatP2P, "p2p-test", true)
+	defer helper.tearDown()
+	helper.mm.EXPECT().Save(types.TenantID(1), gomock.Any(), []string{"/v0/file/s/KJ0Qlrk4Gzw"}, gomock.Any()).
+		Return(&types.MessageSaveResult{SeqId: 1, MarkedReadBySender: true}, nil)
+
+	from := helper.uids[0].UserId()
+	msg := &ClientComMessage{
+		AsUser:   from,
+		Original: from,
+		Pub: &MsgClientPub{
+			Topic: "p2p",
+			Content: map[string]any{
+				"file_id":     "KJ0Qlrk4Gzw",
+				"mime":        "audio/mpeg",
+				"size":        14143,
+				"duration_ms": 3719,
+			},
+			Head: map[string]any{
+				"x-im-type": "voice",
+			},
+			NoEcho: true,
+		},
+		sess: helper.sessions[0],
+	}
+	helper.topic.handleClientMsg(msg)
+	helper.finish()
+
+	if errorMsgs, hasError := helper.hubMessages["__ERROR__"]; hasError {
+		t.Fatal(errorMsgs[0].Ctrl.Text)
+	}
+	if len(helper.results[1].messages) != 1 {
+		t.Fatalf("Uid2: expected 1 message, got %d", len(helper.results[1].messages))
+	}
+}
+
+func TestMessageAttachmentURLsDeduplicatesExplicitAndContentFileID(t *testing.T) {
+	msg := &ClientComMessage{
+		Extra: &MsgClientExtra{Attachments: []string{"/v0/file/s/KJ0Qlrk4Gzw"}},
+		Pub: &MsgClientPub{
+			Content: map[string]any{
+				"file_id": "KJ0Qlrk4Gzw",
+				"mime":    "audio/mpeg",
+			},
+		},
+	}
+
+	attachments := messageAttachmentURLs(msg)
+	if len(attachments) != 1 || attachments[0] != "/v0/file/s/KJ0Qlrk4Gzw" {
+		t.Fatalf("unexpected attachments: %#v", attachments)
+	}
+}
+
 func TestHandleBroadcastInfoChannelWithMultipleReaders(t *testing.T) {
 	topicName := "grpTest"
 	chanName := "chnTest"

server/user.go

@@ -298,7 +298,6 @@ func replyCreateUser(s *Session, msg *ClientComMessage, rec *auth.Rec) {
 			}
 			return
 		}
-		dispatchIMBusinessEvents(s.tenantID)
 	}
 
 	if msg.Extra != nil && len(msg.Extra.Attachments) > 0 {
@@ -826,6 +825,7 @@ func changeUserState(s *Session, uid types.Uid, user *types.User, msg *ClientCom
 	if state != types.StateOK {
 		// Terminate all sessions.
 		globals.sessionStore.EvictUser(uid, "")
+		publishRealtimeEvictUser(s.tenantID, uid, "user_state_changed", "")
 	}
 
 	err = store.Users.UpdateState(s.tenantID, uid, state)
@@ -901,6 +901,7 @@ func replyDelUser(s *Session, msg *ClientComMessage) {
 
 	// Terminate all sessions. Skip the current session so the requester gets a response.
 	globals.sessionStore.EvictUser(uid, s.sid)
+	publishRealtimeEvictUser(s.tenantID, uid, "user_deleted", s.sid)
 	// Remove user from cache and announce to cluster that the user is deleted.
 	usersRemoveUser(msg.TenantID, uid)