Cyrene_v2/internal/services/channelbridge/channelbridge.go

// Package channelbridge wires the registered IM channel adapters into the Cyrene
// runtime: it connects each configured adapter, installs the inbound message
// handler, and routes every normalized inbound message through the
// channel→session→actor→loop→reply pipeline (DevDocs/design/08 §0-§2).
//
// The bridge is deliberately generic over adapter.PlatformAdapter. It owns the
// per-platform Connect lifecycle and the inbound→reply path; the concrete
// adapters themselves (and their configuration, sourced from env) are assembled
// in cmd/cyrene/wire.go and handed in via Add. It never reads message text to
// decide identity — that is ResolveActor's job (actor metadata + owner whitelist,
// a project red line).
package channelbridge

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"sync"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
)

// webhookAdapter is the optional capability a channel adapter may expose to
// source an HTTP inbound callback. *webhook.Adapter implements it via
// InboundHandler(). The bridge reports these so the gateway can mount them.
type webhookAdapter interface {
	// InboundHandler returns the http.Handler the gateway mounts at the adapter's
	// configured endpoint path to receive external callbacks.
	InboundHandler() http.Handler
	// Config returns the adapter config snapshot; the endpoint path is read from
	// it so the gateway can register the correct route.
	Config() map[string]any
}

// Knob constants. Compile-time so no bare magic number appears in the run path
// (project rule §10).
const (
	// connectTimeout bounds a single adapter Connect so a hung dial cannot stall
	// boot (a failed connect logs a warning and is skipped, never fatal).
	connectTimeout = 10 * time.Second
	// maxConcurrent bounds how many inbound messages run the agent loop at once,
	// so a burst of channel messages cannot spawn unbounded goroutines.
	maxConcurrent = 16
)

// ErrNilLoop is returned when the bridge is built without a chat runner.
var ErrNilLoop = errors.New("channelbridge: nil loop runner")

// ErrNilSessions is returned when the bridge is built without a session store.
var ErrNilSessions = errors.New("channelbridge: nil session store")

// Chatter runs a single agent turn. *loop.Loop is the production implementation;
// tests inject a stub.
type Chatter interface {
	Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error)
}

// SessionStore resolves the channel↔session binding for an inbound message.
// *session.SessionManager is the production implementation.
type SessionStore interface {
	GetOrCreate(ctx context.Context, msg adapter.UnifiedMessage, owner *user.OwnerProfile) (session.Session, error)
}

// Options carries the bridge's injected dependencies. Loop and Sessions are
// required; Owner may be nil (then no account is recognized as owner, the safe
// default-deny posture from design/11 §0). History and OnPersisted are optional
// (design/16 §4.4): when a history store is wired, owner-private messages are
// routed into the main web session and tagged with the channel platform, and every
// persisted message is handed to OnPersisted so the gateway can broadcast a
// `message_persisted` event; when both are absent the bridge runs exactly as
// before (isolated sess-xxx, no chat_messages writes).
type Options struct {
	Log         *slog.Logger
	Loop        Chatter
	Sessions    SessionStore
	Owner       *user.OwnerProfile
	History     history.HistoryStore
	OnPersisted func(history.Message)
}

// Bridge owns the channel adapter lifecycle and the inbound message pipeline. It
// is safe for concurrent use.
type Bridge struct {
	log         *slog.Logger
	run         Chatter
	sessions    SessionStore
	owner       *user.OwnerProfile
	history     history.HistoryStore
	onPersisted func(history.Message)

	mu        sync.RWMutex
	adapters  map[string]adapter.PlatformAdapter
	connected map[string]bool
	closed    bool

	ctx    context.Context
	cancel context.CancelFunc
	wg     sync.WaitGroup

	sem chan struct{}
}

// New builds a Bridge from opts. It returns an error when a required dependency
// is missing so the caller fails fast rather than silently dropping messages.
func New(opts Options) (*Bridge, error) {
	if opts.Loop == nil {
		return nil, ErrNilLoop
	}
	if opts.Sessions == nil {
		return nil, ErrNilSessions
	}
	log := opts.Log
	if log == nil {
		log = logger.New(logger.Options{Out: nil})
	}
	return &Bridge{
		log:         logger.WithSource(log, "services/channelbridge"),
		run:         opts.Loop,
		sessions:    opts.Sessions,
		owner:       opts.Owner,
		history:     opts.History,
		onPersisted: opts.OnPersisted,
		adapters:    make(map[string]adapter.PlatformAdapter),
		connected:   make(map[string]bool),
		cancel:      func() {},
		sem:         make(chan struct{}, maxConcurrent),
	}, nil
}

// Add registers an adapter by its Name. It is safe to call before Start; adding
// after Start has no effect on already-running adapters.
func (b *Bridge) Add(a adapter.PlatformAdapter) {
	if a == nil {
		return
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.adapters[a.Name()] = a
}

// WebhookHandlers returns the inbound HTTP handlers of every registered adapter
// that exposes an HTTP callback (e.g. *webhook.Adapter), keyed by the endpoint
// path from the adapter's config. The gateway mounts these to receive external
// webhook callbacks into the runtime. A nil map is returned when no adapter
// provides an inbound handler.
func (b *Bridge) WebhookHandlers() map[string]http.Handler {
	b.mu.RLock()
	defer b.mu.RUnlock()
	out := make(map[string]http.Handler)
	for _, a := range b.adapters {
		wh, ok := a.(webhookAdapter)
		if !ok {
			continue
		}
		cfg := wh.Config()
		path, _ := cfg["endpointPath"].(string)
		if path == "" {
			path = "/webhook"
		}
		out[path] = wh.InboundHandler()
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

// Start connects every configured adapter, installs the inbound handler, and
// launches the read loops. Adapters without credentials (or whose Connect fails)
// are logged and left disconnected; they never fail startup. Start blocks only
// long enough to connect (bounded by connectTimeout per adapter) and returns.
func (b *Bridge) Start() {
	b.mu.Lock()
	if b.closed {
		b.mu.Unlock()
		return
	}
	b.ctx, b.cancel = context.WithCancel(context.Background())
	b.mu.Unlock()

	for _, a := range b.snapshot() {
		if !b.configured(a) {
			b.log.Warn("channel skipped: no credentials", "channel", a.Name())
			continue
		}
		connCtx, cancel := context.WithTimeout(context.Background(), connectTimeout)
		err := a.Connect(connCtx)
		cancel()
		if err != nil {
			b.log.Warn("channel connect failed, leaving disconnected",
				"channel", a.Name(), "error", err)
			continue
		}
		b.mu.Lock()
		b.connected[a.Name()] = true
		b.mu.Unlock()

		a.SetMessageHandler(b.onMessage)
		b.wg.Add(1)
		go b.readLoop(a)
		b.log.Info("channel connected", "channel", a.Name())
	}
	b.log.Info("channel bridge started", "connected", b.ConnectedCount(),
		"registered", len(b.snapshot()))
}

// Stop cancels the read loops, disconnects every adapter and waits for in-flight
// message processing to finish. It is idempotent.
func (b *Bridge) Stop(ctx context.Context) {
	b.mu.Lock()
	if b.closed {
		b.mu.Unlock()
		return
	}
	b.closed = true
	cancel := b.cancel
	b.mu.Unlock()

	cancel()
	for name, a := range b.snapshot() {
		if err := a.Disconnect(ctx); err != nil {
			b.log.Warn("channel disconnect", "channel", name, "error", err)
		}
	}
	b.wg.Wait()
	b.log.Info("channel bridge stopped")
}

// IsConnected reports whether the named adapter is currently connected.
func (b *Bridge) IsConnected(name string) bool {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return b.connected[name]
}

// ConnectedCount reports how many adapters are currently connected.
func (b *Bridge) ConnectedCount() int {
	b.mu.RLock()
	defer b.mu.RUnlock()
	n := 0
	for _, ok := range b.connected {
		if ok {
			n++
		}
	}
	return n
}

// snapshot returns a copy of the registered adapters for a lock-free iteration.
func (b *Bridge) snapshot() []adapter.PlatformAdapter {
	b.mu.RLock()
	defer b.mu.RUnlock()
	out := make([]adapter.PlatformAdapter, 0, len(b.adapters))
	for _, a := range b.adapters {
		out = append(out, a)
	}
	return out
}

// configured reports whether the adapter carries enough configuration to attempt
// a Connect. It inspects the adapter's own Config() snapshot for the credential
// each platform needs, so a bare (empty) adapter is skipped rather than dialed.
// This is what keeps weixin-ilink from falling into its blocking QR-login flow
// when no bot token is present.
func (b *Bridge) configured(a adapter.PlatformAdapter) bool {
	cfg := a.Config()
	switch a.Name() {
	case "telegram", "weixin-ilink":
		return cfgStr(cfg, "bot_token") != ""
	case "qq-obv11":
		if cfgStr(cfg, "mode") == "client" {
			return cfgStr(cfg, "remoteURL") != ""
		}
		return cfgStr(cfg, "listenAddr") != ""
	case "webhook":
		// Inbound-only: no external credential is required to hold the session.
		return true
	default:
		return true
	}
}

// readLoop runs an adapter's blocking read path until the bridge stops. The emit
// sink is wired to onMessage so poll-style adapters (Telegram getUpdates, weixin
// long poll) deliver through the same pipeline as push-style adapters.
func (b *Bridge) readLoop(a adapter.PlatformAdapter) {
	defer b.wg.Done()
	ctx := b.ctx
	a.ReadMessages(ctx, func(msg adapter.UnifiedMessage) {
		b.onMessage(ctx, msg)
	})
}

// onMessage is the adapter-facing inbound callback. It runs the pipeline in a
// bounded goroutine so the adapter's read loop never blocks on a slow agent turn.
func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage) {
	b.mu.Lock()
	if b.closed {
		b.mu.Unlock()
		return
	}
	b.mu.Unlock()

	if err := msg.Validate(); err != nil {
		b.log.Warn("inbound message invalid, dropped", "platform", msg.Platform, "error", err)
		return
	}

	select {
	case b.sem <- struct{}{}:
	case <-ctx.Done():
		return
	}

	// Add the worker under the same lock that Stop uses to set closed, so a
	// concurrent Stop either sees our Add before it waits, or we observe closed
	// and bail without leaking a goroutine.
	b.mu.Lock()
	if b.closed {
		b.mu.Unlock()
		<-b.sem
		return
	}
	b.wg.Add(1)
	b.mu.Unlock()

	go func() {
		defer b.wg.Done()
		defer func() { <-b.sem }()
		b.process(ctx, msg)
	}()
}

// process runs the full inbound pipeline: resolve the session/actor, drive the
// agent loop with the chat text, and send the loop's reply back through the
// channel. Identity is resolved structurally by the session layer; the bridge
// never reads message content to decide who the sender is.
func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) {
	text := strings.TrimSpace(msg.Content)
	if text == "" {
		b.log.Debug("inbound message has no chat text, skipped",
			"platform", msg.Platform, "channel", msg.ChannelID)
		return
	}

	sess, err := b.sessions.GetOrCreate(ctx, msg, b.owner)
	if err != nil {
		b.log.Warn("session resolve failed", "platform", msg.Platform,
			"channel", msg.ChannelID, "error", err)
		return
	}

	b.log.Debug("inbound → session",
		"platform", msg.Platform, "channel", msg.ChannelID,
		"session", sess.SessionID, "actor_role", sess.ActorRole,
		"env", sess.Env(), "mem_writable", sess.MemWritable)

	// design/16 §4.4 + design/02 §3: an owner private chat is routed INTO the single
	// main web session (shared across platforms) and persisted to chat_messages with
	// the channel platform, so the Web client mirrors it; group chats and strangers
	// keep their isolated sess-xxx context and are NOT persisted to the main session.
	sessID := sess.SessionID
	isOwnerPrivate := !sess.IsGroup && sess.Owner
	platform := ""
	if isOwnerPrivate {
		sessID = history.MainSessionID
		platform = msg.Platform
	}

	// Persist the inbound user message into the main web session before the turn, so
	// the Web client sees it even if the channel reply fails (design/16 §4.4).
	if isOwnerPrivate {
		b.persistToMain(ctx, history.RoleUser, text, platform, deviceIDFor(msg))
	}

	// Thread the resolved isolation boundary into the loop's memory gates: the
	// owner's private conversation reads+writes memory, group chats and strangers
	// are read-only isolation (design/02 §3, design/11 §2).
	runEnv := loop.SessionEnv{ActorEnv: sess.Env(), MemWritable: sess.MemWritable}
	res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), sessID, []model.Message{
		model.TextMessage(model.RoleUser, text),
	})
	if err != nil {
		b.log.Warn("agent run failed", "platform", msg.Platform,
			"session", sessID, "error", err)
		return
	}

	reply := strings.TrimSpace(res.Text)
	if reply == "" {
		b.log.Debug("no reply produced", "platform", msg.Platform, "session", sessID)
		return
	}
	// Persist the assistant reply into the main web session (design/16 §4.4) and
	// broadcast, then deliver it back to the channel — so the Web client mirrors the
	// IM reply too.
	if isOwnerPrivate {
		b.persistToMain(ctx, history.RoleAssistant, reply, platform, deviceIDFor(msg))
	}
	if err := b.sendReply(ctx, msg, reply); err != nil {
		if errors.Is(err, adapter.ErrUnsupported) {
			b.log.Debug("channel does not support outbound reply, skipped",
				"platform", msg.Platform, "session", sessID)
			return
		}
		b.log.Warn("send reply failed", "platform", msg.Platform, "error", err)
	}
}

// sendReply delivers the loop's text back to the channel's source conversation.
// It mirrors the inbound message's channel targeting fields so the adapter
// resolves the same peer the message came from.
func (b *Bridge) sendReply(ctx context.Context, in adapter.UnifiedMessage, text string) error {
	a := b.adapter(in.Platform)
	if a == nil {
		return fmt.Errorf("channelbridge: no adapter for platform %q", in.Platform)
	}
	out := adapter.UnifiedMessage{
		Platform:    in.Platform,
		ChannelType: in.ChannelType,
		ChannelID:   in.ChannelID,
		GroupID:     in.GroupID,
		GroupName:   in.GroupName,
		SenderID:    in.SenderID,
		Sender:      in.Sender,
		Content:     text,
		ContentType: adapter.ContentText,
		Timestamp:   time.Now().UTC(),
	}
	return a.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{out}})
}

// adapter returns the registered adapter for a platform name, or nil.
func (b *Bridge) adapter(platform string) adapter.PlatformAdapter {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return b.adapters[platform]
}

// cfgStr reads a string field from an adapter config map, returning "" when
// absent.
func cfgStr(m map[string]any, key string) string {
	if m == nil {
		return ""
	}
	if s, ok := m[key].(string); ok {
		return s
	}
	return ""
}

// deviceIDFor builds the client deviceID tag for an inbound IM message
// (design/16 §4.5): <im平台名>:<本地配置唯一id>:<用户平台唯一uid>. UnifiedMessage has no
// explicit deviceId field yet, so the local config id and the user platform uid are
// taken from the normalized channel binding: ChannelID stands in for the local
// adapter binding and SenderID is the owner's platform uid. When SenderID is absent
// the tag degrades to <platform>:<channelID> so it stays non-empty and readable.
//
// Extension point: when an adapter/UnifiedMessage grows a concrete deviceId (or the
// bridge gains a local config id), replace the ChannelID stand-in here and return
// that value directly.
func deviceIDFor(msg adapter.UnifiedMessage) string {
	if msg.SenderID != "" {
		return msg.Platform + ":" + msg.ChannelID + ":" + msg.SenderID
	}
	return msg.Platform + ":" + msg.ChannelID
}