kefu/im/.deploy/random-refresh-20260902-090437/source/rollback-backend/internal/app/im.go

package app

import (
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/url"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
	"unicode/utf8"

	"github.com/gorilla/websocket"
)

type messageView struct {
	ID             int64     `json:"id"`
	ConversationID int64     `json:"conversationId"`
	Seq            int64     `json:"seq"`
	SenderID       int64     `json:"senderId"`
	ClientMsgID    string    `json:"clientMsgId"`
	Type           int       `json:"type"`
	Content        any       `json:"content"`
	Recalled       bool      `json:"recalled"`
	CreatedAt      time.Time `json:"createdAt"`
}

const (
	maxMessageBodyBytes = 16 << 10
	maxMessageTextRunes = 2000
)

var clientMessageIDPattern = regexp.MustCompile(`^[A-Za-z0-9_.:-]{1,64}$`)

func validMessageMediaURL(raw string, production bool) bool {
	raw = strings.TrimSpace(raw)
	if raw == "" || len(raw) > 2048 {
		return false
	}
	if strings.HasPrefix(raw, "/uploads/") {
		return !strings.Contains(raw, "..")
	}
	parsed, err := url.Parse(raw)
	if err != nil || parsed.Host == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") {
		return false
	}
	return !production || parsed.Scheme == "https"
}

func numericDuration(value any) (int, bool) {
	switch duration := value.(type) {
	case float64:
		return int(duration), duration == float64(int(duration))
	case int:
		return duration, true
	case int64:
		return int(duration), true
	case json.Number:
		parsed, err := strconv.Atoi(duration.String())
		return parsed, err == nil
	default:
		return 0, false
	}
}

func (a *App) validateMessagePayload(clientMsgID string, messageType int, content any) (string, any, []byte, error) {
	clientMsgID = strings.TrimSpace(clientMsgID)
	if clientMsgID == "" {
		clientMsgID = randomToken()[:26]
	}
	if !clientMessageIDPattern.MatchString(clientMsgID) {
		return "", nil, nil, fmt.Errorf("客户端消息 ID 格式无效")
	}
	if messageType == 0 {
		messageType = 1
	}
	data, ok := content.(map[string]any)
	if !ok {
		return "", nil, nil, fmt.Errorf("消息内容格式错误")
	}
	switch messageType {
	case 1:
		text, ok := data["text"].(string)
		text = strings.TrimSpace(text)
		if !ok || text == "" || !utf8.ValidString(text) || utf8.RuneCountInString(text) > maxMessageTextRunes {
			return "", nil, nil, fmt.Errorf("文本消息应为 1 至 %d 个字符", maxMessageTextRunes)
		}
		data = map[string]any{"text": text}
	case 2:
		mediaURL, ok := data["url"].(string)
		if !ok || !validMessageMediaURL(mediaURL, a.config.Environment == "production") {
			return "", nil, nil, fmt.Errorf("图片地址无效")
		}
		data = map[string]any{"url": strings.TrimSpace(mediaURL)}
	case 3:
		mediaURL, ok := data["url"].(string)
		duration, durationOK := numericDuration(data["duration"])
		if !ok || !validMessageMediaURL(mediaURL, a.config.Environment == "production") || !durationOK || duration < 1 || duration > 60 {
			return "", nil, nil, fmt.Errorf("语音消息地址或时长无效")
		}
		data = map[string]any{"duration": duration, "url": strings.TrimSpace(mediaURL)}
	default:
		return "", nil, nil, fmt.Errorf("不支持的消息类型")
	}
	body, err := json.Marshal(data)
	if err != nil || len(body) > maxMessageBodyBytes {
		return "", nil, nil, fmt.Errorf("消息内容过大")
	}
	return clientMsgID, data, body, nil
}

type wsClient struct {
	userID int64
	conn   *websocket.Conn
	mu     sync.Mutex
}

func (c *wsClient) writeJSON(payload any) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	_ = c.conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
	return c.conn.WriteJSON(payload)
}

type Hub struct {
	mu      sync.RWMutex
	clients map[int64]map[*wsClient]struct{}
}

func NewHub() *Hub { return &Hub{clients: make(map[int64]map[*wsClient]struct{})} }

func (h *Hub) add(client *wsClient) {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.clients[client.userID] == nil {
		h.clients[client.userID] = make(map[*wsClient]struct{})
	}
	h.clients[client.userID][client] = struct{}{}
}

func (h *Hub) remove(client *wsClient) {
	h.mu.Lock()
	defer h.mu.Unlock()
	delete(h.clients[client.userID], client)
	if len(h.clients[client.userID]) == 0 {
		delete(h.clients, client.userID)
	}
}

func (h *Hub) broadcast(userIDs []int64, payload any) {
	h.mu.RLock()
	targets := []*wsClient{}
	for _, userID := range userIDs {
		for client := range h.clients[userID] {
			targets = append(targets, client)
		}
	}
	h.mu.RUnlock()
	for _, client := range targets {
		_ = client.writeJSON(payload)
	}
}

func (h *Hub) disconnect(userID int64) {
	h.mu.RLock()
	targets := []*wsClient{}
	for client := range h.clients[userID] {
		targets = append(targets, client)
	}
	h.mu.RUnlock()
	for _, client := range targets {
		client.mu.Lock()
		_ = client.conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.ClosePolicyViolation, "登录状态已失效"), time.Now().Add(time.Second))
		_ = client.conn.Close()
		client.mu.Unlock()
	}
}

func (a *App) directConversation(w http.ResponseWriter, r *http.Request) {
	var req struct {
		UserID int64 `json:"userId"`
	}
	if decode(r, &req) != nil || req.UserID == 0 || req.UserID == current(r).ID {
		fail(w, 400, 20001, "无效的聊天对象")
		return
	}
	first, second := current(r).ID, req.UserID
	if first > second {
		first, second = second, first
	}
	var targetStatus, allowStranger int
	if err := a.db.QueryRowContext(r.Context(), `SELECT u.status,p.allow_stranger_message FROM users u JOIN user_privacy_settings p ON p.user_id=u.id WHERE u.id=? AND u.deleted_at IS NULL`, req.UserID).Scan(&targetStatus, &allowStranger); err != nil || targetStatus != 1 {
		fail(w, http.StatusNotFound, 30001, "聊天对象不存在或不可用")
		return
	}
	var blocked int
	if err := a.db.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM user_blocks WHERE (user_id=? AND blocked_user_id=?) OR (user_id=? AND blocked_user_id=?))`, current(r).ID, req.UserID, req.UserID, current(r).ID).Scan(&blocked); err != nil || blocked == 1 {
		fail(w, http.StatusForbidden, 30002, "当前无法与该用户聊天")
		return
	}
	var id int64
	err := a.db.QueryRowContext(r.Context(), `SELECT conversation_id FROM im_direct_conversations WHERE user1_id=? AND user2_id=?`, first, second).Scan(&id)
	if err == nil {
		reply(w, map[string]any{"id": id})
		return
	}
	if allowStranger == 0 {
		var mutualFollow int
		_ = a.db.QueryRowContext(r.Context(), `SELECT EXISTS(
			SELECT 1 FROM user_follows mine JOIN user_follows target
			ON target.user_id=mine.target_user_id AND target.target_user_id=mine.user_id
			WHERE mine.user_id=? AND mine.target_user_id=?)`, current(r).ID, req.UserID).Scan(&mutualFollow)
		if mutualFollow != 1 {
			fail(w, http.StatusForbidden, 30002, "对方仅允许互相关注的人发起私信")
			return
		}
	}
	tx, beginErr := a.db.BeginTx(r.Context(), nil)
	if beginErr != nil {
		fail(w, http.StatusInternalServerError, 50001, "创建会话失败")
		return
	}
	defer func() { _ = tx.Rollback() }()
	result, err := tx.ExecContext(r.Context(), `INSERT INTO im_conversations(conversation_type)VALUES(1)`)
	if err != nil {
		_ = tx.Rollback()
		// Concurrent requests may have created the same unique direct pair.
		if a.db.QueryRowContext(r.Context(), `SELECT conversation_id FROM im_direct_conversations WHERE user1_id=? AND user2_id=?`, first, second).Scan(&id) == nil {
			reply(w, map[string]any{"id": id})
			return
		}
		fail(w, 500, 50001, "创建会话失败")
		return
	}
	id, _ = result.LastInsertId()
	_, err = tx.ExecContext(r.Context(), `INSERT INTO im_direct_conversations(conversation_id,user1_id,user2_id)VALUES(?,?,?)`, id, first, second)
	if err == nil {
		_, err = tx.ExecContext(r.Context(), `INSERT INTO im_conversation_members(conversation_id,user_id)VALUES(?,?),(?,?)`, id, first, id, second)
	}
	if err != nil {
		_ = tx.Rollback()
		fail(w, 500, 50001, "创建会话失败")
		return
	}
	_ = tx.Commit()
	reply(w, map[string]any{"id": id})
}

func (a *App) conversations(w http.ResponseWriter, r *http.Request) {
	who := current(r)
	rows, err := a.db.QueryContext(r.Context(), `SELECT c.id,c.last_seq,c.last_message_at,m.read_seq,m.pinned,m.muted,
		other.user_id,p.nickname,p.avatar_url,p.is_vip,p.last_active_at,privacy.online_visible,COALESCE(CAST(msg.body AS CHAR CHARACTER SET utf8mb4),''),msg.recalled_at,msg.admin_removed_at
		FROM im_conversation_members m JOIN im_conversations c ON c.id=m.conversation_id
		JOIN im_conversation_members other ON other.conversation_id=c.id AND other.user_id<>m.user_id
		JOIN user_profiles p ON p.user_id=other.user_id JOIN user_privacy_settings privacy ON privacy.user_id=other.user_id LEFT JOIN im_messages msg ON msg.id=c.last_message_id
		WHERE m.user_id=? AND m.status=1 ORDER BY m.pinned DESC,c.last_message_at DESC`, who.ID)
	if err != nil {
		fail(w, 500, 50001, err.Error())
		return
	}
	defer rows.Close()
	items := []map[string]any{}
	for rows.Next() {
		var id, lastSeq, readSeq, otherID int64
		var lastAt sql.NullTime
		var pinned, muted, vip int
		var nick, avatar, body string
		var active sql.NullTime
		var recalledAt, adminRemovedAt sql.NullTime
		var onlineVisible int
		_ = rows.Scan(&id, &lastSeq, &lastAt, &readSeq, &pinned, &muted, &otherID, &nick, &avatar, &vip, &active, &onlineVisible, &body, &recalledAt, &adminRemovedAt)
		preview := "开始聊天吧"
		var content map[string]any
		if recalledAt.Valid || adminRemovedAt.Valid {
			preview = "消息已撤回"
		} else if json.Unmarshal([]byte(body), &content) == nil {
			if text, ok := content["text"].(string); ok {
				preview = text
			}
		}
		items = append(items, map[string]any{"id": id, "lastSeq": lastSeq, "unread": max64(lastSeq-readSeq, 0), "lastMessageAt": lastAt, "pinned": pinned == 1, "muted": muted == 1, "lastMessage": preview, "user": map[string]any{"id": otherID, "nickname": nick, "avatar": avatar, "vip": vip == 1, "online": onlineVisible == 1 && active.Valid && time.Since(active.Time) < 15*time.Minute}})
	}
	reply(w, map[string]any{"items": items, "total": len(items)})
}

func (a *App) messages(w http.ResponseWriter, r *http.Request) {
	id, err := pathID(r)
	if err != nil {
		fail(w, 400, 20001, "invalid conversation")
		return
	}
	if !a.isMember(r, id, current(r).ID) {
		fail(w, 403, 30002, "不是会话成员")
		return
	}
	beforeSeq, _ := strconv.ParseInt(r.URL.Query().Get("beforeSeq"), 10, 64)
	limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
	if limit <= 0 || limit > 100 {
		limit = 50
	}
	query := `SELECT id,conversation_id,seq,sender_id,client_msg_id,message_type,body,recalled_at,admin_removed_at,created_at FROM im_messages WHERE conversation_id=?`
	args := []any{id}
	if beforeSeq > 0 {
		query += ` AND seq<?`
		args = append(args, beforeSeq)
	}
	query += ` ORDER BY seq DESC LIMIT ?`
	args = append(args, limit+1)
	rows, err := a.db.QueryContext(r.Context(), query, args...)
	if err != nil {
		fail(w, 500, 50001, "查询消息失败")
		return
	}
	defer rows.Close()
	items := []messageView{}
	for rows.Next() {
		var item messageView
		var body []byte
		var recalledAt, adminRemovedAt sql.NullTime
		_ = rows.Scan(&item.ID, &item.ConversationID, &item.Seq, &item.SenderID, &item.ClientMsgID, &item.Type, &body, &recalledAt, &adminRemovedAt, &item.CreatedAt)
		var content any
		item.Recalled = recalledAt.Valid || adminRemovedAt.Valid
		if !item.Recalled {
			_ = json.Unmarshal(body, &content)
		}
		item.Content = content
		items = append(items, item)
	}
	hasMore := len(items) > limit
	if hasMore {
		items = items[:limit]
	}
	sort.Slice(items, func(i, j int) bool { return items[i].Seq < items[j].Seq })
	if len(items) > 0 {
		last := items[len(items)-1].Seq
		_, _ = a.db.ExecContext(r.Context(), `UPDATE im_conversation_members SET read_seq=GREATEST(read_seq,?) WHERE conversation_id=? AND user_id=?`, last, id, current(r).ID)
	}
	nextBeforeSeq := int64(0)
	if len(items) > 0 {
		nextBeforeSeq = items[0].Seq
	}
	reply(w, map[string]any{"items": items, "hasMore": hasMore, "nextBeforeSeq": nextBeforeSeq})
}

func (a *App) sendMessageHTTP(w http.ResponseWriter, r *http.Request) {
	conversationID, err := pathID(r)
	if err != nil {
		fail(w, 400, 20001, "invalid conversation")
		return
	}
	var req struct {
		ClientMsgID string `json:"clientMsgId"`
		Type        int    `json:"type"`
		Content     any    `json:"content"`
	}
	if decode(r, &req) != nil {
		fail(w, 400, 20001, "消息格式错误")
		return
	}
	item, members, err := a.persistMessage(r, conversationID, current(r).ID, req.ClientMsgID, req.Type, req.Content)
	if err != nil {
		var limitErr *dailyActiveChatLimitError
		if errors.As(err, &limitErr) {
			fail(w, http.StatusTooManyRequests, 30005, limitErr.Error())
			return
		}
		fail(w, 400, 30004, err.Error())
		return
	}
	a.hub.broadcast(members, map[string]any{"command": "MESSAGE_PUSH", "data": item})
	reply(w, item)
}

func (a *App) recallMessage(w http.ResponseWriter, r *http.Request) {
	messageID, err := pathID(r)
	if err != nil {
		fail(w, http.StatusBadRequest, 20001, "消息编号无效")
		return
	}
	window, _ := strconv.Atoi(a.configPlain(r.Context(), "im.recall_seconds", "120"))
	if window < 1 || window > 86400 {
		window = 120
	}
	tx, err := a.db.BeginTx(r.Context(), nil)
	if err != nil {
		fail(w, 500, 50001, "撤回失败")
		return
	}
	defer func() { _ = tx.Rollback() }()
	var conversationID, seq int64
	var senderID int64
	var created time.Time
	var recalled sql.NullTime
	if err = tx.QueryRowContext(r.Context(), `SELECT conversation_id,seq,sender_id,created_at,recalled_at FROM im_messages WHERE id=? FOR UPDATE`, messageID).Scan(&conversationID, &seq, &senderID, &created, &recalled); err != nil {
		fail(w, 404, 30001, "消息不存在")
		return
	}
	if senderID != current(r).ID {
		fail(w, 403, 30002, "只能撤回自己发送的消息")
		return
	}
	if recalled.Valid {
		reply(w, map[string]bool{"success": true})
		return
	}
	if time.Since(created) > time.Duration(window)*time.Second {
		fail(w, 409, 30004, "已超过消息撤回时限")
		return
	}
	_, err = tx.ExecContext(r.Context(), `UPDATE im_messages SET recalled_at=NOW(3) WHERE id=?`, messageID)
	if err != nil || tx.Commit() != nil {
		fail(w, 500, 50001, "撤回失败")
		return
	}
	rows, _ := a.db.QueryContext(r.Context(), `SELECT user_id FROM im_conversation_members WHERE conversation_id=? AND status=1`, conversationID)
	members := []int64{}
	if rows != nil {
		defer rows.Close()
		for rows.Next() {
			var id int64
			_ = rows.Scan(&id)
			members = append(members, id)
		}
	}
	a.hub.broadcast(members, map[string]any{"command": "MESSAGE_RECALLED", "data": map[string]any{"id": messageID, "conversationId": conversationID, "seq": seq}})
	reply(w, map[string]bool{"success": true})
}

func (a *App) persistMessage(r *http.Request, conversationID, senderID int64, clientMsgID string, messageType int, content any) (messageView, []int64, error) {
	item := messageView{}
	clientMsgID, content, body, err := a.validateMessagePayload(clientMsgID, messageType, content)
	if err != nil {
		return item, nil, err
	}
	if messageType == 0 {
		messageType = 1
	}
	if messageType == 2 || messageType == 3 {
		contentMap, _ := content.(map[string]any)
		mediaURL, _ := contentMap["url"].(string)
		expectedType := "image"
		if messageType == 3 {
			expectedType = "audio"
		}
		var owned int
		if queryErr := a.db.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM media_assets WHERE owner_user_id=? AND public_url=? AND media_type=? AND status=1 AND moderation_status=1)`, senderID, mediaURL, expectedType).Scan(&owned); queryErr != nil || owned != 1 {
			return item, nil, fmt.Errorf("消息媒体必须由当前账号上传")
		}
	}
	if !a.allowRequest(r.Context(), "message_send", fmt.Sprintf("%d", senderID), 120, time.Minute) {
		return item, nil, fmt.Errorf("消息发送过于频繁，请稍后再试")
	}
	if a.isSanctionActive(r.Context(), senderID, "MUTE") {
		return item, nil, fmt.Errorf("账号处于禁言期，暂时无法发送消息")
	}
	tx, err := a.db.BeginTx(r.Context(), nil)
	if err != nil {
		return item, nil, err
	}
	defer func() { _ = tx.Rollback() }()
	var lastSeq int64
	if err = tx.QueryRowContext(r.Context(), `SELECT last_seq FROM im_conversations WHERE id=? AND status=1 FOR UPDATE`, conversationID).Scan(&lastSeq); err != nil {
		return item, nil, fmt.Errorf("会话不存在")
	}
	var memberCount int
	if err = tx.QueryRowContext(r.Context(), `SELECT COUNT(*) FROM im_conversation_members WHERE conversation_id=? AND user_id=? AND status=1`, conversationID, senderID).Scan(&memberCount); err != nil || memberCount == 0 {
		return item, nil, fmt.Errorf("不是会话成员")
	}
	var blocked int
	if err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(
		SELECT 1 FROM im_direct_conversations d JOIN user_blocks b
		ON (b.user_id=d.user1_id AND b.blocked_user_id=d.user2_id) OR (b.user_id=d.user2_id AND b.blocked_user_id=d.user1_id)
		WHERE d.conversation_id=?)`, conversationID).Scan(&blocked); err != nil {
		return item, nil, err
	}
	if blocked == 1 {
		return item, nil, fmt.Errorf("当前无法向该用户发送消息")
	}
	if err = a.reserveDailyActiveChat(r.Context(), tx, conversationID, senderID); err != nil {
		return item, nil, err
	}
	seq := lastSeq + 1
	result, err := tx.ExecContext(r.Context(), `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 {
		var existingID, existingSeq int64
		existingErr := tx.QueryRowContext(r.Context(), `SELECT id,seq FROM im_messages WHERE sender_id=? AND client_msg_id=?`, senderID, clientMsgID).Scan(&existingID, &existingSeq)
		if existingErr == nil {
			_ = tx.Rollback()
			return a.loadMessage(r, existingID), nil, nil
		}
		return item, nil, err
	}
	messageID, _ := result.LastInsertId()
	_, err = tx.ExecContext(r.Context(), `UPDATE im_conversations SET last_seq=?,last_message_id=?,last_message_at=NOW(3) WHERE id=?`, seq, messageID, conversationID)
	if err != nil {
		return item, nil, err
	}
	if _, err = tx.ExecContext(r.Context(), `UPDATE im_conversation_members SET delivered_seq=GREATEST(delivered_seq,?),updated_at=NOW(3) WHERE conversation_id=?`, seq, conversationID); err != nil {
		return item, nil, err
	}
	memberRows, err := tx.QueryContext(r.Context(), `SELECT user_id FROM im_conversation_members WHERE conversation_id=? AND status=1`, conversationID)
	if err != nil {
		return item, nil, err
	}
	members := []int64{}
	for memberRows.Next() {
		var userID int64
		if err = memberRows.Scan(&userID); err != nil {
			_ = memberRows.Close()
			return item, nil, err
		}
		members = append(members, userID)
	}
	if err = memberRows.Err(); err != nil {
		_ = memberRows.Close()
		return item, nil, err
	}
	if err = memberRows.Close(); err != nil {
		return item, nil, err
	}
	sort.Slice(members, func(i, j int) bool { return members[i] < members[j] })
	for _, userID := range members {
		var lockedUserID int64
		if err = tx.QueryRowContext(r.Context(), `SELECT id FROM users WHERE id=? FOR UPDATE`, userID).Scan(&lockedUserID); err != nil {
			return item, nil, err
		}
	}
	for _, userID := range members {
		var next int64
		if err = tx.QueryRowContext(r.Context(), `SELECT COALESCE(MAX(event_seq),0)+1 FROM im_user_sync_events WHERE user_id=?`, userID).Scan(&next); err != nil {
			return item, nil, err
		}
		if _, err = tx.ExecContext(r.Context(), `INSERT INTO im_user_sync_events(user_id,event_seq,event_type,conversation_id,message_seq,event_data)VALUES(?, ?,12,?,?,?)`, userID, next, conversationID, seq, body); err != nil {
			return item, nil, err
		}
	}
	if err = tx.Commit(); err != nil {
		return item, nil, err
	}
	return messageView{ID: messageID, ConversationID: conversationID, Seq: seq, SenderID: senderID, ClientMsgID: clientMsgID, Type: messageType, Content: content, CreatedAt: time.Now()}, members, nil
}

func (a *App) loadMessage(r *http.Request, id int64) messageView {
	var item messageView
	var body []byte
	var recalledAt, adminRemovedAt sql.NullTime
	_ = a.db.QueryRowContext(r.Context(), `SELECT id,conversation_id,seq,sender_id,client_msg_id,message_type,body,recalled_at,admin_removed_at,created_at FROM im_messages WHERE id=?`, id).Scan(&item.ID, &item.ConversationID, &item.Seq, &item.SenderID, &item.ClientMsgID, &item.Type, &body, &recalledAt, &adminRemovedAt, &item.CreatedAt)
	item.Recalled = recalledAt.Valid || adminRemovedAt.Valid
	if !item.Recalled {
		_ = json.Unmarshal(body, &item.Content)
	}
	return item
}
func (a *App) isMember(r *http.Request, conversationID, userID int64) bool {
	var count int
	_ = a.db.QueryRowContext(r.Context(), `SELECT COUNT(*) FROM im_conversation_members WHERE conversation_id=? AND user_id=? AND status=1`, conversationID, userID).Scan(&count)
	return count > 0
}

func (a *App) conversationSettings(w http.ResponseWriter, r *http.Request) {
	id, pathErr := pathID(r)
	if pathErr != nil {
		fail(w, http.StatusBadRequest, 20001, "会话 ID 无效")
		return
	}
	var req struct {
		Pinned  *bool `json:"pinned"`
		Muted   *bool `json:"muted"`
		ReadSeq int64 `json:"readSeq"`
	}
	if decode(r, &req) != nil || req.ReadSeq < 0 {
		fail(w, 400, 20001, "invalid settings")
		return
	}
	pinned, muted := -1, -1
	if req.Pinned != nil {
		pinned = btoi(*req.Pinned)
	}
	if req.Muted != nil {
		muted = btoi(*req.Muted)
	}
	var lastSeq int64
	if err := a.db.QueryRowContext(r.Context(), `SELECT c.last_seq FROM im_conversations c JOIN im_conversation_members m ON m.conversation_id=c.id WHERE c.id=? AND m.user_id=? AND m.status=1`, id, current(r).ID).Scan(&lastSeq); err != nil {
		fail(w, http.StatusForbidden, 30002, "不是会话成员")
		return
	}
	readSeq := req.ReadSeq
	if readSeq > lastSeq {
		readSeq = lastSeq
	}
	_, err := a.db.ExecContext(r.Context(), `UPDATE im_conversation_members SET pinned=IF(?>=0,?,pinned),muted=IF(?>=0,?,muted),read_seq=GREATEST(read_seq,?) WHERE conversation_id=? AND user_id=?`, pinned, pinned, muted, muted, readSeq, id, current(r).ID)
	if err != nil {
		fail(w, 500, 50001, "保存失败")
		return
	}
	reply(w, map[string]bool{"success": true})
}

func (a *App) websocket(w http.ResponseWriter, r *http.Request) {
	if !a.originAllowed(r.Header.Get("Origin")) {
		fail(w, http.StatusForbidden, 10006, "WebSocket 来源不允许")
		return
	}
	selectedProtocol := ""
	raw := ""
	for _, protocol := range websocket.Subprotocols(r) {
		if strings.HasPrefix(protocol, "xingyu.jwt.") {
			selectedProtocol = protocol
			raw = strings.TrimPrefix(protocol, "xingyu.jwt.")
			break
		}
	}
	// Query-token compatibility is development-only because URLs may be written to proxy logs.
	if raw == "" && a.config.Environment != "production" {
		raw = r.URL.Query().Get("token")
	}
	who, err := a.parseToken(raw)
	if err != nil || who.Role != "user" {
		fail(w, 401, 10001, "invalid token")
		return
	}
	var userStatus int
	statusErr := a.db.QueryRowContext(r.Context(), `SELECT status FROM users WHERE id=? AND deleted_at IS NULL`, who.ID).Scan(&userStatus)
	if statusErr == nil {
		userStatus = a.normalizeUserStatus(r.Context(), who.ID, userStatus)
	}
	if statusErr != nil || userStatus != 1 {
		fail(w, 403, 10006, "账号已被冻结或封禁")
		return
	}
	var tokenVersion int
	_ = a.db.QueryRowContext(r.Context(), `SELECT token_version FROM user_security_controls WHERE user_id=?`, who.ID).Scan(&tokenVersion)
	if who.Version != tokenVersion {
		fail(w, 401, 10001, "登录状态已失效")
		return
	}
	upgrader := websocket.Upgrader{CheckOrigin: func(request *http.Request) bool {
		return a.originAllowed(request.Header.Get("Origin"))
	}}
	if selectedProtocol != "" {
		upgrader.Subprotocols = []string{selectedProtocol}
	}
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		return
	}
	client := &wsClient{userID: who.ID, conn: conn}
	conn.SetReadLimit(64 << 10)
	_ = conn.SetReadDeadline(time.Now().Add(75 * time.Second))
	a.hub.add(client)
	defer func() { a.hub.remove(client); _ = conn.Close() }()
	_ = client.writeJSON(map[string]any{"command": "AUTH_ACK", "data": map[string]any{"heartbeatSeconds": 25, "serverTime": time.Now().UnixMilli()}})
	for {
		var frame struct {
			Command        string `json:"command"`
			ConversationID int64  `json:"conversationId"`
			ClientMsgID    string `json:"clientMsgId"`
			Type           int    `json:"type"`
			Content        any    `json:"content"`
			ReadSeq        int64  `json:"readSeq"`
		}
		if conn.ReadJSON(&frame) != nil {
			return
		}
		switch frame.Command {
		case "PING":
			_ = conn.SetReadDeadline(time.Now().Add(75 * time.Second))
			_ = client.writeJSON(map[string]any{"command": "PONG", "timestamp": time.Now().UnixMilli()})
		case "SEND_MESSAGE":
			item, members, persistErr := a.persistMessage(r, frame.ConversationID, who.ID, frame.ClientMsgID, frame.Type, frame.Content)
			if persistErr != nil {
				_ = client.writeJSON(map[string]any{"command": "ERROR", "message": persistErr.Error()})
				continue
			}
			a.hub.broadcast(members, map[string]any{"command": "MESSAGE_PUSH", "data": item})
		case "READ":
			if frame.ReadSeq < 0 {
				_ = client.writeJSON(map[string]any{"command": "ERROR", "message": "已读序号无效"})
				continue
			}
			var lastSeq int64
			if a.db.QueryRowContext(r.Context(), `SELECT c.last_seq FROM im_conversations c JOIN im_conversation_members m ON m.conversation_id=c.id WHERE c.id=? AND m.user_id=? AND m.status=1`, frame.ConversationID, who.ID).Scan(&lastSeq) != nil {
				_ = client.writeJSON(map[string]any{"command": "ERROR", "message": "不是会话成员"})
				continue
			}
			if frame.ReadSeq > lastSeq {
				frame.ReadSeq = lastSeq
			}
			_, _ = a.db.ExecContext(r.Context(), `UPDATE im_conversation_members SET read_seq=GREATEST(read_seq,?) WHERE conversation_id=? AND user_id=?`, frame.ReadSeq, frame.ConversationID, who.ID)
			a.hub.broadcast([]int64{who.ID}, map[string]any{"command": "READ_ACK", "data": frame})
		}
	}
}

func max64(a, b int64) int64 {
	if a > b {
		return a
	}
	return b
}