package postgres import ( "context" "fmt" "slices" "strings" "time" "go.kenn.io/agentsview/internal/db" ) const attachToolCallBatchSize = 500 // GetMessages returns paginated messages for a session. func (s *Store) GetMessages( ctx context.Context, sessionID string, from, limit int, asc bool, ) ([]db.Message, error) { if limit <= 0 || limit > db.MaxMessageLimit { limit = db.DefaultMessageLimit } dir := "ASC" op := ">=" if !asc { dir = "DESC" op = "<=" } query := fmt.Sprintf(` SELECT session_id, ordinal, role, content, thinking_text, timestamp, has_thinking, has_tool_use, content_length, is_system, model, token_usage, context_tokens, output_tokens, has_context_tokens, has_output_tokens, claude_message_id, claude_request_id, source_type, source_subtype, source_uuid, source_parent_uuid, is_sidechain, is_compact_boundary FROM messages WHERE session_id = $1 AND ordinal %s $2 ORDER BY ordinal %s LIMIT $3`, op, dir) rows, err := s.pg.QueryContext( ctx, query, sessionID, from, limit, ) if err != nil { return nil, fmt.Errorf( "querying messages: %w", err, ) } defer rows.Close() msgs, err := scanPGMessages(rows) if err != nil { return nil, err } if err := s.attachToolCalls(ctx, msgs); err != nil { return nil, err } return msgs, nil } const pgMessageCols = `session_id, ordinal, role, content, thinking_text, timestamp, has_thinking, has_tool_use, content_length, is_system, model, token_usage, context_tokens, output_tokens, has_context_tokens, has_output_tokens, claude_message_id, claude_request_id, source_type, source_subtype, source_uuid, source_parent_uuid, is_sidechain, is_compact_boundary` // GetMessagesWindow mirrors internal/db's GetMessagesWindow: linear mode // (optionally role-filtered) delegates to GetMessages when Roles is empty; // Around mode merges three queries (before/anchor/after) into one ascending // slice. The anchor query has no role predicate so the anchor row is always // present regardless of Roles; before/after apply the role filter first, so // Before/After count role-matching messages, not raw ordinal distance. func (s *Store) GetMessagesWindow( ctx context.Context, sessionID string, w db.MessageWindow, ) ([]db.Message, error) { if w.Around != nil { return s.getMessagesAroundAnchor(ctx, sessionID, w) } from := 0 if w.From != nil { from = *w.From } if len(w.Roles) == 0 { return s.GetMessages(ctx, sessionID, from, w.Limit, w.Asc) } return s.getMessagesLinearRoleFiltered(ctx, sessionID, from, w.Limit, w.Asc, w.Roles) } func (s *Store) getMessagesLinearRoleFiltered( ctx context.Context, sessionID string, from, limit int, asc bool, roles []string, ) ([]db.Message, error) { if limit <= 0 || limit > db.MaxMessageLimit { limit = db.DefaultMessageLimit } dir := "ASC" op := ">=" if !asc { dir = "DESC" op = "<=" } roleClause, roleArgs := pgRoleFilterClause(roles, 3) query := fmt.Sprintf(` SELECT %s FROM messages WHERE session_id = $1 AND ordinal %s $2%s ORDER BY ordinal %s LIMIT $%d`, pgMessageCols, op, roleClause, dir, len(roleArgs)+3) args := append([]any{sessionID, from}, roleArgs...) args = append(args, limit) rows, err := s.pg.QueryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf("querying role-filtered messages: %w", err) } defer rows.Close() msgs, err := scanPGMessages(rows) if err != nil { return nil, err } if err := s.attachToolCalls(ctx, msgs); err != nil { return nil, err } return msgs, nil } func (s *Store) getMessagesAroundAnchor( ctx context.Context, sessionID string, w db.MessageWindow, ) ([]db.Message, error) { anchor := *w.Around beforeLimit := max(w.Before, 0) afterLimit := max(w.After, 0) roleClause, roleArgs := pgRoleFilterClause(w.Roles, 3) beforeQuery := fmt.Sprintf(` SELECT %s FROM messages WHERE session_id = $1 AND ordinal < $2%s ORDER BY ordinal DESC LIMIT $%d`, pgMessageCols, roleClause, len(roleArgs)+3) beforeArgs := append([]any{sessionID, anchor}, roleArgs...) beforeArgs = append(beforeArgs, beforeLimit) before, err := s.queryMessageRows(ctx, beforeQuery, beforeArgs...) if err != nil { return nil, fmt.Errorf("querying before-window messages: %w", err) } slices.Reverse(before) anchorQuery := fmt.Sprintf(` SELECT %s FROM messages WHERE session_id = $1 AND ordinal = $2`, pgMessageCols) anchorMsgs, err := s.queryMessageRows(ctx, anchorQuery, sessionID, anchor) if err != nil { return nil, fmt.Errorf("querying anchor message: %w", err) } afterQuery := fmt.Sprintf(` SELECT %s FROM messages WHERE session_id = $1 AND ordinal > $2%s ORDER BY ordinal ASC LIMIT $%d`, pgMessageCols, roleClause, len(roleArgs)+3) afterArgs := append([]any{sessionID, anchor}, roleArgs...) afterArgs = append(afterArgs, afterLimit) after, err := s.queryMessageRows(ctx, afterQuery, afterArgs...) if err != nil { return nil, fmt.Errorf("querying after-window messages: %w", err) } msgs := make([]db.Message, 0, len(before)+len(anchorMsgs)+len(after)) msgs = append(msgs, before...) msgs = append(msgs, anchorMsgs...) msgs = append(msgs, after...) if err := s.attachToolCalls(ctx, msgs); err != nil { return nil, err } return msgs, nil } // queryMessageRows runs query and scans the resulting message rows without // attaching tool calls; callers batch that across the merged window set. func (s *Store) queryMessageRows( ctx context.Context, query string, args ...any, ) ([]db.Message, error) { rows, err := s.pg.QueryContext(ctx, query, args...) if err != nil { return nil, err } defer rows.Close() return scanPGMessages(rows) } // pgRoleFilterClause returns an "AND role IN ($n, ...)" clause and its bind // args for the given roles, or ("", nil) when roles is empty. startAt is the // first placeholder ordinal to use (the caller's query already consumes // $1..$(startAt-1)). func pgRoleFilterClause(roles []string, startAt int) (string, []any) { if len(roles) == 0 { return "", nil } placeholders := make([]string, len(roles)) args := make([]any, len(roles)) for i, r := range roles { placeholders[i] = fmt.Sprintf("$%d", startAt+i) args[i] = r } return " AND role IN (" + strings.Join(placeholders, ",") + ")", args } // GetAllMessages returns all messages for a session ordered // by ordinal. func (s *Store) GetAllMessages( ctx context.Context, sessionID string, ) ([]db.Message, error) { rows, err := s.pg.QueryContext(ctx, ` SELECT session_id, ordinal, role, content, thinking_text, timestamp, has_thinking, has_tool_use, content_length, is_system, model, token_usage, context_tokens, output_tokens, has_context_tokens, has_output_tokens, claude_message_id, claude_request_id, source_type, source_subtype, source_uuid, source_parent_uuid, is_sidechain, is_compact_boundary FROM messages WHERE session_id = $1 ORDER BY ordinal ASC`, sessionID) if err != nil { return nil, fmt.Errorf( "querying all messages: %w", err, ) } defer rows.Close() msgs, err := scanPGMessages(rows) if err != nil { return nil, err } if err := s.attachToolCalls(ctx, msgs); err != nil { return nil, err } return msgs, nil } // SearchSession performs ILIKE substring search within a single // session's messages, returning matching ordinals. func (s *Store) SearchSession( ctx context.Context, sessionID, query string, ) ([]int, error) { if query == "" { return nil, nil } like := "%" + escapeLike(query) + "%" rows, err := s.pg.QueryContext(ctx, ` SELECT DISTINCT m.ordinal FROM messages m LEFT JOIN tool_calls tc ON tc.session_id = m.session_id AND tc.message_ordinal = m.ordinal WHERE m.session_id = $1 AND m.is_system = FALSE AND `+db.PostgresSystemPrefixSQL("m.content", "m.role")+` AND (m.content ILIKE $2 OR tc.result_content ILIKE $2) ORDER BY m.ordinal ASC`, sessionID, like, ) if err != nil { return nil, fmt.Errorf( "searching session: %w", err, ) } defer rows.Close() var ordinals []int for rows.Next() { var ord int if err := rows.Scan(&ord); err != nil { return nil, fmt.Errorf( "scanning ordinal: %w", err, ) } ordinals = append(ordinals, ord) } return ordinals, rows.Err() } // HasFTS returns true because ILIKE search is available. func (s *Store) HasFTS() bool { return true } // HasSemantic reports whether a PG vector searcher was wired at startup // (pg serve found a generation matching its embeddings fingerprint). When // false, SearchContent rejects "semantic"/"hybrid" modes up front with // db.ErrSemanticUnavailable. func (s *Store) HasSemantic() bool { return s.getVectorSearcher() != nil } // escapeLike escapes SQL LIKE metacharacters so the bind // parameter is treated as a literal substring. func escapeLike(v string) string { return db.EscapeLikePattern(v) } // Search performs ILIKE-based full-text search across messages, // grouped to one result per session via DISTINCT ON, UNION'd with a // session name (display_name / first_message) branch. func (s *Store) Search( ctx context.Context, f db.SearchFilter, ) (db.SearchPage, error) { if f.Limit <= 0 || f.Limit > db.MaxSearchLimit { f.Limit = db.DefaultSearchLimit } // plainTerm is the de-quoted query joined back into one string. It feeds // the name-branch ILIKE (matching the typed text against the short session // name) and centers the message snippet, mirroring SQLite's plainQuery. // terms is the per-term decomposition: every term must appear in the // message content (AND), matching SQLite FTS5's implicit AND so the same // user query behaves identically across backends. An explicit exact phrase // (user-supplied leading quote) collapses to a single term, preserving the // exact-phrase opt-in. plainTerm := db.StripFTSQuotes(f.Query) terms := db.FTSTerms(f.Query) if plainTerm == "" || len(terms) == 0 { return db.SearchPage{}, nil } // firstTerm anchors POSITION-based ordering and snippet centering. firstTerm := terms[0] // Validate Sort before interpolating into ORDER BY. // session_id ASC is a deterministic tie-breaker for both modes, // preventing pagination instability when sort keys are equal. // NULLS LAST ensures sessions with NULL timestamps sort after // sessions with real timestamps under DESC ordering. // match_priority: 1 = message content match, 2 = name-only match. // This ensures content matches always rank above name-only fallbacks // regardless of match_pos (name-only rows have match_pos=0 which would // otherwise sort them before content matches under match_pos ASC alone). // match_priority: 1 = message content match, 2 = name-only match. // Only applied in relevance mode so content matches rank above name-only // fallbacks. Recency mode orders purely by time so the newest session // wins regardless of match type. outerOrderBy := "match_priority ASC, match_pos ASC, session_ended_at DESC NULLS LAST, session_id ASC" if f.Sort == "recency" { outerOrderBy = "session_ended_at DESC NULLS LAST, session_id ASC" } // $1 = escaped ILIKE pattern for the name branch (full plain term) // $2 = raw first term (for POSITION — case folded in expression) args := []any{escapeLike(plainTerm), firstTerm} argIdx := 3 // Message branch matches every term (AND). Each term gets its own escaped // ILIKE placeholder so a multi-word query requires all terms to be present // without demanding they be contiguous, exactly like SQLite FTS5. termClauses := make([]string, len(terms)) for i, t := range terms { termClauses[i] = fmt.Sprintf( "m.content ILIKE '%%' || $%d || '%%' ESCAPE E'\\\\'", argIdx) args = append(args, escapeLike(t)) argIdx++ } msgTermPredicate := strings.Join(termClauses, "\n\t\t\t\t\tAND ") msgProjectClause := "" nameProjectClause := "" if f.Project != "" { msgProjectClause = fmt.Sprintf("AND s.project = $%d", argIdx) nameProjectClause = fmt.Sprintf("AND s.project = $%d", argIdx) args = append(args, f.Project) argIdx++ } query := fmt.Sprintf(` WITH msg_matches AS ( SELECT DISTINCT ON (m.session_id) m.session_id, s.project, s.agent, COALESCE(s.display_name, s.session_name, s.first_message, '') AS name, COALESCE(s.ended_at, s.started_at) AS session_ended_at, m.ordinal, POSITION(LOWER($2) IN LOWER(m.content)) AS match_pos, CASE WHEN POSITION(LOWER($2) IN LOWER(m.content)) > 100 THEN '...' || SUBSTRING(m.content FROM GREATEST(1, POSITION( LOWER($2) IN LOWER(m.content) ) - 50) FOR 200) || '...' ELSE SUBSTRING(m.content FROM 1 FOR 200) || CASE WHEN LENGTH(m.content) > 200 THEN '...' ELSE '' END END AS snippet FROM messages m JOIN sessions s ON m.session_id = s.id WHERE %s AND s.deleted_at IS NULL AND m.is_system = FALSE AND `+db.PostgresSystemPrefixSQL("m.content", "m.role")+` %s ORDER BY m.session_id, POSITION(LOWER($2) IN LOWER(m.content)) ASC, m.ordinal ASC ), name_matches AS ( SELECT s.id AS session_id, s.project, s.agent, COALESCE(s.display_name, s.session_name, s.first_message, '') AS name, COALESCE(s.ended_at, s.started_at) AS session_ended_at, -1 AS ordinal, 0 AS match_pos, CASE WHEN COALESCE(s.display_name, s.session_name) ILIKE '%%' || $1 || '%%' ESCAPE E'\\' THEN COALESCE(s.display_name, s.session_name, '') WHEN s.first_message ILIKE '%%' || $1 || '%%' ESCAPE E'\\' THEN COALESCE(s.first_message, '') ELSE COALESCE(s.display_name, s.session_name, s.first_message, '') END AS snippet FROM sessions s WHERE (COALESCE(s.display_name, s.session_name) ILIKE '%%' || $1 || '%%' ESCAPE E'\\' OR s.first_message ILIKE '%%' || $1 || '%%' ESCAPE E'\\') AND s.deleted_at IS NULL AND EXISTS ( SELECT 1 FROM messages mx WHERE mx.session_id = s.id AND mx.is_system = FALSE AND `+db.PostgresSystemPrefixSQL("mx.content", "mx.role")+` ) AND s.id NOT IN (SELECT session_id FROM msg_matches) %s ) -- rank is a constant 1.0 because PostgreSQL ILIKE has no -- relevance scoring engine (unlike SQLite FTS5). Ordering -- uses match_pos and session_ended_at instead. SELECT session_id, project, agent, name, session_ended_at, ordinal, snippet, 1.0 AS rank, match_pos FROM ( SELECT *, 1 AS match_priority FROM msg_matches UNION ALL SELECT *, 2 AS match_priority FROM name_matches ) combined ORDER BY %s LIMIT $%d OFFSET $%d`, msgTermPredicate, msgProjectClause, nameProjectClause, outerOrderBy, argIdx, argIdx+1, ) args = append(args, f.Limit+1, f.Cursor) rows, err := s.pg.QueryContext(ctx, query, args...) if err != nil { return db.SearchPage{}, fmt.Errorf("searching: %w", err) } defer rows.Close() results := []db.SearchResult{} for rows.Next() { var r db.SearchResult var endedAt *time.Time var matchPos int if err := rows.Scan( &r.SessionID, &r.Project, &r.Agent, &r.Name, &endedAt, &r.Ordinal, &r.Snippet, &r.Rank, &matchPos, ); err != nil { return db.SearchPage{}, fmt.Errorf( "scanning search result: %w", err, ) } if endedAt != nil { r.SessionEndedAt = FormatISO8601(*endedAt) } results = append(results, r) } if err := rows.Err(); err != nil { return db.SearchPage{}, err } page := db.SearchPage{Results: results} if len(results) > f.Limit { page.Results = results[:f.Limit] page.NextCursor = f.Cursor + f.Limit } return page, nil } // attachToolCalls loads tool_calls for the given messages and // attaches them to each message's ToolCalls field. func (s *Store) attachToolCalls( ctx context.Context, msgs []db.Message, ) error { if len(msgs) == 0 { return nil } ordToIdx := make(map[int]int, len(msgs)) sessionID := msgs[0].SessionID ordinals := make([]int, 0, len(msgs)) for i, m := range msgs { ordToIdx[m.Ordinal] = i ordinals = append(ordinals, m.Ordinal) } for i := 0; i < len(ordinals); i += attachToolCallBatchSize { end := min(i+attachToolCallBatchSize, len(ordinals)) if err := s.attachToolCallsBatch( ctx, msgs, ordToIdx, sessionID, ordinals[i:end], ); err != nil { return err } } if err := s.attachToolResultEvents( ctx, msgs, ordToIdx, sessionID, ordinals, ); err != nil { return err } return nil } func (s *Store) attachToolCallsBatch( ctx context.Context, msgs []db.Message, ordToIdx map[int]int, sessionID string, batch []int, ) error { if len(batch) == 0 { return nil } args := []any{sessionID} phs := make([]string, len(batch)) for i, ord := range batch { args = append(args, ord) phs[i] = fmt.Sprintf("$%d", i+2) } query := fmt.Sprintf(` SELECT message_ordinal, session_id, tool_name, category, COALESCE(tool_use_id, ''), COALESCE(input_json, ''), COALESCE(skill_name, ''), COALESCE(result_content_length, 0), COALESCE(result_content, ''), COALESCE(subagent_session_id, ''), COALESCE(file_path, ''), COALESCE(call_index, 0) FROM tool_calls WHERE session_id = $1 AND message_ordinal IN (%s) ORDER BY message_ordinal, call_index`, strings.Join(phs, ",")) rows, err := s.pg.QueryContext(ctx, query, args...) if err != nil { return fmt.Errorf( "querying tool_calls: %w", err, ) } defer rows.Close() for rows.Next() { var tc db.ToolCall var msgOrdinal int if err := rows.Scan( &msgOrdinal, &tc.SessionID, &tc.ToolName, &tc.Category, &tc.ToolUseID, &tc.InputJSON, &tc.SkillName, &tc.ResultContentLength, &tc.ResultContent, &tc.SubagentSessionID, &tc.FilePath, &tc.CallIndex, ); err != nil { return fmt.Errorf( "scanning tool_call: %w", err, ) } if idx, ok := ordToIdx[msgOrdinal]; ok { msgs[idx].ToolCalls = append( msgs[idx].ToolCalls, tc, ) } } return rows.Err() } func (s *Store) attachToolResultEvents( ctx context.Context, msgs []db.Message, ordToIdx map[int]int, sessionID string, ordinals []int, ) error { for i := 0; i < len(ordinals); i += attachToolCallBatchSize { end := min(i+attachToolCallBatchSize, len(ordinals)) if err := s.attachToolResultEventsBatch( ctx, msgs, ordToIdx, sessionID, ordinals[i:end], ); err != nil { return err } } return nil } func (s *Store) attachToolResultEventsBatch( ctx context.Context, msgs []db.Message, ordToIdx map[int]int, sessionID string, ordinals []int, ) error { if len(ordinals) == 0 { return nil } args := []any{sessionID} phs := make([]string, len(ordinals)) for i, ord := range ordinals { args = append(args, ord) phs[i] = fmt.Sprintf("$%d", i+2) } query := fmt.Sprintf(` SELECT tool_call_message_ordinal, call_index, COALESCE(tool_use_id, ''), COALESCE(agent_id, ''), COALESCE(subagent_session_id, ''), source, status, content, content_length, timestamp, event_index FROM tool_result_events WHERE session_id = $1 AND tool_call_message_ordinal IN (%s) ORDER BY tool_call_message_ordinal, call_index, event_index`, strings.Join(phs, ",")) rows, err := s.pg.QueryContext(ctx, query, args...) if err != nil { return fmt.Errorf("querying tool_result_events: %w", err) } defer rows.Close() for rows.Next() { var ( msgOrdinal int callIndex int ev db.ToolResultEvent ts *time.Time ) if err := rows.Scan( &msgOrdinal, &callIndex, &ev.ToolUseID, &ev.AgentID, &ev.SubagentSessionID, &ev.Source, &ev.Status, &ev.Content, &ev.ContentLength, &ts, &ev.EventIndex, ); err != nil { return fmt.Errorf("scanning tool_result_event: %w", err) } if ts != nil { ev.Timestamp = FormatISO8601(*ts) } idx, ok := ordToIdx[msgOrdinal] if !ok { continue } if callIndex < 0 || callIndex >= len(msgs[idx].ToolCalls) { continue } msgs[idx].ToolCalls[callIndex].ResultEvents = append( msgs[idx].ToolCalls[callIndex].ResultEvents, ev, ) } return rows.Err() } // scanPGMessages scans message rows from PostgreSQL, // converting TIMESTAMPTZ to string. // // The PG messages table has no id column (composite PK on // session_id, ordinal), so we synthesize Message.ID = int64(ordinal) // to match the convention used by TurnRow.MessageID and // CallRow.MessageID in session_timing.go. The frontend keys // {#each messages (message.id)} and looks up turns via // turnByMessage.get(message.id); both depend on Message.ID being // non-zero, unique within a session, and equal to int64(ordinal) // so it joins with TurnRow.MessageID. func scanPGMessages(rows interface { Next() bool Scan(dest ...any) error Err() error }, ) ([]db.Message, error) { msgs := []db.Message{} for rows.Next() { var m db.Message var ts *time.Time var tokenUsage string if err := rows.Scan( &m.SessionID, &m.Ordinal, &m.Role, &m.Content, &m.ThinkingText, &ts, &m.HasThinking, &m.HasToolUse, &m.ContentLength, &m.IsSystem, &m.Model, &tokenUsage, &m.ContextTokens, &m.OutputTokens, &m.HasContextTokens, &m.HasOutputTokens, &m.ClaudeMessageID, &m.ClaudeRequestID, &m.SourceType, &m.SourceSubtype, &m.SourceUUID, &m.SourceParentUUID, &m.IsSidechain, &m.IsCompactBoundary, ); err != nil { return nil, fmt.Errorf( "scanning message: %w", err, ) } m.ID = int64(m.Ordinal) if ts != nil { m.Timestamp = FormatISO8601(*ts) } if tokenUsage != "" { m.TokenUsage = []byte(tokenUsage) } msgs = append(msgs, m) } return msgs, rows.Err() }