diff --git a/server/db/mysql/group_schema_test.go b/server/db/mysql/group_schema_test.go index 76f6dca..90447d0 100644 --- a/server/db/mysql/group_schema_test.go +++ b/server/db/mysql/group_schema_test.go @@ -4,6 +4,7 @@ package mysql import ( "database/sql" + "os" "strings" "testing" @@ -150,3 +151,23 @@ func TestTenantShardSQL(t *testing.T) { t.Fatalf("tenant shard query rewrote index name: %s", query) } } + +func TestSharedDBGroupFileLinkIncludesTenantID(t *testing.T) { + source, err := os.ReadFile("tenant_groups.go") + if err != nil { + t.Fatal(err) + } + body := sourceBetween(t, string(source), "func (a *adapter) TenantGroupFileCreate", "\nfunc (a *adapter) enrichGroupFileFromUpload") + for _, required := range []string{ + "INSERT INTO im_filemsglinks(tenant_id,createdat,fileid,topic)", + "SELECT ?,?,?,? WHERE NOT EXISTS", + "SELECT 1 FROM im_filemsglinks WHERE tenant_id=? AND fileid=? AND topic=?)", + } { + if !strings.Contains(body, required) { + t.Fatalf("TenantGroupFileCreate shared-db file link must include tenant_id; missing %q", required) + } + } + if strings.Contains(body, "INSERT INTO im_filemsglinks(createdat,fileid,topic)") { + t.Fatal("TenantGroupFileCreate must not rely on im_filemsglinks.tenant_id default") + } +} diff --git a/server/db/mysql/tenant_groups.go b/server/db/mysql/tenant_groups.go index cf6efd9..66c87a6 100644 --- a/server/db/mysql/tenant_groups.go +++ b/server/db/mysql/tenant_groups.go @@ -1132,10 +1132,10 @@ func (a *adapter) TenantGroupFileCreate(id t.TenantID, file *t.GroupFile) (*t.Gr SELECT 1 FROM im_filemsglinks WHERE tenant_id=? AND fileid=? AND topic=?)`, id, now, decodeUID(fid), file.Topic, id, decodeUID(fid), file.Topic) } else { - _, err = tx.ExecContext(ctx, `INSERT INTO im_filemsglinks(createdat,fileid,topic) - SELECT ?,?,? WHERE NOT EXISTS ( - SELECT 1 FROM im_filemsglinks WHERE fileid=? AND topic=?)`, - now, decodeUID(fid), file.Topic, decodeUID(fid), file.Topic) + _, err = tx.ExecContext(ctx, `INSERT INTO im_filemsglinks(tenant_id,createdat,fileid,topic) + SELECT ?,?,?,? WHERE NOT EXISTS ( + SELECT 1 FROM im_filemsglinks WHERE tenant_id=? AND fileid=? AND topic=?)`, + id, now, decodeUID(fid), file.Topic, id, decodeUID(fid), file.Topic) } if err != nil { return nil, err diff --git a/server/im_http.go b/server/im_http.go index 0ed06a4..bb8aa68 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -2637,6 +2637,10 @@ func serveIMSystemNotificationRead(wrt http.ResponseWriter, req *http.Request) { return } readAt = parsed.UTC().Round(time.Millisecond) + if readAt.After(time.Now().UTC().Round(time.Millisecond)) { + writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "read_at", "reason": "future_read_at"}) + return + } } notification, err := store.Security.MarkSystemNotificationRead(principal.TenantID, principal.Uid, eventID, readAt) if err != nil { diff --git a/server/im_http_test.go b/server/im_http_test.go index c0ff3e9..6fcd5f3 100644 --- a/server/im_http_test.go +++ b/server/im_http_test.go @@ -3166,6 +3166,18 @@ func TestServeIMSystemNotificationsListAndRead(t *testing.T) { if readAgainData["id"] != "evt-2" || readAgainData["read_at"] != readAt.UTC().Format(time.RFC3339Nano) { t.Fatalf("read again response should keep persisted read_at: %#v", readAgainData) } + + expectIMBearer(t, ctrl, "token-read-future", 7, 42) + req = httptest.NewRequest(http.MethodPost, "/v1/im/system-notifications/evt-1/read", bytes.NewBufferString(`{"read_at":"2099-01-01T00:00:00Z"}`)) + req.Header.Set("Authorization", "Bearer token-read-future") + req.Header.Set("Idempotency-Key", "read-future") + rec = httptest.NewRecorder() + + serveIMSystemNotificationRead(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("future read_at status: got %d, want %d; body=%s", rec.Code, http.StatusBadRequest, rec.Body.String()) + } } func TestServeIMMessageReceipts(t *testing.T) { diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index cf4e479..9851e95 100644 --- a/server/store/group_message_runtime.go +++ b/server/store/group_message_runtime.go @@ -1411,7 +1411,8 @@ func (r *redisGroupMessageRuntime) hotSeqIDs(tenantID types.TenantID, topic stri if err != nil { return nil, false, err } - if len(raw) < limit { + expected := hotSeqExpectedCount(opts, limit) + if len(raw) < expected { return nil, false, nil } seqs := make([]int64, 0, len(raw)) @@ -1425,6 +1426,23 @@ func (r *redisGroupMessageRuntime) hotSeqIDs(tenantID types.TenantID, topic stri return seqs, true, nil } +func hotSeqExpectedCount(opts *types.QueryOpt, limit int) int { + if limit <= 0 { + return 0 + } + if opts == nil || opts.Since <= 0 || opts.Before <= opts.Since { + return limit + } + expected := opts.Before - opts.Since + if expected <= 0 { + return limit + } + if expected > int64(limit) { + return limit + } + return int(expected) +} + func expandRanges(ranges []types.Range, limit int) []int64 { if limit <= 0 { return nil diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index d16d1f3..6b6ae1e 100644 --- a/server/store/group_message_runtime_test.go +++ b/server/store/group_message_runtime_test.go @@ -488,6 +488,21 @@ func TestTopicAndSubscriptionCacheTTLIsShort(t *testing.T) { } } +func TestHotSeqExpectedCountUsesRangeSizeWhenSmallerThanLimit(t *testing.T) { + if got := hotSeqExpectedCount(&types.QueryOpt{Since: 41, Before: 54, Limit: 20}, 20); got != 13 { + t.Fatalf("expected range-sized count, got %d", got) + } + if got := hotSeqExpectedCount(&types.QueryOpt{Since: 41, Before: 80, Limit: 20}, 20); got != 20 { + t.Fatalf("expected limit-sized count, got %d", got) + } + if got := hotSeqExpectedCount(&types.QueryOpt{Since: 0, Before: 54, Limit: 20}, 20); got != 20 { + t.Fatalf("expected fallback limit count, got %d", got) + } + if got := hotSeqExpectedCount(nil, 20); got != 20 { + t.Fatalf("expected nil opts limit count, got %d", got) + } +} + func TestInvalidateSubscriptionCacheWithZeroUserDeletesTopicPrefix(t *testing.T) { oldRuntime := groupMsgRuntime runtime := newTestGroupMessageRuntime() diff --git a/server/topic.go b/server/topic.go index 417fa79..b394c5b 100644 --- a/server/topic.go +++ b/server/topic.go @@ -3535,6 +3535,13 @@ func (t *Topic) replyGetData(sess *Session, asUid types.Uid, asChan bool, req *M } else if ok { messages = filterHotMessagesForUser(hotMessages, asUid) } + } else if shouldTryHotMessageSupplement(t.cat, opts, len(messages)) { + hotMessages, ok, cacheErr := store.GetHotMessages(t.tenantID, t.name, opts, hotMessageRangeLimit(opts)) + if cacheErr != nil { + logs.Warn.Printf("get.data hot message supplement failed topic=%s user=%s: %v", t.name, asUid.UserId(), cacheErr) + } else if ok { + messages = mergeHotMessageSupplement(messages, filterHotMessagesForUser(hotMessages, asUid), opts) + } } // Push the list of messages to the client as {data}. @@ -3595,6 +3602,28 @@ func shouldTryHotMessageFallback(cat types.TopicCat, lastID int64, opts *types.Q return lastID <= 0 || opts.Since >= floor } +func shouldTryHotMessageSupplement(cat types.TopicCat, opts *types.QueryOpt, got int) bool { + if cat != types.TopicCatGrp || opts == nil || opts.IfModifiedSince != nil || + !opts.User.IsZero() || opts.Topic != "" || len(opts.IdRanges) > 0 || + opts.Since <= 0 || opts.Before <= opts.Since { + return false + } + expected := hotMessageRangeLimit(opts) + return expected > 0 && got > 0 && got < expected +} + +func hotMessageRangeLimit(opts *types.QueryOpt) int { + if opts == nil || opts.Since <= 0 || opts.Before <= opts.Since { + return hotMessageFallbackLimit(opts) + } + expected := opts.Before - opts.Since + limit := hotMessageFallbackLimit(opts) + if limit <= 0 || int64(limit) > expected { + return int(expected) + } + return limit +} + func hotMessageFallbackLimit(opts *types.QueryOpt) int { if opts != nil && opts.Limit > 0 { return opts.Limit @@ -3602,6 +3631,36 @@ func hotMessageFallbackLimit(opts *types.QueryOpt) int { return 1 } +func mergeHotMessageSupplement(messages, hotMessages []types.Message, opts *types.QueryOpt) []types.Message { + if len(messages) == 0 || len(hotMessages) == 0 { + return messages + } + bySeq := make(map[int64]types.Message, len(hotMessages)+len(messages)) + for _, msg := range hotMessages { + bySeq[msg.SeqId] = msg + } + for _, msg := range messages { + bySeq[msg.SeqId] = msg + } + merged := make([]types.Message, 0, len(bySeq)) + for _, msg := range bySeq { + if opts != nil { + if opts.Since > 0 && msg.SeqId < opts.Since { + continue + } + if opts.Before > 0 && msg.SeqId >= opts.Before { + continue + } + } + merged = append(merged, msg) + } + sort.Slice(merged, func(i, j int) bool { return merged[i].SeqId > merged[j].SeqId }) + if opts != nil && opts.Limit > 0 && len(merged) > opts.Limit { + merged = merged[:opts.Limit] + } + return merged +} + func filterHotMessagesForUser(messages []types.Message, asUid types.Uid) []types.Message { if len(messages) == 0 { return messages diff --git a/server/topic_test.go b/server/topic_test.go index bf463bb..0412cfd 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -1272,6 +1272,55 @@ func TestShouldTryHotMessageFallback(t *testing.T) { } } +func TestShouldTryHotMessageSupplement(t *testing.T) { + opts := &types.QueryOpt{Since: 41, Before: 54, Limit: 20} + if !shouldTryHotMessageSupplement(types.TopicCatGrp, opts, 9) { + t.Fatal("expected hot message supplement for partial recent group range") + } + if shouldTryHotMessageSupplement(types.TopicCatGrp, opts, 13) { + t.Fatal("did not expect supplement when range is complete") + } + if shouldTryHotMessageSupplement(types.TopicCatP2P, opts, 9) { + t.Fatal("did not expect supplement for p2p topic") + } + if shouldTryHotMessageSupplement(types.TopicCatGrp, &types.QueryOpt{Since: 41, Before: 54, Limit: 20, IdRanges: []types.Range{{Low: 41, Hi: 54}}}, 9) { + t.Fatal("did not expect supplement for explicit id ranges") + } +} + +func TestMergeHotMessageSupplementFillsRecentRangeGaps(t *testing.T) { + opts := &types.QueryOpt{Since: 41, Before: 54, Limit: 20} + messages := []types.Message{ + {SeqId: 53, Content: "db-53"}, + {SeqId: 52, Content: "db-52"}, + {SeqId: 51, Content: "db-51"}, + {SeqId: 50, Content: "db-50"}, + {SeqId: 49, Content: "db-49"}, + {SeqId: 47, Content: "db-47"}, + {SeqId: 46, Content: "db-46"}, + {SeqId: 42, Content: "db-42"}, + {SeqId: 41, Content: "db-41"}, + } + hotMessages := make([]types.Message, 0, 13) + for seq := int64(53); seq >= 41; seq-- { + hotMessages = append(hotMessages, types.Message{SeqId: seq, Content: "hot"}) + } + + merged := mergeHotMessageSupplement(messages, hotMessages, opts) + if len(merged) != 13 { + t.Fatalf("merged len = %d, want 13: %#v", len(merged), merged) + } + for i, msg := range merged { + wantSeq := int64(53 - i) + if msg.SeqId != wantSeq { + t.Fatalf("merged[%d].SeqId = %d, want %d", i, msg.SeqId, wantSeq) + } + if msg.SeqId == 53 && msg.Content != "db-53" { + t.Fatalf("database message should win duplicate seqs, got %#v", msg.Content) + } + } +} + func TestFilterHotMessagesForUserSkipsDeleted(t *testing.T) { uid := types.Uid(1) now := time.Now() @@ -3794,6 +3843,30 @@ func TestMessageAttachmentURLsDeduplicatesExplicitAndContentFileID(t *testing.T) } } +func TestMessageAttachmentURLsFindsDraftyEntityRefs(t *testing.T) { + msg := &ClientComMessage{ + Pub: &MsgClientPub{ + Content: map[string]any{ + "txt": "video", + "ent": []any{ + map[string]any{ + "tp": "EX", + "data": map[string]any{ + "mime": "video/mp4", + "ref": "/v0/file/s/KJ0Qlrk4Gzw.mp4", + }, + }, + }, + }, + }, + } + + attachments := messageAttachmentURLs(msg) + if len(attachments) != 1 || attachments[0] != "/v0/file/s/KJ0Qlrk4Gzw" { + t.Fatalf("unexpected Drafty attachments: %#v", attachments) + } +} + func TestHandleBroadcastInfoChannelWithMultipleReaders(t *testing.T) { topicName := "grpTest" chanName := "chnTest"