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

	botMu       sync.RWMutex
	botID       string
	botUsername string
	botNickname string
}

// 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,
// WebhookVerifyToken and WebhookEncryptKey are sensitive; callers that persist
// this map must treat them as secrets (the admin read path masks these keys via
// the "token"/"key" substring rules).
func (a *Adapter) Config() map[string]any {
	id, username, nickname := a.botIdentitySnapshot()
	return map[string]any{
		"bot_token":            a.cfg.BotToken,
		"base_url":             a.baseURL,
		"receive_mode":         a.cfg.normalizedReceiveMode(),
		"webhook_path":         a.cfg.WebhookPath,
		"webhook_verify_token": a.cfg.WebhookVerifyToken,
		"webhook_encrypt_key":  a.cfg.WebhookEncryptKey,
		"bot_id":               id,
		"bot_username":         username,
		"bot_nickname":         nickname,
	}
}

// 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
	}
	mode := a.cfg.normalizedReceiveMode()
	a.mu.Unlock()

	// Webhook mode: the Cyrene gateway mounts InboundHandler to receive KOOK
	// callbacks (KOOK posts to the configured route). No gateway WebSocket is
	// dialed — webhook and websocket are mutually exclusive. The bot identity
	// lookup is best-effort (a bad token or network hiccup logs a warning and
	// leaves the identity empty, never aborting the connect), and Mark connected
	// so the REST reply path and HealthCheck see a live session.
	if mode == modeWebhook {
		a.refreshBotIdentity(ctx)
		a.mu.Lock()
		a.connected = true
		a.mu.Unlock()
		a.log.Info("connected", "mode", mode, "token_len", len(a.cfg.BotToken))
		return nil
	}

	if _, err := a.fetchGatewayURL(ctx); err != nil {
		return fmt.Errorf("kook: connect: %w", err)
	}

	// Best-effort bot identity (user/me) so the admin list shows who the bot is.
	// Failure here is non-fatal: the gateway URL above already validated the token.
	a.refreshBotIdentity(ctx)

	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", "mode", mode, "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()
}