Cyrene_v2/plugins/channels/kook/adapter.go

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