diff --git a/.env.example b/.env.example index c526b84..1986300 100644 --- a/.env.example +++ b/.env.example @@ -90,6 +90,13 @@ SESSION_IDLE_TIMEOUT_MIN=30 # ---- 跨域白名单 ---- ALLOWED_ORIGINS=http://localhost:5173,http://localhost:3000 +# ---- KOOK（KaiHeiLa）渠道 ---- +# 单账号走环境变量；多账号走下方 CHANNELS_JSON。 +# CHANNEL_KOOK_TOKEN 为 KOOK 开发者中心机器人 token（敏感，勿写日志/提交）。 +# CHANNEL_KOOK_BASE_URL 可选，默认 https://www.kookapp.cn/api/v3。 +# CHANNEL_KOOK_TOKEN= +# CHANNEL_KOOK_BASE_URL= + # ---- IM 多账号 / 黑白名单（可选，design/16 §4.6 §5-6） ---- # 单账号仍走上面 CHANNEL__* 环境变量；这里提供独立的 JSON 注册表， # 支持同一平台配多个接入配置（多账号），并配群聊/私聊黑白名单（代码层拦截， diff --git a/cmd/cyrene/channels.go b/cmd/cyrene/channels.go index 7396143..97a9683 100644 --- a/cmd/cyrene/channels.go +++ b/cmd/cyrene/channels.go @@ -15,6 +15,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" "git.yeij.top/AskaEth/Cyrene_v2/internal/config" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/kook" "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/qq_obv11" "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/telegram" "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/webhook" @@ -53,6 +54,11 @@ func buildChannelAdapters(log *slog.Logger) []adapter.PlatformAdapter { WebhookURL: os.Getenv("CHANNEL_TELEGRAM_WEBHOOK_URL"), }, telegram.WithLogger(log)) + kook := kook.New(kook.Config{ + BotToken: os.Getenv("CHANNEL_KOOK_TOKEN"), + BaseURL: os.Getenv("CHANNEL_KOOK_BASE_URL"), + }, kook.WithLogger(log)) + // webhook.WithLogger does not tag provenance, so wrap base with the module // source here to satisfy the logging rule. wh := webhook.New( @@ -66,7 +72,7 @@ func buildChannelAdapters(log *slog.Logger) []adapter.PlatformAdapter { webhook.WithLogger(logger.WithSource(log, "plugins/channels/webhook")), ) - out := []adapter.PlatformAdapter{qq, tg, wh} + out := []adapter.PlatformAdapter{qq, tg, wh, kook} if weixin != nil { out = append(out, weixin) } @@ -154,6 +160,12 @@ func buildAccountAdapter(platform string, acct config.ChannelAccount, log *slog. return nil, fmt.Errorf("decode qq-obv11 config: %w", err) } inner = qq_obv11.NewAdapter(cfg) + case "kook": + var cfg kook.Config + if err := adapter.DecodeConfig(acct.Config, &cfg); err != nil { + return nil, fmt.Errorf("decode kook config: %w", err) + } + inner = kook.New(cfg, kook.WithLogger(log)) case "webhook": var cfg webhook.Config if err := adapter.DecodeConfig(acct.Config, &cfg); err != nil { diff --git a/cmd/cyrene/channels_test.go b/cmd/cyrene/channels_test.go new file mode 100644 index 0000000..6a30b94 --- /dev/null +++ b/cmd/cyrene/channels_test.go @@ -0,0 +1,56 @@ +package main + +import ( + "io" + "log/slog" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +func testLog() *slog.Logger { + return logger.New(logger.Options{Out: io.Discard}) +} + +func TestBuildChannelAdapters_IncludesKook(t *testing.T) { + t.Setenv("CHANNEL_KOOK_TOKEN", "kook-secret") + adapters := buildChannelAdapters(testLog()) + if !hasAdapter(adapters, "kook") { + t.Fatalf("expected a kook adapter, got %v", names(adapters)) + } +} + +func TestBuildChannelAdapters_KookMultiAccount(t *testing.T) { + t.Setenv("CHANNELS_JSON", `{"kook":[{"id":"conf0","config":{"bot_token":"kook-secret"},"allowUsers":["u-1"],"denyGroups":["g-9"]}]}`) + adapters := buildChannelAdapters(testLog()) + if !hasAdapter(adapters, "kook:conf0") { + t.Fatalf("expected a kook:conf0 adapter, got %v", names(adapters)) + } + // The named adapter must expose its allow/deny lists via Config(). + for _, a := range adapters { + if a.Name() == "kook:conf0" { + cfg := a.Config() + if got, _ := cfg["allowUsers"].([]string); len(got) != 1 || got[0] != "u-1" { + t.Errorf("allowUsers = %v, want [u-1]", cfg["allowUsers"]) + } + } + } +} + +func hasAdapter(adapters []adapter.PlatformAdapter, name string) bool { + for _, a := range adapters { + if a.Name() == name { + return true + } + } + return false +} + +func names(adapters []adapter.PlatformAdapter) []string { + out := make([]string, 0, len(adapters)) + for _, a := range adapters { + out = append(out, a.Name()) + } + return out +} diff --git a/internal/channel/adapter/httpclient/client.go b/internal/channel/adapter/httpclient/client.go new file mode 100644 index 0000000..6a57fa9 --- /dev/null +++ b/internal/channel/adapter/httpclient/client.go @@ -0,0 +1,169 @@ +// Package httpclient provides a small, reusable HTTP (/JSON) client for IM +// channel adapters that talk to a REST API with bearer-token auth and a +// common `{code,message,data}` response envelope (KOOK, and similarly shaped +// platforms). It is used by the KOOK adapter and is intended to be shared by +// future platforms; existing adapters (Telegram, OneBot) that already carry +// their own transport are left untouched. +// +// It depends only on the standard library and internal/logger. The bearer token +// is never logged or surfaced in errors. +package httpclient + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +// ErrAPI is the sentinel returned when the remote API reports a non-zero code or +// a non-2xx HTTP status. Tests compare with errors.Is. +var ErrAPI = errors.New("httpclient: api error") + +// maxBodyBytes caps how large a response body the client will read. +const maxBodyBytes = 1 << 20 // 1 MiB + +// Client is a bearer-auth JSON HTTP client bound to a base URL. +type Client struct { + http *http.Client + baseURL string + token string + log *slog.Logger +} + +// Option customizes a Client at construction time. +type Option func(*Client) + +// WithHTTPClient injects the *http.Client used for requests. Tests pass the +// client of an httptest.Server. +func WithHTTPClient(h *http.Client) Option { + return func(c *Client) { c.http = h } +} + +// WithBaseURL sets the URL root that paths are joined to. A base ending in "/" +// has its trailing slash trimmed. +func WithBaseURL(base string) Option { + return func(c *Client) { c.baseURL = strings.TrimRight(base, "/") } +} + +// WithToken permits an Authorization: Bearer header on every request. The token +// is stored in memory only and never logged. +func WithToken(token string) Option { + return func(c *Client) { c.token = token } +} + +// WithLogger injects the logger. A nil logger stays silent. +func WithLogger(l *slog.Logger) Option { + return func(c *Client) { + if l == nil { + c.log = logger.New(logger.Options{Out: io.Discard}) + } else { + c.log = l + } + } +} + +// New builds a Client. With no base URL the caller must pass absolute URLs (via +// the Do/Get helpers' path being an absolute URL). +func New(opts ...Option) *Client { + c := &Client{ + http: &http.Client{}, + log: logger.New(logger.Options{Out: io.Discard}), + } + for _, opt := range opts { + opt(c) + } + return c +} + +// APIEnvelope is the common `{code,message,data}` response envelope used by the +// KOOK REST API. Code==0 means success; a non-zero code is mapped to ErrAPI. +type APIEnvelope struct { + Code int `json:"code"` + Message string `json:"message"` + Data json.RawMessage `json:"data"` +} + +// DoJSON performs a request with the given method against path (joined to the +// base URL) and returns the decoded data payload of the envelope. A non-zero +// code or a non-2xx status is wrapped with ErrAPI. A 2xx body that is not an +// envelope (e.g. a raw payload) is returned verbatim as data. +func (c *Client) DoJSON(ctx context.Context, method, path string, body any) (json.RawMessage, error) { + return c.do(ctx, method, c.resolve(path), body) +} + +// GetJSON performs a GET against path with the given query parameters and +// returns the decoded data payload. A nil query adds nothing. +func (c *Client) GetJSON(ctx context.Context, path string, query url.Values) (json.RawMessage, error) { + target := c.resolve(path) + if len(query) > 0 { + target += "?" + query.Encode() + } + return c.do(ctx, http.MethodGet, target, nil) +} + +// resolve joins path to the base URL, passing through an absolute path as-is. +func (c *Client) resolve(path string) string { + if c.baseURL == "" { + return path + } + return c.baseURL + path +} + +// do performs the request, injecting auth and decoding the envelope. +func (c *Client) do(ctx context.Context, method, target string, body any) (json.RawMessage, error) { + var buf bytes.Buffer + if body != nil { + if err := json.NewEncoder(&buf).Encode(body); err != nil { + return nil, fmt.Errorf("httpclient: encode %s request: %w", method, err) + } + } + req, err := http.NewRequestWithContext(ctx, method, target, &buf) + if err != nil { + return nil, fmt.Errorf("httpclient: build %s %s request: %w", method, target, err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + if c.token != "" { + req.Header.Set("Authorization", "Bearer "+c.token) + } + + resp, err := c.http.Do(req) + if err != nil { + return nil, fmt.Errorf("httpclient: %s %s: %w", method, target, err) + } + defer resp.Body.Close() + + raw, err := io.ReadAll(io.LimitReader(resp.Body, maxBodyBytes)) + if err != nil { + return nil, fmt.Errorf("httpclient: read %s %s response: %w", method, target, err) + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("httpclient: %s %s: status %d: %s: %w", method, target, resp.StatusCode, truncate(raw), ErrAPI) + } + + var envelope APIEnvelope + if err := json.Unmarshal(raw, &envelope); err != nil { + return raw, nil + } + if envelope.Code != 0 { + return nil, fmt.Errorf("httpclient: %s %s: code %d: %s: %w", method, target, envelope.Code, envelope.Message, ErrAPI) + } + return envelope.Data, nil +} + +// truncate caps a byte slice to 512 bytes for inclusion in an error message. +func truncate(b []byte) string { + if len(b) > 512 { + return strings.TrimSpace(string(b[:512])) + "…" + } + return strings.TrimSpace(string(b)) +} diff --git a/internal/channel/adapter/httpclient/client_test.go b/internal/channel/adapter/httpclient/client_test.go new file mode 100644 index 0000000..073fcbd --- /dev/null +++ b/internal/channel/adapter/httpclient/client_test.go @@ -0,0 +1,96 @@ +package httpclient + +import ( + "context" + "encoding/json" + "errors" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +func discardLogger() *slog.Logger { return logger.New(logger.Options{Out: io.Discard}) } + +func TestGetJSON_EnvelopeAndAuth(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var auth string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + auth = r.Header.Get("Authorization") + mu.Unlock() + _ = json.NewEncoder(w).Encode(map[string]any{ + "code": 0, "message": "ok", + "data": map[string]any{"url": "wss://gw.example.com/gateway"}, + }) + })) + defer srv.Close() + + c := New( + WithBaseURL(srv.URL), + WithToken("secret-token"), + WithHTTPClient(srv.Client()), + WithLogger(discardLogger()), + ) + data, err := c.GetJSON(context.Background(), "/gateway/index", url.Values{"compress": {"0"}}) + if err != nil { + t.Fatalf("GetJSON: %v", err) + } + if string(data) != `{"url":"wss://gw.example.com/gateway"}` { + t.Errorf("data = %s", string(data)) + } + mu.Lock() + defer mu.Unlock() + if auth != "Bearer secret-token" { + t.Errorf("Authorization = %q, want Bearer secret-token", auth) + } +} + +func TestDoJSON_NonZeroCode(t *testing.T) { + t.Parallel() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{"code": 40101, "message": "invalid token", "data": struct{}{}}) + })) + defer srv.Close() + c := New(WithBaseURL(srv.URL), WithHTTPClient(srv.Client()), WithLogger(discardLogger())) + _, err := c.DoJSON(context.Background(), http.MethodPost, "/message/create", map[string]any{"a": 1}) + if !errors.Is(err, ErrAPI) { + t.Fatalf("err = %v, want ErrAPI", err) + } +} + +func TestDoJSON_StatusError(t *testing.T) { + t.Parallel() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + _, _ = w.Write([]byte("unauthorized")) + })) + defer srv.Close() + c := New(WithBaseURL(srv.URL), WithHTTPClient(srv.Client()), WithLogger(discardLogger())) + _, err := c.DoJSON(context.Background(), http.MethodGet, "/gateway/index", nil) + if !errors.Is(err, ErrAPI) { + t.Fatalf("err = %v, want ErrAPI", err) + } +} + +func TestGetJSON_NonEnvelopeBodyPassedThrough(t *testing.T) { + t.Parallel() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`[1,2,3]`)) + })) + defer srv.Close() + c := New(WithBaseURL(srv.URL), WithHTTPClient(srv.Client()), WithLogger(discardLogger())) + data, err := c.GetJSON(context.Background(), "/list", nil) + if err != nil { + t.Fatalf("GetJSON: %v", err) + } + if string(data) != "[1,2,3]" { + t.Errorf("data = %s, want [1,2,3]", string(data)) + } +} diff --git a/internal/channel/adapter/wsutil/backoff.go b/internal/channel/adapter/wsutil/backoff.go new file mode 100644 index 0000000..f6faa9b --- /dev/null +++ b/internal/channel/adapter/wsutil/backoff.go @@ -0,0 +1,54 @@ +// Package wsutil provides a small, reusable WebSocket gateway client for +// stateful IM channels (KOOK, QQ OneBot, and similar). It owns the transport +// mechanics that those channels share — dialing a gateway, running a read loop, +// sending an application-level heartbeat, and reconnecting with exponential +// backoff (optionally splicing resume parameters onto the gateway URL) — while +// leaving the platform-specific signaling to the adapter via Handler hooks. +// +// It depends only on the standard library plus github.com/gorilla/websocket and +// internal/logger, never on a specific channel adapter. +package wsutil + +import "time" + +// Backoff is a resettable exponential backoff delay scheduler. It yields delays +// starting at base and doubling up to cap (for example 2s, 4s, 8s, ... capped at +// 60s), which is the WebSocket gateway reconnect pattern used by stateful IM +// channels. It is safe for concurrent use by a single goroutine. +type Backoff struct { + base time.Duration + max time.Duration + attempts int +} + +// NewBackoff creates a backoff with the given base and cap. A non-positive base +// defaults to 1s and a non-positive cap to 60s. +func NewBackoff(base, cap time.Duration) *Backoff { + if base <= 0 { + base = time.Second + } + if cap <= 0 { + cap = 60 * time.Second + } + return &Backoff{base: base, max: cap} +} + +// Next returns the delay to sleep before the next attempt and advances the +// internal attempt counter. It never exceeds the configured cap. +func (b *Backoff) Next() time.Duration { + var d time.Duration + if b.attempts >= 60 { + d = b.max + } else { + d = b.base << uint(b.attempts) + } + if d > b.max || d <= 0 { + d = b.max + } + b.attempts++ + return d +} + +// Reset restores the backoff to its initial state after a successful connect, so +// a later drop starts the ramp over from the base delay. +func (b *Backoff) Reset() { b.attempts = 0 } diff --git a/internal/channel/adapter/wsutil/backoff_test.go b/internal/channel/adapter/wsutil/backoff_test.go new file mode 100644 index 0000000..50e11e9 --- /dev/null +++ b/internal/channel/adapter/wsutil/backoff_test.go @@ -0,0 +1,56 @@ +package wsutil + +import ( + "testing" + "time" +) + +func TestBackoffSequence(t *testing.T) { + t.Parallel() + b := NewBackoff(2*time.Second, 60*time.Second) + want := []time.Duration{2, 4, 8, 16, 32, 60, 60, 60} + for i, exp := range want { + if got := b.Next(); got != exp*time.Second { + t.Fatalf("Next[%d] = %v, want %v", i, got, exp*time.Second) + } + } +} + +func TestBackoffReset(t *testing.T) { + t.Parallel() + b := NewBackoff(time.Second, time.Minute) + _ = b.Next() + _ = b.Next() + b.Reset() + if got := b.Next(); got != time.Second { + t.Errorf("after Reset, Next = %v, want 1s", got) + } +} + +func TestBackoffDefaults(t *testing.T) { + t.Parallel() + b := NewBackoff(0, 0) + if got := b.Next(); got != time.Second { + t.Errorf("default base Next = %v, want 1s", got) + } + if got := b.Next(); got != 2*time.Second { + t.Errorf("default Next2 = %v, want 2s", got) + } + // cap of 60s by default: 1,2,4,8,16,32,60... + for i := 0; i < 7; i++ { + _ = b.Next() + } + if got := b.Next(); got != 60*time.Second { + t.Errorf("default cap Next = %v, want 60s", got) + } +} + +func TestBackoffNoOverflow(t *testing.T) { + t.Parallel() + b := NewBackoff(time.Second, time.Second) + for i := 0; i < 200; i++ { + if got := b.Next(); got != time.Second { + t.Fatalf("Next[%d] = %v, want 1s (cap)", i, got) + } + } +} diff --git a/internal/channel/adapter/wsutil/client.go b/internal/channel/adapter/wsutil/client.go new file mode 100644 index 0000000..4b871f4 --- /dev/null +++ b/internal/channel/adapter/wsutil/client.go @@ -0,0 +1,359 @@ +package wsutil + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "sync" + "time" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +const ( + // defaultHeartbeatInterval is used when a client enables heartbeat without a + // configured interval. + defaultHeartbeatInterval = 30 * time.Second + // defaultPongTimeout is the read deadline margin added to the heartbeat + // interval; if the peer sends nothing within interval+pongTimeout the socket + // is considered dead and the client reconnects. + defaultPongTimeout = 6 * time.Second + // writeWait bounds a single control-frame write. + writeWait = 10 * time.Second +) + +// DialFunc returns the gateway URL to connect to. On a reconnect the platform may +// return a resume-augmented URL (splice session/sn via AppendQuery) so the +// gateway can resynchronize the offline message window. +type DialFunc func(ctx context.Context) (string, error) + +// Handler carries the platform-specific protocol hooks the Client invokes. A +// nil field disables that hook. The hooks run on the Client's read goroutine +// (OnConnect / OnMessage) or a dedicated heartbeat goroutine (Heartbeat). +type Handler struct { + // OnConnect is called after the WebSocket connection is established and the + // application handshake (for example KOOK's hello) succeeds. Returning an + // error aborts the session and triggers a reconnect with backoff. + OnConnect func(ctx context.Context, c *Conn) error + // OnMessage is called for every application text frame received. Returning an + // error forces a reconnect. + OnMessage func(ctx context.Context, payload []byte) error + // Heartbeat is called on the heartbeat ticker (when HeartbeatInterval > 0) to + // send an application-level keepalive frame. A nil Heartbeat disables it. + Heartbeat func(ctx context.Context, c *Conn) error +} + +// Client is a reusable WebSocket gateway connection manager. It dials a gateway, +// runs the read loop, maintains an optional application-level heartbeat with a +// liveness read deadline, and reconnects with exponential backoff until the +// context is canceled. Only the platform signaling is left to the Handler. +type Client struct { + dial DialFunc + handler Handler + dialer *websocket.Dialer + header http.Header + interval time.Duration + pongWait time.Duration + backoff *Backoff + log *slog.Logger + + mu sync.Mutex + conn *Conn + cancel context.CancelFunc + closed bool +} + +// Option customizes a Client at construction time. +type Option func(*Client) + +// WithDialer injects the WebSocket dialer used for the handshake. Tests pass a +// dialer whose net dial is redirected to a fake server. +func WithDialer(d *websocket.Dialer) Option { + return func(c *Client) { c.dialer = d } +} + +// WithHeader injects extra HTTP headers on the handshake (for example an +// Authorization: Bearer token). +func WithHeader(h http.Header) Option { + return func(c *Client) { c.header = h } +} + +// WithHeartbeat enables an application-level heartbeat at the given interval. +// A nil interval disables the heartbeat ticker (default off). +func WithHeartbeat(interval time.Duration) Option { + return func(c *Client) { c.interval = interval } +} + +// WithPongTimeout sets the read deadline margin added to the heartbeat interval. +// A non-positive value uses the default (6s). +func WithPongTimeout(d time.Duration) Option { + return func(c *Client) { c.pongWait = d } +} + +// WithBackoff injects the reconnect backoff scheduler. +func WithBackoff(b *Backoff) Option { + return func(c *Client) { c.backoff = b } +} + +// WithLogger injects the logger used by the client. A nil logger stays silent. +func WithLogger(l *slog.Logger) Option { + return func(c *Client) { + if l == nil { + c.log = logger.New(logger.Options{Out: io.Discard}) + } else { + c.log = l + } + } +} + +// New builds a Client. dial must be non-nil. +func New(dial DialFunc, h Handler, opts ...Option) *Client { + if dial == nil { + dial = func(context.Context) (string, error) { return "", nil } + } + c := &Client{ + dial: dial, + handler: h, + dialer: &websocket.Dialer{HandshakeTimeout: 15 * time.Second}, + pongWait: defaultPongTimeout, + backoff: NewBackoff(2*time.Second, 60*time.Second), + log: logger.New(logger.Options{Out: io.Discard}), + } + for _, opt := range opts { + opt(c) + } + return c +} + +// Run executes the connect/reconnect loop until ctx is canceled. It returns +// nil when ctx is canceled and a non-nil error on a non-context termination. It +// is intended to be run in its own goroutine by the adapter's Connect. +func (c *Client) Run(ctx context.Context) error { + for { + if ctx.Err() != nil { + return ctx.Err() + } + url, err := c.dial(ctx) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + c.log.Warn("ws gateway dial failed", "err", err) + if !c.sleepBackoff(ctx) { + return ctx.Err() + } + continue + } + if err := c.connectOnce(ctx, url); err != nil { + c.log.Warn("ws session ended", "err", err) + } + if ctx.Err() != nil { + return ctx.Err() + } + } +} + +// connectOnce dials one URL and runs the connect hook plus the read loop until +// the connection drops, the read loop errors, or ctx is canceled. +func (c *Client) connectOnce(ctx context.Context, url string) error { + if c.header == nil { + c.header = http.Header{} + } + ws, _, err := c.dialer.DialContext(ctx, url, c.header) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + c.log.Warn("ws dial failed", "err", err) + c.sleepBackoff(ctx) + return err + } + conn := &Conn{ws: ws} + defer conn.Close() + + // A watcher closes the connection when ctx ends so a pending read (including + // the OnConnect handshake) unblocks and Run can exit promptly. + stop := make(chan struct{}) + defer close(stop) + go c.watchCtx(ctx, conn, stop) + + if c.handler.OnConnect != nil { + if cerr := c.handler.OnConnect(ctx, conn); cerr != nil { + if ctx.Err() != nil { + return ctx.Err() + } + c.log.Warn("ws connect hook failed", "err", cerr) + c.sleepBackoff(ctx) + return cerr + } + } + + c.setConn(conn) + c.backoff.Reset() + c.runReadLoop(ctx, conn) + c.setConn(nil) + return nil +} + +// watchCtx closes the connection when ctx is canceled so a pending read +// unblocks, letting Run exit promptly. +func (c *Client) watchCtx(ctx context.Context, conn *Conn, stop <-chan struct{}) { + select { + case <-ctx.Done(): + _ = conn.Close() + case <-stop: + } +} + +// runReadLoop consumes text frames until the read errors, the platform handler +// returns an error, or ctx is canceled. When heartbeat is enabled it also runs a +// heartbeat ticker and refreshes a read deadline so a silent-but-live socket +// stays open while a dead one is detected. +func (c *Client) runReadLoop(ctx context.Context, conn *Conn) { + hbCtx, hbCancel := context.WithCancel(ctx) + defer hbCancel() + if c.handler.Heartbeat != nil && c.interval > 0 { + go c.heartbeatLoop(hbCtx, conn) + } + + // A liveness read deadline is only meaningful when an application heartbeat + // keeps the socket active; without one, an idle-but-open connection must stay + // open indefinitely. + deadline := time.Duration(0) + if c.handler.Heartbeat != nil && c.interval > 0 { + deadline = c.interval + c.pongWait + } + for { + if ctx.Err() != nil { + return + } + if deadline > 0 { + _ = conn.SetReadDeadline(time.Now().Add(deadline)) + } else { + _ = conn.SetReadDeadline(time.Time{}) + } + payload, err := conn.ReadMessage() + if err != nil { + return + } + if c.handler.OnMessage != nil { + if err := c.handler.OnMessage(ctx, payload); err != nil { + if ctx.Err() == nil { + c.log.Warn("ws on-message failed", "err", err) + } + return + } + } + } +} + +// heartbeatLoop sends an application-level keepalive on every interval. A +// failing heartbeat closes the connection to force a reconnect. +func (c *Client) heartbeatLoop(ctx context.Context, conn *Conn) { + t := time.NewTicker(c.interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + if err := c.handler.Heartbeat(ctx, conn); err != nil { + c.log.Warn("ws heartbeat failed", "err", err) + _ = conn.Close() + return + } + } + } +} + +// sleepBackoff sleeps the current backoff delay or until ctx is canceled, +// returning false (and leaving the connection for Run to exit) when ctx fires. +func (c *Client) sleepBackoff(ctx context.Context) bool { + d := c.backoff.Next() + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} + +// setConn installs the live connection so Close can tear it down. +func (c *Client) setConn(conn *Conn) { + c.mu.Lock() + defer c.mu.Unlock() + c.conn = conn +} + +// Close tears down the client, closing the live connection and its context. +func (c *Client) Close() error { + c.mu.Lock() + if c.closed { + c.mu.Unlock() + return nil + } + c.closed = true + conn := c.conn + cancel := c.cancel + c.mu.Unlock() + if conn != nil { + _ = conn.Close() + } + if cancel != nil { + cancel() + } + return nil +} + +// BindCancel associates a context cancel func so Close can stop a running Run. +func (c *Client) BindCancel(cancel context.CancelFunc) { + c.mu.Lock() + defer c.mu.Unlock() + c.cancel = cancel +} + +// Conn is a thin, concurrency-safe wrapper over one *websocket.Conn. It +// serializes writes (gorilla allows one reader plus one writer) and exposes only +// the JSON text-frame surface the gateway clients need. +type Conn struct { + ws *websocket.Conn + writeMu sync.Mutex +} + +// ReadMessage reads the next complete message, returning its text payload. It +// honors any read deadline set via SetReadDeadline. +func (c *Conn) ReadMessage() ([]byte, error) { + _, p, err := c.ws.ReadMessage() + if err != nil { + return nil, err + } + return p, nil +} + +// WriteMessage writes a text frame with the given payload. +func (c *Conn) WriteMessage(payload []byte) error { + c.writeMu.Lock() + defer c.writeMu.Unlock() + return c.ws.WriteMessage(websocket.TextMessage, payload) +} + +// WriteJSON marshals v and writes it as a text frame. +func (c *Conn) WriteJSON(v any) error { + data, err := json.Marshal(v) + if err != nil { + return err + } + return c.WriteMessage(data) +} + +// SetReadDeadline bounds the next ReadMessage by the given absolute time. +func (c *Conn) SetReadDeadline(t time.Time) error { return c.ws.SetReadDeadline(t) } + +// Close closes the underlying connection exactly once. +func (c *Conn) Close() error { return c.ws.Close() } diff --git a/internal/channel/adapter/wsutil/client_test.go b/internal/channel/adapter/wsutil/client_test.go new file mode 100644 index 0000000..0bc3b63 --- /dev/null +++ b/internal/channel/adapter/wsutil/client_test.go @@ -0,0 +1,147 @@ +package wsutil + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +var wsu = websocket.Upgrader{} + +// TestClient_DialReadReconnect exercises the connect hook, the read loop, and the +// reconnect path: the server accepts a connection, sends one frame, drops it; +// the client re-reads a second frame on the next connection, then the test cancels +// the context and the client exits cleanly. +func TestClient_DialReadReconnect(t *testing.T) { + var conns atomic.Int64 + var mu sync.Mutex + var got []string + received := make(chan string, 2) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := wsu.Upgrade(w, r, nil) + if err != nil { + t.Errorf("upgrade: %v", err) + return + } + defer conn.Close() + n := conns.Add(1) + msg := "msg-" + string(rune('0'+n)) + _ = conn.WriteMessage(websocket.TextMessage, []byte(msg)) + if n < 2 { + return // close the first connection to trigger a reconnect + } + // Hold the second connection open until the client cancels the context. + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + })) + defer srv.Close() + + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + "/testws" + client := New( + func(context.Context) (string, error) { return url, nil }, + Handler{ + OnConnect: func(ctx context.Context, c *Conn) error { return nil }, + OnMessage: func(ctx context.Context, p []byte) error { + mu.Lock() + got = append(got, string(p)) + mu.Unlock() + received <- string(p) + return nil + }, + }, + WithBackoff(NewBackoff(10*time.Millisecond, 50*time.Millisecond)), + WithLogger(logger.New(logger.Options{Out: io.Discard})), + ) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- client.Run(ctx) }() + + for i := 0; i < 2; i++ { + select { + case m := <-received: + if m != "msg-" + string(rune('1'+i)) { + t.Errorf("message[%d] = %q", i, m) + } + case <-time.After(3 * time.Second): + t.Fatalf("timed out waiting for message %d", i+1) + } + } + cancel() + select { + case err := <-done: + if err == nil { + t.Error("Run should return a non-nil error on cancel") + } + case <-time.After(3 * time.Second): + t.Fatal("Run did not exit after cancel") + } + + mu.Lock() + defer mu.Unlock() + if len(got) != 2 { + t.Fatalf("received %d messages, want 2", len(got)) + } + if conns.Load() < 2 { + t.Errorf("connection count = %d, want >= 2 (reconnect)", conns.Load()) + } +} + +// TestClient_HeartbeatVerifiesPing sends an s=2-like heartbeat every 20ms and +// asserts the dial URL is still the one provided. +func TestClient_OnConnectErrorTriggersRedial(t *testing.T) { + var dials atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := wsu.Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + })) + defer srv.Close() + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + "/testws" + + // OnConnect always errors so the client keeps rejecting each session and the + // dial fires repeatedly. + client := New( + func(context.Context) (string, error) { dials.Add(1); return url, nil }, + Handler{ + OnConnect: func(ctx context.Context, c *Conn) error { return context.DeadlineExceeded }, + }, + WithBackoff(NewBackoff(5*time.Millisecond, 20*time.Millisecond)), + WithLogger(logger.New(logger.Options{Out: io.Discard})), + ) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- client.Run(ctx) }() + + time.Sleep(60 * time.Millisecond) + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not exit") + } + if dials.Load() < 2 { + t.Errorf("dials = %d, want >= 2 (reconnect on failed connect hook)", dials.Load()) + } +} diff --git a/internal/channel/adapter/wsutil/url.go b/internal/channel/adapter/wsutil/url.go new file mode 100644 index 0000000..07d877e --- /dev/null +++ b/internal/channel/adapter/wsutil/url.go @@ -0,0 +1,85 @@ +package wsutil + +import ( + "net/url" + "sort" + "strings" +) + +// AppendQuery returns rawURL with the given query parameters added, preserving +// any existing query string. Parameters with an empty value are omitted. It is +// used by stateful gateway clients to splice resume parameters (resume/sn/ +// session_id) and access tokens onto a gateway URL before reconnecting. Keys are +// appended in sorted order so the result is deterministic for tests. +func AppendQuery(rawURL string, params map[string]string) string { + if len(params) == 0 { + return rawURL + } + u, err := url.Parse(rawURL) + if err != nil { + return appendNaive(rawURL, params) + } + + keys := make([]string, 0, len(params)) + for k := range params { + keys = append(keys, k) + } + sort.Strings(keys) + + pairs := existingPairs(u.Query()) + for _, k := range keys { + if params[k] == "" { + continue + } + pairs = append(pairs, url.QueryEscape(k)+"="+url.QueryEscape(params[k])) + } + u.RawQuery = strings.Join(pairs, "&") + return u.String() +} + +// existingPairs flattens url.Values into ordered "k=v" pairs so that appending +// new parameters produces a deterministic query string. Multi-valued keys keep +// their original value order. +func existingPairs(v url.Values) []string { + keys := make([]string, 0, len(v)) + for k := range v { + keys = append(keys, k) + } + sort.Strings(keys) + pairs := make([]string, 0, len(v)) + for _, k := range keys { + for _, val := range v[k] { + pairs = append(pairs, url.QueryEscape(k)+"="+url.QueryEscape(val)) + } + } + return pairs +} + +// appendNaive appends query parameters to a URL that could not be parsed, using +// '?' or '&' based on what is already present. +func appendNaive(rawURL string, params map[string]string) string { + sep := "?" + if strings.Contains(rawURL, "?") { + sep = "&" + } + var b strings.Builder + b.WriteString(rawURL) + b.WriteString(sep) + keys := make([]string, 0, len(params)) + for k := range params { + keys = append(keys, k) + } + sort.Strings(keys) + first := true + for _, k := range keys { + if params[k] == "" { + continue + } + if !first { + b.WriteString("&") + } + first = false + b.WriteString(url.QueryEscape(k) + "=" + url.QueryEscape(params[k])) + } + return b.String() +} diff --git a/internal/channel/adapter/wsutil/url_test.go b/internal/channel/adapter/wsutil/url_test.go new file mode 100644 index 0000000..793076c --- /dev/null +++ b/internal/channel/adapter/wsutil/url_test.go @@ -0,0 +1,40 @@ +package wsutil + +import "testing" + +func TestAppendQuery_PreservesExisting(t *testing.T) { + t.Parallel() + got := AppendQuery("wss://gw.example.com/gateway?compress=0&token=abc", map[string]string{ + "resume": "1", "sn": "5", "session_id": "sess-123", + }) + want := "wss://gw.example.com/gateway?compress=0&token=abc&resume=1&session_id=sess-123&sn=5" + if got != want { + t.Errorf("AppendQuery = %q, want %q", got, want) + } +} + +func TestAppendQuery_NoExisting(t *testing.T) { + t.Parallel() + got := AppendQuery("wss://gw.example.com/gateway", map[string]string{"a": "1", "b": "2"}) + want := "wss://gw.example.com/gateway?a=1&b=2" + if got != want { + t.Errorf("AppendQuery = %q, want %q", got, want) + } +} + +func TestAppendQuery_OmitsEmptyValue(t *testing.T) { + t.Parallel() + got := AppendQuery("wss://gw.example.com/gateway?x=1", map[string]string{"sn": "", "resume": "1"}) + want := "wss://gw.example.com/gateway?x=1&resume=1" + if got != want { + t.Errorf("AppendQuery = %q, want %q", got, want) + } +} + +func TestAppendQuery_EmptyParams(t *testing.T) { + t.Parallel() + got := AppendQuery("wss://gw.example.com/gateway?x=1", nil) + if got != "wss://gw.example.com/gateway?x=1" { + t.Errorf("AppendQuery nil = %q, want unchanged", got) + } +} diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index 3d91f8b..4b56be1 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -287,7 +287,7 @@ func (b *Bridge) snapshot() []adapter.PlatformAdapter { func (b *Bridge) configured(a adapter.PlatformAdapter) bool { cfg := a.Config() switch basePlatform(a.Name()) { - case "telegram", "weixin-ilink": + case "telegram", "weixin-ilink", "kook": return cfgStr(cfg, "bot_token") != "" case "qq-obv11": if cfgStr(cfg, "mode") == "client" { diff --git a/plugins/channels/kook/adapter.go b/plugins/channels/kook/adapter.go new file mode 100644 index 0000000..57d7a05 --- /dev/null +++ b/plugins/channels/kook/adapter.go @@ -0,0 +1,298 @@ +package kook + +import ( + "context" + "fmt" + "io" + "log/slog" + "net/http" + "sort" + "sync" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter/httpclient" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter/wsutil" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +// Option customizes an Adapter at construction time. +type Option func(*Adapter) + +// WithHTTPClient injects the *http.Client used for REST calls. Tests pass the +// client of an httptest.Server. +func WithHTTPClient(c *http.Client) Option { + return func(a *Adapter) { a.client = c } +} + +// WithBaseURL overrides the KOOK REST API root (default https://www.kookapp.cn/api/v3). +// Tests point it at an httptest.Server URL. +func WithBaseURL(base string) Option { + return func(a *Adapter) { + if base != "" { + a.baseURL = base + } + } +} + +// WithLogger injects the logger used by the adapter. When nil the adapter logs +// through a discard logger so a bare adapter stays silent. +func WithLogger(l *slog.Logger) Option { + return func(a *Adapter) { + if l == nil { + l = logger.New(logger.Options{Out: io.Discard}) + } + a.log = logger.WithSource(l, "plugins/channels/kook") + } +} + +// Adapter is the KOOK (KaiHeiLa) implementation of adapter.PlatformAdapter. It +// is safe for concurrent use: the gateway client runs in its own goroutine while +// REST sends and inbound dispatch use their own mutexes. +type Adapter struct { + cfg Config + baseURL string + client *http.Client + http *httpclient.Client + log *slog.Logger + + wsc *wsutil.Client + wsMu sync.Mutex + mu sync.RWMutex + + connected bool + closed bool + cancel context.CancelFunc + + handlerMu sync.RWMutex + handler adapter.MessageHandler + emitMu sync.RWMutex + emit func(adapter.UnifiedMessage) + + sessMu sync.RWMutex + sessionID string + lastSN int64 + resumeOK bool + + chatsMu sync.RWMutex + chats map[string]adapter.ChannelInfo +} + +// Compile-time assertion that Adapter satisfies the shared contract. +var _ adapter.PlatformAdapter = (*Adapter)(nil) + +// New builds a KOOK adapter. It performs no network I/O; call Connect to +// establish the session. +func New(cfg Config, opts ...Option) *Adapter { + a := &Adapter{ + cfg: cfg, + client: &http.Client{Timeout: defaultHTTPTimeout}, + baseURL: "", + log: logger.WithSource(logger.New(logger.Options{Out: io.Discard}), "plugins/channels/kook"), + chats: map[string]adapter.ChannelInfo{}, + } + for _, opt := range opts { + opt(a) + } + if a.baseURL == "" { + if cfg.BaseURL != "" { + a.baseURL = cfg.BaseURL + } else { + a.baseURL = defaultBaseURL + } + } + a.http = httpclient.New( + httpclient.WithBaseURL(a.baseURL), + httpclient.WithToken(cfg.BotToken), + httpclient.WithHTTPClient(a.client), + httpclient.WithLogger(a.log), + ) + return a +} + +// Name returns the platform identifier. +func (a *Adapter) Name() string { return platformName } + +// Config returns a snapshot of the adapter's configuration. The BotToken is the +// sensitive field; callers that persist this map must treat it as a secret. +func (a *Adapter) Config() map[string]any { + return map[string]any{ + "bot_token": a.cfg.BotToken, + "base_url": a.baseURL, + } +} + +// Connect establishes the session and starts the gateway client. It validates +// the token first via a gateway-address lookup (failing fast on a bad token), +// then launches the WS read loop in its own goroutine. It is idempotent. +func (a *Adapter) Connect(ctx context.Context) error { + a.mu.Lock() + if a.connected { + a.mu.Unlock() + return nil + } + if a.closed { + a.mu.Unlock() + return fmt.Errorf("kook: %w", adapter.ErrClosed) + } + if err := a.cfg.Validate(); err != nil { + a.mu.Unlock() + return err + } + a.mu.Unlock() + + if _, err := a.fetchGatewayURL(ctx); err != nil { + return fmt.Errorf("kook: connect: %w", err) + } + + lifeCtx, cancel := context.WithCancel(context.Background()) + a.mu.Lock() + a.cancel = cancel + a.connected = true + a.mu.Unlock() + + go a.runWS(lifeCtx) + a.log.Info("connected", "token_len", len(a.cfg.BotToken)) + return nil +} + +// Disconnect tears down the session, canceling the gateway client. It is +// idempotent. +func (a *Adapter) Disconnect(ctx context.Context) error { + a.mu.Lock() + if !a.connected { + a.mu.Unlock() + return nil + } + a.connected = false + cancel := a.cancel + a.cancel = nil + a.mu.Unlock() + + if cancel != nil { + cancel() + } + a.wsMu.Lock() + wsc := a.wsc + a.wsMu.Unlock() + if wsc != nil { + _ = wsc.Close() + } + a.log.Info("disconnected") + return nil +} + +// IsConnected reports whether the session is currently live. +func (a *Adapter) IsConnected() bool { + a.mu.RLock() + defer a.mu.RUnlock() + return a.connected +} + +// HealthCheck verifies the session is usable. The connection state is the +// liveness signal; KOOK has no cheap keepalive call used here. +func (a *Adapter) HealthCheck(ctx context.Context) error { + a.mu.RLock() + defer a.mu.RUnlock() + if a.closed { + return fmt.Errorf("kook: health check: %w", adapter.ErrClosed) + } + if !a.connected { + return fmt.Errorf("kook: health check: %w", adapter.ErrNotConnected) + } + return nil +} + +// ReadMessages registers emit as the pull-based sink and blocks until ctx is +// canceled. Inbound events arrive via the gateway read loop and are delivered to +// SetMessageHandler (push) when set, otherwise to emit. +func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMessage)) { + a.emitMu.Lock() + a.emit = emit + a.emitMu.Unlock() + defer func() { + a.emitMu.Lock() + a.emit = nil + a.emitMu.Unlock() + }() + <-ctx.Done() +} + +// SetMessageHandler installs the push-style inbound callback. A nil fn clears it. +func (a *Adapter) SetMessageHandler(fn adapter.MessageHandler) { + a.handlerMu.Lock() + a.handler = fn + a.handlerMu.Unlock() +} + +// Capabilities declares what the KOOK platform supports. Proactive push is +// enabled (KOOK permits arbitrary private/channel sends); markdown is sent as +// kmarkdown (type 9). Media reply is not advertised because KOOK only accepts +// bot-uploaded media URLs. +func (a *Adapter) Capabilities() adapter.Capabilities { + return adapter.Capabilities{ + MaxMessageLength: maxMessageLen, + Markdown: true, + Image: false, + Voice: false, + Reaction: false, + TypingHint: false, + RecommendBurstMax: 1, + Proactive: true, + }.Normalize() +} + +// ListChannels returns the conversations observed so far, ordered by ChannelID. +func (a *Adapter) ListChannels(ctx context.Context) ([]adapter.ChannelInfo, error) { + a.chatsMu.RLock() + defer a.chatsMu.RUnlock() + out := make([]adapter.ChannelInfo, 0, len(a.chats)) + for _, c := range a.chats { + out = append(out, c) + } + sort.Slice(out, func(i, j int) bool { return out[i].ChannelID < out[j].ChannelID }) + return out, nil +} + +// requireConnected guards operations that need a live session. +func (a *Adapter) requireConnected() error { + a.mu.RLock() + defer a.mu.RUnlock() + if a.closed { + return fmt.Errorf("kook: %w", adapter.ErrClosed) + } + if !a.connected { + return fmt.Errorf("kook: %w", adapter.ErrNotConnected) + } + return nil +} + +// rememberChannel caches a conversation so ListChannels can report it. +func (a *Adapter) rememberChannel(msg adapter.UnifiedMessage) { + if msg.ChannelID == "" { + return + } + info := adapter.ChannelInfo{ + ChannelID: msg.ChannelID, + ChannelType: msg.ChannelType, + GroupID: msg.GroupID, + GroupName: msg.GroupName, + ChannelName: msg.GroupName, + } + if msg.ChannelType == adapter.ChannelGroup { + if info.ChannelName == "" { + info.ChannelName = info.ChannelID + } + if info.GroupName == "" { + info.GroupName = info.GroupID + } + } else { + // A private conversation is the peer user; show their name. + info.ChannelName = msg.SenderName + if info.ChannelName == "" { + info.ChannelName = info.ChannelID + } + } + a.chatsMu.Lock() + a.chats[info.ChannelID] = info + a.chatsMu.Unlock() +} diff --git a/plugins/channels/kook/config.go b/plugins/channels/kook/config.go new file mode 100644 index 0000000..fbe9a62 --- /dev/null +++ b/plugins/channels/kook/config.go @@ -0,0 +1,55 @@ +// Package kook implements the built-in KOOK (KaiHeiLa) channel adapter for +// Cyrene v2. KOOK is a stateful WebSocket-gateway channel: inbound events are +// delivered over a gateway WebSocket (hello/heartbeat/event/resume signaling), +// while outbound messages are sent over a bearer-authenticated REST API. +// +// The adapter conforms to internal/channel/adapter.PlatformAdapter, translating +// native KOOK events into the normalized adapter.UnifiedMessage model and +// delivering replies via /api/v3/message/create (channels) and +// /api/v3/direct-message/create (private chat). Identity is carried as channel +// actor metadata (author id + nickname) only — owner/member resolution is the +// channel layer's job, never a prompt-text heuristic. +// +// The design authority is DevDocs/design/08-渠道接入.md §4 (kook) and §7 +// (framework reusability). It depends on internal/channel/adapter (including the +// shared wsutil and httpclient primitives), internal/logger, and the standard +// library. The bot token is never logged. +package kook + +import ( + "strings" + "time" +) + +// Defaults and knobs for the KOOK integration. +const ( + // defaultBaseURL is the KOOK REST API root. + defaultBaseURL = "https://www.kookapp.cn/api/v3" + // defaultHTTPTimeout bounds a single REST request. + defaultHTTPTimeout = 30 * time.Second + // gatewayPath is the gateway-address fetch endpoint relative to baseURL. + gatewayPath = "/gateway/index" + // maxMessageLen is KOOK's documented per-message content limit. + maxMessageLen = 8000 + // heartbeatInterval is how often the client sends a ping (KOOK: 30s). + heartbeatInterval = 30 * time.Second + // helloTimeout bounds the wait for the handshake (hello/resume ack). + helloTimeout = 6 * time.Second +) + +// Config holds the settings required to reach the KOOK API. +type Config struct { + // BotToken is the secret bot token issued by the KOOK developer center. It is + // surfaced through Config() but never written to logs. + BotToken string `json:"bot_token"` + // BaseURL is the KOOK REST API root; empty uses the production default. + BaseURL string `json:"base_url"` +} + +// Validate checks the configuration for values needed to connect. +func (c Config) Validate() error { + if strings.TrimSpace(c.BotToken) == "" { + return ErrMissingToken + } + return nil +} diff --git a/plugins/channels/kook/config_test.go b/plugins/channels/kook/config_test.go new file mode 100644 index 0000000..22e7aad --- /dev/null +++ b/plugins/channels/kook/config_test.go @@ -0,0 +1,115 @@ +package kook + +import ( + "context" + "errors" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestName(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + if got := a.Name(); got != "kook" { + t.Fatalf("Name() = %q, want kook", got) + } +} + +func TestConfig(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "abc123", BaseURL: "https://example.com/api/v3"}) + got := a.Config() + if got["bot_token"] != "abc123" { + t.Fatalf("bot_token = %v, want abc123", got["bot_token"]) + } + if got["base_url"] != "https://example.com/api/v3" { + t.Fatalf("base_url = %v, want https://example.com/api/v3", got["base_url"]) + } + var round Config + if err := adapter.DecodeConfig(got, &round); err != nil { + t.Fatalf("DecodeConfig: %v", err) + } + if round.BotToken != "abc123" || round.BaseURL != "https://example.com/api/v3" { + t.Fatalf("round trip mismatch: %+v", round) + } +} + +func TestValidate(t *testing.T) { + t.Parallel() + if got := (Config{BotToken: "t"}).Validate(); got != nil { + t.Fatalf("Validate with token = %v, want nil", got) + } + if err := (Config{}).Validate(); !errors.Is(err, ErrMissingToken) { + t.Fatalf("Validate blank = %v, want ErrMissingToken", err) + } + if err := (Config{BotToken: " "}).Validate(); !errors.Is(err, ErrMissingToken) { + t.Fatalf("Validate whitespace = %v, want ErrMissingToken", err) + } +} + +func TestCapabilities(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + caps := a.Capabilities() + if !caps.Proactive { + t.Error("Proactive should be true for kook") + } + if !caps.Markdown { + t.Error("kook should support kmarkdown") + } + if caps.Image || caps.Voice || caps.Reaction || caps.TypingHint { + t.Errorf("expected Image/Voice/Reaction/TypingHint unsupported, got %+v", caps) + } + if caps.MaxMessageLength != maxMessageLen { + t.Errorf("MaxMessageLength = %d, want %d", caps.MaxMessageLength, maxMessageLen) + } + if caps.RecommendBurstMax <= 0 { + t.Errorf("RecommendBurstMax should be positive, got %d", caps.RecommendBurstMax) + } +} + +func TestListChannels_Empty(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + got, err := a.ListChannels(context.Background()) + if err != nil { + t.Fatalf("ListChannels: %v", err) + } + if got == nil || len(got) != 0 { + t.Fatalf("expected empty non-nil slice, got %#v", got) + } +} + +func TestConnect_MissingToken(t *testing.T) { + t.Parallel() + a := New(Config{}) + err := a.Connect(context.Background()) + if !errors.Is(err, ErrMissingToken) { + t.Fatalf("err = %v, want ErrMissingToken", err) + } + if a.IsConnected() { + t.Fatal("should not be connected") + } +} + +func TestHealthCheck_NotConnected(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + err := a.HealthCheck(context.Background()) + if !errors.Is(err, adapter.ErrNotConnected) { + t.Fatalf("err = %v, want ErrNotConnected", err) + } +} + +func TestSendMessage_NotConnected(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + err := a.SendMessage(context.Background(), adapter.UnifiedResponse{}) + if !errors.Is(err, adapter.ErrNotConnected) { + t.Fatalf("err = %v, want ErrNotConnected", err) + } + if err := a.SendProactive(context.Background(), adapter.UnifiedMessage{}); !errors.Is(err, adapter.ErrNotConnected) { + t.Fatalf("SendProactive err = %v, want ErrNotConnected", err) + } +} diff --git a/plugins/channels/kook/errors.go b/plugins/channels/kook/errors.go new file mode 100644 index 0000000..4c920e0 --- /dev/null +++ b/plugins/channels/kook/errors.go @@ -0,0 +1,17 @@ +package kook + +import "errors" + +// Sentinel errors returned by the KOOK adapter. Consumers test for them with +// errors.Is; failures that mean "not connected" wrap adapter.ErrNotConnected so +// the shared channel sentinel still matches. +var ( + // ErrMissingToken is returned when no BotToken is configured. + ErrMissingToken = errors.New("kook: missing bot token") + // ErrBadResponse is returned when the KOOK API replies with a non-zero code, + // an unexpected payload shape, or no usable gateway URL. + ErrBadResponse = errors.New("kook: bad api response") + // ErrNoChatTarget is returned when an outbound message carries no usable + // channel or user identifier. + ErrNoChatTarget = errors.New("kook: message has no target") +) diff --git a/plugins/channels/kook/gateway.go b/plugins/channels/kook/gateway.go new file mode 100644 index 0000000..9389390 --- /dev/null +++ b/plugins/channels/kook/gateway.go @@ -0,0 +1,31 @@ +package kook + +import ( + "context" + "encoding/json" + "fmt" + "net/url" +) + +// gatewayResult is the "data" payload of GET /gateway/index. +type gatewayResult struct { + URL string `json:"url"` +} + +// fetchGatewayURL retrieves the gateway WebSocket address from the KOOK REST +// API. compress=0 requests uncompressed delivery so every server->client frame is +// plain JSON text, matching wsutil's reader. +func (a *Adapter) fetchGatewayURL(ctx context.Context) (string, error) { + data, err := a.http.GetJSON(ctx, gatewayPath, url.Values{"compress": {"0"}}) + if err != nil { + return "", fmt.Errorf("kook: fetch gateway: %w", err) + } + var res gatewayResult + if err := json.Unmarshal(data, &res); err != nil { + return "", fmt.Errorf("kook: decode gateway result: %w", err) + } + if res.URL == "" { + return "", fmt.Errorf("kook: gateway url empty: %w", ErrBadResponse) + } + return res.URL, nil +} diff --git a/plugins/channels/kook/helpers_test.go b/plugins/channels/kook/helpers_test.go new file mode 100644 index 0000000..1edccb7 --- /dev/null +++ b/plugins/channels/kook/helpers_test.go @@ -0,0 +1,57 @@ +package kook + +import ( + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +// discardLogger returns a silent, source-tagged logger for tests. +func discardLogger() *slog.Logger { + return logger.WithSource(logger.New(logger.Options{Out: io.Discard}), "plugins/channels/kook") +} + +// newTestAdapter wires a KOOK adapter to an httptest server whose handler +// answers the REST API (and optionally upgrades a WebSocket). The server is +// closed on test cleanup. +func newTestAdapter(t *testing.T, h http.Handler) (*Adapter, *httptest.Server) { + t.Helper() + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + a := New( + Config{BotToken: "testtoken"}, + WithBaseURL(srv.URL), + WithHTTPClient(srv.Client()), + WithLogger(discardLogger()), + ) + return a, srv +} + +// decodeBody unmarshals an incoming JSON request body into a map for assertions. +func decodeBody(r *http.Request) map[string]any { + var m map[string]any + if err := json.NewDecoder(r.Body).Decode(&m); err != nil { + return nil + } + return m +} + +// kookOK writes a KOOK success envelope with the given data payload. +func kookOK(w http.ResponseWriter, data any) { + _ = json.NewEncoder(w).Encode(map[string]any{"code": 0, "message": "ok", "data": data}) +} + +// kookFail writes a KOOK error envelope with the given code/message. +func kookFail(w http.ResponseWriter, code int, msg string) { + _ = json.NewEncoder(w).Encode(map[string]any{"code": code, "message": msg, "data": struct{}{}}) +} + +// wsUpgrader upgrades an HTTP request to a WebSocket for gateway tests. +var wsUpgrader = websocket.Upgrader{} diff --git a/plugins/channels/kook/message.go b/plugins/channels/kook/message.go new file mode 100644 index 0000000..b95229e --- /dev/null +++ b/plugins/channels/kook/message.go @@ -0,0 +1,202 @@ +package kook + +// This file carries the KOOK protocol types and the conversion between a native +// s=0 event payload and the normalized adapter.UnifiedMessage model. KOOK events +// are delivered over the gateway WebSocket: channel_type distinguishes group +// (GROUP/BROADCAST) from private (PERSON), type declares the content kind +// (1 text / 2 image / 3 video / 4 file / 8 audio / 9 kmarkdown / 10 card / +// 255 system), author_id==1 marks a system sender, and extra.author carries the +// sender's user object (id/username/nickname). + +import ( + "encoding/json" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +const platformName = "kook" + +// KOOK message content kinds (the "type" field of an s=0 event). +const ( + kindText = 1 + kindImage = 2 + kindVideo = 3 + kindFile = 4 + kindAudio = 8 + kindKMarkdown = 9 + kindCard = 10 + kindSystem = 255 +) + +// KOOK channel types (the "channel_type" field of an s=0 event). +const ( + channelGroup = "GROUP" + channelBroadcast = "BROADCAST" +) + +// kookEvent is the "d" payload of an s=0 message event. +type kookEvent struct { + ChannelType string `json:"channel_type"` + Type int `json:"type"` + TargetID string `json:"target_id"` + AuthorID string `json:"author_id"` + Content json.RawMessage `json:"content"` + MsgID string `json:"msg_id"` + MsgTimestamp int64 `json:"msg_timestamp"` + Nonce string `json:"nonce"` + Extra json.RawMessage `json:"extra"` +} + +// kookExtra carries the per-event auxiliary data (author, guild/channel names, +// mentions). KOOK's field set varies by message type, so all fields are optional. +type kookExtra struct { + Type int `json:"type"` + GuildID string `json:"guild_id"` + ChannelName string `json:"channel_name"` + Mention []string `json:"mention"` + Author *kookUser `json:"author"` +} + +// kookUser is the sender's user object carried by extra.author. +type kookUser struct { + ID string `json:"id"` + Username string `json:"username"` + Nickname string `json:"nickname"` +} + +// toUnifiedMessage converts a parsed s=0 event payload into a normalized +// UnifiedMessage. It returns ok=false for events that carry no deliverable chat +// message (system messages, non-chat notifications). +func (a *Adapter) toUnifiedMessage(data json.RawMessage, frame []byte) (adapter.UnifiedMessage, bool) { + var ev kookEvent + if err := json.Unmarshal(data, &ev); err != nil { + return adapter.UnifiedMessage{}, false + } + if ev.Type == kindSystem || ev.AuthorID == "1" { + return adapter.UnifiedMessage{}, false + } + + content := asString(ev.Content) + var ex kookExtra + _ = json.Unmarshal(ev.Extra, &ex) + + senderID, senderName := senderOf(&ev, &ex) + ct, media := messageKind(ev.Type, content) + group := channelIsGroup(ev.ChannelType) + + msg := adapter.UnifiedMessage{ + Platform: platformName, + ChannelType: group, + ChannelID: ev.TargetID, + SenderID: senderID, + SenderName: senderName, + Content: content, + ContentType: ct, + Media: media, + Mentions: mentionsFrom(ex.Mention), + MessageID: ev.MsgID, + Timestamp: time.UnixMilli(ev.MsgTimestamp), + RawData: frame, + Sender: adapter.Actor{ + Role: adapter.RoleUnknown, + UserID: senderID, + DisplayName: senderName, + GroupID: "", + GroupName: ex.ChannelName, + IsGroup: bool(group == adapter.ChannelGroup), + }, + } + + if group == adapter.ChannelGroup { + gid := ex.GuildID + if gid == "" { + gid = ev.TargetID + } + msg.GroupID = gid + msg.GroupName = ex.ChannelName + msg.Sender.GroupID = gid + msg.Sender.GroupName = ex.ChannelName + a.rememberChannel(msg) + } + + // Adjust ChannelID: for private messages target_id is the peer user id, which + // is already set above; for group messages target_id is the channel id. + return msg, true +} + +// senderOf resolves the sender id and display name from the event fields, +// preferring the extra.author nickname (design/16 §4.6: third parties must be +// shown by nickname) with fallbacks to username and the author_id. +func senderOf(ev *kookEvent, ex *kookExtra) (id, name string) { + id = ev.AuthorID + name = ev.AuthorID + if ex.Author == nil { + return id, name + } + if ex.Author.ID != "" { + id = ex.Author.ID + } + switch { + case ex.Author.Nickname != "": + name = ex.Author.Nickname + case ex.Author.Username != "": + name = ex.Author.Username + default: + name = id + } + return id, name +} + +// channelIsGroup maps a KOOK channel_type onto the adapter's ChannelType. +func channelIsGroup(t string) adapter.ChannelType { + if t == channelGroup || t == channelBroadcast { + return adapter.ChannelGroup + } + return adapter.ChannelDirect +} + +// messageKind maps a KOOK content-kind onto the adapter ContentType and the +// media list (media URL is carried by the content field for image/video/file/ +// audio messages). +func messageKind(kind int, content string) (adapter.ContentType, []adapter.Media) { + switch kind { + case kindImage: + return adapter.ContentImage, []adapter.Media{{Kind: adapter.MediaImage, URL: content}} + case kindVideo: + return adapter.ContentFile, []adapter.Media{{Kind: adapter.MediaVideo, URL: content}} + case kindFile: + return adapter.ContentFile, []adapter.Media{{Kind: adapter.MediaFile, URL: content}} + case kindAudio: + return adapter.ContentVoice, []adapter.Media{{Kind: adapter.MediaVoice, URL: content}} + default: + return adapter.ContentText, nil + } +} + +// mentionsFrom converts an array of mentioned user ids into adapter.Mention. +func mentionsFrom(ids []string) []adapter.Mention { + if len(ids) == 0 { + return nil + } + out := make([]adapter.Mention, 0, len(ids)) + for _, id := range ids { + if id != "" { + out = append(out, adapter.Mention{UserID: id}) + } + } + return out +} + +// asString decodes a raw JSON value that is expected to be a string, returning +// "" when it is not. +func asString(raw json.RawMessage) string { + if len(raw) == 0 { + return "" + } + var s string + if err := json.Unmarshal(raw, &s); err != nil { + return "" + } + return s +} diff --git a/plugins/channels/kook/message_test.go b/plugins/channels/kook/message_test.go new file mode 100644 index 0000000..57be1c4 --- /dev/null +++ b/plugins/channels/kook/message_test.go @@ -0,0 +1,162 @@ +package kook + +import ( + "encoding/json" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestToUnified_GroupText(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + data := json.RawMessage(`{ + "channel_type":"GROUP","type":1,"target_id":"chan-1","author_id":"u-1", + "content":"hello world","msg_id":"m-1","msg_timestamp":1700000000000,"nonce":"", + "extra":{"type":1,"guild_id":"g-1","channel_name":"test","mention":[],"author":{"id":"u-1","username":"alice","nickname":"Alice"}} + }`) + msg, ok := a.toUnifiedMessage(data, []byte(`{"s":0,"d":{}}`)) + if !ok { + t.Fatal("expected a deliverable message") + } + if msg.Platform != "kook" { + t.Errorf("Platform = %q, want kook", msg.Platform) + } + if msg.ChannelType != adapter.ChannelGroup { + t.Errorf("ChannelType = %q, want group", msg.ChannelType) + } + if msg.ChannelID != "chan-1" { + t.Errorf("ChannelID = %q, want chan-1", msg.ChannelID) + } + if msg.GroupID != "g-1" { + t.Errorf("GroupID = %q, want g-1", msg.GroupID) + } + if msg.GroupName != "test" { + t.Errorf("GroupName = %q, want test", msg.GroupName) + } + if msg.SenderID != "u-1" || msg.SenderName != "Alice" { + t.Errorf("Sender = %q/%q, want u-1/Alice", msg.SenderID, msg.SenderName) + } + if msg.Sender.DisplayName != "Alice" || msg.Sender.UserID != "u-1" { + t.Errorf("Actor = %+v, want DisplayName Alice/UserID u-1", msg.Sender) + } + if msg.Content != "hello world" || msg.ContentType != adapter.ContentText { + t.Errorf("Content = %q/%q, want hello world/text", msg.Content, msg.ContentType) + } + if msg.MessageID != "m-1" || msg.Timestamp.UnixMilli() != 1700000000000 { + t.Errorf("MessageID/Timestamp = %q/%d", msg.MessageID, msg.Timestamp.UnixMilli()) + } +} + +func TestToUnified_PrivateText(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + data := json.RawMessage(`{ + "channel_type":"PERSON","type":1,"target_id":"user-9","author_id":"user-9", + "content":"hi","msg_id":"m-2","msg_timestamp":1700000000000,"nonce":"", + "extra":{"type":1,"author":{"id":"user-9","username":"bob","nickname":"Bob"}} + }`) + msg, ok := a.toUnifiedMessage(data, nil) + if !ok { + t.Fatal("expected a deliverable message") + } + if msg.ChannelType != adapter.ChannelDirect { + t.Errorf("ChannelType = %q, want direct", msg.ChannelType) + } + if msg.ChannelID != "user-9" { + t.Errorf("ChannelID = %q, want user-9 (peer id)", msg.ChannelID) + } + if msg.GroupID != "" { + t.Errorf("GroupID should be empty for private, got %q", msg.GroupID) + } + if msg.SenderName != "Bob" { + t.Errorf("SenderName = %q, want Bob", msg.SenderName) + } +} + +func TestToUnified_KMarkdown(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + data := json.RawMessage(`{ + "channel_type":"GROUP","type":9,"target_id":"chan-2","author_id":"u-2", + "content":"*bold*","msg_id":"m-3","msg_timestamp":1700000000000,"nonce":"", + "extra":{"type":9,"guild_id":"g-2","channel_name":"c","author":{"id":"u-2","username":"u","nickname":"N"}} + }`) + msg, ok := a.toUnifiedMessage(data, nil) + if !ok { + t.Fatal("expected a deliverable message") + } + if msg.ContentType != adapter.ContentText || msg.Content != "*bold*" { + t.Errorf("kmarkdown should map to text, got %q/%q", msg.Content, msg.ContentType) + } + if msg.GroupID != "g-2" { + t.Errorf("GroupID = %q, want g-2", msg.GroupID) + } +} + +func TestToUnified_Image(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + data := json.RawMessage(`{ + "channel_type":"GROUP","type":2,"target_id":"chan-3","author_id":"u-3", + "content":"https://img/photo.jpg","msg_id":"m-4","msg_timestamp":1700000000000,"nonce":"", + "extra":{"type":2,"guild_id":"g-3","author":{"id":"u-3","username":"u","nickname":"N"}} + }`) + msg, ok := a.toUnifiedMessage(data, nil) + if !ok { + t.Fatal("expected a deliverable message") + } + if msg.ContentType != adapter.ContentImage { + t.Errorf("ContentType = %q, want image", msg.ContentType) + } + if len(msg.Media) != 1 || msg.Media[0].Kind != adapter.MediaImage || msg.Media[0].URL != "https://img/photo.jpg" { + t.Errorf("Media = %+v, want image URL https://img/photo.jpg", msg.Media) + } +} + +func TestToUnified_VideoMapsToFile(t *testing.T) { + t.Parallel() + data := json.RawMessage(`{ + "channel_type":"GROUP","type":3,"target_id":"c","author_id":"u", + "content":"https://img/a.mp4","msg_id":"m","msg_timestamp":1700000000000,"nonce":"", + "extra":{"type":3,"guild_id":"g","author":{"id":"u","username":"u","nickname":"N"}} + }`) + a := New(Config{BotToken: "x"}) + msg, ok := a.toUnifiedMessage(data, nil) + if !ok { + t.Fatal("expected a deliverable message") + } + if len(msg.Media) != 1 || msg.Media[0].Kind != adapter.MediaVideo { + t.Fatalf("Media = %+v, want video", msg.Media) + } +} + +func TestToUnified_SystemSkipped(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + // type 255 system message → not a chat deliverable. + if _, ok := a.toUnifiedMessage(json.RawMessage(`{"channel_type":"GROUP","type":255,"target_id":"c","author_id":"1","content":"{}","msg_id":"m","msg_timestamp":1,"extra":{}}`), nil); ok { + t.Error("system message should be skipped") + } + // author_id 1 = system → skipped even for a normal type. + if _, ok := a.toUnifiedMessage(json.RawMessage(`{"channel_type":"GROUP","type":1,"target_id":"c","author_id":"1","content":"x","msg_id":"m","msg_timestamp":1,"extra":{}}`), nil); ok { + t.Error("author_id 1 (system) should be skipped") + } +} + +func TestMentions(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + data := json.RawMessage(`{ + "channel_type":"GROUP","type":1,"target_id":"c","author_id":"u", + "content":"hello","msg_id":"m","msg_timestamp":1700000000000,"nonce":"", + "extra":{"type":1,"guild_id":"g","mention":["u-1","u-2"],"author":{"id":"u","username":"u","nickname":"N"}} + }`) + msg, ok := a.toUnifiedMessage(data, nil) + if !ok { + t.Fatal("expected a deliverable message") + } + if len(msg.Mentions) != 2 || msg.Mentions[0].UserID != "u-1" || msg.Mentions[1].UserID != "u-2" { + t.Errorf("Mentions = %+v, want u-1/u-2", msg.Mentions) + } +} diff --git a/plugins/channels/kook/send.go b/plugins/channels/kook/send.go new file mode 100644 index 0000000..a046552 --- /dev/null +++ b/plugins/channels/kook/send.go @@ -0,0 +1,119 @@ +package kook + +import ( + "context" + "fmt" + "net/http" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// REST endpoints for sending KOOK messages. +const ( + channelMessagePath = "/message/create" // channels (GROUP/BROADCAST) + privateMessagePath = "/direct-message/create" // private chat +) + +// kookPostRequest is the shared body of the two KOOK send endpoints. +type kookPostRequest struct { + Type int `json:"type"` + TargetID string `json:"target_id"` + Content string `json:"content"` + Quote string `json:"quote,omitempty"` +} + +// SendMessage delivers the reply messages in resp. Each message goes to its own +// channel (GROUP) or private conversation (PERSON/BROADCAST). +func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) error { + if err := a.requireConnected(); err != nil { + return err + } + for i, m := range resp.Messages { + if err := a.sendOne(ctx, m, resp.ReplyTo); err != nil { + return fmt.Errorf("kook: send message #%d: %w", i, err) + } + } + return nil +} + +// SendProactive pushes a message not triggered by a recent inbound. KOOK allows +// arbitrary proactive push, so this is only gated by the connection state. +func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error { + if err := a.requireConnected(); err != nil { + return err + } + return a.sendOne(ctx, msg, "") +} + +// sendOne delivers a single outbound message to its conversation, choosing the +// send endpoint that matches the channel kind and the message type that matches +// the content kind. +func (a *Adapter) sendOne(ctx context.Context, m adapter.UnifiedMessage, replyTo string) error { + group := m.ChannelType == adapter.ChannelGroup + target := m.ChannelID + if group { + if target == "" { + target = m.GroupID + } + } else { + if target == "" { + target = m.SenderID + } + } + if target == "" { + return fmt.Errorf("kook: %w", ErrNoChatTarget) + } + + msgType, content := outboundType(m) + req := kookPostRequest{Type: msgType, TargetID: target, Content: content} + if replyTo != "" { + req.Quote = replyTo + } + + path := channelMessagePath + if !group { + path = privateMessagePath + } + if _, err := a.http.DoJSON(ctx, http.MethodPost, path, req); err != nil { + return err + } + a.log.Debug("sent", "type", msgType, "target", target, "group", group) + return nil +} + +// outboundType maps a unified message to a KOOK message type and content. Text +// falls back to kmarkdown (type 9) because the adapter declares the Markdown +// capability; media carries the first media URL in the content field. +func outboundType(m adapter.UnifiedMessage) (int, string) { + switch m.ContentType { + case adapter.ContentImage: + return kindImage, mediaURL(m.Media) + case adapter.ContentVoice: + return kindAudio, mediaURL(m.Media) + case adapter.ContentFile: + if mediaKind(m.Media) == adapter.MediaVideo { + return kindVideo, mediaURL(m.Media) + } + return kindFile, mediaURL(m.Media) + default: + return kindKMarkdown, m.Content + } +} + +// mediaURL returns the first non-empty media URL. +func mediaURL(media []adapter.Media) string { + for _, m := range media { + if m.URL != "" { + return m.URL + } + } + return "" +} + +// mediaKind returns the kind of the first media entry. +func mediaKind(media []adapter.Media) adapter.MediaKind { + for _, m := range media { + return m.Kind + } + return "" +} diff --git a/plugins/channels/kook/send_test.go b/plugins/channels/kook/send_test.go new file mode 100644 index 0000000..351947e --- /dev/null +++ b/plugins/channels/kook/send_test.go @@ -0,0 +1,178 @@ +package kook + +import ( + "context" + "net/http" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestSendMessage_GroupText(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var body map[string]any + var auth string + h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != channelMessagePath { + http.Error(w, "unexpected path: "+r.URL.Path, http.StatusNotFound) + return + } + mu.Lock() + auth = r.Header.Get("Authorization") + body = decodeBody(r) + mu.Unlock() + kookOK(w, map[string]any{"msg_id": "x-1", "msg_timestamp": 1}) + }) + a, _ := newTestAdapter(t, h) + a.connected = true + resp := adapter.UnifiedResponse{ + ReplyTo: "reply-99", + Messages: []adapter.UnifiedMessage{ + {ChannelID: "chan-1", ChannelType: adapter.ChannelGroup, Content: "hello", ContentType: adapter.ContentText}, + }, + } + if err := a.SendMessage(context.Background(), resp); err != nil { + t.Fatalf("SendMessage: %v", err) + } + mu.Lock() + defer mu.Unlock() + if auth != "Bearer testtoken" { + t.Errorf("Authorization = %q, want Bearer testtoken", auth) + } + if body["target_id"] != "chan-1" { + t.Errorf("target_id = %v, want chan-1", body["target_id"]) + } + if body["type"] != float64(9) { + t.Errorf("type = %v, want 9 (kmarkdown)", body["type"]) + } + if body["content"] != "hello" { + t.Errorf("content = %v, want hello", body["content"]) + } + if body["quote"] != "reply-99" { + t.Errorf("quote = %v, want reply-99", body["quote"]) + } +} + +func TestSendMessage_PrivateText(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var body map[string]any + h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != privateMessagePath { + http.Error(w, "unexpected path: "+r.URL.Path, http.StatusNotFound) + return + } + mu.Lock() + body = decodeBody(r) + mu.Unlock() + kookOK(w, map[string]any{"msg_id": "x-2", "msg_timestamp": 1}) + }) + a, _ := newTestAdapter(t, h) + a.connected = true + resp := adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{ + {ChannelID: "user-9", ChannelType: adapter.ChannelDirect, Content: "private hi", ContentType: adapter.ContentText}, + }, + } + if err := a.SendMessage(context.Background(), resp); err != nil { + t.Fatalf("SendMessage: %v", err) + } + mu.Lock() + defer mu.Unlock() + if body["target_id"] != "user-9" || body["content"] != "private hi" { + t.Errorf("body = %v, want target_id user-9 content private hi", body) + } +} + +func TestSendMessage_MediaTypeMapping(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + ct adapter.ContentType + media adapter.Media + wantType int + }{ + {name: "image", ct: adapter.ContentImage, media: adapter.Media{Kind: adapter.MediaImage, URL: "https://img/a.jpg"}, wantType: 2}, + {name: "voice", ct: adapter.ContentVoice, media: adapter.Media{Kind: adapter.MediaVoice, URL: "https://img/a.mp3"}, wantType: 8}, + {name: "video", ct: adapter.ContentFile, media: adapter.Media{Kind: adapter.MediaVideo, URL: "https://img/a.mp4"}, wantType: 3}, + {name: "file", ct: adapter.ContentFile, media: adapter.Media{Kind: adapter.MediaFile, URL: "https://img/a.txt"}, wantType: 4}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var body map[string]any + h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != channelMessagePath { + http.Error(w, "unexpected path", http.StatusNotFound) + return + } + mu.Lock() + body = decodeBody(r) + mu.Unlock() + kookOK(w, map[string]any{"msg_id": "x", "msg_timestamp": 1}) + }) + a, _ := newTestAdapter(t, h) + a.connected = true + resp := adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{ + {ChannelID: "chan-1", ChannelType: adapter.ChannelGroup, ContentType: tc.ct, Media: []adapter.Media{tc.media}}, + }, + } + if err := a.SendMessage(context.Background(), resp); err != nil { + t.Fatalf("SendMessage: %v", err) + } + mu.Lock() + defer mu.Unlock() + if body["type"] != float64(tc.wantType) { + t.Errorf("type = %v, want %d", body["type"], tc.wantType) + } + if body["content"] != tc.media.URL { + t.Errorf("content = %v, want %s", body["content"], tc.media.URL) + } + }) + } +} + +func TestSendProactive_Private(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var body map[string]any + h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != privateMessagePath { + http.Error(w, "unexpected path", http.StatusNotFound) + return + } + mu.Lock() + body = decodeBody(r) + mu.Unlock() + kookOK(w, map[string]any{"msg_id": "x-3", "msg_timestamp": 1}) + }) + a, _ := newTestAdapter(t, h) + a.connected = true + if err := a.SendProactive(context.Background(), adapter.UnifiedMessage{ + ChannelID: "user-9", ChannelType: adapter.ChannelDirect, Content: "reminder", ContentType: adapter.ContentText, + }); err != nil { + t.Fatalf("SendProactive: %v", err) + } + mu.Lock() + defer mu.Unlock() + if body["target_id"] != "user-9" || body["content"] != "reminder" { + t.Errorf("body = %v, want target_id user-9 content reminder", body) + } +} + +func TestSendMessage_NoTarget(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "unexpected", http.StatusNotFound) + })) + a.connected = true + err := a.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{ContentType: adapter.ContentText, Content: "hi"}}, + }) + if err == nil { + t.Fatal("expected error for missing target") + } +} diff --git a/plugins/channels/kook/websocket.go b/plugins/channels/kook/websocket.go new file mode 100644 index 0000000..7623f8c --- /dev/null +++ b/plugins/channels/kook/websocket.go @@ -0,0 +1,263 @@ +package kook + +// This file wires the stateful KOOK gateway WebSocket protocol onto the shared +// wsutil client. KOOK is a signal-driven gateway: after connecting, the server +// sends s=1 (hello) or s=6 (resume ack); the client heartbeats every ~30s with +// s=2 and expects s=3; inbound events arrive as s=0; the server may force a +// restart with s=5 (reconnect). The reconnect loop, heartbeat scheduling, and +// read-loop skeleton are provided by wsutil; only the signaling lives here. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter/wsutil" +) + +// KOOK gateway signal numbers. +const ( + signalEvent = 0 + signalHello = 1 + signalPing = 2 + signalPong = 3 + signalResume = 4 + signalReconnect = 5 + signalResumeAck = 6 +) + +// kookSignal is the envelope of every gateway frame. +type kookSignal struct { + S int `json:"s"` + D json.RawMessage `json:"d"` + SN int64 `json:"sn"` +} + +// helloData is the "d" payload of an s=1 hello frame. +type helloData struct { + Code int `json:"code"` + SessionID string `json:"session_id"` +} + +// parseSignal decodes a gateway text frame. +func parseSignal(payload []byte) (kookSignal, error) { + var sig kookSignal + if err := json.Unmarshal(payload, &sig); err != nil { + return kookSignal{}, fmt.Errorf("kook: decode signal: %w", err) + } + return sig, nil +} + +// heartbeatFrame builds the s=2 ping payload carrying the current sn. +func heartbeatFrame(sn int64) ([]byte, error) { + return json.Marshal(map[string]any{"s": signalPing, "sn": sn}) +} + +// runWS starts the gateway client goroutine. It is idempotent: a second call +// while one is running is a no-op. The client tears itself down when ctx ends. +func (a *Adapter) runWS(ctx context.Context) { + a.wsMu.Lock() + if a.wsc != nil { + a.wsMu.Unlock() + return + } + client := wsutil.New( + a.gatewayURL, + wsutil.Handler{ + OnConnect: a.onConnect, + OnMessage: a.onMessage, + Heartbeat: a.heartbeat, + }, + wsutil.WithHeartbeat(heartbeatInterval), + wsutil.WithPongTimeout(helloTimeout), + wsutil.WithLogger(a.log), + ) + a.wsc = client + runCtx, cancel := context.WithCancel(ctx) + client.BindCancel(cancel) + a.wsMu.Unlock() + + _ = client.Run(runCtx) + + a.wsMu.Lock() + a.wsc = nil + a.wsMu.Unlock() +} + +// gatewayURL returns the gateway address to dial. On a first connect (or after a +// session reset) it returns a fresh URL; on a drop with an established session it +// appends the resume parameters so the gateway can replay missed messages. +func (a *Adapter) gatewayURL(ctx context.Context) (string, error) { + base, err := a.fetchGatewayURL(ctx) + if err != nil { + return "", err + } + if !a.canResume() { + return base, nil + } + return wsutil.AppendQuery(base, map[string]string{ + "resume": "1", + "sn": strconv.FormatInt(a.currentSN(), 10), + "session_id": a.currentSession(), + }), nil +} + +// onConnect performs the application handshake. It accepts either an s=1 hello +// (fresh or restarted session) or an s=6 resume ack (reconnected with an intact +// session) as the first frame within helloTimeout. +func (a *Adapter) onConnect(ctx context.Context, c *wsutil.Conn) error { + _ = c.SetReadDeadline(time.Now().Add(helloTimeout)) + payload, err := c.ReadMessage() + if err != nil { + return fmt.Errorf("kook: hello read: %w", err) + } + sig, err := parseSignal(payload) + if err != nil { + return err + } + switch sig.S { + case signalHello: + var hd helloData + if err := json.Unmarshal(sig.D, &hd); err != nil { + return fmt.Errorf("kook: hello decode: %w", err) + } + if hd.Code != 0 { + // Rejected (e.g. invalid/expired token or expired session): drop the + // whole session state so the next dial is fresh. + a.resetSession() + return fmt.Errorf("kook: hello rejected code=%d", hd.Code) + } + a.setSession(hd.SessionID, 0) + case signalResumeAck: + // Resume accepted; keep the existing session id and sn. + a.markResumeOK() + default: + return fmt.Errorf("kook: expected hello or resume ack, got s=%d", sig.S) + } + _ = c.SetReadDeadline(time.Time{}) + return nil +} + +// onMessage dispatches every inbound application frame. s=0 events are converted +// to UnifiedMessage and delivered; control signals update session state or force +// a reconnect. +func (a *Adapter) onMessage(ctx context.Context, payload []byte) error { + sig, err := parseSignal(payload) + if err != nil { + return err + } + switch sig.S { + case signalEvent: + if err := a.handleEvent(ctx, sig.D, payload); err != nil { + return err + } + a.advanceSN(sig.SN) + return nil + case signalPong, signalResumeAck: + return nil + case signalReconnect: + // Server invalidated the session: clear sn/session and re-dial fresh. + a.resetSession() + return errors.New("kook: gateway session invalid, reconnecting") + default: + return nil + } +} + +// heartbeat sends the s=2 ping carrying the current sn. +func (a *Adapter) heartbeat(ctx context.Context, c *wsutil.Conn) error { + data, err := heartbeatFrame(a.currentSN()) + if err != nil { + return err + } + return c.WriteMessage(data) +} + +// handleEvent converts a single s=0 event into a UnifiedMessage and dispatches +// it. Malformed or non-chat events are logged and skipped, never fatal. +func (a *Adapter) handleEvent(ctx context.Context, data json.RawMessage, frame []byte) error { + msg, ok := a.toUnifiedMessage(data, frame) + if !ok { + return nil + } + if err := msg.Validate(); err != nil { + a.log.Warn("invalid message dropped", "err", err) + return nil + } + a.dispatch(ctx, msg) + return nil +} + +// dispatch routes an inbound message to the push handler, falling back to the +// ReadMessages sink. +func (a *Adapter) dispatch(ctx context.Context, msg adapter.UnifiedMessage) { + a.handlerMu.RLock() + fn := a.handler + a.handlerMu.RUnlock() + if fn != nil { + fn(ctx, msg) + return + } + a.emitMu.RLock() + emit := a.emit + a.emitMu.RUnlock() + if emit != nil { + emit(msg) + } +} + +// setSession records a fresh session (new id, sn reset) and marks it resumable. +func (a *Adapter) setSession(sessionID string, sn int64) { + a.sessMu.Lock() + defer a.sessMu.Unlock() + a.sessionID = sessionID + a.lastSN = sn + a.resumeOK = sessionID != "" +} + +// markResumeOK marks the current session as resumable after a resume ack. +func (a *Adapter) markResumeOK() { + a.sessMu.Lock() + defer a.sessMu.Unlock() + a.resumeOK = a.sessionID != "" +} + +// resetSession clears all session state so the next connect is a fresh one. +func (a *Adapter) resetSession() { + a.sessMu.Lock() + defer a.sessMu.Unlock() + a.sessionID = "" + a.lastSN = 0 + a.resumeOK = false +} + +// advanceSN raises the tracked sn when a newer event is seen. +func (a *Adapter) advanceSN(sn int64) { + a.sessMu.Lock() + defer a.sessMu.Unlock() + if sn > a.lastSN { + a.lastSN = sn + } +} + +func (a *Adapter) currentSession() string { + a.sessMu.RLock() + defer a.sessMu.RUnlock() + return a.sessionID +} + +func (a *Adapter) currentSN() int64 { + a.sessMu.RLock() + defer a.sessMu.RUnlock() + return a.lastSN +} + +func (a *Adapter) canResume() bool { + a.sessMu.RLock() + defer a.sessMu.RUnlock() + return a.sessionID != "" && a.resumeOK +} diff --git a/plugins/channels/kook/websocket_test.go b/plugins/channels/kook/websocket_test.go new file mode 100644 index 0000000..3fe10f3 --- /dev/null +++ b/plugins/channels/kook/websocket_test.go @@ -0,0 +1,220 @@ +package kook + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestParseSignal(t *testing.T) { + t.Parallel() + s := []byte(`{"s":1,"d":{"code":0,"session_id":"sess-123"}}`) + sig, err := parseSignal(s) + if err != nil { + t.Fatalf("parseSignal: %v", err) + } + if sig.S != signalHello { + t.Errorf("S = %d, want %d (hello)", sig.S, signalHello) + } + if len(sig.D) == 0 { + t.Error("D should carry the hello payload") + } + + s0 := []byte(`{"s":0,"sn":7,"d":{"channel_type":"GROUP"}}`) + sig0, err := parseSignal(s0) + if err != nil { + t.Fatalf("parseSignal event: %v", err) + } + if sig0.S != signalEvent || sig0.SN != 7 { + t.Errorf("event S/SN = %d/%d, want 0/7", sig0.S, sig0.SN) + } + + if sig, err := parseSignal([]byte(`{"s":3}`)); err != nil || sig.S != signalPong { + t.Errorf("pong parse = %+v, %v", sig, err) + } +} + +func TestHeartbeatFrame(t *testing.T) { + t.Parallel() + b, err := heartbeatFrame(42) + if err != nil { + t.Fatalf("heartbeatFrame: %v", err) + } + if string(b) != `{"s":2,"sn":42}` { + t.Errorf("heartbeat frame = %s, want {\"s\":2,\"sn\":42}", string(b)) + } +} + +func TestGatewayURL_FreshNoResume(t *testing.T) { + t.Parallel() + h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != gatewayPath { + http.Error(w, "unexpected", http.StatusNotFound) + return + } + kookOK(w, map[string]any{"url": "wss://gw.example.com/gateway?compress=0&token=abc"}) + }) + a, _ := newTestAdapter(t, h) + got, err := a.gatewayURL(context.Background()) + if err != nil { + t.Fatalf("gatewayURL: %v", err) + } + if got != "wss://gw.example.com/gateway?compress=0&token=abc" { + t.Errorf("fresh url = %q, want unchanged base", got) + } +} + +func TestGatewayURL_ResumeAddsParams(t *testing.T) { + t.Parallel() + h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != gatewayPath { + http.Error(w, "unexpected", http.StatusNotFound) + return + } + kookOK(w, map[string]any{"url": "wss://gw.example.com/gateway?compress=0&token=abc"}) + }) + a, _ := newTestAdapter(t, h) + a.setSession("sess-abc", 5) + assertCanResume(t, a, true) + + got, err := a.gatewayURL(context.Background()) + if err != nil { + t.Fatalf("gatewayURL: %v", err) + } + u, err := url.Parse(got) + if err != nil { + t.Fatalf("parse resume url: %v", err) + } + q := u.Query() + if q.Get("compress") != "0" || q.Get("token") != "abc" { + t.Errorf("existing query params lost: %q", got) + } + if q.Get("resume") != "1" || q.Get("sn") != "5" || q.Get("session_id") != "sess-abc" { + t.Errorf("resume params = %q", got) + } +} + +func TestGatewayURL_ResetDropsResume(t *testing.T) { + t.Parallel() + h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + kookOK(w, map[string]any{"url": "wss://gw.example.com/gateway?compress=0"}) + }) + a, _ := newTestAdapter(t, h) + a.setSession("sess-abc", 5) + a.resetSession() + assertCanResume(t, a, false) + got, err := a.gatewayURL(context.Background()) + if err != nil { + t.Fatalf("gatewayURL: %v", err) + } + if strings.Contains(got, "resume=1") { + t.Errorf("reset session should not emit resume params: %q", got) + } +} + +func assertCanResume(t *testing.T, a *Adapter, want bool) { + t.Helper() + if got := a.canResume(); got != want { + t.Fatalf("canResume = %v, want %v", got, want) + } +} + +// TestConnect_GatewayDeliversEvent is an end-to-end smoke of the gateway path: an +// httptest server serves the REST gateway lookup, upgrades a WebSocket, sends a +// hello then an s=0 event, and the adapter dispatches the normalized message to +// the push handler. +func TestConnect_GatewayDeliversEvent(t *testing.T) { + var ( + mu sync.Mutex + delivered []adapter.UnifiedMessage + ) + got := make(chan adapter.UnifiedMessage, 1) + + var srv *httptest.Server + h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case gatewayPath: + wsURL := "ws://" + strings.TrimPrefix(srv.URL, "http://") + "/ws" + kookOK(w, map[string]any{"url": wsURL}) + case "/ws": + conn, err := wsUpgrader.Upgrade(w, r, nil) + if err != nil { + t.Errorf("ws upgrade: %v", err) + return + } + defer conn.Close() + _ = conn.WriteJSON(map[string]any{"s": 1, "d": map[string]any{"code": 0, "session_id": "sess-123"}}) + _ = conn.WriteJSON(map[string]any{"s": 0, "sn": 7, "d": map[string]any{ + "channel_type": "GROUP", "type": 1, "target_id": "chan-8", "author_id": "u-8", + "content": "gateway hi", "msg_id": "m-8", "msg_timestamp": 1700000000000, "nonce": "", + "extra": map[string]any{ + "guild_id": "g-8", "channel_name": "test", + "author": map[string]any{"id": "u-8", "username": "u", "nickname": "Nick"}, + }, + }}) + // Hold the connection and consume client frames (heartbeat/close). + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + default: + http.NotFound(w, r) + } + }) + srv = httptest.NewServer(h) + defer srv.Close() + + a := New( + Config{BotToken: "testtoken"}, + WithBaseURL(srv.URL), + WithHTTPClient(srv.Client()), + WithLogger(discardLogger()), + ) + a.SetMessageHandler(func(ctx context.Context, msg adapter.UnifiedMessage) { + mu.Lock() + delivered = append(delivered, msg) + mu.Unlock() + select { + case got <- msg: + default: + } + }) + + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + defer func() { _ = a.Disconnect(context.Background()) }() + + select { + case msg := <-got: + if msg.Platform != "kook" { + t.Errorf("Platform = %q, want kook", msg.Platform) + } + if msg.ChannelID != "chan-8" || msg.GroupID != "g-8" { + t.Errorf("ChannelID/GroupID = %q/%q, want chan-8/g-8", msg.ChannelID, msg.GroupID) + } + if msg.SenderName != "Nick" || msg.SenderID != "u-8" { + t.Errorf("Sender = %q/%q, want Nick/u-8", msg.SenderName, msg.SenderID) + } + if msg.Content != "gateway hi" || msg.ContentType != adapter.ContentText { + t.Errorf("Content = %q/%q, want gateway hi/text", msg.Content, msg.ContentType) + } + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for gateway event") + } + + if sn := a.currentSN(); sn != 7 { + t.Errorf("currentSN = %d, want 7", sn) + } + if sid := a.currentSession(); sid != "sess-123" { + t.Errorf("currentSession = %q, want sess-123", sid) + } +}