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 }