From f313979e8881f3ca225c6218e38e1233363c591c Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 4 Sep 2026 18:21:19 +0800 Subject: [PATCH] =?UTF-8?q?=E5=BC=80=E5=9C=BA=E6=B6=88=E6=81=AF=E6=9D=A1?= =?UTF-8?q?=E6=95=B0=E9=99=90=E5=88=B6=EF=BC=8C=E9=99=84=E8=BF=91=E6=8C=89?= =?UTF-8?q?=E8=B7=9D=E7=A6=BB=E5=88=86=E6=A1=A3=E9=9A=8F=E6=9C=BA=E5=88=B7?= =?UTF-8?q?=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一、对方回复前,免费用户最多发送 N 条(默认 3,迁移 034,管理端可改, 0 表示不限)。约束在 persistMessageContext 的事务内,HTTP 与 WebSocket 两条发送路径都覆盖,并发也不会挤过最后一个名额。要点: - 对方一旦开口,限制永久解除;会员始终不受限。 - 撤回自己的消息不会换回名额,否则删了重发即可绕过。 - AI 托管账号的回复是「回答」不是「敲门」,不计入限制。 - 被拒时返回 30007 + HTTP 402,与会员媒体限制区分,客户端据此引导开通。 - 消息列表接口附带 unansweredQuota,聊天页在发送失败之前就把剩余条数 告诉用户。 二、附近下拉刷新按距离分档随机。原先带 seed 的分支排在最前,导致附近 一旦刷新就退化成全城纯随机、距离排序被完全丢弃。现在按 1 公里分档, 档内用同一 seed 做确定性洗牌:刷新会换人,但近处的人永远排在远处之前, 同一 seed 下分页顺序稳定,不会重复或漏人。 两个既有测试显式关闭了开场限制——它们测的是入队规则和媒体会员限制, 连发消息只是手段。 Co-Authored-By: Claude Opus 5 --- .../app/ai_enqueue_integration_test.go | 2 + im/backend/internal/app/im.go | 41 ++++- .../im_membership_gate_integration_test.go | 4 + im/backend/internal/app/membership_limits.go | 47 +++++ .../app/nearby_shuffle_integration_test.go | 108 +++++++++++ im/backend/internal/app/social.go | 15 +- .../app/unanswered_limit_integration_test.go | 167 ++++++++++++++++++ .../034_unanswered_message_limit.sql | 4 + 8 files changed, 386 insertions(+), 2 deletions(-) create mode 100644 im/backend/internal/app/nearby_shuffle_integration_test.go create mode 100644 im/backend/internal/app/unanswered_limit_integration_test.go create mode 100644 im/backend/migrations/034_unanswered_message_limit.sql diff --git a/im/backend/internal/app/ai_enqueue_integration_test.go b/im/backend/internal/app/ai_enqueue_integration_test.go index ffcfb02..0a08699 100644 --- a/im/backend/internal/app/ai_enqueue_integration_test.go +++ b/im/backend/internal/app/ai_enqueue_integration_test.go @@ -20,6 +20,8 @@ func TestEnqueueAIReplyMySQL(t *testing.T) { a := &App{db: db, hub: NewHub(), config: Config{Environment: "development", JWTSecret: "isolated-im-regression-secret-only"}} for _, statement := range []string{ `INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES('ai.enabled','true','boolean','') ON DUPLICATE KEY UPDATE config_value=VALUES(config_value)`, + // 这里测的是入队规则,不是开场消息条数限制;关掉后者以免连发时被拦。 + `INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES('membership.free_unanswered_message_limit','0','integer','') ON DUPLICATE KEY UPDATE config_value=VALUES(config_value)`, `UPDATE users SET is_test=1,test_batch='regression' WHERE id IN (2,3)`, `INSERT INTO ai_agents(user_id,model_id,enabled) VALUES(2,0,1),(3,0,1)`, } { diff --git a/im/backend/internal/app/im.go b/im/backend/internal/app/im.go index ebebf8c..9e910e3 100644 --- a/im/backend/internal/app/im.go +++ b/im/backend/internal/app/im.go @@ -460,7 +460,34 @@ func (a *App) messages(w http.ResponseWriter, r *http.Request) { if len(items) > 0 { nextAfterSeq = items[len(items)-1].Seq } - reply(w, map[string]any{"items": items, "hasMore": hasMore, "nextBeforeSeq": nextBeforeSeq, "nextAfterSeq": nextAfterSeq}) + // The chat page shows how many opening messages are left before the other + // side answers, so the reader learns the rule before a send is refused. + reply(w, map[string]any{"items": items, "hasMore": hasMore, "nextBeforeSeq": nextBeforeSeq, "nextAfterSeq": nextAfterSeq, + "unansweredQuota": a.unansweredQuotaView(r.Context(), id, current(r).ID)}) +} + +// unansweredQuotaView reports the opening-message allowance for this +// conversation: remaining -1 means "no limit applies", either because the other +// side already replied, the reader is a member, or the switch is off. +func (a *App) unansweredQuotaView(ctx context.Context, conversationID, userID int64) map[string]any { + limit := a.resolveUnansweredMessageLimit(ctx, a.db) + unlimited := map[string]any{"limit": limit, "remaining": -1, "unlimited": true} + if limit <= 0 { + return unlimited + } + var mine, theirs int + if err := a.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(sender_id=?),0),COALESCE(SUM(sender_id<>?),0) FROM im_messages WHERE conversation_id=?`, + userID, userID, conversationID).Scan(&mine, &theirs); err != nil { + return unlimited + } + if theirs > 0 || a.hasActiveMembership(ctx, userID) { + return unlimited + } + remaining := limit - mine + if remaining < 0 { + remaining = 0 + } + return map[string]any{"limit": limit, "remaining": remaining, "unlimited": false} } func (a *App) sendMessageHTTP(w http.ResponseWriter, r *http.Request) { @@ -492,6 +519,11 @@ func (a *App) sendMessageHTTP(w http.ResponseWriter, r *http.Request) { fail(w, http.StatusPaymentRequired, 30006, gateErr.Error()) return } + var knockErr *unansweredMessageLimitError + if errors.As(err, &knockErr) { + fail(w, http.StatusPaymentRequired, 30007, knockErr.Error()) + return + } fail(w, 400, 30004, err.Error()) return } @@ -616,6 +648,13 @@ func (a *App) persistMessageContext(ctx context.Context, conversationID, senderI if err = a.reserveDailyActiveChat(ctx, tx, conversationID, senderID); err != nil { return item, nil, err } + // The AI worker answers on behalf of a managed account; its reply is the + // answer, never an unanswered knock. + if !isAIGenerated(ctx) { + if err = a.reserveUnansweredMessage(ctx, tx, conversationID, senderID); err != nil { + return item, nil, err + } + } seq := lastSeq + 1 result, err := tx.ExecContext(ctx, `INSERT INTO im_messages(conversation_id,seq,sender_id,client_msg_id,message_type,body)VALUES(?,?,?,?,?,?)`, conversationID, seq, senderID, clientMsgID, messageType, body) if err != nil { diff --git a/im/backend/internal/app/im_membership_gate_integration_test.go b/im/backend/internal/app/im_membership_gate_integration_test.go index fd16e95..f37d537 100644 --- a/im/backend/internal/app/im_membership_gate_integration_test.go +++ b/im/backend/internal/app/im_membership_gate_integration_test.go @@ -19,6 +19,10 @@ func TestMessageMembershipGateMySQL(t *testing.T) { t.Fatal(err) } } + // 这里测的是语音/图片的会员限制,与开场消息条数无关,后者会干扰连发。 + if _, err := db.Exec(`INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES('membership.free_unanswered_message_limit','0','integer','') ON DUPLICATE KEY UPDATE config_value=VALUES(config_value)`); err != nil { + t.Fatal(err) + } for _, media := range []struct { url string kind string diff --git a/im/backend/internal/app/membership_limits.go b/im/backend/internal/app/membership_limits.go index cefd5af..7e8edd6 100644 --- a/im/backend/internal/app/membership_limits.go +++ b/im/backend/internal/app/membership_limits.go @@ -5,6 +5,7 @@ import ( "database/sql" "fmt" "strconv" + "strings" ) const defaultFreeDailyActiveChatLimit = 5 @@ -161,3 +162,49 @@ func (a *App) ensureMessageMembership(ctx context.Context, senderID int64, messa } return &messageMembershipGateError{Kind: kind} } + +const defaultUnansweredMessageLimit = 3 + +// unansweredMessageLimitError is the opening-message cap: until the other side +// says something back, a free account may only send so many. It is the mirror of +// the daily active-chat limit — that one bounds how many people you may open a +// conversation with, this one bounds how hard you may knock on one door. +type unansweredMessageLimitError struct{ Limit int } + +func (e *unansweredMessageLimitError) Error() string { + return fmt.Sprintf("对方回复前最多发送 %d 条消息,开通会员后不受此限制", e.Limit) +} + +func (a *App) resolveUnansweredMessageLimit(ctx context.Context, queryer rowQuerier) int { + var raw string + if err := queryer.QueryRowContext(ctx, `SELECT config_value FROM system_configs WHERE config_key='membership.free_unanswered_message_limit'`).Scan(&raw); err == nil { + if parsed, parseErr := strconv.Atoi(strings.TrimSpace(raw)); parseErr == nil && parsed >= 0 { + return parsed + } + } + return defaultUnansweredMessageLimit +} + +// reserveUnansweredMessage runs inside the send transaction, so two messages +// racing each other cannot both pass the last slot. +func (a *App) reserveUnansweredMessage(ctx context.Context, tx *sql.Tx, conversationID, senderID int64) error { + limit := a.resolveUnansweredMessageLimit(ctx, tx) + if limit <= 0 { + return nil + } + var mine, theirs int + // Recalled and removed messages still count as knocking: deleting your own + // message must not hand back a slot. + if err := tx.QueryRowContext(ctx, `SELECT + COALESCE(SUM(sender_id=?),0),COALESCE(SUM(sender_id<>?),0) + FROM im_messages WHERE conversation_id=?`, senderID, senderID, conversationID).Scan(&mine, &theirs); err != nil { + return err + } + if theirs > 0 || mine < limit { + return nil + } + if a.hasActiveMembership(ctx, senderID) { + return nil + } + return &unansweredMessageLimitError{Limit: limit} +} diff --git a/im/backend/internal/app/nearby_shuffle_integration_test.go b/im/backend/internal/app/nearby_shuffle_integration_test.go new file mode 100644 index 0000000..9bb1249 --- /dev/null +++ b/im/backend/internal/app/nearby_shuffle_integration_test.go @@ -0,0 +1,108 @@ +package app + +import ( + "encoding/json" + "fmt" + "testing" +) + +// Pull to refresh on 附近 should bring different faces without sending the +// reader across town. Before this, a seed replaced the distance ordering +// entirely and the list became a plain shuffle of the whole city. +func TestNearbyShuffleStaysWithinDistanceBandsMySQL(t *testing.T) { + db := isolatedIMDatabase(t) + a := &App{db: db, hub: NewHub(), config: Config{Environment: "development", JWTSecret: "isolated-im-regression-secret-only"}} + // Viewer at the city centre; then three people ~200m away and three ~40km away. + const centreLat, centreLng = 22.5431, 114.0579 + if _, err := db.Exec(`INSERT INTO user_location_states(user_id,city_code,latitude,longitude) VALUES(1,'440300',?,?)`, centreLat, centreLng); err != nil { + t.Fatal(err) + } + near := map[int64]bool{} + for id := int64(2); id <= 7; id++ { + if id > 3 { + if _, err := db.Exec(`INSERT INTO users(id,public_id,password_hash) VALUES(?,CONCAT('NEARBY',?),'!NO_LOGIN')`, id, id); err != nil { + t.Fatal(err) + } + for _, statement := range []string{ + `INSERT INTO user_profiles(user_id,nickname) VALUES(?,CONCAT('附近',?))`, + `INSERT INTO user_privacy_settings(user_id) VALUES(?)`, + } { + args := []any{id} + if statement == `INSERT INTO user_profiles(user_id,nickname) VALUES(?,CONCAT('附近',?))` { + args = append(args, id) + } + if _, err := db.Exec(statement, args...); err != nil { + t.Fatal(err) + } + } + } + offset := 0.002 * float64(id) // ~200m steps + if id >= 5 { + offset = 0.36 * float64(id) // ~40km and beyond + } else { + near[id] = true + } + if _, err := db.Exec(`INSERT INTO user_location_states(user_id,city_code,latitude,longitude) VALUES(?,'440300',?,?)`, id, centreLat+offset, centreLng); err != nil { + t.Fatal(err) + } + } + page := func(seed int) []int64 { + t.Helper() + var payload struct { + Items []struct { + ID int64 `json:"id"` + } `json:"items"` + } + query := "/api/v1/nearby/users?page=1&pageSize=20" + if seed > 0 { + query += fmt.Sprintf("&seed=%d", seed) + } + if err := json.Unmarshal(imTestCall(t, a.nearby, 1, "GET", query, "", 200), &payload); err != nil { + t.Fatal(err) + } + ids := []int64{} + for _, item := range payload.Items { + ids = append(ids, item.ID) + } + return ids + } + + t.Run("the near band always comes first, whatever the seed", func(t *testing.T) { + for _, seed := range []int{0, 7, 12345, 2147483647} { + ids := page(seed) + if len(ids) < 6 { + t.Fatalf("seed %d 返回 %v", seed, ids) + } + for index, id := range ids[:3] { + if !near[id] { + t.Fatalf("seed %d 第 %d 位是远处的 %d: %v", seed, index+1, id, ids) + } + } + for _, id := range ids[3:] { + if near[id] { + t.Fatalf("seed %d 把近处的人排到了后面: %v", seed, ids) + } + } + } + }) + + t.Run("different seeds reorder people inside the band", func(t *testing.T) { + seen := map[string]bool{} + for seed := 1; seed <= 40; seed++ { + ids := page(seed) + seen[fmt.Sprint(ids[:3])] = true + } + if len(seen) < 2 { + t.Fatalf("刷新应当换顺序,实际只有一种排列: %v", seen) + } + }) + + t.Run("one seed is stable, so paging cannot repeat or skip a row", func(t *testing.T) { + first := page(99) + for attempt := 0; attempt < 3; attempt++ { + if fmt.Sprint(page(99)) != fmt.Sprint(first) { + t.Fatalf("同一 seed 必须给出同一顺序: %v vs %v", page(99), first) + } + } + }) +} diff --git a/im/backend/internal/app/social.go b/im/backend/internal/app/social.go index bf8db5c..ec90be7 100644 --- a/im/backend/internal/app/social.go +++ b/im/backend/internal/app/social.go @@ -208,6 +208,10 @@ func (a *App) userProfile(w http.ResponseWriter, r *http.Request) { reply(w, item) } +// One kilometre: fine enough that a refresh never trades a neighbour for someone +// across the city, coarse enough that there is a real pool to shuffle. +const nearbyDistanceBandMetres = "1000" + func (a *App) discover(w http.ResponseWriter, r *http.Request) { a.discoverList(w, r, false) } func (a *App) nearby(w http.ResponseWriter, r *http.Request) { a.discoverList(w, r, true) } @@ -308,7 +312,16 @@ func (a *App) discoverList(w http.ResponseWriter, r *http.Request, byDistance bo EXISTS(SELECT 1 FROM user_follows f WHERE f.user_id=? AND f.target_user_id=u.id),EXISTS(SELECT 1 FROM user_likes x WHERE x.user_id=? AND x.target_user_id=u.id) ` + fromWhere args := append([]any{who.ID, who.ID}, filterArgs...) - if shuffleSeed > 0 { + distanceExpression := `CASE WHEN privacy.distance_visible=1 THEN COALESCE(ST_Distance_Sphere(POINT(l.longitude,l.latitude),POINT(?,?)),1000000000000000) ELSE 1000000000000000 END` + if byDistance && shuffleSeed > 0 && myLat.Valid && myLng.Valid { + // Pull to refresh on 附近 should bring different faces without sending the + // reader across town: order by a coarse distance band first, then shuffle + // inside it. Everyone within the same kilometre is equally "nearby", so + // that is the granularity worth randomising over. Hidden distances land + // in the last band, as they do without a seed. + query += ` ORDER BY FLOOR(` + distanceExpression + `/` + nearbyDistanceBandMetres + `) ASC,CRC32(CONCAT(u.id,':',?)),u.id DESC LIMIT ? OFFSET ?` + args = append(args, myLng.Float64, myLat.Float64, shuffleSeed) + } else if shuffleSeed > 0 { // CRC32 produces a deterministic shuffle for this seed. The client keeps // the seed while paging, so adjacent pages cannot overlap or skip rows. query += ` ORDER BY CRC32(CONCAT(u.id,':',?)),u.id DESC LIMIT ? OFFSET ?` diff --git a/im/backend/internal/app/unanswered_limit_integration_test.go b/im/backend/internal/app/unanswered_limit_integration_test.go new file mode 100644 index 0000000..2d722bd --- /dev/null +++ b/im/backend/internal/app/unanswered_limit_integration_test.go @@ -0,0 +1,167 @@ +package app + +import ( + "context" + "encoding/json" + "fmt" + "net/http/httptest" + "strings" + "testing" +) + +// Opening a conversation is cheap for the sender and expensive for the reader, +// so a free account may only knock a few times before the other side answers. +// Once there is an answer the conversation is mutual and the cap is gone. +func TestUnansweredMessageLimitMySQL(t *testing.T) { + db := isolatedIMDatabase(t) + a := &App{db: db, hub: NewHub(), config: Config{Environment: "development", JWTSecret: "isolated-im-regression-secret-only"}} + setLimit := func(value string) { + t.Helper() + if _, err := db.Exec(`INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES('membership.free_unanswered_message_limit',?,'integer','') ON DUPLICATE KEY UPDATE config_value=VALUES(config_value)`, value); err != nil { + t.Fatal(err) + } + } + conversation := func(viewer, peer int64) int64 { + t.Helper() + var created struct { + ID int64 `json:"id"` + } + body := fmt.Sprintf(`{"userId":%d}`, peer) + if err := json.Unmarshal(imTestCall(t, a.directConversation, viewer, "POST", "/api/v1/im/conversations", body, 200), &created); err != nil { + t.Fatal(err) + } + return created.ID + } + // A direct conversation between two people is idempotent, so each subtest + // starts from an empty one rather than inheriting the previous knocks. + fresh := func(viewer, peer int64) int64 { + t.Helper() + id := conversation(viewer, peer) + if _, err := db.Exec(`DELETE FROM im_messages WHERE conversation_id=?`, id); err != nil { + t.Fatal(err) + } + return id + } + send := func(conversationID, sender int64, key string) (int, string) { + t.Helper() + w := httptest.NewRecorder() + path := fmt.Sprintf("/api/v1/im/conversations/%d/messages", conversationID) + payload := fmt.Sprintf(`{"clientMsgId":%q,"type":1,"content":{"text":"你好"}}`, key) + a.sendMessageHTTP(w, imTestRequest(sender, "POST", path, payload)) + return w.Code, w.Body.String() + } + member := func(userID int64, active bool) { + t.Helper() + if _, err := db.Exec(`DELETE FROM subscriptions WHERE user_id=?`, userID); err != nil { + t.Fatal(err) + } + if !active { + return + } + if _, err := db.Exec(`INSERT INTO subscriptions(user_id,plan_id,source,status,started_at,expires_at) VALUES(?,1,'test',1,NOW(3),DATE_ADD(NOW(3),INTERVAL 30 DAY))`, userID); err != nil { + t.Fatal(err) + } + } + setLimit("3") + + t.Run("the fourth unanswered message is refused with its own code", func(t *testing.T) { + id := fresh(1, 2) + for index := 1; index <= 3; index++ { + if status, body := send(id, 1, fmt.Sprintf("knock-%d", index)); status != 200 { + t.Fatalf("第 %d 条应当发出: HTTP %d %s", index, status, body) + } + } + status, body := send(id, 1, "knock-4") + if status != 402 || !strings.Contains(body, "30007") || !strings.Contains(body, "对方回复前最多发送 3 条") { + t.Fatalf("HTTP %d: %s", status, body) + } + }) + + t.Run("one answer lifts the cap for good", func(t *testing.T) { + id := fresh(1, 3) + for index := 1; index <= 3; index++ { + send(id, 1, fmt.Sprintf("a-knock-%d", index)) + } + if status, _ := send(id, 1, "a-knock-4"); status != 402 { + t.Fatalf("HTTP %d", status) + } + if status, body := send(id, 3, "the-answer"); status != 200 { + t.Fatalf("对方回复本身不受限: HTTP %d %s", status, body) + } + for index := 5; index <= 8; index++ { + if status, body := send(id, 1, fmt.Sprintf("a-knock-%d", index)); status != 200 { + t.Fatalf("回复之后应当不限: HTTP %d %s", status, body) + } + } + }) + + t.Run("a member is never capped", func(t *testing.T) { + member(2, true) + defer member(2, false) + id := fresh(2, 3) + for index := 1; index <= 6; index++ { + if status, body := send(id, 2, fmt.Sprintf("vip-%d", index)); status != 200 { + t.Fatalf("会员第 %d 条被拦: HTTP %d %s", index, status, body) + } + } + }) + + t.Run("deleting your own message does not hand back a slot", func(t *testing.T) { + id := fresh(1, 2) + for index := 1; index <= 3; index++ { + send(id, 1, fmt.Sprintf("recall-%d", index)) + } + if _, err := db.Exec(`UPDATE im_messages SET recalled_at=NOW(3) WHERE conversation_id=? AND sender_id=1`, id); err != nil { + t.Fatal(err) + } + if status, _ := send(id, 1, "recall-4"); status != 402 { + t.Fatalf("撤回不能换来新的额度: HTTP %d", status) + } + }) + + t.Run("the limit can be switched off entirely", func(t *testing.T) { + setLimit("0") + defer setLimit("3") + id := fresh(1, 3) + for index := 1; index <= 5; index++ { + if status, body := send(id, 1, fmt.Sprintf("off-%d", index)); status != 200 { + t.Fatalf("HTTP %d: %s", status, body) + } + } + }) + + t.Run("an AI reply is an answer, not a knock", func(t *testing.T) { + id := fresh(2, 3) + for index := 1; index <= 4; index++ { + if _, _, err := a.persistMessageContext(markAIGenerated(context.Background()), id, 3, fmt.Sprintf("ai-%d", index), 1, map[string]any{"text": "在的"}); err != nil { + t.Fatalf("托管账号的回复不该受开场限制: %v", err) + } + } + }) + + t.Run("the chat page is told how many opening messages are left", func(t *testing.T) { + id := fresh(1, 2) + read := func(user int64) map[string]any { + t.Helper() + var payload struct { + Quota map[string]any `json:"unansweredQuota"` + } + path := fmt.Sprintf("/api/v1/im/conversations/%d/messages", id) + if err := json.Unmarshal(imTestCall(t, a.messages, user, "GET", path, "", 200), &payload); err != nil { + t.Fatal(err) + } + return payload.Quota + } + if quota := read(1); quota["remaining"] != float64(3) || quota["unlimited"] != false { + t.Fatalf("quota = %v", quota) + } + send(id, 1, "quota-1") + if quota := read(1); quota["remaining"] != float64(2) { + t.Fatalf("quota = %v", read(1)) + } + send(id, 2, "quota-answer") + if quota := read(1); quota["unlimited"] != true || quota["remaining"] != float64(-1) { + t.Fatalf("对方回复后不再有额度概念: %v", quota) + } + }) +} diff --git a/im/backend/migrations/034_unanswered_message_limit.sql b/im/backend/migrations/034_unanswered_message_limit.sql new file mode 100644 index 0000000..8a4518a --- /dev/null +++ b/im/backend/migrations/034_unanswered_message_limit.sql @@ -0,0 +1,4 @@ +-- 对方回复之前,免费用户最多能发几条。0 表示不限制;会员始终不受限。 +INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES + ('membership.free_unanswered_message_limit','3','integer','对方回复前普通用户可发送的消息条数,0 表示不限制') +ON DUPLICATE KEY UPDATE description=VALUES(description);