开场消息条数限制，附近按距离分档随机刷新

一、对方回复前，免费用户最多发送 N 条（默认 3，迁移 034，管理端可改，
0 表示不限）。约束在 persistMessageContext 的事务内，HTTP 与 WebSocket
两条发送路径都覆盖，并发也不会挤过最后一个名额。要点：
- 对方一旦开口，限制永久解除；会员始终不受限。
- 撤回自己的消息不会换回名额，否则删了重发即可绕过。
- AI 托管账号的回复是「回答」不是「敲门」，不计入限制。
- 被拒时返回 30007 + HTTP 402，与会员媒体限制区分，客户端据此引导开通。
- 消息列表接口附带 unansweredQuota，聊天页在发送失败之前就把剩余条数
  告诉用户。

二、附近下拉刷新按距离分档随机。原先带 seed 的分支排在最前，导致附近
一旦刷新就退化成全城纯随机、距离排序被完全丢弃。现在按 1 公里分档，
档内用同一 seed 做确定性洗牌：刷新会换人，但近处的人永远排在远处之前，
同一 seed 下分页顺序稳定，不会重复或漏人。

两个既有测试显式关闭了开场限制——它们测的是入队规则和媒体会员限制，
连发消息只是手段。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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)`,
 	} {

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 {

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

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}
+}

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)
+			}
+		}
+	})
+}

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 ?`

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)
+		}
+	})
+}

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);