feat: 实现 Go 与 JS/TS SDK 连接收发与其余接口

sdk/go/LICENSE

@@ -0,0 +1,10 @@
+Copyright (c) 2026 Nixevol. All rights reserved.
+
+本仓库的源代码、文档、各语言 SDK 和构建产物（包括发布的软件包和 Docker 镜像）均为专有软件。
+源代码和发布物公开可读，不代表授予任何使用许可。未经版权所有者书面许可，不得使用、复制、
+修改、合并、发布、分发、再许可或出售其任何部分。
+
+This repository, including its source code, documentation, SDKs and build artifacts (including
+published packages and Docker images), is proprietary software. Public visibility does not grant
+any license. No part of it may be used, copied, modified, merged, published, distributed,
+sublicensed or sold without prior written permission from the copyright holder.

sdk/go/api.go

@@ -0,0 +1,224 @@
+package nixmsg
+
+import (
+	"context"
+	"encoding/json"
+)
+
+// Recall 撤回消息。
+func (c *Client) Recall(ctx context.Context, id string) (RecallResult, error) {
+	data, err := c.request(ctx, map[string]any{"v": 1, "type": "recall", "rid": c.nextRID(), "id": id}, false)
+	if err != nil {
+		return RecallResult{}, err
+	}
+	var r RecallResult
+	_ = json.Unmarshal(data, &r)
+	return r, nil
+}
+
+// Status 查询发送状态。
+func (c *Client) Status(ctx context.Context, id, cursor string, limit int) (json.RawMessage, error) {
+	req := map[string]any{"v": 1, "type": "status", "rid": c.nextRID(), "id": id}
+	if cursor != "" {
+		req["cursor"] = cursor
+	}
+	if limit > 0 {
+		req["limit"] = limit
+	}
+	return c.request(ctx, req, false)
+}
+
+// Unlock 解锁对话密码。
+func (c *Client) Unlock(ctx context.Context, endpointID, talkPassword string) error {
+	_, err := c.request(ctx, map[string]any{
+		"v": 1, "type": "unlock", "rid": c.nextRID(),
+		"endpoint_id": endpointID, "talk_password": talkPassword,
+	}, false)
+	return err
+}
+
+// Presence 查询在线。
+func (c *Client) Presence(ctx context.Context, ids []string) ([]PresenceItem, error) {
+	data, err := c.request(ctx, map[string]any{"v": 1, "type": "presence.get", "rid": c.nextRID(), "ids": ids}, false)
+	if err != nil {
+		return nil, err
+	}
+	var wrap struct {
+		Items []PresenceItem `json:"items"`
+	}
+	if json.Unmarshal(data, &wrap) == nil && wrap.Items != nil {
+		return wrap.Items, nil
+	}
+	var items []PresenceItem
+	if err := json.Unmarshal(data, &items); err == nil {
+		return items, nil
+	}
+	return nil, nil
+}
+
+// Directory 列目录。
+func (c *Client) Directory(ctx context.Context, cursor, query string, limit int) (json.RawMessage, error) {
+	req := map[string]any{"v": 1, "type": "directory.list", "rid": c.nextRID()}
+	if cursor != "" {
+		req["cursor"] = cursor
+	}
+	if query != "" {
+		req["query"] = query
+	}
+	if limit > 0 {
+		req["limit"] = limit
+	}
+	return c.request(ctx, req, false)
+}
+
+// WatchPresence 订阅上下线；ids 为空且 all 为 true 表示全部。
+func (c *Client) WatchPresence(ctx context.Context, ids []string, all bool) error {
+	req := map[string]any{"v": 1, "type": "presence.watch", "rid": c.nextRID(), "all": all}
+	if len(ids) > 0 {
+		req["ids"] = ids
+	}
+	_, err := c.request(ctx, req, false)
+	return err
+}
+
+// GetSelf 读取自己的资料。
+func (c *Client) GetSelf(ctx context.Context) (SelfInfo, error) {
+	data, err := c.request(ctx, map[string]any{"v": 1, "type": "self.get", "rid": c.nextRID()}, false)
+	if err != nil {
+		return SelfInfo{}, err
+	}
+	var s SelfInfo
+	_ = json.Unmarshal(data, &s)
+	return s, nil
+}
+
+// UpdateSelf 更新资料。
+func (c *Client) UpdateSelf(ctx context.Context, name string, defaultDelayMs *int64) error {
+	req := map[string]any{"v": 1, "type": "self.update", "rid": c.nextRID()}
+	if name != "" {
+		req["name"] = name
+	}
+	if defaultDelayMs != nil {
+		req["default_delay_ms"] = *defaultDelayMs
+	}
+	_, err := c.request(ctx, req, false)
+	return err
+}
+
+// SetTalkPassword 设置对话密码；空字符串清除。
+func (c *Client) SetTalkPassword(ctx context.Context, talkPassword string) error {
+	_, err := c.request(ctx, map[string]any{
+		"v": 1, "type": "self.talk_password", "rid": c.nextRID(), "talk_password": talkPassword,
+	}, false)
+	return err
+}
+
+// ChangeLoginPassword 修改登录密码；成功后若返回新令牌会经 onSession 交出。
+func (c *Client) ChangeLoginPassword(ctx context.Context, oldPassword, newPassword string) error {
+	data, err := c.request(ctx, map[string]any{
+		"v": 1, "type": "self.login_password", "rid": c.nextRID(),
+		"old_password": oldPassword, "new_password": newPassword,
+	}, false)
+	if err != nil {
+		return err
+	}
+	var d struct {
+		SessionToken string `json:"session_token"`
+	}
+	if json.Unmarshal(data, &d) == nil && d.SessionToken != "" {
+		c.mu.Lock()
+		c.session = d.SessionToken
+		if c.transport != nil {
+			c.transport.SetCredential(d.SessionToken)
+		}
+		c.mu.Unlock()
+		if c.onSession != nil {
+			c.cbMu.Lock()
+			c.onSession(d.SessionToken)
+			c.cbMu.Unlock()
+		}
+	}
+	return nil
+}
+
+// CreateGroup 建群。
+func (c *Client) CreateGroup(ctx context.Context, id, name string, members []GroupMemberIn) (json.RawMessage, error) {
+	ms := make([]map[string]any, 0, len(members))
+	for _, m := range members {
+		ms = append(ms, map[string]any{"id": m.ID, "talk_password": m.TalkPassword})
+	}
+	return c.request(ctx, map[string]any{
+		"v": 1, "type": "group.create", "rid": c.nextRID(), "id": id, "name": name, "members": ms,
+	}, false)
+}
+
+// AddGroupMembers 加成员。
+func (c *Client) AddGroupMembers(ctx context.Context, groupID string, members []GroupMemberIn) (json.RawMessage, error) {
+	ms := make([]map[string]any, 0, len(members))
+	for _, m := range members {
+		ms = append(ms, map[string]any{"id": m.ID, "talk_password": m.TalkPassword})
+	}
+	return c.request(ctx, map[string]any{
+		"v": 1, "type": "group.add", "rid": c.nextRID(), "group_id": groupID, "members": ms,
+	}, false)
+}
+
+// RemoveGroupMember 移除成员。
+func (c *Client) RemoveGroupMember(ctx context.Context, groupID, endpointID string) error {
+	_, err := c.request(ctx, map[string]any{
+		"v": 1, "type": "group.remove", "rid": c.nextRID(), "group_id": groupID, "endpoint_id": endpointID,
+	}, false)
+	return err
+}
+
+// LeaveGroup 退群。
+func (c *Client) LeaveGroup(ctx context.Context, groupID string) error {
+	_, err := c.request(ctx, map[string]any{"v": 1, "type": "group.leave", "rid": c.nextRID(), "group_id": groupID}, false)
+	return err
+}
+
+// TransferGroup 转让群主。
+func (c *Client) TransferGroup(ctx context.Context, groupID, endpointID string) error {
+	_, err := c.request(ctx, map[string]any{
+		"v": 1, "type": "group.transfer", "rid": c.nextRID(), "group_id": groupID, "endpoint_id": endpointID,
+	}, false)
+	return err
+}
+
+// RenameGroup 改群名。
+func (c *Client) RenameGroup(ctx context.Context, groupID, name string) error {
+	_, err := c.request(ctx, map[string]any{
+		"v": 1, "type": "group.rename", "rid": c.nextRID(), "group_id": groupID, "name": name,
+	}, false)
+	return err
+}
+
+// DissolveGroup 解散群。
+func (c *Client) DissolveGroup(ctx context.Context, groupID string) error {
+	_, err := c.request(ctx, map[string]any{"v": 1, "type": "group.dissolve", "rid": c.nextRID(), "group_id": groupID}, false)
+	return err
+}
+
+// ListGroups 列出我的群。
+func (c *Client) ListGroups(ctx context.Context, cursor string, limit int) (json.RawMessage, error) {
+	req := map[string]any{"v": 1, "type": "group.list", "rid": c.nextRID()}
+	if cursor != "" {
+		req["cursor"] = cursor
+	}
+	if limit > 0 {
+		req["limit"] = limit
+	}
+	return c.request(ctx, req, false)
+}
+
+// GetGroup 获取群详情。
+func (c *Client) GetGroup(ctx context.Context, groupID, cursor string, limit int) (json.RawMessage, error) {
+	req := map[string]any{"v": 1, "type": "group.get", "rid": c.nextRID(), "group_id": groupID}
+	if cursor != "" {
+		req["cursor"] = cursor
+	}
+	if limit > 0 {
+		req["limit"] = limit
+	}
+	return c.request(ctx, req, false)
+}

sdk/go/backoff.go

@@ -0,0 +1,122 @@
+package nixmsg
+
+import (
+	"math/rand"
+	"sync"
+	"time"
+)
+
+// reconnectBackoff 按 DEVELOPMENT 第 9 节：1s 起、加倍、上限 30s、±30% 抖动；
+// 稳定在线 60s 后恢复到 1s。
+type reconnectBackoff struct {
+	mu       sync.Mutex
+	base     time.Duration
+	onlineAt time.Time
+	online   bool
+	stable   bool
+	timer    *time.Timer
+}
+
+func newReconnectBackoff() *reconnectBackoff {
+	return &reconnectBackoff{base: time.Second}
+}
+
+func (b *reconnectBackoff) Func(attempt int) time.Duration {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if attempt <= 0 {
+		return 0
+	}
+	d := b.base
+	for i := 1; i < attempt; i++ {
+		d *= 2
+		if d > 30*time.Second {
+			d = 30 * time.Second
+			break
+		}
+	}
+	return withJitter(d)
+}
+
+func (b *reconnectBackoff) MarkOnline() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.online = true
+	b.onlineAt = time.Now()
+	b.stable = false
+	if b.timer != nil {
+		b.timer.Stop()
+	}
+	b.timer = time.AfterFunc(60*time.Second, func() {
+		b.mu.Lock()
+		defer b.mu.Unlock()
+		if b.online {
+			b.stable = true
+			b.base = time.Second
+		}
+	})
+}
+
+func (b *reconnectBackoff) MarkOffline() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.timer != nil {
+		b.timer.Stop()
+		b.timer = nil
+	}
+	wasOnline := b.online
+	b.online = false
+	if !wasOnline {
+		// 连接尚未成功就失败：在 Func 内已按 attempt 加倍，这里把 base 提到下次周期的起点。
+		next := b.base * 2
+		if next > 30*time.Second {
+			next = 30 * time.Second
+		}
+		if next < time.Second {
+			next = time.Second
+		}
+		b.base = next
+		return
+	}
+	if b.stable || time.Since(b.onlineAt) >= 60*time.Second {
+		b.base = time.Second
+		b.stable = false
+		return
+	}
+	next := b.base * 2
+	if next > 30*time.Second {
+		next = 30 * time.Second
+	}
+	b.base = next
+	b.stable = false
+}
+
+func (b *reconnectBackoff) Base() time.Duration {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.base
+}
+
+func withJitter(d time.Duration) time.Duration {
+	// ±30%
+	f := 0.7 + rand.Float64()*0.6
+	return time.Duration(float64(d) * f)
+}
+
+// computeBackoffDelay 供单测：无 attempt 与 base 计算无抖动前的标称延迟。
+func computeBackoffDelay(base time.Duration, attempt int) time.Duration {
+	if attempt <= 0 {
+		return 0
+	}
+	d := base
+	for i := 1; i < attempt; i++ {
+		d *= 2
+		if d > 30*time.Second {
+			return 30 * time.Second
+		}
+	}
+	if d > 30*time.Second {
+		return 30 * time.Second
+	}
+	return d
+}

sdk/go/client.go

@@ -0,0 +1,140 @@
+package nixmsg
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+type pendingReq struct {
+	rid string
+	ch  chan respFrame
+}
+
+type sendItem struct {
+	frame    map[string]any
+	payload  []byte
+	id       string
+	sendAtMs *int64
+	result   chan sendOutcome
+	inflight bool
+}
+
+type sendOutcome struct {
+	res SendResult
+	err error
+}
+
+type dedupState int
+
+const (
+	dedupDelivered dedupState = iota + 1
+	dedupAcked
+)
+
+type dedupEntry struct {
+	state dedupState
+	from  string
+	id    string
+}
+
+// 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
+
+	ridSeq   atomic.Uint64
+	pending  map[string]*pendingReq
+	sendQ    []*sendItem
+	inflight int
+
+	dedup    map[string]*dedupEntry
+	dedupOrd []string
+
+	receiptSeen map[string]struct{}
+
+	cbMu sync.Mutex
+
+	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
+}
+
+// New 创建客户端（尚未连接）。
+func New() *Client {
+	return &Client{
+		pending:     make(map[string]*pendingReq),
+		dedup:       make(map[string]*dedupEntry),
+		receiptSeen: make(map[string]struct{}),
+		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"`
+}

sdk/go/client_test.go

@@ -0,0 +1,254 @@
+package nixmsg
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+func connectFake(t *testing.T, fake *FakeTransport) *Client {
+	t.Helper()
+	c := New()
+	opts := Options{transport: fake, ConnectTimeout: 5 * time.Second}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if err := c.Connect(ctx, "ws://example.test/mqtt", "ep1", Credential{Password: "secret"}, opts); err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	return c
+}
+
+func TestCleanStartEveryConnect(t *testing.T) {
+	fake := NewFakeTransport()
+	c := connectFake(t, fake)
+	defer c.Close()
+
+	if err := fake.SimulateReconnect(); err != nil {
+		t.Fatal(err)
+	}
+	if err := fake.SimulateReconnect(); err != nil {
+		t.Fatal(err)
+	}
+	cs := fake.Connects()
+	if len(cs) < 3 {
+		t.Fatalf("connects=%d want >=3", len(cs))
+	}
+	for i, c0 := range cs {
+		if !c0.CleanStart {
+			t.Fatalf("connect %d CleanStart=false", i)
+		}
+		if c0.SessionExpiry != 0 {
+			t.Fatalf("connect %d SessionExpiry=%d", i, c0.SessionExpiry)
+		}
+	}
+}
+
+func TestSessionTokenCallback(t *testing.T) {
+	fake := NewFakeTransport()
+	fake.HelloToken = "nst_abc"
+	c := New()
+	var got string
+	c.OnSession(func(tok string) { got = tok })
+	opts := Options{transport: fake, ConnectTimeout: 5 * time.Second}
+	ctx := context.Background()
+	if err := c.Connect(ctx, "ws://example.test/mqtt", "ep1", Credential{Password: "p"}, opts); err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close()
+	if got != "nst_abc" {
+		t.Fatalf("token=%q", got)
+	}
+}
+
+func TestDedupReack(t *testing.T) {
+	fake := NewFakeTransport()
+	c := connectFake(t, fake)
+	defer c.Close()
+
+	var calls int
+	var mu sync.Mutex
+	c.OnMessage(func(msg Message) error {
+		mu.Lock()
+		calls++
+		mu.Unlock()
+		return nil
+	})
+
+	// 自动回复 ack
+	go func() {
+		for i := 0; i < 50; i++ {
+			for _, fr := range fake.FindUp("ack") {
+				rid, _ := fr["rid"].(string)
+				if rid != "" {
+					fake.ReplyOK(rid, map[string]any{"result": "accepted"})
+				}
+			}
+			time.Sleep(5 * time.Millisecond)
+		}
+	}()
+
+	msg, _ := marshalJSON(map[string]any{
+		"v": 1, "type": "msg", "id": "m1", "from": "a",
+		"to": map[string]any{"kind": "endpoint", "id": "ep1"},
+		"body": map[string]any{"enc": "utf8", "data": "hi"},
+		"send_at_ms": 1,
+	})
+	fake.InjectDown(msg)
+	fake.InjectDown(msg) // 已交给未确认：忽略
+	time.Sleep(50 * time.Millisecond)
+
+	// 等 ack 发出并标记已确认后再推一次
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if len(fake.FindUp("ack")) >= 1 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	time.Sleep(30 * time.Millisecond)
+	fake.InjectDown(msg) // 已确认：再 ack，不回调
+	time.Sleep(80 * time.Millisecond)
+
+	mu.Lock()
+	n := calls
+	mu.Unlock()
+	if n != 1 {
+		t.Fatalf("callbacks=%d want 1", n)
+	}
+	acks := fake.FindUp("ack")
+	if len(acks) < 2 {
+		t.Fatalf("acks=%d want >=2 (re-ack)", len(acks))
+	}
+}
+
+func TestBodyTooLargeLocal(t *testing.T) {
+	fake := NewFakeTransport()
+	fake.MaxBodyBytes = 16
+	c := connectFake(t, fake)
+	defer c.Close()
+
+	body := Body{Enc: "utf8", Data: strings.Repeat("x", 64)}
+	_, err := c.Send(context.Background(), Target{Kind: "endpoint", ID: "b"}, body, SendOptions{})
+	var ae *APIError
+	if !errors.As(err, &ae) || ae.Code != CodeBodyTooLarge {
+		t.Fatalf("err=%v", err)
+	}
+}
+
+func TestFrameTooLargeLocal(t *testing.T) {
+	fake := NewFakeTransport()
+	fake.MaxBodyBytes = 1 << 20
+	fake.MaxFrameBytes = 200
+	c := connectFake(t, fake)
+	defer c.Close()
+
+	body := Body{Enc: "utf8", Data: strings.Repeat("y", 180)}
+	_, err := c.Send(context.Background(), Target{Kind: "endpoint", ID: "b"}, body, SendOptions{})
+	var ae *APIError
+	if !errors.As(err, &ae) || ae.Code != CodeFrameTooLarge {
+		t.Fatalf("err=%v", err)
+	}
+}
+
+func TestResendKeepsIDAndSendAt(t *testing.T) {
+	fake := NewFakeTransport()
+	c := connectFake(t, fake)
+	defer c.Close()
+
+	at := time.UnixMilli(1_700_000_000_000)
+	var firstID string
+	var firstSendAt any
+	var once sync.Once
+	done := make(chan struct{})
+
+	go func() {
+		for {
+			select {
+			case <-done:
+				return
+			default:
+			}
+			sends := fake.FindUp("send")
+			if len(sends) == 0 {
+				time.Sleep(5 * time.Millisecond)
+				continue
+			}
+			last := sends[len(sends)-1]
+			rid, _ := last["rid"].(string)
+			once.Do(func() {
+				firstID, _ = last["id"].(string)
+				firstSendAt = last["send_at_ms"]
+				fake.ReplyErr(rid, CodeRateLimited, "slow")
+			})
+			if len(sends) >= 2 {
+				second := sends[1]
+				if second["id"] != firstID {
+					t.Errorf("id changed %v -> %v", firstID, second["id"])
+				}
+				if second["send_at_ms"] != firstSendAt {
+					t.Errorf("send_at_ms changed %v -> %v", firstSendAt, second["send_at_ms"])
+				}
+				fake.ReplyOK(second["rid"].(string), map[string]any{
+					"id": firstID, "send_at_ms": firstSendAt, "state": "scheduled",
+				})
+				return
+			}
+			time.Sleep(5 * time.Millisecond)
+		}
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	res, err := c.Send(ctx, Target{Kind: "endpoint", ID: "b"}, Body{Enc: "utf8", Data: "hi"}, SendOptions{SendAt: &at})
+	close(done)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.ID == "" || res.ID != firstID {
+		t.Fatalf("result id=%q first=%q", res.ID, firstID)
+	}
+}
+
+func TestRegisterHTTP(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/api/client/register" {
+			t.Fatalf("path %s", r.URL.Path)
+		}
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"ok": true, "data": map[string]any{"id": "e_1", "login_password": "genpass"},
+		})
+	}))
+	defer srv.Close()
+
+	// 从 ws 地址推出
+	ws := "ws" + strings.TrimPrefix(srv.URL, "http") + "/mqtt"
+	res, err := Register(context.Background(), ws, "code", RegisterOptions{Name: "n"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.ID != "e_1" || res.LoginPassword != "genpass" {
+		t.Fatalf("%+v", res)
+	}
+}
+
+func TestBuildCleanConnectFlags(t *testing.T) {
+	clean, exp := buildCleanConnectFlags()
+	if !clean || exp != 0 {
+		t.Fatalf("clean=%v exp=%d", clean, exp)
+	}
+}
+
+func TestBackoffNominal(t *testing.T) {
+	if d := computeBackoffDelay(time.Second, 1); d != time.Second {
+		t.Fatal(d)
+	}
+	if d := computeBackoffDelay(time.Second, 6); d != 30*time.Second {
+		t.Fatal(d)
+	}
+}

sdk/go/connect.go

@@ -0,0 +1,265 @@
+package nixmsg
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+)
+
+// Connect 连接服务器。credential 为密码或会话令牌。
+func (c *Client) Connect(ctx context.Context, rawURL, endpointID string, credential Credential, opts Options) error {
+	c.mu.Lock()
+	if c.closed {
+		c.mu.Unlock()
+		return apiErr(CodeClosed, "已关闭")
+	}
+	if c.transport != nil {
+		c.mu.Unlock()
+		return apiErr(CodeBadRequest, "已在连接中")
+	}
+	o := opts.withDefaults()
+	c.opts = o
+	c.endpointID = endpointID
+	c.url = rawURL
+	c.stopReconnect = false
+	c.handshook = false
+	pass := credential.Password
+	if credential.SessionToken != "" {
+		pass = credential.SessionToken
+		c.credKind = "token"
+	} else {
+		c.credKind = "password"
+	}
+	var tr transport
+	if o.transport != nil {
+		tr = o.transport
+	} else {
+		tr = newMQTTTransport()
+	}
+	c.transport = tr
+	c.backoff = newReconnectBackoff()
+	inner, cancel := context.WithCancel(context.Background())
+	c.ctx = inner
+	c.cancel = cancel
+	c.setStateLocked(StateConnecting, "")
+	c.mu.Unlock()
+
+	tr.SetCredential(pass)
+	cfg := transportConfig{
+		URL:            rawURL,
+		EndpointID:     endpointID,
+		ConnectTimeout: o.ConnectTimeout,
+		AllowTCP:       o.AllowTCP,
+		Backoff:        c.backoff,
+		OnDown:         c.handleDown,
+		OnOffline: func() {
+			c.mu.Lock()
+			c.handshook = false
+			if !c.stopReconnect && !c.closed {
+				c.setStateLocked(StateReconnecting, "")
+			}
+			c.mu.Unlock()
+		},
+		OnAuthFailed: func(reason AuthReason) { c.failAuth(reason) },
+		OnKicked:     func() { c.failKicked() },
+		MQTTReady:    func(readyCtx context.Context) error { return c.doHello(readyCtx) },
+	}
+	if err := tr.Start(inner, cfg); err != nil {
+		return err
+	}
+
+	deadline := time.Now().Add(o.ConnectTimeout)
+	for time.Now().Before(deadline) {
+		c.mu.Lock()
+		ok := c.handshook
+		failed := c.stopReconnect
+		st := c.state
+		c.mu.Unlock()
+		if ok {
+			return nil
+		}
+		if failed || st == StateAuthFailed || st == StateKicked {
+			return apiErr(CodeAuthFailed, string(st))
+		}
+		select {
+		case <-ctx.Done():
+			_ = c.Close()
+			return ctx.Err()
+		case <-time.After(20 * time.Millisecond):
+		}
+	}
+	_ = c.Close()
+	return apiErr(CodeNotConnected, "连接超时")
+}
+
+func (c *Client) failAuth(reason AuthReason) {
+	c.mu.Lock()
+	c.stopReconnect = true
+	c.handshook = false
+	c.setStateLocked(StateAuthFailed, string(reason))
+	c.failQueuedLocked(apiErr(string(reason), "认证失败，停止重连"))
+	cancel := c.cancel
+	c.mu.Unlock()
+	if cancel != nil {
+		cancel()
+	}
+}
+
+func (c *Client) failKicked() {
+	c.mu.Lock()
+	c.stopReconnect = true
+	c.handshook = false
+	c.setStateLocked(StateKicked, "0x8E")
+	c.failQueuedLocked(apiErr(CodeKicked, "被顶号，停止重连"))
+	cancel := c.cancel
+	c.mu.Unlock()
+	if cancel != nil {
+		cancel()
+	}
+}
+
+func (c *Client) setStateLocked(st ConnectionState, reason string) {
+	c.state = st
+	h := c.onConnection
+	ev := ConnectionEvent{State: st, Reason: reason}
+	go func() {
+		c.cbMu.Lock()
+		defer c.cbMu.Unlock()
+		if h != nil {
+			h(ev)
+		}
+	}()
+}
+
+func (c *Client) doHello(ctx context.Context) error {
+	c.mu.Lock()
+	c.helloSentAt = time.Now()
+	sentAt := c.helloSentAt
+	label := c.opts.ClientLabel
+	maxRecv := c.opts.MaxReceiveBytes
+	c.mu.Unlock()
+
+	req := map[string]any{
+		"v":      1,
+		"type":   "hello",
+		"rid":    c.nextRID(),
+		"client": label,
+	}
+	if maxRecv > 0 {
+		req["max_receive_bytes"] = maxRecv
+	}
+	data, err := c.request(ctx, req, true)
+	if err != nil {
+		return err
+	}
+	recvAt := time.Now()
+	var hd struct {
+		ServerTimeMs       int64  `json:"server_time_ms"`
+		ServerVersion      string `json:"server_version"`
+		MaxBodyBytes       int    `json:"max_body_bytes"`
+		MaxMetaBytes       int    `json:"max_meta_bytes"`
+		MaxFrameBytes      int    `json:"max_frame_bytes"`
+		MaxTTLSeconds      int64  `json:"max_ttl_seconds"`
+		MaxScheduleSeconds int64  `json:"max_schedule_seconds"`
+		AckTimeoutSeconds  int64  `json:"ack_timeout_seconds"`
+		SessionToken       string `json:"session_token"`
+	}
+	if err := json.Unmarshal(data, &hd); err != nil {
+		return err
+	}
+	mid := (sentAt.UnixMilli() + recvAt.UnixMilli()) / 2
+	skew := hd.ServerTimeMs - mid
+
+	c.mu.Lock()
+	c.limits = HandshakeLimits{
+		ServerTimeMs:       hd.ServerTimeMs,
+		ServerVersion:      hd.ServerVersion,
+		MaxBodyBytes:       hd.MaxBodyBytes,
+		MaxMetaBytes:       hd.MaxMetaBytes,
+		MaxFrameBytes:      hd.MaxFrameBytes,
+		MaxTTLSeconds:      hd.MaxTTLSeconds,
+		MaxScheduleSeconds: hd.MaxScheduleSeconds,
+		AckTimeoutSeconds:  hd.AckTimeoutSeconds,
+	}
+	c.clockSkew = skew
+	c.handshook = true
+	c.setStateLocked(StateOnline, "")
+	token := hd.SessionToken
+	if token != "" {
+		c.session = token
+		c.transport.SetCredential(token)
+		c.credKind = "token"
+	}
+	c.mu.Unlock()
+
+	if token != "" && c.onSession != nil {
+		c.cbMu.Lock()
+		c.onSession(token)
+		c.cbMu.Unlock()
+	}
+	c.drainSendQueue()
+	return nil
+}
+
+// ClockSkewMs 当前服务器时间偏差（毫秒）。
+func (c *Client) ClockSkewMs() int64 {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.clockSkew
+}
+
+// Limits 握手上限。
+func (c *Client) Limits() HandshakeLimits {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.limits
+}
+
+// Logout 作废会话并停止重连。
+func (c *Client) Logout(ctx context.Context) error {
+	req := map[string]any{"v": 1, "type": "self.logout", "rid": c.nextRID()}
+	_, err := c.request(ctx, req, false)
+	c.mu.Lock()
+	c.stopReconnect = true
+	c.session = ""
+	cancel := c.cancel
+	c.mu.Unlock()
+	if cancel != nil {
+		cancel()
+	}
+	return err
+}
+
+// Close 关闭连接并停止重连。
+func (c *Client) Close() error {
+	c.mu.Lock()
+	c.closed = true
+	c.stopReconnect = true
+	c.failQueuedLocked(apiErr(CodeClosed, "已关闭"))
+	tr := c.transport
+	cancel := c.cancel
+	c.setStateLocked(StateOffline, "")
+	c.mu.Unlock()
+	if cancel != nil {
+		cancel()
+	}
+	if tr != nil {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		return tr.Stop(ctx)
+	}
+	return nil
+}
+
+func (c *Client) failQueuedLocked(err error) {
+	for _, it := range c.sendQ {
+		if it.result != nil {
+			select {
+			case it.result <- sendOutcome{err: err}:
+			default:
+			}
+		}
+	}
+	c.sendQ = nil
+	c.inflight = 0
+}

sdk/go/doc.go

@@ -0,0 +1,7 @@
+// Package nixmsg 是 NixMsg 的 Go SDK。
+//
+// 模块路径：git.asio.asia/nixevol/NixMsg/sdk/go
+// 行为以 docs/DEVELOPMENT.md 第 9 节与第 6 节为准。
+//
+// 许可证：SEE LICENSE（本目录 LICENSE 为仓库根许可证副本）。
+package nixmsg

sdk/go/errors.go

@@ -0,0 +1,46 @@
+package nixmsg
+
+import "fmt"
+
+// APIError 是带协议错误码的错误。
+type APIError struct {
+	Code    string
+	Message string
+}
+
+func (e *APIError) Error() string {
+	if e.Message == "" {
+		return e.Code
+	}
+	return fmt.Sprintf("%s: %s", e.Code, e.Message)
+}
+
+func apiErr(code, msg string) *APIError {
+	return &APIError{Code: code, Message: msg}
+}
+
+// 常用错误码（与 DEVELOPMENT 第 6.10 节一致）。
+const (
+	CodeBodyTooLarge     = "body_too_large"
+	CodeFrameTooLarge    = "frame_too_large"
+	CodeMetaTooLarge     = "meta_too_large"
+	CodeRateLimited      = "rate_limited"
+	CodeBadRequest       = "bad_request"
+	CodeNotReady         = "not_ready"
+	CodeBusy             = "busy"
+	CodeSessionInvalid   = "session_invalid"
+	CodeBadCredentials   = "bad_credentials"
+	CodeQueueFull        = "queue_full"
+	CodeNotConnected     = "not_connected"
+	CodeClosed           = "closed"
+	CodeAuthFailed       = "auth_failed"
+	CodeKicked           = "kicked"
+)
+
+// AuthReason 是认证失败原因。
+type AuthReason string
+
+const (
+	AuthSessionInvalid AuthReason = "session_invalid"
+	AuthBadCredentials AuthReason = "bad_credentials"
+)

sdk/go/go.mod

@@ -0,0 +1,13 @@
+module git.asio.asia/nixevol/NixMsg/sdk/go
+
+go 1.27.1
+
+require (
+	github.com/eclipse/paho.golang v0.23.0
+	github.com/google/uuid v1.6.0
+)
+
+require (
+	github.com/gorilla/websocket v1.5.3 // indirect
+	golang.org/x/net v0.43.0 // indirect
+)

sdk/go/go.sum

@@ -0,0 +1,20 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/eclipse/paho.golang v0.23.0 h1:KHgl2wz6EJo7cMBmkuhpt7C576vP+kpPv7jjvSyR6Mk=
+github.com/eclipse/paho.golang v0.23.0/go.mod h1:nQRhTkoZv8EAiNs5UU0/WdQIx2NrnWUpL9nsGJTQN04=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE=
+golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

sdk/go/json.go

@@ -0,0 +1,36 @@
+package nixmsg
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+)
+
+func marshalJSON(v any) ([]byte, error) {
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	enc.SetEscapeHTML(false)
+	if err := enc.Encode(v); err != nil {
+		return nil, err
+	}
+	b := buf.Bytes()
+	if n := len(b); n > 0 && b[n-1] == '\n' {
+		b = b[:n-1]
+	}
+	return b, nil
+}
+
+func unmarshalJSON(data []byte, v any) error {
+	dec := json.NewDecoder(bytes.NewReader(data))
+	dec.UseNumber()
+	return dec.Decode(v)
+}
+
+func writeJSON(w io.Writer, v any) error {
+	b, err := marshalJSON(v)
+	if err != nil {
+		return err
+	}
+	_, err = w.Write(b)
+	return err
+}

sdk/go/receive.go

@@ -0,0 +1,326 @@
+package nixmsg
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"time"
+)
+
+func (c *Client) handleDown(payload []byte) {
+	var head struct {
+		Type string `json:"type"`
+		RID  string `json:"rid"`
+	}
+	if err := unmarshalJSON(payload, &head); err != nil {
+		return
+	}
+	switch head.Type {
+	case "resp":
+		var rf respFrame
+		if err := unmarshalJSON(payload, &rf); err != nil {
+			return
+		}
+		c.mu.Lock()
+		p := c.pending[head.RID]
+		if p != nil {
+			delete(c.pending, head.RID)
+		}
+		c.mu.Unlock()
+		if p != nil {
+			select {
+			case p.ch <- rf:
+			default:
+			}
+		}
+	case "msg":
+		c.handleMsg(payload)
+	case "receipt":
+		c.handleReceipt(payload)
+	case "revoked":
+		c.handleRevoked(payload)
+	case "presence":
+		c.handlePresence(payload)
+	case "group_event":
+		c.handleGroupEvent(payload)
+	case "fatal":
+		var f struct {
+			Reason string `json:"reason"`
+		}
+		_ = unmarshalJSON(payload, &f)
+		c.mu.Lock()
+		c.stopReconnect = true
+		c.setStateLocked(StateAuthFailed, f.Reason)
+		c.failQueuedLocked(apiErr("fatal", f.Reason))
+		cancel := c.cancel
+		c.mu.Unlock()
+		if cancel != nil {
+			cancel()
+		}
+	}
+}
+
+func (c *Client) handleMsg(payload []byte) {
+	var m struct {
+		ID       string         `json:"id"`
+		From     string         `json:"from"`
+		To       Target         `json:"to"`
+		Body     Body           `json:"body"`
+		Meta     map[string]any `json:"meta"`
+		SendAtMs int64          `json:"send_at_ms"`
+	}
+	if err := unmarshalJSON(payload, &m); err != nil {
+		return
+	}
+	key := m.From + "\x00" + m.ID
+	c.mu.Lock()
+	ent := c.dedup[key]
+	manual := c.opts.ManualAck
+	if ent != nil && ent.state == dedupAcked {
+		c.mu.Unlock()
+		_ = c.sendAckFrame(m.From, m.ID)
+		return
+	}
+	if ent != nil && ent.state == dedupDelivered {
+		c.mu.Unlock()
+		return
+	}
+	c.rememberDedupLocked(key, m.From, m.ID, dedupDelivered)
+	c.mu.Unlock()
+
+	msg := Message{ID: m.ID, From: m.From, To: m.To, Body: m.Body, Meta: m.Meta, SendAtMs: m.SendAtMs}
+	var cbErr error
+	c.cbMu.Lock()
+	if c.onMessage != nil {
+		cbErr = c.onMessage(msg)
+	}
+	c.cbMu.Unlock()
+
+	if manual {
+		return
+	}
+	if cbErr != nil {
+		c.mu.Lock()
+		delete(c.dedup, key)
+		c.mu.Unlock()
+		return
+	}
+	_ = c.sendAckFrame(m.From, m.ID)
+	c.mu.Lock()
+	if e := c.dedup[key]; e != nil {
+		e.state = dedupAcked
+	}
+	c.mu.Unlock()
+}
+
+func (c *Client) rememberDedupLocked(key, from, id string, st dedupState) {
+	if _, ok := c.dedup[key]; !ok {
+		c.dedupOrd = append(c.dedupOrd, key)
+		for len(c.dedupOrd) > c.opts.DedupCapacity {
+			old := c.dedupOrd[0]
+			c.dedupOrd = c.dedupOrd[1:]
+			delete(c.dedup, old)
+		}
+	}
+	c.dedup[key] = &dedupEntry{state: st, from: from, id: id}
+}
+
+// Ack 手动确认。
+func (c *Client) Ack(msg Message) error {
+	if err := c.sendAckFrame(msg.From, msg.ID); err != nil {
+		return err
+	}
+	key := msg.From + "\x00" + msg.ID
+	c.mu.Lock()
+	c.rememberDedupLocked(key, msg.From, msg.ID, dedupAcked)
+	c.mu.Unlock()
+	return nil
+}
+
+func (c *Client) sendAckFrame(from, id string) error {
+	req := map[string]any{
+		"v": 1, "type": "ack", "rid": c.nextRID(), "from": from, "id": id,
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	data, err := c.request(ctx, req, true)
+	if err != nil {
+		return err
+	}
+	if len(data) > 0 {
+		var d struct {
+			Result string `json:"result"`
+		}
+		if json.Unmarshal(data, &d) == nil && d.Result != "" && d.Result != "accepted" {
+			c.emitRevoked(RevokedEvent{ID: id, From: from, Reason: d.Result})
+		}
+	}
+	return nil
+}
+
+func (c *Client) handleReceipt(payload []byte) {
+	var r struct {
+		ReceiptID  string `json:"receipt_id"`
+		ID         string `json:"id"`
+		EndpointID string `json:"endpoint_id"`
+		State      string `json:"state"`
+		Reason     string `json:"reason"`
+		AtMs       int64  `json:"at_ms"`
+	}
+	if err := unmarshalJSON(payload, &r); err != nil {
+		return
+	}
+	c.mu.Lock()
+	if _, ok := c.receiptSeen[r.ReceiptID]; ok {
+		c.mu.Unlock()
+		_ = c.sendReceiptAck(r.ReceiptID)
+		return
+	}
+	c.receiptSeen[r.ReceiptID] = struct{}{}
+	c.mu.Unlock()
+	ev := Receipt{ReceiptID: r.ReceiptID, ID: r.ID, EndpointID: r.EndpointID, State: r.State, Reason: r.Reason, AtMs: r.AtMs}
+	c.cbMu.Lock()
+	if c.onReceipt != nil {
+		c.onReceipt(ev)
+	}
+	c.cbMu.Unlock()
+	_ = c.sendReceiptAck(r.ReceiptID)
+}
+
+func (c *Client) sendReceiptAck(receiptID string) error {
+	req := map[string]any{"v": 1, "type": "receipt_ack", "rid": c.nextRID(), "receipt_id": receiptID}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	_, err := c.request(ctx, req, true)
+	return err
+}
+
+func (c *Client) handleRevoked(payload []byte) {
+	var r struct {
+		ID     string `json:"id"`
+		From   string `json:"from"`
+		Reason string `json:"reason"`
+	}
+	if err := unmarshalJSON(payload, &r); err != nil {
+		return
+	}
+	key := r.From + "\x00" + r.ID
+	c.mu.Lock()
+	ent := c.dedup[key]
+	if ent == nil || ent.state == dedupAcked {
+		c.mu.Unlock()
+		return
+	}
+	delete(c.dedup, key)
+	c.mu.Unlock()
+	c.emitRevoked(RevokedEvent{ID: r.ID, From: r.From, Reason: r.Reason})
+}
+
+func (c *Client) emitRevoked(e RevokedEvent) {
+	c.cbMu.Lock()
+	defer c.cbMu.Unlock()
+	if c.onRevoked != nil {
+		c.onRevoked(e)
+	}
+}
+
+func (c *Client) handlePresence(payload []byte) {
+	var p struct {
+		ID     string `json:"id"`
+		Online bool   `json:"online"`
+		AtMs   int64  `json:"at_ms"`
+	}
+	if err := unmarshalJSON(payload, &p); err != nil {
+		return
+	}
+	c.cbMu.Lock()
+	defer c.cbMu.Unlock()
+	if c.onPresence != nil {
+		c.onPresence(PresenceEvent{ID: p.ID, Online: p.Online, AtMs: p.AtMs})
+	}
+}
+
+func (c *Client) handleGroupEvent(payload []byte) {
+	var g struct {
+		GroupID    string `json:"group_id"`
+		Event      string `json:"event"`
+		EndpointID string `json:"endpoint_id"`
+		AtMs       int64  `json:"at_ms"`
+	}
+	if err := unmarshalJSON(payload, &g); err != nil {
+		return
+	}
+	c.cbMu.Lock()
+	defer c.cbMu.Unlock()
+	if c.onGroupEvent != nil {
+		c.onGroupEvent(GroupEvent{GroupID: g.GroupID, Event: g.Event, EndpointID: g.EndpointID, AtMs: g.AtMs})
+	}
+}
+
+func (c *Client) request(ctx context.Context, frame map[string]any, allowUnready bool) (json.RawMessage, error) {
+	c.mu.Lock()
+	if c.closed {
+		c.mu.Unlock()
+		return nil, apiErr(CodeClosed, "已关闭")
+	}
+	if !allowUnready && !c.handshook {
+		c.mu.Unlock()
+		return nil, apiErr(CodeNotConnected, "未握手")
+	}
+	tr := c.transport
+	c.mu.Unlock()
+	if tr == nil {
+		return nil, apiErr(CodeNotConnected, "未连接")
+	}
+	rid, _ := frame["rid"].(string)
+	if rid == "" {
+		rid = c.nextRID()
+		frame["rid"] = rid
+	}
+	payload, err := marshalJSON(frame)
+	if err != nil {
+		return nil, err
+	}
+	ch := make(chan respFrame, 1)
+	c.mu.Lock()
+	c.pending[rid] = &pendingReq{rid: rid, ch: ch}
+	c.mu.Unlock()
+	if err := tr.PublishUp(payload); err != nil {
+		c.mu.Lock()
+		delete(c.pending, rid)
+		c.mu.Unlock()
+		return nil, err
+	}
+	select {
+	case <-ctx.Done():
+		c.mu.Lock()
+		delete(c.pending, rid)
+		c.mu.Unlock()
+		return nil, ctx.Err()
+	case rf := <-ch:
+		if !rf.OK {
+			code, msg := CodeBadRequest, "请求失败"
+			if rf.Error != nil {
+				code, msg = rf.Error.Code, rf.Error.Message
+			}
+			return nil, apiErr(code, msg)
+		}
+		return rf.Data, nil
+	}
+}
+
+// bodyDecodedLen 按解码后字节计正文大小。
+func bodyDecodedLen(b Body) (int, error) {
+	switch b.Enc {
+	case "utf8", "":
+		return len([]byte(b.Data)), nil
+	case "base64":
+		raw, err := base64.StdEncoding.DecodeString(b.Data)
+		if err != nil {
+			return 0, apiErr(CodeBadRequest, "base64 正文无效")
+		}
+		return len(raw), nil
+	default:
+		return 0, apiErr(CodeBadRequest, "未知 enc")
+	}
+}

sdk/go/register.go

@@ -0,0 +1,65 @@
+package nixmsg
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+)
+
+// Register 静态方法：HTTP 自助注册。url 可为连接地址或注册地址。
+func Register(ctx context.Context, connectOrRegisterURL, registrationCode string, opt RegisterOptions) (RegisterResult, error) {
+	regURL := connectOrRegisterURL
+	if u, err := RegisterURLFromConnect(connectOrRegisterURL); err == nil {
+		regURL = u
+	}
+	body := map[string]any{
+		"registration_code": registrationCode,
+		"id":                opt.ID,
+		"login_password":    opt.LoginPassword,
+		"name":              opt.Name,
+		"talk_password":     opt.TalkPassword,
+	}
+	payload, err := marshalJSON(body)
+	if err != nil {
+		return RegisterResult{}, err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, regURL, bytes.NewReader(payload))
+	if err != nil {
+		return RegisterResult{}, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return RegisterResult{}, err
+	}
+	defer resp.Body.Close()
+	raw, err := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
+	if err != nil {
+		return RegisterResult{}, err
+	}
+	var wrap struct {
+		OK    bool `json:"ok"`
+		Data  struct {
+			ID            string `json:"id"`
+			LoginPassword string `json:"login_password"`
+		} `json:"data"`
+		Error *struct {
+			Code    string `json:"code"`
+			Message string `json:"message"`
+		} `json:"error"`
+	}
+	if err := json.Unmarshal(raw, &wrap); err != nil {
+		return RegisterResult{}, fmt.Errorf("注册响应无效: %w", err)
+	}
+	if !wrap.OK {
+		code, msg := CodeBadRequest, "注册失败"
+		if wrap.Error != nil {
+			code, msg = wrap.Error.Code, wrap.Error.Message
+		}
+		return RegisterResult{}, apiErr(code, msg)
+	}
+	return RegisterResult{ID: wrap.Data.ID, LoginPassword: wrap.Data.LoginPassword}, nil
+}

sdk/go/send.go

@@ -0,0 +1,238 @@
+package nixmsg
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+)
+
+// Send 发送消息；可在未连接时入队，重连后按原内容重交。
+func (c *Client) Send(ctx context.Context, to Target, body Body, opt SendOptions) (SendResult, error) {
+	c.mu.Lock()
+	limits := c.limits
+	skew := c.clockSkew
+	handshook := c.handshook
+	maxQ := c.opts.SendQueueSize
+	qLen := len(c.sendQ)
+	c.mu.Unlock()
+
+	if body.Enc == "" {
+		body.Enc = "utf8"
+	}
+	if body.ContentType == "" && opt.ContentType != "" {
+		body.ContentType = opt.ContentType
+	} else if body.ContentType == "" {
+		if body.Enc == "base64" {
+			body.ContentType = "application/octet-stream"
+		} else {
+			body.ContentType = "text/plain; charset=utf-8"
+		}
+	} else if opt.ContentType != "" {
+		body.ContentType = opt.ContentType
+	}
+
+	n, err := bodyDecodedLen(body)
+	if err != nil {
+		return SendResult{}, err
+	}
+	maxBody := limits.MaxBodyBytes
+	if maxBody <= 0 {
+		maxBody = 262144
+	}
+	if handshook && n > maxBody {
+		return SendResult{}, apiErr(CodeBodyTooLarge, "正文超限")
+	}
+	if !handshook && n > 262144 {
+		return SendResult{}, apiErr(CodeBodyTooLarge, "正文超限")
+	}
+
+	id := opt.ID
+	if id == "" {
+		id = newMessageID()
+	}
+
+	frame := map[string]any{
+		"v":    1,
+		"type": "send",
+		"rid":  c.nextRID(),
+		"id":   id,
+		"to":   to,
+		"body": body,
+	}
+	if opt.Meta != nil {
+		frame["meta"] = opt.Meta
+	}
+	if opt.TalkPassword != "" {
+		frame["talk_password"] = opt.TalkPassword
+	}
+	if opt.Receipt != nil {
+		frame["receipt"] = *opt.Receipt
+	}
+	if opt.Keep {
+		off := map[string]any{"keep": true}
+		if opt.TTL != nil {
+			off["ttl_seconds"] = *opt.TTL
+		}
+		frame["offline"] = off
+	}
+
+	var sendAtMs *int64
+	if opt.SendAt != nil && opt.Delay != nil {
+		return SendResult{}, apiErr(CodeBadRequest, "sendAt 与 delay 互斥")
+	}
+	if opt.SendAt != nil {
+		// sendAt 使用本机时间 + 服务器偏差，换算后写入，重交不重算
+		ms := opt.SendAt.UnixMilli() + skew
+		sendAtMs = &ms
+		frame["send_at_ms"] = ms
+	} else if opt.Delay != nil {
+		frame["delay_ms"] = opt.Delay.Milliseconds()
+	}
+
+	payload, err := marshalJSON(frame)
+	if err != nil {
+		return SendResult{}, err
+	}
+	maxFrame := limits.MaxFrameBytes
+	if maxFrame <= 0 {
+		maxFrame = 786432
+	}
+	if handshook && len(payload) > maxFrame {
+		return SendResult{}, apiErr(CodeFrameTooLarge, "整帧超限")
+	}
+
+	item := &sendItem{
+		frame:    frame,
+		payload:  payload,
+		id:       id,
+		sendAtMs: sendAtMs,
+		result:   make(chan sendOutcome, 1),
+	}
+
+	c.mu.Lock()
+	if c.closed || c.stopReconnect {
+		c.mu.Unlock()
+		return SendResult{}, apiErr(CodeClosed, "已关闭")
+	}
+	if qLen >= maxQ {
+		c.mu.Unlock()
+		return SendResult{}, apiErr(CodeQueueFull, "发送队列已满")
+	}
+	c.sendQ = append(c.sendQ, item)
+	c.mu.Unlock()
+
+	c.drainSendQueue()
+
+	select {
+	case <-ctx.Done():
+		return SendResult{}, ctx.Err()
+	case out := <-item.result:
+		return out.res, out.err
+	}
+}
+
+func (c *Client) drainSendQueue() {
+	for {
+		c.mu.Lock()
+		if !c.handshook || c.transport == nil {
+			c.mu.Unlock()
+			return
+		}
+		var next *sendItem
+		for _, it := range c.sendQ {
+			if !it.inflight {
+				next = it
+				break
+			}
+		}
+		if next == nil || c.inflight >= c.opts.MaxInflight {
+			c.mu.Unlock()
+			return
+		}
+		next.inflight = true
+		c.inflight++
+		tr := c.transport
+		payload := next.payload
+		rid, _ := next.frame["rid"].(string)
+		item := next
+		c.mu.Unlock()
+
+		go c.dispatchSend(tr, item, rid, payload)
+	}
+}
+
+func (c *Client) dispatchSend(tr transport, item *sendItem, rid string, payload []byte) {
+	ch := make(chan respFrame, 1)
+	c.mu.Lock()
+	c.pending[rid] = &pendingReq{rid: rid, ch: ch}
+	c.mu.Unlock()
+
+	if err := tr.PublishUp(payload); err != nil {
+		c.mu.Lock()
+		delete(c.pending, rid)
+		item.inflight = false
+		c.inflight--
+		c.mu.Unlock()
+		// 网络错误：保留队列等重连
+		return
+	}
+
+	rf := <-ch
+	if !rf.OK {
+		code, msg := CodeBadRequest, "发送失败"
+		if rf.Error != nil {
+			code, msg = rf.Error.Code, rf.Error.Message
+		}
+		if code == CodeRateLimited {
+			// 自动重交：保持同一 payload（含 id / send_at_ms）
+			c.mu.Lock()
+			item.inflight = false
+			c.inflight--
+			delete(c.pending, rid)
+			c.mu.Unlock()
+			time.AfterFunc(time.Second, func() { c.drainSendQueue() })
+			return
+		}
+		c.finishSend(item, SendResult{}, apiErr(code, msg))
+		return
+	}
+	var sd SendResult
+	_ = json.Unmarshal(rf.Data, &sd)
+	if sd.ID == "" {
+		sd.ID = item.id
+	}
+	c.finishSend(item, sd, nil)
+}
+
+func (c *Client) finishSend(item *sendItem, res SendResult, err error) {
+	c.mu.Lock()
+	// 从队列移除
+	out := item.result
+	nq := c.sendQ[:0]
+	for _, it := range c.sendQ {
+		if it != item {
+			nq = append(nq, it)
+		}
+	}
+	c.sendQ = nq
+	if item.inflight {
+		c.inflight--
+		item.inflight = false
+	}
+	c.mu.Unlock()
+	select {
+	case out <- sendOutcome{res: res, err: err}:
+	default:
+	}
+	c.drainSendQueue()
+}
+
+// ResendPayloadForTest 返回队列中第一条发送帧的编码（测试用）。
+func (c *Client) ResendPayloadForTest() []byte {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if len(c.sendQ) == 0 {
+		return nil
+	}
+	return append([]byte(nil), c.sendQ[0].payload...)
+}

sdk/go/transport.go

@@ -0,0 +1,40 @@
+package nixmsg
+
+import (
+	"context"
+	"time"
+)
+
+// transport 抽象 MQTT 应用层通道，便于单测注入假实现。
+type transport interface {
+	// Start 开始连接循环（含重连）。凭据在每次 CONNECT 时读取。
+	Start(ctx context.Context, cfg transportConfig) error
+	// PublishUp 向 up 主题发布一帧。
+	PublishUp(payload []byte) error
+	// Stop 停止并断开，不再重连。
+	Stop(ctx context.Context) error
+	// Credential 当前用于 CONNECT 的密码字段（密码或令牌）。
+	SetCredential(passwordOrToken string)
+}
+
+type transportConfig struct {
+	URL            string // ws(s):// 或 mqtt(s)://
+	EndpointID     string
+	ConnectTimeout time.Duration
+	AllowTCP       bool
+	Backoff        *reconnectBackoff
+	OnDown         func(payload []byte)
+	OnOnline       func() // MQTT 已连上且可发（订阅完成前也会调；hello 由上层做）
+	OnOffline      func()
+	OnAuthFailed   func(reason AuthReason)
+	OnKicked       func()
+	// OnConnectPacket 每次 CONNECT 构建后回调（单测核对 Clean Start）。
+	OnConnectPacket func(cleanStart bool, sessionExpiry uint32)
+	// MQTTConnected 在 CONNACK 成功、订阅 down 完成后调用，上层在此发 hello。
+	MQTTReady func(ctx context.Context) error
+}
+
+// buildCleanConnect 供真实与假传输共用：每次连接 Clean Start=true，会话过期=0。
+func buildCleanConnectFlags() (cleanStart bool, sessionExpiry uint32) {
+	return true, 0
+}

sdk/go/transport_fake.go

@@ -0,0 +1,266 @@
+package nixmsg
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+	"sync/atomic"
+)
+
+// FakeTransport 单测用假 MQTT：不启真实网络。
+type FakeTransport struct {
+	mu       sync.Mutex
+	cfg      transportConfig
+	cred     atomic.Value
+	up       [][]byte
+	stopped  atomic.Bool
+	online   bool
+	connects []fakeConnect
+
+	// AutoHello 为 true 时自动回复 hello（默认 true）。
+	AutoHello bool
+	// HelloToken 自动 hello 响应里的 session_token。
+	HelloToken string
+	// HelloServerTimeMs 自动 hello 的 server_time_ms。
+	HelloServerTimeMs int64
+	MaxBodyBytes      int
+	MaxMetaBytes      int
+	MaxFrameBytes     int
+}
+
+type fakeConnect struct {
+	CleanStart    bool
+	SessionExpiry uint32
+	Password      string
+}
+
+// NewFakeTransport 创建假传输；默认自动回复 hello。
+func NewFakeTransport() *FakeTransport {
+	f := &FakeTransport{
+		AutoHello:         true,
+		HelloToken:        "nst_test_token",
+		HelloServerTimeMs: 1_750_000_000_000,
+		MaxBodyBytes:      262144,
+		MaxMetaBytes:      4096,
+		MaxFrameBytes:     786432,
+	}
+	f.cred.Store("")
+	return f
+}
+
+func (f *FakeTransport) SetCredential(passwordOrToken string) {
+	f.cred.Store(passwordOrToken)
+}
+
+func (f *FakeTransport) Start(_ context.Context, cfg transportConfig) error {
+	f.mu.Lock()
+	f.cfg = cfg
+	f.mu.Unlock()
+	return f.SimulateConnectOK()
+}
+
+func (f *FakeTransport) PublishUp(payload []byte) error {
+	f.mu.Lock()
+	cp := append([]byte(nil), payload...)
+	f.up = append(f.up, cp)
+	auto := f.AutoHello
+	f.mu.Unlock()
+
+	var head struct {
+		Type string `json:"type"`
+		RID  string `json:"rid"`
+	}
+	_ = unmarshalJSON(payload, &head)
+	if auto && head.Type == "hello" && head.RID != "" {
+		f.replyHello(head.RID)
+	}
+	return nil
+}
+
+func (f *FakeTransport) replyHello(rid string) {
+	f.mu.Lock()
+	token := f.HelloToken
+	st := f.HelloServerTimeMs
+	mb, mm, mf := f.MaxBodyBytes, f.MaxMetaBytes, f.MaxFrameBytes
+	f.mu.Unlock()
+	resp, _ := marshalJSON(map[string]any{
+		"v": 1, "type": "resp", "rid": rid, "ok": true,
+		"data": map[string]any{
+			"server_time_ms":       st,
+			"server_version":       "0.1.0",
+			"max_body_bytes":       mb,
+			"max_meta_bytes":       mm,
+			"max_frame_bytes":      mf,
+			"max_ttl_seconds":      2592000,
+			"max_schedule_seconds": 31536000,
+			"ack_timeout_seconds":  300,
+			"session_token":        token,
+		},
+	})
+	f.InjectDown(resp)
+}
+
+func (f *FakeTransport) Stop(_ context.Context) error {
+	f.stopped.Store(true)
+	f.mu.Lock()
+	cfg := f.cfg
+	f.online = false
+	f.mu.Unlock()
+	if cfg.OnOffline != nil {
+		cfg.OnOffline()
+	}
+	return nil
+}
+
+// SimulateConnectOK 模拟一次成功 CONNECT（含 Clean Start 记录）并完成订阅就绪。
+func (f *FakeTransport) SimulateConnectOK() error {
+	clean, expiry := buildCleanConnectFlags()
+	pass, _ := f.cred.Load().(string)
+	f.mu.Lock()
+	f.connects = append(f.connects, fakeConnect{CleanStart: clean, SessionExpiry: expiry, Password: pass})
+	cfg := f.cfg
+	f.online = true
+	f.mu.Unlock()
+	if cfg.OnConnectPacket != nil {
+		cfg.OnConnectPacket(clean, expiry)
+	}
+	if cfg.Backoff != nil {
+		cfg.Backoff.MarkOnline()
+	}
+	if cfg.MQTTReady != nil {
+		if err := cfg.MQTTReady(context.Background()); err != nil {
+			return err
+		}
+	}
+	if cfg.OnOnline != nil {
+		cfg.OnOnline()
+	}
+	return nil
+}
+
+// SimulateReconnect 模拟断线再连（再次 Clean Start）。
+func (f *FakeTransport) SimulateReconnect() error {
+	f.mu.Lock()
+	cfg := f.cfg
+	f.online = false
+	f.mu.Unlock()
+	if cfg.Backoff != nil {
+		cfg.Backoff.MarkOffline()
+	}
+	if cfg.OnOffline != nil {
+		cfg.OnOffline()
+	}
+	return f.SimulateConnectOK()
+}
+
+// SimulateAuthFail 模拟 CONNACK 认证失败。
+func (f *FakeTransport) SimulateAuthFail(reason AuthReason) {
+	f.stopped.Store(true)
+	f.mu.Lock()
+	cfg := f.cfg
+	f.mu.Unlock()
+	if cfg.OnAuthFailed != nil {
+		cfg.OnAuthFailed(reason)
+	}
+}
+
+// SimulateKick 模拟 0x8E。
+func (f *FakeTransport) SimulateKick() {
+	f.stopped.Store(true)
+	f.mu.Lock()
+	cfg := f.cfg
+	f.mu.Unlock()
+	if cfg.OnKicked != nil {
+		cfg.OnKicked()
+	}
+}
+
+// InjectDown 注入下行帧。
+func (f *FakeTransport) InjectDown(payload []byte) {
+	f.mu.Lock()
+	cfg := f.cfg
+	f.mu.Unlock()
+	if cfg.OnDown != nil {
+		cfg.OnDown(payload)
+	}
+}
+
+// ReplyOK 对指定 rid 回成功 resp。
+func (f *FakeTransport) ReplyOK(rid string, data any) {
+	var raw json.RawMessage
+	if data != nil {
+		raw, _ = marshalJSON(data)
+	} else {
+		raw = json.RawMessage(`{}`)
+	}
+	resp, _ := marshalJSON(map[string]any{
+		"v": 1, "type": "resp", "rid": rid, "ok": true, "data": raw,
+	})
+	f.InjectDown(resp)
+}
+
+// ReplyErr 对指定 rid 回失败 resp。
+func (f *FakeTransport) ReplyErr(rid, code, message string) {
+	resp, _ := marshalJSON(map[string]any{
+		"v": 1, "type": "resp", "rid": rid, "ok": false,
+		"error": map[string]any{"code": code, "message": message},
+	})
+	f.InjectDown(resp)
+}
+
+// UpFrames 已发布的上行帧副本。
+func (f *FakeTransport) UpFrames() [][]byte {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	out := make([][]byte, len(f.up))
+	for i, p := range f.up {
+		out[i] = append([]byte(nil), p...)
+	}
+	return out
+}
+
+// Connects 历次 CONNECT 参数。
+func (f *FakeTransport) Connects() []fakeConnect {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	out := make([]fakeConnect, len(f.connects))
+	copy(out, f.connects)
+	return out
+}
+
+// DrainUp 取出并清空上行。
+func (f *FakeTransport) DrainUp() [][]byte {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	out := f.up
+	f.up = nil
+	return out
+}
+
+// LastUp 最后一条上行（解析为通用 map）。
+func (f *FakeTransport) LastUp() map[string]any {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if len(f.up) == 0 {
+		return nil
+	}
+	var m map[string]any
+	_ = unmarshalJSON(f.up[len(f.up)-1], &m)
+	return m
+}
+
+// FindUp 找指定 type 的上行帧。
+func (f *FakeTransport) FindUp(typ string) []map[string]any {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	var out []map[string]any
+	for _, p := range f.up {
+		var m map[string]any
+		if unmarshalJSON(p, &m) == nil {
+			if m["type"] == typ {
+				out = append(out, m)
+			}
+		}
+	}
+	return out
+}

sdk/go/transport_mqtt.go

@@ -0,0 +1,258 @@
+package nixmsg
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/eclipse/paho.golang/autopaho"
+	"github.com/eclipse/paho.golang/paho"
+)
+
+type mqttTransport struct {
+	mu       sync.Mutex
+	cm       *autopaho.ConnectionManager
+	cancel   context.CancelFunc
+	cfg      transportConfig
+	cred     atomic.Value // string
+	upTopic  string
+	downTopic string
+	stopped  atomic.Bool
+	ready    chan struct{}
+}
+
+func newMQTTTransport() *mqttTransport {
+	t := &mqttTransport{ready: make(chan struct{})}
+	t.cred.Store("")
+	return t
+}
+
+func (t *mqttTransport) SetCredential(passwordOrToken string) {
+	t.cred.Store(passwordOrToken)
+}
+
+func (t *mqttTransport) Start(ctx context.Context, cfg transportConfig) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.cm != nil {
+		return errors.New("transport already started")
+	}
+	t.cfg = cfg
+	t.upTopic = fmt.Sprintf("nix/c/%s/up", cfg.EndpointID)
+	t.downTopic = fmt.Sprintf("nix/c/%s/down", cfg.EndpointID)
+
+	u, err := normalizeMQTTURL(cfg.URL, cfg.AllowTCP)
+	if err != nil {
+		return err
+	}
+
+	innerCtx, cancel := context.WithCancel(ctx)
+	t.cancel = cancel
+
+	var sessionExpiry uint32 // 0；由 ConnectPacketBuilder 显式写入 Properties
+	cliCfg := autopaho.ClientConfig{
+		ServerUrls:                    []*url.URL{u},
+		KeepAlive:                     30,
+		ConnectTimeout:                cfg.ConnectTimeout,
+		CleanStartOnInitialConnection: false, // 不要只靠这个；每次用 ConnectPacketBuilder
+		SessionExpiryInterval:         sessionExpiry,
+		ConnectUsername:               cfg.EndpointID,
+		ReconnectBackoff:              cfg.Backoff.Func,
+		OnConnectError: func(err error) {
+			var ce *autopaho.ConnackError
+			if errors.As(err, &ce) {
+				if isAuthCONNACK(ce.ReasonCode) {
+					reason := AuthBadCredentials
+					if tok, _ := t.cred.Load().(string); strings.HasPrefix(tok, "nst_") {
+						reason = AuthSessionInvalid
+					}
+					if cfg.OnAuthFailed != nil {
+						cfg.OnAuthFailed(reason)
+					}
+					cancel()
+				}
+			}
+		},
+		OnConnectionDown: func() bool {
+			if t.stopped.Load() {
+				return false
+			}
+			if cfg.Backoff != nil {
+				cfg.Backoff.MarkOffline()
+			}
+			if cfg.OnOffline != nil {
+				cfg.OnOffline()
+			}
+			return !t.stopped.Load()
+		},
+		OnConnectionUp: func(cm *autopaho.ConnectionManager, _ *paho.Connack) {
+			if cfg.Backoff != nil {
+				cfg.Backoff.MarkOnline()
+			}
+			go func() {
+				_, err := cm.Subscribe(innerCtx, &paho.Subscribe{
+					Subscriptions: []paho.SubscribeOptions{{
+						Topic: t.downTopic,
+						QoS:   1,
+					}},
+				})
+				if err != nil {
+					return
+				}
+				if cfg.MQTTReady != nil {
+					_ = cfg.MQTTReady(innerCtx)
+				}
+				if cfg.OnOnline != nil {
+					cfg.OnOnline()
+				}
+			}()
+		},
+		ClientConfig: paho.ClientConfig{
+			ClientID: cfg.EndpointID,
+			OnServerDisconnect: func(d *paho.Disconnect) {
+				if d != nil && d.ReasonCode == 0x8E {
+					if cfg.OnKicked != nil {
+						cfg.OnKicked()
+					}
+					cancel()
+				}
+			},
+			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
+				func(pr paho.PublishReceived) (bool, error) {
+					if cfg.OnDown != nil && pr.Packet != nil {
+						cfg.OnDown(pr.Packet.Payload)
+					}
+					return true, nil
+				},
+			},
+		},
+	}
+
+	cliCfg.ConnectPacketBuilder = func(c *paho.Connect, _ *url.URL) (*paho.Connect, error) {
+		c.CleanStart = true
+		zero := uint32(0)
+		if c.Properties == nil {
+			c.Properties = &paho.ConnectProperties{}
+		}
+		c.Properties.SessionExpiryInterval = &zero
+		pass, _ := t.cred.Load().(string)
+		c.UsernameFlag = true
+		c.Username = cfg.EndpointID
+		c.PasswordFlag = true
+		c.Password = []byte(pass)
+		if cfg.OnConnectPacket != nil {
+			cfg.OnConnectPacket(c.CleanStart, zero)
+		}
+		return c, nil
+	}
+
+	cm, err := autopaho.NewConnection(innerCtx, cliCfg)
+	if err != nil {
+		cancel()
+		return err
+	}
+	t.cm = cm
+	return nil
+}
+
+func (t *mqttTransport) PublishUp(payload []byte) error {
+	t.mu.Lock()
+	cm := t.cm
+	topic := t.upTopic
+	t.mu.Unlock()
+	if cm == nil {
+		return apiErr(CodeNotConnected, "未连接")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	_, err := cm.Publish(ctx, &paho.Publish{
+		Topic:   topic,
+		QoS:     1,
+		Payload: payload,
+	})
+	return err
+}
+
+func (t *mqttTransport) Stop(ctx context.Context) error {
+	t.stopped.Store(true)
+	t.mu.Lock()
+	cm := t.cm
+	cancel := t.cancel
+	t.mu.Unlock()
+	if cancel != nil {
+		cancel()
+	}
+	if cm != nil {
+		return cm.Disconnect(ctx)
+	}
+	return nil
+}
+
+func isAuthCONNACK(code byte) bool {
+	switch code {
+	case 0x86, 0x87, 0x8A, // MQTT 5
+		4, 5: // MQTT 3.1.1 bad user/pass, not authorized
+		return true
+	default:
+		return false
+	}
+}
+
+func normalizeMQTTURL(raw string, allowTCP bool) (*url.URL, error) {
+	u, err := url.Parse(raw)
+	if err != nil {
+		return nil, err
+	}
+	switch strings.ToLower(u.Scheme) {
+	case "ws", "wss":
+		if u.Path == "" || u.Path == "/" {
+			u.Path = "/mqtt"
+		}
+		return u, nil
+	case "http":
+		u.Scheme = "ws"
+		if u.Path == "" || u.Path == "/" {
+			u.Path = "/mqtt"
+		}
+		return u, nil
+	case "https":
+		u.Scheme = "wss"
+		if u.Path == "" || u.Path == "/" {
+			u.Path = "/mqtt"
+		}
+		return u, nil
+	case "mqtt", "tcp", "mqtts", "ssl", "tls":
+		if !allowTCP {
+			return nil, apiErr(CodeBadRequest, "裸 TCP 需在选项中显式打开 AllowTCP")
+		}
+		return u, nil
+	default:
+		return nil, fmt.Errorf("不支持的 URL scheme: %s", u.Scheme)
+	}
+}
+
+// RegisterURLFromConnect 从连接地址推出注册 HTTP 地址（第 6.9 节）。
+func RegisterURLFromConnect(connectURL string) (string, error) {
+	u, err := url.Parse(connectURL)
+	if err != nil {
+		return "", err
+	}
+	out := *u
+	switch strings.ToLower(u.Scheme) {
+	case "wss", "https", "mqtts", "ssl", "tls":
+		out.Scheme = "https"
+	case "ws", "http", "mqtt", "tcp":
+		out.Scheme = "http"
+	default:
+		return "", fmt.Errorf("无法从 %s 推出注册地址", u.Scheme)
+	}
+	out.Path = "/api/client/register"
+	out.RawQuery = ""
+	out.Fragment = ""
+	return out.String(), nil
+}

sdk/go/types.go

@@ -0,0 +1,231 @@
+package nixmsg
+
+import "time"
+
+// ConnectionState 连接状态（onConnection）。
+type ConnectionState string
+
+const (
+	StateConnecting   ConnectionState = "connecting"
+	StateOnline       ConnectionState = "online"
+	StateReconnecting ConnectionState = "reconnecting"
+	StateOffline      ConnectionState = "offline"
+	StateKicked       ConnectionState = "kicked"
+	StateAuthFailed   ConnectionState = "auth_failed"
+)
+
+// Credential 登录凭据：密码或会话令牌二选一。
+type Credential struct {
+	Password     string
+	SessionToken string
+}
+
+// Options 连接选项。
+type Options struct {
+	// ManualAck 为 true 时手动确认模式。
+	ManualAck bool
+	// AllowTCP 显式允许裸 TCP（mqtt:// / mqtts://）。默认只用 WebSocket。
+	AllowTCP bool
+	// ConnectTimeout 连接超时，默认 30s。
+	ConnectTimeout time.Duration
+	// ClientLabel hello.client，默认 go-sdk/0.1。
+	ClientLabel string
+	// MaxReceiveBytes hello.max_receive_bytes；0 表示省略（不限）。
+	MaxReceiveBytes int
+	// SendQueueSize 发送队列上限，默认 1000。
+	SendQueueSize int
+	// MaxInflight 在途上限，默认 100。
+	MaxInflight int
+	// DedupCapacity from+id 去重容量，默认 10000。
+	DedupCapacity int
+	// HTTPClient 注册用；nil 用默认。
+	// 测试可注入 transport。
+	transport transport
+}
+
+func (o *Options) withDefaults() Options {
+	out := *o
+	if out.ConnectTimeout <= 0 {
+		out.ConnectTimeout = 30 * time.Second
+	}
+	if out.ClientLabel == "" {
+		out.ClientLabel = "go-sdk/0.1"
+	}
+	if out.SendQueueSize <= 0 {
+		out.SendQueueSize = 1000
+	}
+	if out.MaxInflight <= 0 {
+		out.MaxInflight = 100
+	}
+	if out.DedupCapacity <= 0 {
+		out.DedupCapacity = 10000
+	}
+	return out
+}
+
+// Target 发送目标。
+type Target struct {
+	Kind string `json:"kind"`
+	ID   string `json:"id"`
+}
+
+// Body 消息正文。
+type Body struct {
+	Enc         string `json:"enc"`
+	ContentType string `json:"content_type,omitempty"`
+	Data        string `json:"data"`
+}
+
+// SendOptions 发送选项。
+type SendOptions struct {
+	SendAt       *time.Time
+	Delay        *time.Duration
+	Keep         bool
+	TTL          *int64
+	Receipt      *bool
+	TalkPassword string
+	ContentType  string
+	Meta         map[string]any
+	ID           string // 可选；空则生成 UUIDv7
+}
+
+// SendResult 发送结果。
+type SendResult struct {
+	ID       string
+	SendAtMs int64
+	State    string
+}
+
+// Message 下行消息（交给应用）。
+type Message struct {
+	ID       string
+	From     string
+	To       Target
+	Body     Body
+	Meta     map[string]any
+	SendAtMs int64
+}
+
+// RecallResult 撤回结果。
+type RecallResult struct {
+	Result   string
+	Recalled int
+	Accepted int
+	Other    int
+}
+
+// Receipt 回执。
+type Receipt struct {
+	ReceiptID  string
+	ID         string
+	EndpointID string
+	State      string
+	Reason     string
+	AtMs       int64
+}
+
+// RevokedEvent 撤回/作废事件。
+type RevokedEvent struct {
+	ID     string
+	From   string
+	Reason string
+}
+
+// PresenceItem 在线查询项。
+type PresenceItem struct {
+	ID      string `json:"id"`
+	Online  bool   `json:"online"`
+	SinceMs int64  `json:"since_ms"`
+}
+
+// DirectoryItem 目录项。
+type DirectoryItem struct {
+	ID              string `json:"id"`
+	Name            string `json:"name"`
+	Online          bool   `json:"online"`
+	OnlineSinceMs   int64  `json:"online_since_ms"`
+	OfflineSinceMs  int64  `json:"offline_since_ms"`
+	TalkPasswordSet bool   `json:"talk_password_set"`
+}
+
+// PresenceEvent 上下线通知。
+type PresenceEvent struct {
+	ID     string
+	Online bool
+	AtMs   int64
+}
+
+// GroupEvent 群事件。
+type GroupEvent struct {
+	GroupID    string
+	Event      string
+	EndpointID string
+	AtMs       int64
+}
+
+// SelfInfo 自己的资料。
+type SelfInfo struct {
+	ID             string `json:"id"`
+	Name           string `json:"name"`
+	DefaultDelayMs int64  `json:"default_delay_ms"`
+	TalkPasswordSet bool  `json:"talk_password_set"`
+}
+
+// GroupInfo 群摘要。
+type GroupInfo struct {
+	ID          string `json:"id"`
+	Name        string `json:"name"`
+	OwnerID     string `json:"owner_id"`
+	MemberCount int    `json:"member_count"`
+}
+
+// GroupMember 群成员。
+type GroupMember struct {
+	ID     string `json:"id"`
+	Name   string `json:"name"`
+	Online bool   `json:"online"`
+}
+
+// GroupMemberIn 建群/加群成员。
+type GroupMemberIn struct {
+	ID           string
+	TalkPassword string
+}
+
+// MemberFailure 部分成员失败。
+type MemberFailure struct {
+	ID   string `json:"id"`
+	Code string `json:"code"`
+}
+
+// RegisterOptions 注册选项。
+type RegisterOptions struct {
+	ID            string
+	LoginPassword string
+	Name          string
+	TalkPassword  string
+}
+
+// RegisterResult 注册结果。
+type RegisterResult struct {
+	ID            string
+	LoginPassword string // 仅请求未带密码时返回
+}
+
+// HandshakeLimits 握手参数。
+type HandshakeLimits struct {
+	ServerTimeMs       int64
+	ServerVersion      string
+	MaxBodyBytes       int
+	MaxMetaBytes       int
+	MaxFrameBytes      int
+	MaxTTLSeconds      int64
+	MaxScheduleSeconds int64
+	AckTimeoutSeconds  int64
+}
+
+// ConnectionEvent 连接状态事件。
+type ConnectionEvent struct {
+	State  ConnectionState
+	Reason string // auth_failed / kicked 时的原因
+}