mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-15 20:15:48 +00:00
feat(sessions): canonical WS session key format + channel/user filtering
- Add BuildWSSessionKey() and IsWSSession() helpers for ws:direct:{convId} format
- Change ChatID to userId (not clientID) for consistent team/workspace isolation
- Add Channel and UserID fields to SessionListOpts with dynamic WHERE builder
- Extract shared buildSessionFilter() for ListPaged and ListPagedRich
- Chat sidebar sends channel:"ws" to filter WS-only sessions per user
- Sessions admin page remains unfiltered (no channel param)
This commit is contained in:
1 parent
3f9401257c
commit
642797f079
5 files changed
+76
-45
No files matched your search
@@ -119,7 +119,7 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re
|
||||
runID := uuid.NewString()
|
||||
sessionKey := params.SessionKey
|
||||
if sessionKey == "" {
|
||||
sessionKey = sessions.SessionKey(params.AgentID, "ws-"+client.ID())
|
||||
sessionKey = sessions.BuildWSSessionKey(params.AgentID, uuid.NewString())
|
||||
}
|
||||
|
||||
// Detach from HTTP request context so agent runs survive page navigation/reconnect.
|
||||
@@ -190,7 +190,7 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re
|
||||
Message: message,
|
||||
Media: mediaFiles,
|
||||
Channel: "ws",
|
||||
ChatID: client.ID(),
|
||||
ChatID: userID, // use stable userID for team/workspace isolation (not ephemeral client.ID())
|
||||
RunID: runID,
|
||||
UserID: userID,
|
||||
Stream: params.Stream,
|
||||
@@ -237,7 +237,7 @@ func (m *ChatMethods) handleHistory(ctx context.Context, client *gateway.Client,
|
||||
|
||||
sessionKey := params.SessionKey
|
||||
if sessionKey == "" {
|
||||
sessionKey = sessions.SessionKey(params.AgentID, "ws-"+client.ID())
|
||||
sessionKey = sessions.BuildWSSessionKey(params.AgentID, uuid.NewString())
|
||||
}
|
||||
|
||||
history := m.sessions.GetHistory(sessionKey)
|
||||
|
||||
@@ -31,6 +31,7 @@ func (m *SessionsMethods) Register(router *gateway.MethodRouter) {
|
||||
|
||||
type sessionsListParams struct {
|
||||
AgentID string `json:"agentId"`
|
||||
Channel string `json:"channel"` // optional: filter by channel prefix ("ws", "telegram")
|
||||
Limit int `json:"limit"`
|
||||
Offset int `json:"offset"`
|
||||
}
|
||||
@@ -45,11 +46,19 @@ func (m *SessionsMethods) handleList(_ context.Context, client *gateway.Client,
|
||||
params.Limit = 20
|
||||
}
|
||||
|
||||
result := m.sessions.ListPagedRich(store.SessionListOpts{
|
||||
opts := store.SessionListOpts{
|
||||
AgentID: params.AgentID,
|
||||
Channel: params.Channel,
|
||||
Limit: params.Limit,
|
||||
Offset: params.Offset,
|
||||
})
|
||||
}
|
||||
// Only filter by UserID when a channel filter is specified (e.g. chat sidebar sends channel="ws").
|
||||
// Sessions admin page omits channel → sees all sessions unfiltered.
|
||||
if params.Channel != "" {
|
||||
opts.UserID = client.UserID()
|
||||
}
|
||||
|
||||
result := m.sessions.ListPagedRich(opts)
|
||||
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{
|
||||
"sessions": result.Sessions,
|
||||
"total": result.Total,
|
||||
|
||||
@@ -183,6 +183,19 @@ func IsHeartbeatSession(key string) bool {
|
||||
return strings.HasPrefix(rest, "heartbeat")
|
||||
}
|
||||
|
||||
// BuildWSSessionKey builds the canonical WS session key for a web conversation.
|
||||
//
|
||||
// agent:{agentId}:ws:direct:{conversationId}
|
||||
func BuildWSSessionKey(agentID, conversationID string) string {
|
||||
return BuildSessionKey(agentID, "ws", PeerDirect, conversationID)
|
||||
}
|
||||
|
||||
// IsWSSession checks if a session key is a WS session (new or legacy format).
|
||||
func IsWSSession(key string) bool {
|
||||
_, rest := ParseSessionKey(key)
|
||||
return strings.HasPrefix(rest, "ws:") || strings.HasPrefix(rest, "ws-")
|
||||
}
|
||||
|
||||
// PeerKindFromGroup returns PeerGroup if isGroup is true, PeerDirect otherwise.
|
||||
func PeerKindFromGroup(isGroup bool) PeerKind {
|
||||
if isGroup {
|
||||
|
||||
@@ -3,6 +3,7 @@ package pg
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -12,6 +13,41 @@ import (
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
// buildSessionFilter builds a dynamic WHERE clause from SessionListOpts.
|
||||
// Returns the WHERE string (with leading " WHERE ") and the positional args.
|
||||
// The tableAlias is prepended to column names (e.g. "s" → "s.session_key").
|
||||
func buildSessionFilter(opts store.SessionListOpts, tableAlias string) (string, []any) {
|
||||
prefix := ""
|
||||
if tableAlias != "" {
|
||||
prefix = tableAlias + "."
|
||||
}
|
||||
var conditions []string
|
||||
var args []any
|
||||
idx := 1
|
||||
|
||||
if opts.AgentID != "" {
|
||||
conditions = append(conditions, fmt.Sprintf("%ssession_key LIKE $%d", prefix, idx))
|
||||
args = append(args, "agent:"+opts.AgentID+":%")
|
||||
idx++
|
||||
}
|
||||
if opts.Channel != "" {
|
||||
// Match canonical format: agent:X:{channel}:...
|
||||
conditions = append(conditions, fmt.Sprintf("%ssession_key LIKE $%d", prefix, idx))
|
||||
args = append(args, "agent:%:"+opts.Channel+":%")
|
||||
idx++
|
||||
}
|
||||
if opts.UserID != "" {
|
||||
conditions = append(conditions, fmt.Sprintf("%suser_id = $%d", prefix, idx))
|
||||
args = append(args, opts.UserID)
|
||||
idx++
|
||||
}
|
||||
|
||||
if len(conditions) == 0 {
|
||||
return "", nil
|
||||
}
|
||||
return " WHERE " + strings.Join(conditions, " AND "), args
|
||||
}
|
||||
|
||||
func (s *PGSessionStore) List(agentID string) []store.SessionInfo {
|
||||
var rows *sql.Rows
|
||||
var err error
|
||||
@@ -65,13 +101,7 @@ func (s *PGSessionStore) ListPaged(opts store.SessionListOpts) store.SessionList
|
||||
}
|
||||
offset := max(opts.Offset, 0)
|
||||
|
||||
var where string
|
||||
var whereArgs []any
|
||||
|
||||
if opts.AgentID != "" {
|
||||
where = " WHERE session_key LIKE $1"
|
||||
whereArgs = append(whereArgs, "agent:"+opts.AgentID+":%")
|
||||
}
|
||||
where, whereArgs := buildSessionFilter(opts, "")
|
||||
|
||||
// Count total
|
||||
var total int
|
||||
@@ -81,18 +111,10 @@ func (s *PGSessionStore) ListPaged(opts store.SessionListOpts) store.SessionList
|
||||
}
|
||||
|
||||
// Fetch page using jsonb_array_length to avoid loading full messages
|
||||
var selectQ string
|
||||
var selectArgs []any
|
||||
|
||||
if opts.AgentID != "" {
|
||||
selectQ = `SELECT session_key, jsonb_array_length(messages), created_at, updated_at, label, channel, user_id, COALESCE(metadata, '{}')
|
||||
FROM sessions WHERE session_key LIKE $1 ORDER BY updated_at DESC LIMIT $2 OFFSET $3`
|
||||
selectArgs = []any{whereArgs[0], limit, offset}
|
||||
} else {
|
||||
selectQ = `SELECT session_key, jsonb_array_length(messages), created_at, updated_at, label, channel, user_id, COALESCE(metadata, '{}')
|
||||
FROM sessions ORDER BY updated_at DESC LIMIT $1 OFFSET $2`
|
||||
selectArgs = []any{limit, offset}
|
||||
}
|
||||
nextIdx := len(whereArgs) + 1
|
||||
selectQ := fmt.Sprintf(`SELECT session_key, jsonb_array_length(messages), created_at, updated_at, label, channel, user_id, COALESCE(metadata, '{}')
|
||||
FROM sessions%s ORDER BY updated_at DESC LIMIT $%d OFFSET $%d`, where, nextIdx, nextIdx+1)
|
||||
selectArgs := append(append([]any{}, whereArgs...), limit, offset)
|
||||
|
||||
rows, err := s.db.Query(selectQ, selectArgs...)
|
||||
if err != nil {
|
||||
@@ -139,13 +161,7 @@ func (s *PGSessionStore) ListPagedRich(opts store.SessionListOpts) store.Session
|
||||
}
|
||||
offset := max(opts.Offset, 0)
|
||||
|
||||
var where string
|
||||
var whereArgs []any
|
||||
|
||||
if opts.AgentID != "" {
|
||||
where = " WHERE s.session_key LIKE $1"
|
||||
whereArgs = append(whereArgs, "agent:"+opts.AgentID+":%")
|
||||
}
|
||||
where, whereArgs := buildSessionFilter(opts, "s")
|
||||
|
||||
// Count total
|
||||
var total int
|
||||
@@ -155,9 +171,6 @@ func (s *PGSessionStore) ListPagedRich(opts store.SessionListOpts) store.Session
|
||||
}
|
||||
|
||||
// Fetch page with agent name via LEFT JOIN
|
||||
var selectQ string
|
||||
var selectArgs []any
|
||||
|
||||
const richCols = `s.session_key, jsonb_array_length(s.messages), s.created_at, s.updated_at,
|
||||
s.label, s.channel, s.user_id, COALESCE(s.metadata, '{}'),
|
||||
s.model, s.provider, s.input_tokens, s.output_tokens,
|
||||
@@ -166,17 +179,11 @@ func (s *PGSessionStore) ListPagedRich(opts store.SessionListOpts) store.Session
|
||||
COALESCE(a.context_window, 200000), -- config.DefaultContextWindow
|
||||
s.compaction_count`
|
||||
|
||||
if opts.AgentID != "" {
|
||||
selectQ = `SELECT ` + richCols + `
|
||||
FROM sessions s LEFT JOIN agents a ON s.agent_id = a.id
|
||||
WHERE s.session_key LIKE $1 ORDER BY s.updated_at DESC LIMIT $2 OFFSET $3`
|
||||
selectArgs = []any{whereArgs[0], limit, offset}
|
||||
} else {
|
||||
selectQ = `SELECT ` + richCols + `
|
||||
FROM sessions s LEFT JOIN agents a ON s.agent_id = a.id
|
||||
ORDER BY s.updated_at DESC LIMIT $1 OFFSET $2`
|
||||
selectArgs = []any{limit, offset}
|
||||
}
|
||||
nextIdx := len(whereArgs) + 1
|
||||
selectQ := fmt.Sprintf(`SELECT %s
|
||||
FROM sessions s LEFT JOIN agents a ON s.agent_id = a.id
|
||||
%s ORDER BY s.updated_at DESC LIMIT $%d OFFSET $%d`, richCols, where, nextIdx, nextIdx+1)
|
||||
selectArgs := append(append([]any{}, whereArgs...), limit, offset)
|
||||
|
||||
rows, err := s.db.Query(selectQ, selectArgs...)
|
||||
if err != nil {
|
||||
|
||||
@@ -53,6 +53,8 @@ type SessionInfo struct {
|
||||
// SessionListOpts holds pagination options for ListPaged.
|
||||
type SessionListOpts struct {
|
||||
AgentID string
|
||||
Channel string // optional: filter by channel prefix ("ws", "telegram", etc.)
|
||||
UserID string // optional: filter by user_id
|
||||
Limit int
|
||||
Offset int
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user