Cyrene_v2/internal/channel/session/manager.go

package session

import (
	"context"
	"database/sql"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
)

// DDL for the channel↔session mapping. The composite primary key over
// (platform, channel_id) is the unique key the design requires: each real-world
// channel maps to exactly one internal session_id. created_at/updated_at follow
// the project's timestamp convention; actor_role records the identity lane so the
// mapping survives restarts.
//
// The created_at/updated_at DEFAULT expressions are a fallback only: every write
// path (insertOrIgnoreSessionSQL) supplies explicit fixed-width UTC timestamps
// via formatSessionTime, so the 3-digit-millisecond strftime defaults are never
// exercised and no row mixes timestamp widths.
const createTableSQL = `
CREATE TABLE IF NOT EXISTS channel_sessions (
    platform    TEXT NOT NULL,
    channel_id  TEXT NOT NULL,
    session_id  TEXT NOT NULL,
    actor_role  TEXT NOT NULL,
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
    updated_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
    PRIMARY KEY (platform, channel_id)
)`

const createIndexSQL = `
CREATE INDEX IF NOT EXISTS idx_channel_sessions_actor_role ON channel_sessions(actor_role)`

const insertOrIgnoreSessionSQL = `
INSERT OR IGNORE INTO channel_sessions
    (platform, channel_id, session_id, actor_role, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?)`

const selectSessionSQL = `
SELECT session_id, actor_role FROM channel_sessions WHERE platform = ? AND channel_id = ?`

// updateActorRoleSQL 只改身份车道列（不碰 session_id / created_at）：渠道↔会话的
// 绑定不因身份变化而改变，同一段对话继续落进同一个 session。
const updateActorRoleSQL = `
UPDATE channel_sessions SET actor_role = ?, updated_at = ?
WHERE platform = ? AND channel_id = ?`

// listSessionsSQL selects every persisted channel mapping, newest first. Group
// sessions key on a channel_id carrying groupChannelPrefix; direct sessions use
// their channel id. created_at is fixed-width UTC text (see sessionTimeLayout),
// so its text order is its time order; rowid breaks created_at ties so mappings
// written in the same coarse clock tick still list in a stable order
// (channel_sessions is an ordinary rowid table, not WITHOUT ROWID).
const listSessionSQL = `
SELECT platform, channel_id, session_id, actor_role FROM channel_sessions
ORDER BY created_at DESC, rowid DESC`

// groupChannelPrefix marks a channel id as a group conversation so a session
// reconstructed from the backing table (where no IsGroup flag is persisted) can
// still report group membership.
const groupChannelPrefix = "group:"

// SessionManager owns the channel↔session mapping. It backs the map with the
// SQLite table channel_sessions so a channel keeps its internal session across
// restarts, and it caches lookups in memory to avoid a query per message. The
// cache is guarded by an RWMutex for concurrent adapter use.
type SessionManager struct {
	mu    sync.RWMutex
	db    *sql.DB
	log   *slog.Logger
	cache map[string]Session
}

// New builds a SessionManager, ensuring the channel_sessions table and its index
// exist. It performs no migration bookkeeping: the schema is created here so the
// module is self-contained. A nil db is rejected. ctx is used for DDL.
func New(ctx context.Context, db *sql.DB, log *slog.Logger) (*SessionManager, error) {
	if db == nil {
		return nil, ErrNilDB
	}
	log = logger.WithSource(log, "channel/session")
	if _, err := db.ExecContext(ctx, createTableSQL); err != nil {
		return nil, fmt.Errorf("session: ensure channel_sessions: %w", err)
	}
	if _, err := db.ExecContext(ctx, createIndexSQL); err != nil {
		return nil, fmt.Errorf("session: ensure channel_sessions index: %w", err)
	}
	return &SessionManager{db: db, log: log, cache: make(map[string]Session)}, nil
}

// GetOrCreate resolves the actor for msg and returns the Session bound to the
// (platform, channel) the message represents, creating the mapping on first
// sight. The owner whitelist is applied here (ResolveActor), so the owner flag
// is set structurally rather than from prompt text. Concurrent calls for the
// same channel return the same session.
//
// 已绑定的渠道若解析出的身份车道与缓存里的不一致（典型：主人刚把该账号加进
// 白名单并热重载，见 rebindLocked 的说明），本次调用会把车道刷新到会话上——
// 否则这段对话会一直留在绑定时的旧车道上（陌生人：隔离会话 + 记忆只读）。
func (m *SessionManager) GetOrCreate(ctx context.Context, msg adapter.UnifiedMessage, owner *user.OwnerProfile) (Session, error) {
	actor := ResolveActor(msg, owner)
	cid := channelID(msg)
	if cid == "" {
		return Session{}, fmt.Errorf("session: get-or-create %s: %w", msg.Platform, ErrInvalidChannel)
	}
	key := sessionKey(msg.Platform, cid)

	// 快路径：已绑定且车道与本次解析一致——只读锁即可返回。
	if s, ok := m.lookupCached(key); ok && s.ActorRole == actor.Role {
		return s, nil
	}

	m.mu.Lock()
	defer m.mu.Unlock()
	if s, ok := m.cache[key]; ok {
		if s.ActorRole == actor.Role {
			return s, nil
		}
		return m.rebindLocked(ctx, key, s, actor), nil
	}

	sessionID, err := m.persist(ctx, msg.Platform, cid, actor)
	if err != nil {
		return Session{}, err
	}

	s := sessionWithActor(Session{Platform: msg.Platform, ChannelID: cid, SessionID: sessionID}, actor)
	m.cache[key] = s
	return s, nil
}

// List returns every known session, newest first, across the backing table and
// the in-memory cache. Rows that are still in the cache are returned with their
// full derived metadata (Owner/IsGroup/MemWritable from the original resolve);
// rows recovered from the table (for example from a previous process) are
// reconstructed from the persisted actor role. The returned slice is non-nil. It
// is safe for concurrent use.
func (m *SessionManager) List(ctx context.Context) ([]Session, error) {
	rows, err := m.db.QueryContext(ctx, listSessionSQL)
	if err != nil {
		return nil, fmt.Errorf("session: list: %w", err)
	}
	defer func() { _ = rows.Close() }()

	sessions := make([]Session, 0)
	for rows.Next() {
		var platform, cid, sessionID, roleStr string
		if err := rows.Scan(&platform, &cid, &sessionID, &roleStr); err != nil {
			return nil, fmt.Errorf("session: scan list: %w", err)
		}
		key := sessionKey(platform, cid)
		if cached, ok := m.lookupCached(key); ok {
			sessions = append(sessions, cached)
			continue
		}
		role := adapter.Role(roleStr)
		owner := role == adapter.RoleOwner
		isGroup := strings.HasPrefix(cid, groupChannelPrefix)
		sessions = append(sessions, Session{
			Platform:    platform,
			ChannelID:   cid,
			SessionID:   sessionID,
			ActorRole:   role,
			Owner:       owner,
			IsGroup:     isGroup,
			MemWritable: owner && !isGroup,
		})
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("session: iterate list: %w", err)
	}
	return sessions, nil
}

// lookupCached returns the cached session for key under a read lock.
func (m *SessionManager) lookupCached(key string) (Session, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	s, ok := m.cache[key]
	return s, ok
}

// persist writes the channel mapping if it does not exist yet and reads back the
// canonical session_id. INSERT OR IGNORE makes it safe under concurrency: the
// winning writer's session_id becomes the binding. It is called under the write
// lock so the in-memory cache and the table stay in agreement.
//
// 读回的落盘车道与本次解析不一致时就地纠正：这条绑定可能是**上一个进程**留下的
// （那时该账号还没进白名单，落盘的是陌生人车道）。不纠正的话，重启后第一条消息虽然
// 按新车道跑，面板/诊断读到的落盘车道仍是旧值，直到这条渠道再次发生身份变化。
func (m *SessionManager) persist(ctx context.Context, platform, cid string, actor adapter.Actor) (string, error) {
	now := formatSessionTime(time.Now())
	if _, err := m.db.ExecContext(ctx, insertOrIgnoreSessionSQL, platform, cid, newSessionID(), string(actor.Role), now, now); err != nil {
		return "", fmt.Errorf("session: persist %s/%s: %w", platform, cid, err)
	}
	var sessionID, storedRole string
	if err := m.db.QueryRowContext(ctx, selectSessionSQL, platform, cid).Scan(&sessionID, &storedRole); err != nil {
		return "", fmt.Errorf("session: read back %s/%s: %w", platform, cid, err)
	}
	if storedRole != string(actor.Role) {
		m.syncActorRole(ctx, platform, cid, actor)
	}
	if m.log != nil {
		m.log.Debug("channel session bound",
			"platform", platform, "channel_id", cid, "session_id", sessionID, "actor_role", actor.Role)
	}
	return sessionID, nil
}

// channelID derives the stable channel key used for session mapping. Group
// messages key on the group (prefixed so a group never collides with a direct
// channel of the same id); direct messages key on their channel id, falling back
// to the sender id when no explicit channel id is present.
func channelID(msg adapter.UnifiedMessage) string {
	if msg.IsGroup() {
		if msg.GroupID != "" {
			return groupChannelPrefix + msg.GroupID
		}
		return msg.ChannelID
	}
	if msg.ChannelID != "" {
		return msg.ChannelID
	}
	return msg.SenderID
}

// sessionWithActor 按本次结构化解析出的 actor 派生会话的身份字段。Platform /
// ChannelID / SessionID 原样保留——它们描述的是渠道绑定，不是身份：身份变了，
// 这段对话仍属于同一个 session（绝不因判主结果不同而搬家、换 session id）。
func sessionWithActor(s Session, actor adapter.Actor) Session {
	s.ActorRole = actor.Role
	s.Owner = actor.IsOwner()
	s.IsGroup = actor.IsGroupChat()
	s.MemWritable = actor.IsOwner() && !actor.IsGroupChat()
	return s
}

// rebindLocked 用本次解析出的 actor 刷新一个已绑定会话的身份车道，并把新车道写回
// channel_sessions。调用者必须持有写锁。
//
// 为什么必须刷新（真机事故 2026-10-05，QQ 私聊被当陌生人）：主人把 QQ 号经面板加进
// 白名单、热重载成功后，ResolveActor 立刻命中 owner，但这条渠道**早已在它是陌生人
// 时绑定过**。旧实现直接返回缓存里的旧车道，于是主人私聊仍按陌生人处理——不镜像进
// Web 主会话（前端看不到）、MemWritable 恒 false（记忆只读的隔离上下文），要等进程
// 重启才恢复。刷新后新车道立即对后续消息生效，且与 owner.yaml / channels.json 的
// 热重载承诺（design/11 §9）一致。
//
// 车道只由 ResolveActor 的结构化判定（白名单 + 每渠道 adminIds）给出：本函数不读
// 消息正文，也不看适配器上报的 role，因此刷新不引入任何「靠提示词判主」的通道。
func (m *SessionManager) rebindLocked(ctx context.Context, key string, s Session, actor adapter.Actor) Session {
	from := s.ActorRole
	s = sessionWithActor(s, actor)
	m.cache[key] = s
	m.syncActorRole(ctx, s.Platform, s.ChannelID, actor)
	if m.log != nil {
		m.log.Debug("channel session identity lane updated",
			"platform", s.Platform, "channel_id", s.ChannelID, "session_id", s.SessionID,
			"from", from, "to", actor.Role)
	}
	return s
}

// syncActorRole 把身份车道写回绑定表。落盘只为让车道跨重启存活，失败不能把消息整条
// 丢掉：内存里的车道已经是这次结构化判定的结论，如实记一条 WARN，下一条消息会再试。
func (m *SessionManager) syncActorRole(ctx context.Context, platform, cid string, actor adapter.Actor) {
	if err := m.updateActorRole(ctx, platform, cid, actor.Role); err != nil && m.log != nil {
		m.log.Warn("channel session actor role persist failed",
			"platform", platform, "channel_id", cid, "err", err)
	}
}

// updateActorRole 把刷新后的身份车道写回绑定表。
func (m *SessionManager) updateActorRole(ctx context.Context, platform, cid string, role adapter.Role) error {
	if _, err := m.db.ExecContext(ctx, updateActorRoleSQL, string(role), formatSessionTime(time.Now()), platform, cid); err != nil {
		return fmt.Errorf("session: update actor role %s/%s: %w", platform, cid, err)
	}
	return nil
}