From a9afc6ff6ac856bdb1f7a0555f5c988eed111185 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sun, 13 Sep 2026 16:02:23 +0800 Subject: [PATCH] fix: align group management APIs --- docs/IM-前后端对接文档.md | 11 +- loadtest/realtime_fanout_probe.go | 377 ++++++++++++++++++ server/db/adapter.go | 6 +- server/db/mysql/infrastructure_schema.go | 12 - server/db/mysql/infrastructure_schema_test.go | 8 +- server/db/mysql/registration_invites.go | 8 +- server/db/mysql/schema.sql | 2 +- server/db/mysql/tenant_adapter.go | 40 -- server/db/mysql/tenant_infrastructure.go | 101 ----- server/db/mysql/tests/mysql_test.go | 4 +- server/im_business_events.go | 182 ++------- server/im_business_events_test.go | 120 ------ server/im_group_events.go | 99 ++--- server/internal_admin_query_resources.go | 15 +- server/internal_api.go | 90 ++++- server/kafka_admin_commands.go | 60 ++- server/kafka_workbench_audit_test.go | 5 - server/main.go | 9 +- server/meetingim.conf | 1 + server/metrics/metrics.go | 34 +- server/realtime_dispatcher.go | 238 +++++++++++ server/realtime_dispatcher_test.go | 126 ++++++ server/session.go | 17 + server/sessionstore.go | 84 ++++ server/store/group_message_runtime.go | 85 +++- server/store/group_message_runtime_test.go | 8 + server/store/infrastructure.go | 27 -- server/store/store.go | 44 +- server/store/types/groups.go | 18 +- server/store/types/infrastructure.go | 20 - server/tinode.conf | 1 + server/topic.go | 110 ++++- server/topic_test.go | 55 +++ server/user.go | 3 +- 34 files changed, 1381 insertions(+), 639 deletions(-) create mode 100644 loadtest/realtime_fanout_probe.go delete mode 100644 server/im_business_events_test.go create mode 100644 server/realtime_dispatcher.go create mode 100644 server/realtime_dispatcher_test.go diff --git a/docs/IM-前后端对接文档.md b/docs/IM-前后端对接文档.md index 8d71d36..19c4d3d 100644 --- a/docs/IM-前后端对接文档.md +++ b/docs/IM-前后端对接文档.md @@ -931,11 +931,11 @@ X-MeetingIM-APIKey: <预置的普通 API Key> **状态：新增。** -- 敏感操作先调用 `POST /v1/im/security/challenges`；开启青少年模式设置独立密码时 Body 为 `{"scene":"set_independent_password","channel":"sms"}`，关闭青少年模式时 Body 为 `{"scene":"close_youth_mode","channel":"sms"}`。手机号由登录身份读取，客户端不提交目标手机号。 +- 敏感操作先调用 `POST /v1/im/security/challenges`；开启青少年模式设置独立密码时 Body 为 `{"scene":"set_independent_password","channel":"sms"}`，关闭青少年模式时 Body 为 `{"scene":"close_youth_mode","channel":"sms"}`；群主转让使用 `{"scene":"group_owner_transfer","channel":"sms"}`；解散群聊使用 `{"scene":"group_dissolve","channel":"sms"}`。手机号由登录身份读取，客户端不提交目标手机号。 - 发码/校验：服务端返回 `challenge_id` 和脱敏手机号；提交 `POST /v1/im/security/challenges/{id}/verify`，Body `{"code":"..."}`。 - 重复发码：同一用户、同一手机号、同一敏感场景存在未过期 challenge 且仍在 `resend_after_seconds` 冷却期内时，服务端返回 HTTP `429 VERIFICATION_RATE_LIMITED`，`details` 带 `challenge_id` 和 `retry_after_seconds`。客户端应复用该 `challenge_id` 继续倒计时，不要当成接口不可达或重新登录。 - 小时限流：超过一小时创建次数上限时同样返回 HTTP `429 VERIFICATION_RATE_LIMITED`，`details.retry_after_seconds` 可能大于 60 秒且不一定带 `challenge_id`。客户端只展示倒计时，倒计时结束后允许用户重新获取验证码。 -- 成功返回 `action_token`、`action`、`expires_at`；Token 只能用于指定动作且单次消费。设置独立密码接口通过 Body `action_token` 携带；关闭青少年模式接口通过 Header `X-Action-Token` 携带。 +- 成功返回 `action_token`、`action`、`expires_at`；Token 只能用于指定动作且单次消费。设置独立密码接口通过 Body `action_token` 携带；关闭青少年模式、群主转让、解散群聊接口通过 Header `X-Action-Token` 携带。 - 错误：`VERIFICATION_RATE_LIMITED`、`CHALLENGE_EXPIRED`、`CODE_INVALID`、`TOO_MANY_ATTEMPTS`；前端不得把短信验证码或 action token 落普通存储。 ### 1.8 实名认证.png @@ -1418,7 +1418,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `PATCH /v1/im/conversations/{topic}/settings` 待实现 | `{"pinned":true,"mute_notifications":false,"version":1}` | `{"pinned":true,"mute_notifications":false,"message_retention":"default","blocked":false,"remark":"","version":2}` | | `GET /v1/im/conversations/{topic}/messages` 待实现 | Query `?since=12&limit=20` | `{"topic":"p2p...","items":[{"seq":13,"from":"usr...","ts":"2026-08-20T09:00:00Z","head":{"x-im-type":"text"},"content":{"text":"你好"}}],"next_cursor":""}` | | `PUT /v1/im/topics/{topic}/draft` 待实现 | `{"content":"未发送草稿","version":1}` | `{"topic":"grp...","content":"未发送草稿","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | -| `POST /v1/im/topics/{topic}/messages/{seq}/recall` | `{"version":1,"reason":"mistake"}` | `{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | +| `POST /v1/im/topics/{topic}/messages/{seq}/recall` | `{}`；可选 `{"version":1,"reason":"mistake"}` | `{"topic":"grp...","seq":12,"status":"recalled","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `POST /v1/im/topics/{topic}/messages/{seq}/revisions` | `{"content":{"text":"修改后的文本"},"client_mid":"mid_1","version":1}` | `{"topic":"grp...","seq":12,"status":"edited","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `GET /v1/im/topics/{topic}/messages/{seq}/receipts?cursor=` | Query only | `{"topic":"grp...","seq":12,"items":[{"user_id":"usr...","read_at":"2026-08-20T09:00:00Z","recv_at":"2026-08-20T08:59:00Z"}],"next_cursor":""}` | | `GET /v1/im/topics/{topic}/messages/search?q=&type=text&cursor=` | Query only | `{"topic":"grp...","items":[{"seq":12,"from":"usr...","ts":"2026-08-20T09:00:00Z","snippet":"命中的文本","matches":[{"start":0,"end":2}]}],"next_cursor":""}` | @@ -1901,7 +1901,7 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | `POST /v1/im/groups` | `{"name":"项目群","description":"项目沟通","member_user_ids":["usr..."],"avatar_file_id":""}` | `{"topic":"grp...","name":"项目群","avatar_file_id":"","description":"项目沟通","owner":"usr_me","member_count":2,"version":1,"my_role":"owner","permissions":{"can_edit_profile":true,"can_invite_members":true,"can_manage_members":true,"can_manage_settings":true,"can_leave":false}}` | | `GET /v1/im/groups/{topic}` | Path topic | `{"topic":"grp...","name":"项目群","avatar_file_id":"","description":"项目沟通","owner":"usr_me","member_count":2,"version":1,"my_role":"owner","permissions":{"can_edit_profile":true}}` | | `PATCH /v1/im/groups/{topic}` | `{"name":"新群名","avatar_file_id":"file_avatar","description":"新简介","version":1}` | `{"topic":"grp...","name":"新群名","avatar_file_id":"file_avatar","description":"新简介","version":2,"permissions":{"can_edit_profile":true}}` | -| `DELETE /v1/im/groups/{topic}` 待实现 | Header `X-Action-Token: action_token`；Body `{}` | `{"topic":"grp...","deleted":true}` | +| `DELETE /v1/im/groups/{topic}` | Header `X-Action-Token: action_token`；Body `{}` | `{"topic":"grp...","deleted":true,"status":"dissolved","version":2}` | | `GET /v1/im/groups/{topic}/settings` | Path topic | `{"topic":"grp...","profile":{"name":"项目群","description":"项目沟通"},"pinned":false,"mute_notifications":false,"member_count":2,"permissions":{"can_manage_settings":true},"governance":{"join_approval":false,"mute_all":false,"member_invite":true}}` | | `PATCH /v1/im/groups/{topic}/member-settings` | `{"pinned":true,"mute_notifications":false,"version":1}` | `{"topic":"grp...","pinned":true,"mute_notifications":false,"version":2,"updated_at":"2026-08-20T09:00:00Z"}` | | `PUT /v1/im/groups/{topic}/members/me/profile` | `{"nickname":"我在群里的昵称","version":1}` | `{"nickname":"我在群里的昵称","version":2,"updated_at":"2026-08-20T09:00:00Z"}` | @@ -2721,8 +2721,11 @@ X-MeetingIM-APIKey: <预置的普通 API Key> **状态：新增 App 配置接口。** +完整接口与 Android OSS APK 对接流程见：[IM-App版本更新前后端对接文档](./IM-App版本更新前后端对接文档.md)。 + - `GET /v1/im/app/releases/latest?platform=ios|android¤t_version=1.0.0&channel=stable`，全局公共接口，不依赖租户解析。`current_version` 必传，使用数字点分版本号（如 `1.2.0`）。 - 服务端返回 `has_update`、`force_update` 和 `update_type=none|recommended|force`；iOS 客户端使用 `store_url` 跳转 App Store。 +- Android 客户端使用 `package_url` 下载 OSS/CDN 上的 APK，并在安装前使用 `checksum` 完成 SHA-256 校验；强制更新时不得绕过进入主业务。 - 版本策略由 Admin 在 `hotgo.im_app_releases` 统一管理，IM Server 通过 HMAC 内部配置接口读取，不从租户库查询。 - 返回 `current_version,has_update,force_update,latest_version,min_supported_version,update_type=none|recommended|force,release_notes,store_url,package_url,checksum,published_at`。 - 客户端按平台只打开受信任商店/签名下载地址；Android 包必须校验签名和 checksum，服务端 URL 使用允许域名。 diff --git a/loadtest/realtime_fanout_probe.go b/loadtest/realtime_fanout_probe.go new file mode 100644 index 0000000..02cc6db --- /dev/null +++ b/loadtest/realtime_fanout_probe.go @@ -0,0 +1,377 @@ +package main + +import ( + "bufio" + "context" + "crypto/tls" + "encoding/base64" + "encoding/csv" + "encoding/json" + "flag" + "fmt" + "log" + "math" + "net/http" + "net/url" + "os" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/gorilla/websocket" +) + +type account struct { + Username string + Password string + Token string +} + +type client struct { + id int + account account + wsURL string + conn *websocket.Conn + sendMu sync.Mutex + stats *probeStats +} + +type probeStats struct { + expected atomic.Int64 + received atomic.Int64 + acks atomic.Int64 + errors atomic.Int64 + + mu sync.Mutex + latencies []time.Duration + seen map[string]struct{} +} + +func main() { + var wsList string + var apiKey string + var accountsPath string + var topic string + var subscribers int + var publishers int + var rate int + var duration time.Duration + var ramp time.Duration + var insecure bool + + flag.StringVar(&wsList, "ws", "ws://localhost:6060/v0/channels", "Comma-separated websocket endpoints, one or more pods/LBs.") + flag.StringVar(&apiKey, "apikey", "", "Tinode API key appended to websocket URL.") + flag.StringVar(&accountsPath, "accounts", "loadtest/users.csv", "CSV: username,password[,token].") + flag.StringVar(&topic, "topic", "", "Group topic to subscribe and publish to, e.g. grpABC.") + flag.IntVar(&subscribers, "subscribers", 100, "Number of websocket subscribers.") + flag.IntVar(&publishers, "publishers", 1, "Number of publishers selected from subscribers.") + flag.IntVar(&rate, "rate", 10, "Total publish rate, messages per second.") + flag.DurationVar(&duration, "duration", time.Minute, "Publish duration.") + flag.DurationVar(&ramp, "ramp", 10*time.Second, "Subscriber connection ramp duration.") + flag.BoolVar(&insecure, "insecure", false, "Skip TLS verification for wss endpoints.") + flag.Parse() + + if topic == "" || subscribers <= 0 || publishers <= 0 || rate <= 0 || duration <= 0 { + log.Fatal("topic, subscribers, publishers, rate, and duration are required and must be positive") + } + endpoints := parseEndpoints(wsList, apiKey) + accounts, err := readAccounts(accountsPath) + if err != nil { + log.Fatal(err) + } + if len(accounts) < subscribers { + log.Fatalf("need at least %d accounts, got %d", subscribers, len(accounts)) + } + if publishers > subscribers { + publishers = subscribers + } + + stats := &probeStats{seen: make(map[string]struct{})} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + clients := make([]*client, subscribers) + var wg sync.WaitGroup + for i := 0; i < subscribers; i++ { + c := &client{id: i, account: accounts[i], wsURL: endpoints[i%len(endpoints)], stats: stats} + clients[i] = c + wg.Add(1) + go func() { + defer wg.Done() + if err := c.connectAndSubscribe(ctx, topic, insecure); err != nil { + stats.errors.Add(1) + log.Printf("client %d connect failed: %v", c.id, err) + } + }() + if ramp > 0 { + time.Sleep(ramp / time.Duration(subscribers)) + } + } + wg.Wait() + active := 0 + for _, c := range clients { + if c.conn != nil { + active++ + } + } + if active == 0 { + log.Fatal("no active websocket clients") + } + log.Printf("subscribed clients=%d publishers=%d endpoints=%d topic=%s", active, publishers, len(endpoints), topic) + + publishDone := make(chan struct{}) + go publishLoop(ctx, clients[:publishers], topic, rate, duration, stats, publishDone) + <-publishDone + log.Printf("publish finished; waiting 10s for fanout drain") + time.Sleep(10 * time.Second) + cancel() + for _, c := range clients { + if c.conn != nil { + _ = c.conn.Close() + } + } + printSummary(stats, active) +} + +func (c *client) connectAndSubscribe(ctx context.Context, topic string, insecure bool) error { + dialer := websocket.DefaultDialer + if insecure { + dialer.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} + } + conn, _, err := dialer.DialContext(ctx, c.wsURL, http.Header{}) + if err != nil { + return err + } + c.conn = conn + if err = c.sendAndWaitCtrl("hi", map[string]any{"hi": map[string]any{"id": c.msgID("hi"), "ver": "0.22.8", "ua": "realtime-fanout-probe/1.0"}}); err != nil { + return err + } + if c.account.Token != "" { + if err = c.sendAndWaitCtrl("login", map[string]any{"login": map[string]any{"id": c.msgID("login"), "scheme": "token", "secret": c.account.Token}}); err != nil { + return err + } + } else { + secret := base64.StdEncoding.EncodeToString([]byte(c.account.Username + ":" + c.account.Password)) + if err = c.sendAndWaitCtrl("login", map[string]any{"login": map[string]any{"id": c.msgID("login"), "scheme": "basic", "secret": secret}}); err != nil { + return err + } + } + if err = c.sendAndWaitCtrl("sub", map[string]any{"sub": map[string]any{"id": c.msgID("sub"), "topic": topic, "get": map[string]any{"what": "desc"}}}); err != nil { + return err + } + go c.readLoop(ctx) + return nil +} + +func (c *client) sendAndWaitCtrl(name string, msg map[string]any) error { + c.sendMu.Lock() + if err := c.conn.WriteJSON(msg); err != nil { + c.sendMu.Unlock() + return err + } + c.sendMu.Unlock() + deadline := time.Now().Add(15 * time.Second) + for time.Now().Before(deadline) { + var raw map[string]json.RawMessage + if err := c.conn.ReadJSON(&raw); err != nil { + return err + } + if ctrlRaw, ok := raw["ctrl"]; ok { + var ctrl struct { + Code int `json:"code"` + Text string `json:"text"` + } + _ = json.Unmarshal(ctrlRaw, &ctrl) + if ctrl.Code >= 200 && ctrl.Code < 300 { + return nil + } + return fmt.Errorf("%s ctrl code=%d text=%s", name, ctrl.Code, ctrl.Text) + } + } + return fmt.Errorf("%s ctrl timeout", name) +} + +func (c *client) readLoop(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + default: + } + var raw map[string]json.RawMessage + if err := c.conn.ReadJSON(&raw); err != nil { + return + } + if _, ok := raw["ctrl"]; ok { + c.stats.acks.Add(1) + continue + } + dataRaw, ok := raw["data"] + if !ok { + continue + } + var data struct { + Content json.RawMessage `json:"content"` + } + if err := json.Unmarshal(dataRaw, &data); err != nil || len(data.Content) == 0 { + continue + } + var content struct { + ProbeID string `json:"probe_id"` + SentAt int64 `json:"sent_at_unix_nano"` + } + if err := json.Unmarshal(data.Content, &content); err != nil || content.ProbeID == "" || content.SentAt <= 0 { + continue + } + c.stats.record(content.ProbeID, time.Since(time.Unix(0, content.SentAt))) + } +} + +func publishLoop(ctx context.Context, publishers []*client, topic string, rate int, duration time.Duration, stats *probeStats, done chan<- struct{}) { + defer close(done) + interval := time.Second / time.Duration(rate) + if interval <= 0 { + interval = time.Nanosecond + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + deadline := time.Now().Add(duration) + seq := int64(0) + for time.Now().Before(deadline) { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + n := atomic.AddInt64(&seq, 1) + publisher := publishers[int(n)%len(publishers)] + probeID := fmt.Sprintf("probe-%d-%d", time.Now().UnixNano(), n) + msg := map[string]any{"pub": map[string]any{ + "id": publisher.msgID("pub-" + strconv.FormatInt(n, 10)), + "topic": topic, + "head": map[string]any{"mime": "application/json"}, + "content": map[string]any{ + "probe_id": probeID, + "sent_at_unix_nano": time.Now().UnixNano(), + "seq": n, + }, + }} + publisher.sendMu.Lock() + err := publisher.conn.WriteJSON(msg) + publisher.sendMu.Unlock() + if err != nil { + stats.errors.Add(1) + log.Printf("publish failed client=%d seq=%d: %v", publisher.id, n, err) + continue + } + stats.expected.Add(1) + } +} + +func (s *probeStats) record(probeID string, latency time.Duration) { + s.mu.Lock() + defer s.mu.Unlock() + s.received.Add(1) + s.seen[probeID] = struct{}{} + s.latencies = append(s.latencies, latency) +} + +func printSummary(stats *probeStats, activeSubscribers int) { + stats.mu.Lock() + latencies := append([]time.Duration(nil), stats.latencies...) + unique := len(stats.seen) + stats.mu.Unlock() + sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] }) + published := stats.expected.Load() + expectedDeliveries := published * int64(activeSubscribers) + received := stats.received.Load() + lost := expectedDeliveries - received + lossRate := 0.0 + if expectedDeliveries > 0 { + lossRate = float64(lost) / float64(expectedDeliveries) * 100 + } + fmt.Println("==== realtime fanout probe summary ====") + fmt.Printf("published=%d active_subscribers=%d expected_deliveries=%d received=%d lost=%d loss_rate=%.4f%% unique_messages_seen=%d errors=%d ctrl_acks=%d\n", + published, activeSubscribers, expectedDeliveries, received, lost, lossRate, unique, stats.errors.Load(), stats.acks.Load()) + fmt.Printf("latency p50=%s p95=%s p99=%s max=%s samples=%d\n", + percentile(latencies, 50), percentile(latencies, 95), percentile(latencies, 99), maxDuration(latencies), len(latencies)) +} + +func percentile(values []time.Duration, p int) time.Duration { + if len(values) == 0 { + return 0 + } + idx := int(math.Ceil(float64(p)/100*float64(len(values)))) - 1 + if idx < 0 { + idx = 0 + } + if idx >= len(values) { + idx = len(values) - 1 + } + return values[idx] +} + +func maxDuration(values []time.Duration) time.Duration { + if len(values) == 0 { + return 0 + } + return values[len(values)-1] +} + +func (c *client) msgID(suffix string) string { + return fmt.Sprintf("rtp-%d-%s", c.id, suffix) +} + +func parseEndpoints(raw string, apiKey string) []string { + parts := strings.Split(raw, ",") + out := make([]string, 0, len(parts)) + for _, part := range parts { + endpoint := strings.TrimSpace(part) + if endpoint == "" { + continue + } + u, err := url.Parse(endpoint) + if err != nil { + log.Fatalf("bad ws endpoint %q: %v", endpoint, err) + } + q := u.Query() + if apiKey != "" && q.Get("apikey") == "" { + q.Set("apikey", apiKey) + } + u.RawQuery = q.Encode() + out = append(out, u.String()) + } + if len(out) == 0 { + log.Fatal("no websocket endpoints") + } + return out +} + +func readAccounts(path string) ([]account, error) { + file, err := os.Open(path) + if err != nil { + return nil, err + } + defer file.Close() + reader := csv.NewReader(bufio.NewReader(file)) + reader.FieldsPerRecord = -1 + rows, err := reader.ReadAll() + if err != nil { + return nil, err + } + accounts := make([]account, 0, len(rows)) + for _, row := range rows { + if len(row) < 2 || strings.HasPrefix(strings.TrimSpace(row[0]), "#") { + continue + } + acc := account{Username: strings.TrimSpace(row[0]), Password: strings.TrimSpace(row[1])} + if len(row) > 2 { + acc.Token = strings.TrimSpace(row[2]) + } + accounts = append(accounts, acc) + } + return accounts, nil +} diff --git a/server/db/adapter.go b/server/db/adapter.go index 92eb7c4..53b96f1 100644 --- a/server/db/adapter.go +++ b/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) diff --git a/server/db/mysql/infrastructure_schema.go b/server/db/mysql/infrastructure_schema.go index 179ca9d..362c7e2 100644 --- a/server/db/mysql/infrastructure_schema.go +++ b/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, diff --git a/server/db/mysql/infrastructure_schema_test.go b/server/db/mysql/infrastructure_schema_test.go index e8d8f24..45eadaa 100644 --- a/server/db/mysql/infrastructure_schema_test.go +++ b/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) + } } diff --git a/server/db/mysql/registration_invites.go b/server/db/mysql/registration_invites.go index 3bba7ab..92e9131 100644 --- a/server/db/mysql/registration_invites.go +++ b/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 { diff --git a/server/db/mysql/schema.sql b/server/db/mysql/schema.sql index f233ab5..a79084f 100644 --- a/server/db/mysql/schema.sql +++ b/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 # diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index cef1606..945df5e 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/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 } diff --git a/server/db/mysql/tenant_infrastructure.go b/server/db/mysql/tenant_infrastructure.go index b11092e..fdb1083 100644 --- a/server/db/mysql/tenant_infrastructure.go +++ b/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 { diff --git a/server/db/mysql/tests/mysql_test.go b/server/db/mysql/tests/mysql_test.go index 0cf7461..1407690 100644 --- a/server/db/mysql/tests/mysql_test.go +++ b/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 diff --git a/server/im_business_events.go b/server/im_business_events.go index 99bb4f4..0b83bf9 100644 --- a/server/im_business_events.go +++ b/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 -} diff --git a/server/im_business_events_test.go b/server/im_business_events_test.go deleted file mode 100644 index bf50a18..0000000 --- a/server/im_business_events_test.go +++ /dev/null @@ -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") - } -} diff --git a/server/im_group_events.go b/server/im_group_events.go index 6cb57e0..a8397e6 100644 --- a/server/im_group_events.go +++ b/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 } diff --git a/server/internal_admin_query_resources.go b/server/internal_admin_query_resources.go index b457219..f860bf6 100644 --- a/server/internal_admin_query_resources.go +++ b/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'", diff --git a/server/internal_api.go b/server/internal_api.go index 480a18d..77395c9 100644 --- a/server/internal_api.go +++ b/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) diff --git a/server/kafka_admin_commands.go b/server/kafka_admin_commands.go index ce2d9f8..653f832 100644 --- a/server/kafka_admin_commands.go +++ b/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) diff --git a/server/kafka_workbench_audit_test.go b/server/kafka_workbench_audit_test.go index 2aba4ea..aba1041 100644 --- a/server/kafka_workbench_audit_test.go +++ b/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) - } } diff --git a/server/main.go b/server/main.go index 8ffb6b2..9e71119 100644 --- a/server/main.go +++ b/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 { diff --git a/server/meetingim.conf b/server/meetingim.conf index b1d3e88..b4af5c5 100644 --- a/server/meetingim.conf +++ b/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", diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index 48d6dbc..70213e5 100644 --- a/server/metrics/metrics.go +++ b/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 { diff --git a/server/realtime_dispatcher.go b/server/realtime_dispatcher.go new file mode 100644 index 0000000..abfccde --- /dev/null +++ b/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 "" +} diff --git a/server/realtime_dispatcher_test.go b/server/realtime_dispatcher_test.go new file mode 100644 index 0000000..9404d75 --- /dev/null +++ b/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 +} diff --git a/server/session.go b/server/session.go index 8ddd3fb..029c76f 100644 --- a/server/session.go +++ b/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" } diff --git a/server/sessionstore.go b/server/sessionstore.go index 692ee97..20c00de 100644 --- a/server/sessionstore.go +++ b/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() { diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index 0d4c017..3f0260a 100644 --- a/server/store/group_message_runtime.go +++ b/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) } diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index dfa1df7..bb57127 100644 --- a/server/store/group_message_runtime_test.go +++ b/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 } diff --git a/server/store/infrastructure.go b/server/store/infrastructure.go index 52a9a18..d0233e6 100644 --- a/server/store/infrastructure.go +++ b/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 { diff --git a/server/store/store.go b/server/store/store.go index c6ea0d3..2947c8f 100644 --- a/server/store/store.go +++ b/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() } diff --git a/server/store/types/groups.go b/server/store/types/groups.go index e9c7bcd..7cd174c 100644 --- a/server/store/types/groups.go +++ b/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"` -} diff --git a/server/store/types/infrastructure.go b/server/store/types/infrastructure.go index 261a9a2..5d97649 100644 --- a/server/store/types/infrastructure.go +++ b/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 diff --git a/server/tinode.conf b/server/tinode.conf index b1d3e88..b4af5c5 100644 --- a/server/tinode.conf +++ b/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", diff --git a/server/topic.go b/server/topic.go index c99d902..d7112cd 100644 --- a/server/topic.go +++ b/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 diff --git a/server/topic_test.go b/server/topic_test.go index bdcfe8d..2870ab4 100644 --- a/server/topic_test.go +++ b/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" diff --git a/server/user.go b/server/user.go index 372cc55..06528fd 100644 --- a/server/user.go +++ b/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)