fix: resolve P1 backend consistency issues

This commit is contained in:
richjohnrambo
2026-09-08 16:23:21 +08:00
parent 06945a4050
commit 0b3c0b1c78
13 changed files with 587 additions and 57 deletions
+4
View File
@@ -167,6 +167,10 @@ func ensureIMGroupBusinessTables(exec sqlExecutor) error {
PRIMARY KEY(id), UNIQUE INDEX uk_im_group_events_outbox_event(event_id),
INDEX idx_im_group_events_outbox_pending(status, next_retry_at, id), INDEX idx_im_group_events_outbox_topic(topic, id DESC)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
`UPDATE im_topics SET public=JSON_SET(public, '$.fn', JSON_UNQUOTE(JSON_EXTRACT(public, '$.name')))
WHERE name LIKE 'grp%' AND JSON_VALID(public)
AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(public, '$.fn')), '') = ''
AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(public, '$.name')), '') <> ''`,
}
for _, statement := range statements {
if _, err := exec.Exec(statement); err != nil {
+5 -2
View File
@@ -22,8 +22,8 @@ func TestEnsureIMGroupBusinessTablesUsesSelectedTenantDatabase(t *testing.T) {
if err := ensureIMGroupBusinessTables(exec); err != nil {
t.Fatal(err)
}
if len(exec.statements) != 13 {
t.Fatalf("created %d group tables, want 13", len(exec.statements))
if len(exec.statements) != 14 {
t.Fatalf("executed %d group schema statements, want 14", len(exec.statements))
}
for _, statement := range exec.statements {
if strings.Contains(strings.ToLower(statement), "tinode.") || strings.Contains(strings.ToLower(statement), "create database") {
@@ -40,4 +40,7 @@ func TestEnsureIMGroupBusinessTablesUsesSelectedTenantDatabase(t *testing.T) {
t.Fatalf("group schema is missing snowflake primary-key definition %q", expected)
}
}
if !strings.Contains(joined, "json_set(public, '$.fn'") {
t.Fatal("group schema is missing historical public.fn backfill")
}
}
+12 -3
View File
@@ -7051,15 +7051,17 @@ func (a *adapter) TenantTopicDraftPut(id t.TenantID, uid t.Uid, topic, content s
return nil, err
}
now := t.TimeNow()
_, err = a.db.ExecContext(ctx, `INSERT INTO im_topic_drafts
res, err := a.db.ExecContext(ctx, `INSERT INTO im_topic_drafts
(id,user_id,topic,content,entities_json,version,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE content=VALUES(content),entities_json=VALUES(entities_json),
version=version+1,updated_at=VALUES(updated_at)`,
ON DUPLICATE KEY UPDATE id=id`,
draftID, decodeUID(uid), strings.TrimSpace(topic), content, entities, 1, now, now)
if err != nil {
return nil, err
}
if affected, _ := res.RowsAffected(); affected == 0 {
return nil, t.ErrInvalidResponse
}
return a.topicDraftByKey(ctx, uid, strings.TrimSpace(topic))
}
res, err := a.db.ExecContext(ctx, `UPDATE im_topic_drafts
@@ -9210,6 +9212,13 @@ func (a *adapter) TenantFriendRequestAccept(id t.TenantID, requestID string, act
row.HandledAt = &handledAt
row.UpdatedAt = handledAt
}
if _, err = tx.ExecContext(ctx, `UPDATE im_friend_requests SET status=?,handled_at=?,updated_at=?
WHERE request_id<>? AND status=? AND
((applicant_id=? AND target_id=?) OR (applicant_id=? AND target_id=?))`,
t.FriendRequestExpired, handledAt, handledAt, requestID, t.FriendRequestPending,
decodeUID(applicant), decodeUID(target), decodeUID(target), decodeUID(applicant)); err != nil {
return nil, err
}
for _, pair := range [][2]t.Uid{{applicant, target}, {target, applicant}} {
friendshipID, idErr := nextSnowflakeID()
if idErr != nil {
+30 -2
View File
@@ -936,11 +936,16 @@ func (a *adapter) TenantGroupFileCreate(id t.TenantID, file *t.GroupFile) (*t.Gr
}
a.enrichGroupFileFromUpload(id, file)
now := t.TimeNow()
ctx, cancel := a.getContext()
ctx, cancel := a.getContextForTx()
if cancel != nil {
defer cancel()
}
_, err := a.db.ExecContext(ctx, `INSERT INTO im_group_files
tx, err := a.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer tx.Rollback()
_, err = tx.ExecContext(ctx, `INSERT INTO im_group_files
(group_file_id,topic,file_id,source_topic,source_seq,uploader_id,file_name,mime_type,file_type,file_size,status,idempotency_key,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE
@@ -954,6 +959,29 @@ func (a *adapter) TenantGroupFileCreate(id t.TenantID, file *t.GroupFile) (*t.Gr
if err != nil && !isDupe(err) {
return nil, err
}
uploadID := strings.TrimSpace(file.FileID)
if dot := strings.IndexByte(uploadID, '.'); dot > 0 {
uploadID = uploadID[:dot]
}
if fid := t.ParseUid(uploadID); !fid.IsZero() {
if a.tenantScopedDB {
_, 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)
} else {
_, 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
}
}
if err = tx.Commit(); err != nil {
return nil, err
}
created, err := a.groupFileByQuery(ctx, "SELECT * FROM im_group_files WHERE topic=? AND file_id=?", file.Topic, file.FileID)
if err != nil {
return nil, err
+119 -10
View File
@@ -62,7 +62,7 @@ const (
imVerificationResendAfterSeconds = 60
imVerificationMaxAttempts = 5
imVerificationCreateRateWindow = time.Hour
imSensitiveChallengeRateLimitEnabled = false
imSensitiveChallengeRateLimitEnabled = true
imActionTokenExpiresInSeconds = 300
imIndependentPasswordMaxAttempts = 5
imIndependentPasswordLockSeconds = 300
@@ -1503,7 +1503,7 @@ func serveIMSecurityChallenges(wrt http.ResponseWriter, req *http.Request) {
return
}
action := strings.TrimSpace(body.Scene)
if action == "" || len(action) > 64 || strings.TrimSpace(body.Channel) != "sms" {
if !validIMSecurityChallengeScene(action) || strings.TrimSpace(body.Channel) != "sms" {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
return
}
@@ -2530,7 +2530,8 @@ func serveIMMeetingReminders(wrt http.ResponseWriter, req *http.Request, meeting
meetingParticipantUserIDs(participants), "会议提醒", meetingID, map[string]any{"meeting_id": meetingID, "reminded_count": len(participants)})
writeIMSuccess(wrt, req, map[string]any{
"meeting_id": meetingID,
"reminded": meetingParticipantPayloads(participants),
"reminded": meetingParticipantPayloadsWithSummaries(participants,
meetingUserSummaries(principal.TenantID, []types.Meeting{{Participants: participants}})),
})
}
@@ -3706,9 +3707,10 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) {
results := make([]map[string]any, 0, len(body.TargetTopics)*len(body.FavoriteIDs))
comment := strings.TrimSpace(body.Comment)
for _, target := range body.TargetTopics {
target = strings.TrimSpace(target)
if target == "" {
results = append(results, map[string]any{"target_topic": target, "status": "failed", "reason": imReasonMalformedRequest})
rawTarget := strings.TrimSpace(target)
target, targetOK := normalizeIMForwardTopic(principal.Uid, rawTarget)
if target == "" || !targetOK {
results = append(results, map[string]any{"target_topic": rawTarget, "status": "failed", "reason": imReasonMalformedRequest})
continue
}
if err := requireTopicWriter(principal, target); err != nil {
@@ -3733,6 +3735,10 @@ func serveIMFavoriteForward(wrt http.ResponseWriter, req *http.Request) {
},
Content: favoriteForwardContent(item, comment),
}
if err := prepareHTTPForwardMessage(principal.TenantID, msg); err != nil {
results = append(results, map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "failed", "reason": relationErrorReason(err)})
continue
}
saved, err := saveMessageMaybeKafka(req.Context(), principal.TenantID, msg, nil, true)
if err != nil {
results = append(results, map[string]any{"target_topic": target, "favorite_id": favoriteID, "status": "failed", "reason": relationErrorReason(err)})
@@ -4402,6 +4408,60 @@ func serveIMFriendRequestCreate(wrt http.ResponseWriter, req *http.Request) {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
return
}
relation, err := store.Relations.GetRelationshipContext(principal.TenantID, principal.Uid, target, principal.Uid.P2PName(target))
if err != nil {
writeIMRelationError(wrt, req, err)
return
}
if relation == nil || relation.Relationship != types.RelationshipNone {
reason := "FRIEND_REQUEST_CONFLICT"
if relation != nil && (relation.Relationship == types.RelationshipBlockedByMe || relation.Relationship == types.RelationshipBlockedByPeer) {
reason = imReasonPermissionDenied
}
writeIMError(wrt, req, http.StatusConflict, 30006, reason, "friend request is not allowed for current relationship", nil)
return
}
source := strings.TrimSpace(body.Source)
sourceTopic := strings.TrimSpace(body.SourceTopic)
switch source {
case "search", "qr":
if sourceTopic != "" {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "source_topic is only valid for group source", nil)
return
}
case "group":
if types.GetTopicCat(sourceTopic) != types.TopicCatGrp {
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "valid group source_topic is required", nil)
return
}
requesterSub, requesterErr := store.Subs.Get(principal.TenantID, sourceTopic, principal.Uid, false)
targetSub, targetErr := store.Subs.Get(principal.TenantID, sourceTopic, target, false)
if requesterErr != nil || targetErr != nil {
if requesterErr != nil {
writeIMRelationError(wrt, req, requesterErr)
} else {
writeIMRelationError(wrt, req, targetErr)
}
return
}
if requesterSub == nil || targetSub == nil ||
!(requesterSub.ModeGiven & requesterSub.ModeWant).IsJoiner() || !(targetSub.ModeGiven & targetSub.ModeWant).IsJoiner() {
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "group membership required", nil)
return
}
settings, settingsErr := store.GroupBusiness.GetSettings(principal.TenantID, sourceTopic)
if settingsErr != nil {
writeIMRelationError(wrt, req, settingsErr)
return
}
if settings == nil || !settings.MemberPrivateChat {
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "member private chat is disabled", nil)
return
}
default:
writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "invalid friend request source", nil)
return
}
privacy, err := store.Profiles.GetPrivacy(principal.TenantID, target)
if err != nil {
writeIMRelationError(wrt, req, err)
@@ -4424,8 +4484,8 @@ func serveIMFriendRequestCreate(wrt http.ResponseWriter, req *http.Request) {
ApplicantID: principal.Uid,
TargetID: target,
Message: message,
Source: strings.TrimSpace(body.Source),
SourceTopic: strings.TrimSpace(body.SourceTopic),
Source: source,
SourceTopic: sourceTopic,
Status: types.FriendRequestPending,
IdempotencyKey: idempotencyKey,
ExpiresAt: now.Add(7 * 24 * time.Hour),
@@ -5038,7 +5098,7 @@ func serveIMGroupMyProfile(wrt http.ResponseWriter, req *http.Request, topic str
writeIMRelationError(wrt, req, err)
return
}
if _, err := store.GroupBusiness.UpdateMember(principal.TenantID, topic, principal.Uid, types.GroupMemberPatch{Nickname: &nickname, Version: body.Version}); err != nil {
if _, err := store.GroupBusiness.UpdateMember(principal.TenantID, topic, principal.Uid, types.GroupMemberPatch{Nickname: &nickname}); err != nil {
if errors.Is(err, types.ErrNotFound) {
logs.Warn.Printf("group[%s]: member profile saved to subscription only; business member row missing for user %s", topic, principal.Uid.UserId())
dispatchIMGroupEvents(principal.TenantID)
@@ -5129,6 +5189,17 @@ func serveIMGroupInvitations(wrt http.ResponseWriter, req *http.Request, topic s
return
}
access := selfSub.ModeGiven & selfSub.ModeWant
if !access.IsAdmin() {
settings, err := store.GroupBusiness.GetSettings(principal.TenantID, topic)
if err != nil {
writeIMRelationError(wrt, req, err)
return
}
if settings == nil || !settings.MemberInvite {
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "member invitation is disabled", nil)
return
}
}
if !access.IsSharer() && !access.IsAdmin() {
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil)
return
@@ -5232,6 +5303,17 @@ func serveIMGroupFiles(wrt http.ResponseWriter, req *http.Request, topic, tail s
"has_more": nextCursor != "",
})
case tail == "files" && req.Method == http.MethodPost:
if !access.IsAdmin() {
settings, err := store.GroupBusiness.GetSettings(principal.TenantID, topic)
if err != nil {
writeIMRelationError(wrt, req, err)
return
}
if settings == nil || !settings.MemberUploadFiles {
writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "member file upload is disabled", nil)
return
}
}
idempotencyKey, ok := requireIMIdempotencyKey(wrt, req)
if !ok {
return
@@ -5344,6 +5426,18 @@ func serveIMGroupJoinRequests(wrt http.ResponseWriter, req *http.Request, topic
writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "group not found", nil)
return
}
if !settings.JoinApproval {
now := time.Now().UTC().Round(time.Millisecond)
joined, joinErr := store.GroupBusiness.InviteMember(principal.TenantID, topic, principal.Uid, principal.Uid,
"", strings.TrimSpace(req.Header.Get("Idempotency-Key")), groupMemberPrivate("", false, false, 1, now), now)
if joinErr != nil {
writeIMRelationError(wrt, req, joinErr)
return
}
dispatchIMGroupEvents(principal.TenantID)
writeIMSuccess(wrt, req, map[string]any{"topic": topic, "status": "joined", "joined": joined, "source": "open_group"})
return
}
sub, err := store.Subs.Get(principal.TenantID, topic, principal.Uid, false)
if err != nil {
writeIMRelationError(wrt, req, err)
@@ -7353,7 +7447,9 @@ func mergeGroupSettingsPayload(payload map[string]any, settings *types.GroupSett
payload["description"] = settings.Description
payload["owner"] = settings.OwnerUserID.UserId()
payload["member_count"] = settings.MemberCount
payload["version"] = settings.Version
if _, exists := payload["version"]; !exists {
payload["version"] = settings.Version
}
return payload
}
@@ -8682,6 +8778,7 @@ func userProfilePayload10(profile *types.UserProfile) map[string]any {
"field_rules": map[string]any{
"nickname": map[string]any{"min_length": 1, "max_length": 32},
"signature": map[string]any{"max_length": 120},
"region": map[string]any{"max_length": 128},
"birth_date": map[string]any{"format": "YYYY-MM-DD"},
"gender": []string{"male", "female", "other", ""},
"email": map[string]any{"max_length": 128},
@@ -8690,6 +8787,15 @@ func userProfilePayload10(profile *types.UserProfile) map[string]any {
return data
}
func validIMSecurityChallengeScene(scene string) bool {
switch strings.TrimSpace(scene) {
case "set_independent_password", "enable_youth_mode", "close_youth_mode", "reset_independent_password":
return true
default:
return false
}
}
func userPrivacyPayload(settings *types.UserPrivacySettings) map[string]any {
return map[string]any{
"add_me_methods": settings.AddMeMethods,
@@ -9437,6 +9543,9 @@ func normalizeIMUserVisibleTopic(viewer types.Uid, topic string) (string, bool)
if topic == "" {
return "", false
}
if topic == "slf" {
return viewer.SlfName(), !viewer.IsZero()
}
if len(topic) < 3 {
return "", false
}
+125 -6
View File
@@ -2364,7 +2364,7 @@ func TestServeIMSecurityChallengeVerifyIssuesActionToken(t *testing.T) {
}
}
func TestServeIMSecurityChallengeCreateSkipsRateLimitWhenDisabled(t *testing.T) {
func TestServeIMSecurityChallengeCreateEnforcesResendLimit(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
expectIMBearer(t, ctrl, "token-1", 7, 42)
@@ -2405,11 +2405,26 @@ func TestServeIMSecurityChallengeCreateSkipsRateLimitWhenDisabled(t *testing.T)
serveIMSecurityChallenges(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
if rec.Code != http.StatusTooManyRequests {
t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusTooManyRequests, rec.Body.String())
}
if verifications.created == nil || verifications.created.Action != "close_youth_mode" {
t.Fatalf("expected a new challenge while sensitive challenge rate limit is disabled: %#v", verifications.created)
if verifications.created != nil {
t.Fatalf("rate-limited request created a challenge: %#v", verifications.created)
}
}
func TestServeIMSecurityChallengeCreateRejectsUnknownScene(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
expectIMBearer(t, ctrl, "token-1", 7, 42)
req := httptest.NewRequest(http.MethodPost, "/v1/im/security/challenges",
bytes.NewBufferString(`{"scene":"made_up_action","channel":"sms"}`))
req.Header.Set("Authorization", "Bearer token-1")
rec := httptest.NewRecorder()
serveIMSecurityChallenges(rec, req)
if rec.Code != http.StatusBadRequest {
t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusBadRequest, rec.Body.String())
}
}
@@ -4868,6 +4883,71 @@ func TestServeIMFavoritesListDetailBatchDelete(t *testing.T) {
}
}
func TestServeIMFavoritesListSupportsRefOnlyAttachment(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
expectIMBearer(t, ctrl, "token-1", 7, 42)
oldFavorites := store.Favorites
oldSubs := store.Subs
oldMessages := store.Messages
now := time.Date(2026, 9, 6, 9, 0, 0, 0, time.UTC)
store.Favorites = &testFavoritesStore{items: []types.FavoriteItem{{
TenantID: 7, FavoriteID: "fav_ref", UserID: 42, ItemType: "file",
SourceTopic: "grpAlpha", SourceSeq: 8, Availability: "available", CollectedAt: now,
}}}
subs := mock_store.NewMockSubsPersistenceInterface(ctrl)
messages := mock_store.NewMockMessagesPersistenceInterface(ctrl)
store.Subs = subs
store.Messages = messages
defer func() {
store.Favorites = oldFavorites
store.Subs = oldSubs
store.Messages = oldMessages
}()
subs.EXPECT().Get(types.TenantID(7), "grpAlpha", types.Uid(42), false).Return(&types.Subscription{
TenantID: 7, User: types.Uid(42).String(), Topic: "grpAlpha",
ModeWant: types.ModeCPublic, ModeGiven: types.ModeCPublic,
}, nil)
messages.EXPECT().GetOne(types.TenantID(7), "grpAlpha", int64(8)).Return(&types.Message{
TenantID: 7, Topic: "grpAlpha", SeqId: 8, From: types.Uid(42).String(),
Head: types.KVMap{"type": "file"}, Content: map[string]any{
"txt": " ",
"fmt": []any{map[string]any{"at": 0, "len": 1, "key": 0}},
"ent": []any{map[string]any{"tp": "EX", "data": map[string]any{
"mime": "application/pdf", "name": "large.pdf", "ref": "https://oss.example/large.pdf",
}}},
}, ObjHeader: types.ObjHeader{CreatedAt: now},
}, nil)
req := httptest.NewRequest(http.MethodGet, "/v1/im/favorites?type=all&limit=1", nil)
req.Header.Set("Authorization", "Bearer token-1")
rec := httptest.NewRecorder()
serveIMFavorites(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
data := decodeIMHTTPResponse(t, rec).Data.(map[string]any)
if items := data["items"].([]any); len(items) != 1 || items[0].(map[string]any)["favorite_id"] != "fav_ref" {
t.Fatalf("unexpected ref-only favorite response: %#v", data)
}
}
func TestNormalizeIMUserVisibleTopicSelf(t *testing.T) {
uid := types.Uid(42)
got, ok := normalizeIMUserVisibleTopic(uid, "slf")
if !ok || got != uid.SlfName() {
t.Fatalf("normalize slf: got=%q ok=%v want=%q", got, ok, uid.SlfName())
}
}
func TestMergeGroupSettingsPayloadPreservesTopicVersion(t *testing.T) {
payload := map[string]any{"version": 7}
mergeGroupSettingsPayload(payload, &types.GroupSettings{Version: 2})
if payload["version"] != 7 {
t.Fatalf("topic version was overwritten: %#v", payload)
}
}
func TestServeIMFavoriteBatchCreateCountsExisting(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
@@ -6806,7 +6886,7 @@ func TestServeIMFriendRequestCreate(t *testing.T) {
mockStore.EXPECT().GetUidString().Return("request-1")
oldRelations := store.Relations
oldProfiles := store.Profiles
relations := &testRelationsStore{}
relations := &testRelationsStore{context: &types.RelationshipContext{Relationship: types.RelationshipNone}}
store.Relations = relations
store.Profiles = &testProfilesStore{}
defer func() {
@@ -6831,6 +6911,45 @@ func TestServeIMFriendRequestCreate(t *testing.T) {
}
}
func TestServeIMFriendRequestCreateHonorsGroupPrivateChatSetting(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
expectIMBearer(t, ctrl, "token-1", 7, 1)
oldRelations := store.Relations
oldProfiles := store.Profiles
oldSubs := store.Subs
oldGroups := store.GroupBusiness
store.Relations = &testRelationsStore{context: &types.RelationshipContext{Relationship: types.RelationshipNone}}
store.Profiles = &testProfilesStore{}
subs := mock_store.NewMockSubsPersistenceInterface(ctrl)
store.Subs = subs
store.GroupBusiness = &testGroupBusinessStore{settings: &types.GroupSettings{
Topic: "grpAlpha", MemberPrivateChat: false, GroupStatus: types.GroupStatusActive,
}}
defer func() {
store.Relations = oldRelations
store.Profiles = oldProfiles
store.Subs = oldSubs
store.GroupBusiness = oldGroups
}()
memberMode := types.ModeCPublic
for _, uid := range []types.Uid{1, 2} {
subs.EXPECT().Get(types.TenantID(7), "grpAlpha", uid, false).Return(&types.Subscription{
TenantID: 7, Topic: "grpAlpha", User: uid.String(), ModeWant: memberMode, ModeGiven: memberMode,
}, nil)
}
req := httptest.NewRequest(http.MethodPost, "/v1/im/friend-requests",
bytes.NewBufferString(`{"target_user_id":"`+types.Uid(2).UserId()+`","source":"group","source_topic":"grpAlpha"}`))
req.Header.Set("Authorization", "Bearer token-1")
req.Header.Set("Idempotency-Key", "fr-group-1")
rec := httptest.NewRecorder()
serveIMFriendRequests(rec, req)
if rec.Code != http.StatusForbidden {
t.Fatalf("status: got %d, want %d; body=%s", rec.Code, http.StatusForbidden, rec.Body.String())
}
}
func TestServeIMFriendRequestList(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
+29
View File
@@ -71,6 +71,7 @@ type groupMessageRuntime interface {
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
DeleteCachedObjectsByPrefix(tenantID types.TenantID, namespace, idPrefix string) error
AddCachedSetMember(tenantID types.TenantID, namespace, id, member string) error
RemoveCachedSetMember(tenantID types.TenantID, namespace, id, member string) error
CachedSetContains(tenantID types.TenantID, namespace, id, member string) (bool, error)
@@ -226,6 +227,13 @@ func DeleteCachedObject(tenantID types.TenantID, namespace, id string) error {
return groupMsgRuntime.DeleteCachedObject(tenantID, namespace, id)
}
func DeleteCachedObjectsByPrefix(tenantID types.TenantID, namespace, idPrefix string) error {
if groupMsgRuntime == nil || !tenantID.IsValid() || namespace == "" || idPrefix == "" {
return nil
}
return groupMsgRuntime.DeleteCachedObjectsByPrefix(tenantID, namespace, idPrefix)
}
func CacheHotMessage(tenantID types.TenantID, msg *types.Message) error {
if groupMsgRuntime == nil || !tenantID.IsValid() || msg == nil || msg.Topic == "" || msg.SeqId <= 0 {
return nil
@@ -788,6 +796,27 @@ func (r *redisGroupMessageRuntime) DeleteCachedObject(tenantID types.TenantID, n
return r.client.Del(context.Background(), r.objectCacheKey(tenantID, namespace, id)).Err()
}
func (r *redisGroupMessageRuntime) DeleteCachedObjectsByPrefix(tenantID types.TenantID, namespace, idPrefix string) error {
ctx := context.Background()
pattern := r.objectCacheKey(tenantID, namespace, idPrefix+"*")
var cursor uint64
for {
keys, next, err := r.client.Scan(ctx, cursor, pattern, 100).Result()
if err != nil {
return err
}
if len(keys) > 0 {
if err = r.client.Del(ctx, keys...).Err(); err != nil {
return err
}
}
cursor = next
if cursor == 0 {
return nil
}
}
}
func (r *redisGroupMessageRuntime) AddCachedSetMember(tenantID types.TenantID, namespace, id, member string) error {
key := r.setCacheKey(tenantID, namespace, id)
ctx := context.Background()
+20 -2
View File
@@ -9,8 +9,9 @@ import (
)
type testGroupMessageRuntime struct {
claims map[string]*groupMessageUniqueRecord
seqs map[string]int64
claims map[string]*groupMessageUniqueRecord
seqs map[string]int64
deletedPrefixes []string
}
func newTestGroupMessageRuntime() *testGroupMessageRuntime {
@@ -116,6 +117,11 @@ func (r *testGroupMessageRuntime) DeleteCachedObject(types.TenantID, string, str
return nil
}
func (r *testGroupMessageRuntime) DeleteCachedObjectsByPrefix(_ types.TenantID, _ string, prefix string) error {
r.deletedPrefixes = append(r.deletedPrefixes, prefix)
return nil
}
func (r *testGroupMessageRuntime) AddCachedSetMember(types.TenantID, string, string, string) error {
return nil
}
@@ -176,3 +182,15 @@ func TestMessageRuntimeSeqAndClientMidIdempotency(t *testing.T) {
t.Fatalf("unexpected second seq: seq=%d ready=%v err=%v", seq, ready, err)
}
}
func TestInvalidateGroupMemberPagesUsesTopicKeyPrefix(t *testing.T) {
oldRuntime := groupMsgRuntime
runtime := newTestGroupMessageRuntime()
groupMsgRuntime = runtime
t.Cleanup(func() { groupMsgRuntime = oldRuntime })
invalidateGroupMemberPages(7, "grpAlpha")
if len(runtime.deletedPrefixes) != 1 || runtime.deletedPrefixes[0] != "grpAlpha|" {
t.Fatalf("unexpected deleted prefixes: %#v", runtime.deletedPrefixes)
}
}
+68 -24
View File
@@ -1760,6 +1760,16 @@ func GetCachedGroupMemberPage(tenantID types.TenantID, key string) (map[string]a
return nil, false
}
func invalidateGroupMemberPages(tenantID types.TenantID, topic string) {
topic = strings.TrimSpace(topic)
if topic == "" {
return
}
if err := DeleteCachedObjectsByPrefix(tenantID, cacheNamespaceGroupPage, topic+"|"); err != nil {
logs.Warn.Printf("cache group member pages[%s]: delete failed: %v", topic, err)
}
}
func (groupBusinessMapper) UpdateMember(tenantID types.TenantID, topic string, uid types.Uid, patch types.GroupMemberPatch) (*types.GroupMember, error) {
if strings.TrimSpace(topic) == "" || uid.IsZero() {
return nil, types.ErrMalformed
@@ -1775,6 +1785,7 @@ func (groupBusinessMapper) UpdateMember(tenantID types.TenantID, topic string, u
} else {
_ = DeleteCachedObject(tenantID, cacheNamespaceGroupMember, groupMemberCacheID(strings.TrimSpace(topic), uid))
}
invalidateGroupMemberPages(tenantID, topic)
}
return member, err
}
@@ -1798,7 +1809,11 @@ func (groupBusinessMapper) AcceptInvitation(tenantID types.TenantID, invitationI
if err != nil {
return nil, err
}
return tenantAdp.TenantGroupInvitationAccept(tenantID, strings.TrimSpace(invitationID), uid, memberPrivate, now)
invitation, err := tenantAdp.TenantGroupInvitationAccept(tenantID, strings.TrimSpace(invitationID), uid, memberPrivate, now)
if err == nil && invitation != nil {
invalidateGroupMemberPages(tenantID, invitation.Topic)
}
return invitation, err
}
func (groupBusinessMapper) CreateJoinRequest(tenantID types.TenantID, req *types.GroupJoinRequest) (*types.GroupJoinRequest, error) {
@@ -1834,7 +1849,11 @@ func (groupBusinessMapper) ResolveJoinRequest(tenantID types.TenantID, topic, re
if err != nil {
return nil, err
}
return tenantAdp.TenantGroupJoinRequestResolve(tenantID, strings.TrimSpace(topic), strings.TrimSpace(requestID), actor, strings.TrimSpace(action), strings.TrimSpace(rejectReason), version, memberPrivate, now)
resolved, err := tenantAdp.TenantGroupJoinRequestResolve(tenantID, strings.TrimSpace(topic), strings.TrimSpace(requestID), actor, strings.TrimSpace(action), strings.TrimSpace(rejectReason), version, memberPrivate, now)
if err == nil && resolved != nil && resolved.Status == types.GroupJoinRequestStatusAccepted {
invalidateGroupMemberPages(tenantID, topic)
}
return resolved, err
}
func (groupBusinessMapper) LatestAnnouncement(tenantID types.TenantID, topic string) (*types.GroupAnnouncement, error) {
@@ -1951,8 +1970,12 @@ func (groupBusinessMapper) InviteMember(tenantID types.TenantID, topic string, i
if err != nil {
return false, err
}
return tenantAdp.TenantGroupInviteMember(tenantID, strings.TrimSpace(topic), inviter, target,
created, err := tenantAdp.TenantGroupInviteMember(tenantID, strings.TrimSpace(topic), inviter, target,
strings.TrimSpace(message), strings.TrimSpace(idempotencyKey), private, now)
if err == nil {
invalidateGroupMemberPages(tenantID, topic)
}
return created, err
}
func (groupBusinessMapper) Leave(tenantID types.TenantID, topic string, uid types.Uid, now time.Time) error {
@@ -1969,6 +1992,7 @@ func (groupBusinessMapper) Leave(tenantID types.TenantID, topic string, uid type
_ = DeleteCachedObject(tenantID, cacheNamespaceGroupMember, groupMemberCacheID(strings.TrimSpace(topic), uid))
_ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(strings.TrimSpace(topic), uid, false))
_ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(strings.TrimSpace(topic), uid, true))
invalidateGroupMemberPages(tenantID, topic)
return nil
}
@@ -1987,6 +2011,7 @@ func (groupBusinessMapper) UpdateRole(tenantID types.TenantID, topic string, act
_ = DeleteCachedObject(tenantID, cacheNamespaceGroupMember, groupMemberCacheID(strings.TrimSpace(topic), target))
_ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(strings.TrimSpace(topic), target, false))
_ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(strings.TrimSpace(topic), target, true))
invalidateGroupMemberPages(tenantID, topic)
return nil
}
@@ -1999,7 +2024,17 @@ func (groupBusinessMapper) TransferOwner(tenantID types.TenantID, topic string,
if err != nil {
return err
}
return tenantAdp.TenantGroupOwnerTransfer(tenantID, strings.TrimSpace(topic), previousOwner, newOwner, version, now)
topic = strings.TrimSpace(topic)
if err = tenantAdp.TenantGroupOwnerTransfer(tenantID, topic, previousOwner, newOwner, version, now); err != nil {
return err
}
for _, uid := range []types.Uid{previousOwner, newOwner} {
_ = DeleteCachedObject(tenantID, cacheNamespaceGroupMember, groupMemberCacheID(topic, uid))
_ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, uid, false))
_ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, uid, true))
}
invalidateGroupMemberPages(tenantID, topic)
return nil
}
func (groupBusinessMapper) CanPost(tenantID types.TenantID, topic string, uid types.Uid, admin bool, now time.Time) (bool, error) {
@@ -2848,6 +2883,22 @@ func refreshUserProfileProjections(tenantID types.TenantID, uid types.Uid, profi
return
}
public := userProfilePublicProjection(profile)
if user, err := Users.Get(tenantID, uid); err != nil {
logs.Warn.Printf("profile %s: load native user projection failed: %v", uid, err)
} else if user != nil {
merged := make(map[string]any)
if encoded, marshalErr := json.Marshal(user.Public); marshalErr == nil {
_ = json.Unmarshal(encoded, &merged)
}
for key, value := range public {
merged[key] = value
}
if err = Users.Update(tenantID, uid, map[string]any{"Public": merged}); err != nil {
logs.Warn.Printf("profile %s: update native user projection failed: %v", uid, err)
} else {
public = merged
}
}
cursor := ""
for {
items, nextCursor, err := Relations.ListFriends(tenantID, uid, "", "", cursor, 100)
@@ -2871,27 +2922,17 @@ func refreshUserProfileProjections(tenantID types.TenantID, uid types.Uid, profi
}
func userProfilePublicProjection(profile *types.UserProfile) map[string]any {
public := make(map[string]any)
nickname := strings.TrimSpace(profile.Nickname)
if nickname != "" {
for _, key := range []string{"display_name", "nickname", "fn", "name"} {
public[key] = nickname
}
}
if value := strings.TrimSpace(profile.AvatarFileID); value != "" {
public["avatar_file_id"] = value
}
if value := strings.TrimSpace(profile.Signature); value != "" {
public["signature"] = value
}
if value := strings.TrimSpace(profile.Gender); value != "" {
public["gender"] = value
}
if value := strings.TrimSpace(profile.Region); value != "" {
public["region"] = value
}
if value := strings.TrimSpace(profile.PublicAccount); value != "" {
public["public_account"] = value
public := map[string]any{
"display_name": nickname,
"nickname": nickname,
"fn": nickname,
"name": nickname,
"avatar_file_id": strings.TrimSpace(profile.AvatarFileID),
"signature": strings.TrimSpace(profile.Signature),
"gender": strings.TrimSpace(profile.Gender),
"region": strings.TrimSpace(profile.Region),
"public_account": strings.TrimSpace(profile.PublicAccount),
}
return public
}
@@ -2959,6 +3000,9 @@ func validUserProfilePatch(patch types.UserProfilePatch) bool {
if patch.Email != nil && len([]rune(strings.TrimSpace(*patch.Email))) > 128 {
return false
}
if patch.Region != nil && len([]rune(strings.TrimSpace(*patch.Region))) > 128 {
return false
}
return true
}
+87 -7
View File
@@ -12,6 +12,7 @@ import (
"context"
"errors"
"sort"
"strconv"
"strings"
"sync/atomic"
"time"
@@ -1215,6 +1216,18 @@ func (t *Topic) handlePubBroadcast(msg *ClientComMessage) {
msg.sess.queueOut(ErrPermissionDenied(msg.Id, t.original(asUid), msg.Timestamp))
return
}
if err := t.validateMessageReplacement(asUid, msg.Pub.Head, msg.Timestamp); err != nil {
if errors.Is(err, types.ErrMalformed) {
msg.sess.queueOut(ErrMalformedReply(msg, types.TimeNow()))
} else {
msg.sess.queueOut(ErrPermissionDeniedReply(msg, types.TimeNow()))
}
return
}
if err := t.consumeRiskActionToken(asUid, msg.Pub.Head); err != nil {
msg.sess.queueOut(ErrPermissionDeniedReply(msg, types.TimeNow()))
return
}
isCall := msg.Pub.Head != nil && msg.Pub.Head["webrtc"] != nil
if isCall {
@@ -1262,16 +1275,67 @@ func (t *Topic) handlePubBroadcast(msg *ClientComMessage) {
}
}
func (t *Topic) validateMessageReplacement(asUid types.Uid, head map[string]any, now time.Time) error {
if head == nil || head["replace"] == nil {
return nil
}
replace, ok := head["replace"].(string)
if !ok {
return types.ErrMalformed
}
seq, err := strconv.ParseInt(strings.TrimPrefix(strings.TrimSpace(replace), ":"), 10, 64)
if err != nil || seq <= 0 {
return types.ErrMalformed
}
original, err := store.Messages.GetOne(t.tenantID, t.name, seq)
if err != nil {
return err
}
if original == nil || types.ParseUid(original.From) != asUid {
return types.ErrPermissionDenied
}
if now.IsZero() {
now = types.TimeNow()
}
if original.CreatedAt.IsZero() || now.Sub(original.CreatedAt) > 5*time.Minute {
return types.ErrPermissionDenied
}
return nil
}
func (t *Topic) consumeRiskActionToken(asUid types.Uid, head map[string]any) error {
if head == nil {
return nil
}
token, _ := head["risk_action_token"].(string)
challengeID, _ := head["risk_challenge_id"].(string)
token = strings.TrimSpace(token)
challengeID = strings.TrimSpace(challengeID)
if token == "" && challengeID == "" {
return nil
}
if token == "" || challengeID == "" || store.Security == nil {
return types.ErrPermissionDenied
}
if _, err := store.Security.ConsumeActionToken(t.tenantID, asUid, token, "risk_continue", challengeID, types.TimeNow()); err != nil {
return err
}
delete(head, "risk_action_token")
return nil
}
func (t *Topic) validateAndNormalizeMentions(asUid types.Uid, head map[string]any) (map[string]any, error) {
mentionAll, _ := head["mention_all"].(bool)
rawMentions, hasMentions := head["mentions"]
if !mentionAll && !hasMentions {
return head, nil
}
pud := t.perUser[asUid]
access := pud.modeGiven & pud.modeWant
if mentionAll && !access.IsAdmin() {
return nil, types.ErrPermissionDenied
if mentionAll {
if sub, err := store.Subs.Get(t.tenantID, t.name, asUid, false); err != nil {
return nil, err
} else if sub == nil || !(sub.ModeGiven & sub.ModeWant).IsAdmin() {
return nil, types.ErrPermissionDenied
}
}
validMembers := make(map[string]struct{}, len(t.perUser))
for uid, pud := range t.perUser {
@@ -2318,9 +2382,25 @@ func (t *Topic) replyGetDesc(sess *Session, asUid types.Uid, _ bool, opts *MsgGe
desc.Public = t.public
desc.Trusted = t.trusted
} else if full && t.cat == types.TopicCatP2P {
// FIXME: when a P2P participant updates desc at 'me', these cached values are not updated.
desc.Public = pud.public
desc.Trusted = pud.trusted
uid1, uid2, err := types.ParseP2P(t.name)
if err != nil {
return err
}
peer := uid1
if peer == asUid {
peer = uid2
}
user, err := store.Users.Get(t.tenantID, peer)
if err != nil {
return err
}
if user != nil {
desc.Public = user.Public
desc.Trusted = user.Trusted
} else {
desc.Public = pud.public
desc.Trusted = pud.trusted
}
}
}
+37
View File
@@ -3,6 +3,7 @@ package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"os"
@@ -941,6 +942,9 @@ func TestHandleBroadcastDataRejectsMentionAllByMember(t *testing.T) {
pud.modeWant = types.ModeCPublic
pud.modeGiven = types.ModeCPublic
helper.topic.perUser[helper.uids[0]] = pud
helper.ss.EXPECT().Get(types.TenantID(1), "grp-test", helper.uids[0], false).Return(&types.Subscription{
TenantID: 1, Topic: "grp-test", User: helper.uids[0].String(), ModeWant: types.ModeCPublic, ModeGiven: types.ModeCPublic,
}, nil)
from := helper.uids[0].UserId()
msg := &ClientComMessage{
@@ -963,6 +967,39 @@ func TestHandleBroadcastDataRejectsMentionAllByMember(t *testing.T) {
}
}
func TestValidateMessageReplacementChecksAuthorAndWindow(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
oldMessages := store.Messages
messages := mock_store.NewMockMessagesPersistenceInterface(ctrl)
store.Messages = messages
t.Cleanup(func() { store.Messages = oldMessages })
now := time.Date(2026, 9, 6, 10, 0, 0, 0, time.UTC)
topic := &Topic{tenantID: 7, name: "grpAlpha"}
author := types.Uid(42)
messages.EXPECT().GetOne(types.TenantID(7), "grpAlpha", int64(8)).Return(&types.Message{
Topic: "grpAlpha", SeqId: 8, From: author.String(), ObjHeader: types.ObjHeader{CreatedAt: now.Add(-time.Minute)},
}, nil)
if err := topic.validateMessageReplacement(author, map[string]any{"replace": ":8"}, now); err != nil {
t.Fatalf("valid replacement rejected: %v", err)
}
messages.EXPECT().GetOne(types.TenantID(7), "grpAlpha", int64(9)).Return(&types.Message{
Topic: "grpAlpha", SeqId: 9, From: types.Uid(99).String(), ObjHeader: types.ObjHeader{CreatedAt: now.Add(-time.Minute)},
}, nil)
if err := topic.validateMessageReplacement(author, map[string]any{"replace": ":9"}, now); !errors.Is(err, types.ErrPermissionDenied) {
t.Fatalf("replacement of peer message: got %v", err)
}
messages.EXPECT().GetOne(types.TenantID(7), "grpAlpha", int64(10)).Return(&types.Message{
Topic: "grpAlpha", SeqId: 10, From: author.String(), ObjHeader: types.ObjHeader{CreatedAt: now.Add(-6 * time.Minute)},
}, nil)
if err := topic.validateMessageReplacement(author, map[string]any{"replace": ":10"}, now); !errors.Is(err, types.ErrPermissionDenied) {
t.Fatalf("expired replacement: got %v", err)
}
}
func TestHandleBroadcastDataFiltersInvalidMentions(t *testing.T) {
numUsers := 2
helper := TopicTestHelper{}
+6 -1
View File
@@ -3,6 +3,7 @@ package main
import (
"container/heap"
crand "crypto/rand"
"errors"
"math/rand"
"strings"
"time"
@@ -216,7 +217,11 @@ func replyCreateUser(s *Session, msg *ClientComMessage, rec *auth.Rec) {
// Create user record in the database.
if _, err := store.Users.Create(s.tenantID, &user, private); err != nil {
logs.Warn.Println("create user: failed to create user", err, "sid=", s.sid)
s.queueOut(ErrUnknown(msg.Id, "", msg.Timestamp))
params := map[string]any(nil)
if errors.Is(err, types.ErrDuplicate) && phoneRegistration {
params = map[string]any{"what": "phone", "reason": "PHONE_ALREADY_REGISTERED"}
}
s.queueOut(decodeStoreError(err, msg.Id, msg.Timestamp, params))
return
}
+45
View File
@@ -2,6 +2,7 @@ package main
import (
"bytes"
"errors"
"net/http"
"net/http/httptest"
"strings"
@@ -65,6 +66,50 @@ func TestConsumeProductVerificationCred(t *testing.T) {
}
}
func TestConsumeProductVerificationCredReturnsDuplicateCredential(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
oldUsers := store.Users
oldVerifications := store.Verifications
defer func() {
store.Users = oldUsers
store.Verifications = oldVerifications
}()
makeTestAPIKey(t)
token := "verified-token"
tokenHash := hashIMSecret(7, "register", "challenge-duplicate", "phone-hash", token)
store.Verifications = &testVerificationStore{get: &types.VerificationCode{
TenantID: 7,
ChallengeID: "challenge-duplicate",
Scene: "register",
PhoneHash: "phone-hash",
VerificationTokenHash: &tokenHash,
State: types.VerificationStateVerified,
ExpiresAt: time.Now().UTC().Add(time.Minute),
}}
users := mock_store.NewMockUsersPersistenceInterface(ctrl)
users.EXPECT().UpsertCred(types.TenantID(7), gomock.Any()).Return(true, types.ErrDuplicate)
store.Users = users
_, ok, err := consumeProductVerificationCred(7, 42, &MsgCredClient{
Method: "tel",
Value: "+8613800138000",
Response: token,
Params: map[string]any{"challenge_id": "challenge-duplicate"},
}, func(value string, _ map[string]any) (string, error) {
return "tel:" + value, nil
})
if !ok || !errors.Is(err, types.ErrDuplicate) {
t.Fatalf("duplicate phone: ok=%v err=%v", ok, err)
}
reply := decodeStoreError(err, "acc-1", time.Now(), nil)
if reply.Ctrl == nil || reply.Ctrl.Code != http.StatusConflict || reply.Ctrl.Text != "duplicate credential" {
t.Fatalf("unexpected duplicate reply: %#v", reply)
}
}
func TestConsumeProductVerificationCredRejectsBadToken(t *testing.T) {
oldUsers := store.Users
oldVerifications := store.Verifications