NixMsg/sdk/go/client.go

package nixmsg

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"
)

type pendingReq struct {
	rid    string
	ch     chan respFrame
	isSend bool
}

type sendItem struct {
	frame     map[string]any
	payload   []byte
	id        string
	sendAtMs  *int64
	result    chan sendOutcome
	inflight  bool
	rateN     int
	epoch     uint64
	abandoned bool
}

type sendOutcome struct {
	res SendResult
	err error
}

type dedupState int

const (
	dedupDelivered dedupState = iota + 1
	dedupAcked
	dedupRevoked
)

type dedupEntry struct {
	state dedupState
	from  string
	id    string
}

type queuedFrame struct {
	payload []byte
	kind    string
}

const eventQueueCap = 10000

// Client NixMsg 客户端。
type Client struct {
	opts       Options
	endpointID string
	url        string
	credKind   string
	transport  transport
	backoff    *reconnectBackoff

	mu            sync.Mutex
	state         ConnectionState
	handshook     bool
	limits        HandshakeLimits
	clockSkew     int64
	session       string
	stopReconnect bool
	closed        bool
	lastStopCode  string
	lastStopErr   error

	ridSeq   atomic.Uint64
	pending  map[string]*pendingReq
	sendQ    []*sendItem
	inflight int

	store *lruCache

	onSession    func(token string)
	onMessage    func(msg Message) error
	onReceipt    func(r Receipt)
	onRevoked    func(e RevokedEvent)
	onPresence   func(e PresenceEvent)
	onGroupEvent func(e GroupEvent)
	onConnection func(ev ConnectionEvent)

	helloSentAt time.Time
	ctx         context.Context
	cancel      context.CancelFunc

	incoming chan queuedFrame
	cbQ      chan func()
	connLost chan struct{}

	watchIDs []string
	watchAll bool
	watchSet bool
}

// New 创建客户端（尚未连接）。
func New() *Client {
	return &Client{
		pending: make(map[string]*pendingReq),
		store:   newLRU(10000),
		state:   StateOffline,
		backoff: newReconnectBackoff(),
	}
}

// OnSession 会话令牌回调。
func (c *Client) OnSession(h func(token string)) { c.onSession = h }

// OnMessage 消息回调。自动模式下返回 error 则不发 ack 并删去重记录。
func (c *Client) OnMessage(h func(msg Message) error) { c.onMessage = h }

// OnReceipt 回执回调。
func (c *Client) OnReceipt(h func(r Receipt)) { c.onReceipt = h }

// OnRevoked 撤回/作废回调。
func (c *Client) OnRevoked(h func(e RevokedEvent)) { c.onRevoked = h }

// OnPresence 上下线回调。
func (c *Client) OnPresence(h func(e PresenceEvent)) { c.onPresence = h }

// OnGroupEvent 群事件回调。
func (c *Client) OnGroupEvent(h func(e GroupEvent)) { c.onGroupEvent = h }

// OnConnection 连接状态回调。
func (c *Client) OnConnection(h func(ev ConnectionEvent)) { c.onConnection = h }

func newMessageID() string {
	id, err := uuid.NewV7()
	if err != nil {
		return uuid.NewString()
	}
	return id.String()
}

func (c *Client) nextRID() string {
	return fmt.Sprintf("%d", c.ridSeq.Add(1))
}

type respFrame struct {
	OK    bool            `json:"ok"`
	Data  json.RawMessage `json:"data"`
	Error *struct {
		Code    string `json:"code"`
		Message string `json:"message"`
	} `json:"error"`
}

func (c *Client) dispatch(fn func()) {
	if fn == nil {
		return
	}
	c.mu.Lock()
	c.dispatchLocked(fn)
	c.mu.Unlock()
}

func (c *Client) dispatchLocked(fn func()) {
	if fn == nil {
		return
	}
	ch := c.cbQ
	ctx := c.ctx
	if ch == nil {
		go fn()
		return
	}
	select {
	case ch <- fn:
	default:
		go func() {
			if ctx == nil {
				ch <- fn
				return
			}
			select {
			case ch <- fn:
			case <-ctx.Done():
			}
		}()
	}
}

func (c *Client) cbLoop(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case fn := <-c.cbQ:
			if fn != nil {
				fn()
			}
		}
	}
}

func (c *Client) stopErrLocked() error {
	if c.lastStopErr != nil {
		return c.lastStopErr
	}
	if c.lastStopCode != "" {
		return apiErr(c.lastStopCode, "已停止重连")
	}
	return apiErr(CodeClosed, "已关闭")
}

func (c *Client) failPendingLocked(err error, sendsToo bool) {
	for rid, p := range c.pending {
		if p == nil {
			continue
		}
		if p.isSend && !sendsToo {
			continue
		}
		select {
		case p.ch <- respFrame{OK: false, Error: &struct {
			Code    string `json:"code"`
			Message string `json:"message"`
		}{Code: errCode(err), Message: err.Error()}}:
		default:
		}
		delete(c.pending, rid)
	}
}

func errCode(err error) string {
	if err == nil {
		return CodeNotConnected
	}
	if ae, ok := err.(*APIError); ok {
		return ae.Code
	}
	return CodeNotConnected
}

func (c *Client) closeConnLostLocked() {
	if c.connLost != nil {
		select {
		case <-c.connLost:
		default:
			close(c.connLost)
		}
		c.connLost = nil
	}
}

func (c *Client) regenerateSendLocked(it *sendItem) {
	rid := c.nextRID()
	it.frame["rid"] = rid
	payload, err := marshalJSON(it.frame)
	if err != nil {
		return
	}
	it.payload = payload
}

func (c *Client) requeueInflightLocked() {
	for _, it := range c.sendQ {
		if !it.inflight {
			continue
		}
		it.epoch++
		it.inflight = false
		rid, _ := it.frame["rid"].(string)
		delete(c.pending, rid)
		c.regenerateSendLocked(it)
	}
	c.inflight = 0
}

func (c *Client) onTransportOffline() {
	c.mu.Lock()
	c.handshook = false
	c.closeConnLostLocked()
	if c.stopReconnect || c.closed {
		c.failPendingLocked(apiErr(CodeNotConnected, "未连接"), true)
		c.mu.Unlock()
		return
	}
	c.requeueInflightLocked()
	c.failPendingLocked(apiErr(CodeNotConnected, "未连接"), false)
	c.setStateLocked(StateReconnecting, "")
	c.mu.Unlock()
}