Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
726479a57f | ||
|
|
5dd5a6f23c |
@@ -4,7 +4,6 @@ import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -583,8 +582,8 @@ func (h *MessagesHandler) DMMessages(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
// DMPartners returns a list of agents the user has DM conversations with,
|
||||
// ordered by most recent message. Each entry includes the peer agent name,
|
||||
// last message preview, timestamp, and unread count.
|
||||
// ordered by most recent message. Queries ALL messages (not just inbox)
|
||||
// so historical conversations always appear.
|
||||
func (h *MessagesHandler) DMPartners(w http.ResponseWriter, r *http.Request) {
|
||||
ownerID, ok := OwnerIDFromContext(r.Context())
|
||||
if !ok {
|
||||
@@ -603,86 +602,41 @@ func (h *MessagesHandler) DMPartners(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
ownedNames := make(map[string]bool, len(ownedAgents))
|
||||
for _, a := range ownedAgents {
|
||||
ownedNames[a.Name] = true
|
||||
agentNames := make([]string, len(ownedAgents))
|
||||
for i, a := range ownedAgents {
|
||||
agentNames[i] = a.Name
|
||||
}
|
||||
|
||||
// Get recent DMs for all owned agents
|
||||
type partner struct {
|
||||
partners, err := h.msgService.GetDMPartners(r.Context(), agentNames)
|
||||
if err != nil {
|
||||
h.logger.Error("get dm partners failed", "error", err)
|
||||
writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to get DM partners"))
|
||||
return
|
||||
}
|
||||
|
||||
// Resolve display names
|
||||
type partnerWithDisplay struct {
|
||||
Name string `json:"name"`
|
||||
DisplayName string `json:"display_name"`
|
||||
LastMessage string `json:"last_message"`
|
||||
LastTime string `json:"last_time"`
|
||||
Unread int `json:"unread"`
|
||||
}
|
||||
partnerMap := make(map[string]*partner)
|
||||
|
||||
for _, agent := range ownedAgents {
|
||||
opts := messaging.ReadOptions{
|
||||
Limit: 200,
|
||||
IncludeRead: true,
|
||||
result := make([]partnerWithDisplay, len(partners))
|
||||
for i, p := range partners {
|
||||
result[i] = partnerWithDisplay{
|
||||
Name: p.Name,
|
||||
DisplayName: p.Name,
|
||||
LastMessage: p.LastMessage,
|
||||
LastTime: p.LastTime,
|
||||
Unread: p.Unread,
|
||||
}
|
||||
result, err := h.msgService.ReadInbox(r.Context(), agent.Name, opts)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for _, msg := range result.Messages {
|
||||
if msg.ChannelID != nil {
|
||||
continue // skip channel messages
|
||||
}
|
||||
// Determine the peer (the other party in the DM)
|
||||
peer := msg.FromAgent
|
||||
if ownedNames[peer] {
|
||||
peer = msg.ToAgent
|
||||
}
|
||||
if peer == "" || ownedNames[peer] {
|
||||
continue // skip self-to-self
|
||||
}
|
||||
|
||||
existing, exists := partnerMap[peer]
|
||||
if !exists {
|
||||
partnerMap[peer] = &partner{
|
||||
Name: peer,
|
||||
DisplayName: peer,
|
||||
LastMessage: truncateStr(msg.Body, 80),
|
||||
LastTime: msg.CreatedAt.Format(time.RFC3339),
|
||||
Unread: 0,
|
||||
}
|
||||
existing = partnerMap[peer]
|
||||
}
|
||||
|
||||
// Track newest message
|
||||
lt, _ := time.Parse(time.RFC3339, existing.LastTime)
|
||||
if msg.CreatedAt.After(lt) {
|
||||
existing.LastMessage = truncateStr(msg.Body, 80)
|
||||
existing.LastTime = msg.CreatedAt.Format(time.RFC3339)
|
||||
}
|
||||
|
||||
// Count unread (pending messages TO owned agents)
|
||||
if ownedNames[msg.ToAgent] && (msg.Status == "pending" || msg.Status == "processing") {
|
||||
existing.Unread++
|
||||
}
|
||||
if a, err := h.agentService.GetAgent(r.Context(), p.Name); err == nil {
|
||||
result[i].DisplayName = a.DisplayName
|
||||
}
|
||||
}
|
||||
|
||||
// Resolve display names
|
||||
for peer, p := range partnerMap {
|
||||
if a, err := h.agentService.GetAgent(r.Context(), peer); err == nil {
|
||||
p.DisplayName = a.DisplayName
|
||||
}
|
||||
}
|
||||
|
||||
// Sort by last_time descending
|
||||
partners := make([]*partner, 0, len(partnerMap))
|
||||
for _, p := range partnerMap {
|
||||
partners = append(partners, p)
|
||||
}
|
||||
sort.Slice(partners, func(i, j int) bool {
|
||||
return partners[i].LastTime > partners[j].LastTime
|
||||
})
|
||||
|
||||
writeJSON(w, http.StatusOK, map[string]any{"partners": partners})
|
||||
writeJSON(w, http.StatusOK, map[string]any{"partners": result})
|
||||
}
|
||||
|
||||
func (h *MessagesHandler) isAgentOwnedBy(r *http.Request, agentName string, ownerID int64) bool {
|
||||
|
||||
@@ -531,6 +531,11 @@ func (s *MessagingService) GetDMMessages(ctx context.Context, ownedAgents []stri
|
||||
return messages, nil
|
||||
}
|
||||
|
||||
// GetDMPartners returns all DM conversation partners for the owned agents.
|
||||
func (s *MessagingService) GetDMPartners(ctx context.Context, ownedAgents []string) ([]DMPartner, error) {
|
||||
return s.store.GetDMPartners(ctx, ownedAgents)
|
||||
}
|
||||
|
||||
// GetDMUnreadCounts returns unread DM counts grouped by peer agent.
|
||||
func (s *MessagingService) GetDMUnreadCounts(ctx context.Context, agentName string) ([]DMUnreadCount, error) {
|
||||
return s.store.GetDMUnreadCounts(ctx, agentName)
|
||||
|
||||
@@ -29,6 +29,7 @@ type MessageStore interface {
|
||||
GetChannelMessages(ctx context.Context, channelID int64, limit, offset int) ([]*Message, error)
|
||||
CountChannelMessages(ctx context.Context, channelID int64) (int, error)
|
||||
GetDMMessages(ctx context.Context, agents []string, peerAgent string, limit int) ([]*Message, error)
|
||||
GetDMPartners(ctx context.Context, agents []string) ([]DMPartner, error)
|
||||
AgentExists(ctx context.Context, agentName string) (bool, error)
|
||||
CountPendingDMs(ctx context.Context, agentName string) (int64, error)
|
||||
GetPendingDMs(ctx context.Context, agentName string, limit int) ([]*Message, error)
|
||||
@@ -687,6 +688,108 @@ func (s *SQLiteMessageStore) GetDMMessages(ctx context.Context, agents []string,
|
||||
return scanMessages(rows)
|
||||
}
|
||||
|
||||
// GetDMPartners returns all unique DM conversation partners for the given human agent,
|
||||
// with the most recent message preview and unread count. Queries ALL messages (not just
|
||||
// inbox) so historical conversations always appear. The humanAgent is the primary viewer;
|
||||
// allAgents includes all owned agents to capture DMs sent to/from any of them.
|
||||
func (s *SQLiteMessageStore) GetDMPartners(ctx context.Context, allAgents []string) ([]DMPartner, error) {
|
||||
if len(allAgents) == 0 {
|
||||
return []DMPartner{}, nil
|
||||
}
|
||||
|
||||
placeholders := make([]string, len(allAgents))
|
||||
for i := range allAgents {
|
||||
placeholders[i] = "?"
|
||||
}
|
||||
inClause := strings.Join(placeholders, ",")
|
||||
|
||||
// For each DM, the "peer" is the other party. When from_agent is owned,
|
||||
// peer = to_agent. When to_agent is owned, peer = from_agent.
|
||||
// We want to see ALL conversation partners, including other owned agents
|
||||
// (e.g., research-mcpproxy sending DMs to algis — both owned by same user).
|
||||
query := fmt.Sprintf(`
|
||||
SELECT
|
||||
peer,
|
||||
body as last_body,
|
||||
created_at as last_time,
|
||||
0 as unread
|
||||
FROM (
|
||||
SELECT
|
||||
CASE
|
||||
WHEN from_agent IN (%s) THEN to_agent
|
||||
ELSE from_agent
|
||||
END as peer,
|
||||
body,
|
||||
created_at,
|
||||
ROW_NUMBER() OVER (
|
||||
PARTITION BY CASE WHEN from_agent IN (%s) THEN to_agent ELSE from_agent END
|
||||
ORDER BY created_at DESC
|
||||
) as rn
|
||||
FROM messages
|
||||
WHERE channel_id IS NULL
|
||||
AND (from_agent IN (%s) OR to_agent IN (%s))
|
||||
) sub
|
||||
WHERE rn = 1 AND peer IS NOT NULL AND peer != ''
|
||||
ORDER BY last_time DESC
|
||||
LIMIT 50`,
|
||||
inClause, inClause, inClause, inClause,
|
||||
)
|
||||
|
||||
fullArgs := make([]any, 0, len(allAgents)*4)
|
||||
for i := 0; i < 4; i++ {
|
||||
for _, a := range allAgents {
|
||||
fullArgs = append(fullArgs, a)
|
||||
}
|
||||
}
|
||||
|
||||
rows, err := s.db.QueryContext(ctx, query, fullArgs...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get dm partners: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
// Track owned agent names for unread counting
|
||||
ownedSet := make(map[string]bool, len(allAgents))
|
||||
for _, a := range allAgents {
|
||||
ownedSet[a] = true
|
||||
}
|
||||
|
||||
var partners []DMPartner
|
||||
for rows.Next() {
|
||||
var p DMPartner
|
||||
var body sql.NullString
|
||||
if err := rows.Scan(&p.Name, &body, &p.LastTime, &p.Unread); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if body.Valid && len(body.String) > 80 {
|
||||
p.LastMessage = body.String[:80] + "..."
|
||||
} else if body.Valid {
|
||||
p.LastMessage = body.String
|
||||
}
|
||||
partners = append(partners, p)
|
||||
}
|
||||
if partners == nil {
|
||||
partners = []DMPartner{}
|
||||
}
|
||||
|
||||
// Count unread messages per partner (pending DMs TO any owned agent FROM each partner)
|
||||
for i, p := range partners {
|
||||
if ownedSet[p.Name] {
|
||||
continue // Skip unread count for inter-agent DMs
|
||||
}
|
||||
var count int
|
||||
err := s.db.QueryRowContext(ctx,
|
||||
fmt.Sprintf(`SELECT COUNT(*) FROM messages WHERE channel_id IS NULL AND from_agent = ? AND to_agent IN (%s) AND status IN ('pending', 'processing')`, inClause),
|
||||
append([]any{p.Name}, fullArgs[:len(allAgents)]...)...,
|
||||
).Scan(&count)
|
||||
if err == nil {
|
||||
partners[i].Unread = count
|
||||
}
|
||||
}
|
||||
|
||||
return partners, rows.Err()
|
||||
}
|
||||
|
||||
func (s *SQLiteMessageStore) CountPendingDMs(ctx context.Context, agentName string) (int64, error) {
|
||||
var count int64
|
||||
err := s.db.QueryRowContext(ctx,
|
||||
|
||||
@@ -101,3 +101,11 @@ type DMUnreadCount struct {
|
||||
UnreadCount int `json:"unread_count"`
|
||||
LastMessageID int64 `json:"last_message_id"`
|
||||
}
|
||||
|
||||
// DMPartner represents a DM conversation partner with summary info.
|
||||
type DMPartner struct {
|
||||
Name string `json:"name"`
|
||||
LastMessage string `json:"last_message"`
|
||||
LastTime string `json:"last_time"`
|
||||
Unread int `json:"unread"`
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user