feat: web UI notifications — unread badges, new message line, auto mark-as-read
Backend: - GET /api/notifications/unread returns channel + DM unread counts - POST /api/notifications/mark-read updates inbox_state for channels/DMs - SSE broadcaster wired into message send for real-time push - last_read_message_id added to channel/DM message responses Frontend: - Notification store tracks unread counts per channel/DM - SSE listener for new_message and unread_update events - Red circular badges in sidebar (Slack-style) - "New messages" separator line in channel/DM views - Auto mark-as-read after 2 seconds of viewing Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Algis Dumbris
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
68b7d946b3
commit
eec06f5b5f
@@ -0,0 +1,109 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
|
||||
"github.com/synapbus/synapbus/internal/agents"
|
||||
"github.com/synapbus/synapbus/internal/channels"
|
||||
)
|
||||
|
||||
// NewMessageEvent is broadcast when a new message is sent.
|
||||
type NewMessageEvent struct {
|
||||
Channel string `json:"channel,omitempty"` // set for channel messages
|
||||
FromAgent string `json:"from_agent,omitempty"` // set for DMs
|
||||
ToAgent string `json:"to_agent,omitempty"` // set for DMs
|
||||
MessageID int64 `json:"message_id"`
|
||||
}
|
||||
|
||||
// UnreadUpdateEvent is broadcast when unread counts change (e.g. mark-read).
|
||||
type UnreadUpdateEvent struct {
|
||||
Channel string `json:"channel,omitempty"`
|
||||
Agent string `json:"agent,omitempty"`
|
||||
UnreadCount int `json:"unread_count"`
|
||||
}
|
||||
|
||||
// EventBroadcaster broadcasts real-time events to connected SSE clients.
|
||||
type EventBroadcaster interface {
|
||||
BroadcastNewMessage(ctx context.Context, ownerID int64, event NewMessageEvent)
|
||||
BroadcastUnreadUpdate(ctx context.Context, ownerID int64, event UnreadUpdateEvent)
|
||||
}
|
||||
|
||||
// SSEBroadcaster implements EventBroadcaster using the SSEHub.
|
||||
type SSEBroadcaster struct {
|
||||
hub *SSEHub
|
||||
agentService *agents.AgentService
|
||||
channelService *channels.Service
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
// NewSSEBroadcaster creates a broadcaster that sends events via SSE.
|
||||
func NewSSEBroadcaster(hub *SSEHub, agentService *agents.AgentService, channelService *channels.Service) *SSEBroadcaster {
|
||||
return &SSEBroadcaster{
|
||||
hub: hub,
|
||||
agentService: agentService,
|
||||
channelService: channelService,
|
||||
logger: slog.Default().With("component", "api.broadcaster"),
|
||||
}
|
||||
}
|
||||
|
||||
// BroadcastNewMessage sends a new_message event to the given owner.
|
||||
func (b *SSEBroadcaster) BroadcastNewMessage(_ context.Context, ownerID int64, event NewMessageEvent) {
|
||||
b.hub.Broadcast(ownerID, SSEEvent{
|
||||
Type: "new_message",
|
||||
Data: event,
|
||||
})
|
||||
}
|
||||
|
||||
// BroadcastUnreadUpdate sends an unread_update event to the given owner.
|
||||
func (b *SSEBroadcaster) BroadcastUnreadUpdate(_ context.Context, ownerID int64, event UnreadUpdateEvent) {
|
||||
b.hub.Broadcast(ownerID, SSEEvent{
|
||||
Type: "unread_update",
|
||||
Data: event,
|
||||
})
|
||||
}
|
||||
|
||||
// BroadcastDM broadcasts a new_message event for a direct message.
|
||||
// It resolves the recipient agent's owner and sends the event to them.
|
||||
func (b *SSEBroadcaster) BroadcastDM(ctx context.Context, msg NewMessageEvent) {
|
||||
if msg.ToAgent == "" {
|
||||
return
|
||||
}
|
||||
|
||||
agent, err := b.agentService.GetAgent(ctx, msg.ToAgent)
|
||||
if err != nil {
|
||||
b.logger.Debug("could not resolve recipient owner for SSE broadcast",
|
||||
"to_agent", msg.ToAgent, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
b.BroadcastNewMessage(ctx, agent.OwnerID, msg)
|
||||
}
|
||||
|
||||
// BroadcastChannelMessage broadcasts a new_message event for a channel message.
|
||||
// It resolves all channel members' owners and sends the event to each unique owner.
|
||||
func (b *SSEBroadcaster) BroadcastChannelMessage(ctx context.Context, channelID int64, msg NewMessageEvent) {
|
||||
if b.channelService == nil {
|
||||
return
|
||||
}
|
||||
|
||||
members, err := b.channelService.GetMembers(ctx, channelID)
|
||||
if err != nil {
|
||||
b.logger.Debug("could not get channel members for SSE broadcast",
|
||||
"channel_id", channelID, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Collect unique owner IDs to avoid duplicate broadcasts.
|
||||
seen := make(map[int64]bool)
|
||||
for _, m := range members {
|
||||
agent, err := b.agentService.GetAgent(ctx, m.AgentName)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if !seen[agent.OwnerID] {
|
||||
seen[agent.OwnerID] = true
|
||||
b.BroadcastNewMessage(ctx, agent.OwnerID, msg)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -201,7 +201,7 @@ func (h *ChannelsHandler) JoinChannel(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
// ChannelMessages handles GET /api/channels/{name}/messages.
|
||||
func (h *ChannelsHandler) ChannelMessages(w http.ResponseWriter, r *http.Request) {
|
||||
_, ok := OwnerIDFromContext(r.Context())
|
||||
ownerID, ok := OwnerIDFromContext(r.Context())
|
||||
if !ok {
|
||||
writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required"))
|
||||
return
|
||||
@@ -228,9 +228,22 @@ func (h *ChannelsHandler) ChannelMessages(w http.ResponseWriter, r *http.Request
|
||||
return
|
||||
}
|
||||
|
||||
// Compute last_read_message_id across owned agents
|
||||
var lastReadMessageID int64
|
||||
ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID)
|
||||
if err == nil {
|
||||
for _, agent := range ownedAgents {
|
||||
lr, err := h.msgService.GetLastReadForChannel(r.Context(), agent.Name, ch.ID)
|
||||
if err == nil && lr > lastReadMessageID {
|
||||
lastReadMessageID = lr
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"messages": paginated.Messages,
|
||||
"total": paginated.Total,
|
||||
"messages": paginated.Messages,
|
||||
"total": paginated.Total,
|
||||
"last_read_message_id": lastReadMessageID,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -18,9 +18,15 @@ import (
|
||||
type MessagesHandler struct {
|
||||
msgService *messaging.MessagingService
|
||||
agentService *agents.AgentService
|
||||
broadcaster *SSEBroadcaster
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
// SetBroadcaster sets the event broadcaster for real-time SSE notifications.
|
||||
func (h *MessagesHandler) SetBroadcaster(b *SSEBroadcaster) {
|
||||
h.broadcaster = b
|
||||
}
|
||||
|
||||
// NewMessagesHandler creates a new messages handler.
|
||||
func NewMessagesHandler(msgService *messaging.MessagingService, agentService *agents.AgentService) *MessagesHandler {
|
||||
return &MessagesHandler{
|
||||
@@ -309,6 +315,24 @@ func (h *MessagesHandler) SendMessage(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// Broadcast real-time event to connected SSE clients
|
||||
if h.broadcaster != nil {
|
||||
event := NewMessageEvent{
|
||||
MessageID: msg.ID,
|
||||
FromAgent: msg.FromAgent,
|
||||
ToAgent: msg.ToAgent,
|
||||
}
|
||||
if msg.ChannelID != nil && h.broadcaster.channelService != nil {
|
||||
ch, chErr := h.broadcaster.channelService.GetChannel(r.Context(), *msg.ChannelID)
|
||||
if chErr == nil {
|
||||
event.Channel = ch.Name
|
||||
}
|
||||
h.broadcaster.BroadcastChannelMessage(r.Context(), *msg.ChannelID, event)
|
||||
} else {
|
||||
h.broadcaster.BroadcastDM(r.Context(), event)
|
||||
}
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusCreated, msg)
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,251 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
|
||||
"github.com/synapbus/synapbus/internal/agents"
|
||||
"github.com/synapbus/synapbus/internal/channels"
|
||||
"github.com/synapbus/synapbus/internal/messaging"
|
||||
)
|
||||
|
||||
// NotificationsHandler handles REST API requests for notification badges.
|
||||
type NotificationsHandler struct {
|
||||
msgService *messaging.MessagingService
|
||||
agentService *agents.AgentService
|
||||
channelService *channels.Service
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
// NewNotificationsHandler creates a new notifications handler.
|
||||
func NewNotificationsHandler(msgService *messaging.MessagingService, agentService *agents.AgentService, channelService *channels.Service) *NotificationsHandler {
|
||||
return &NotificationsHandler{
|
||||
msgService: msgService,
|
||||
agentService: agentService,
|
||||
channelService: channelService,
|
||||
logger: slog.Default().With("component", "api.notifications"),
|
||||
}
|
||||
}
|
||||
|
||||
// channelUnread is the JSON shape for a channel's unread info.
|
||||
type channelUnread struct {
|
||||
Name string `json:"name"`
|
||||
UnreadCount int `json:"unread_count"`
|
||||
LastMessageID int64 `json:"last_message_id"`
|
||||
}
|
||||
|
||||
// dmUnread is the JSON shape for a DM peer's unread info.
|
||||
type dmUnread struct {
|
||||
Agent string `json:"agent"`
|
||||
UnreadCount int `json:"unread_count"`
|
||||
LastMessageID int64 `json:"last_message_id"`
|
||||
}
|
||||
|
||||
// UnreadCounts handles GET /api/notifications/unread.
|
||||
func (h *NotificationsHandler) UnreadCounts(w http.ResponseWriter, r *http.Request) {
|
||||
ownerID, ok := OwnerIDFromContext(r.Context())
|
||||
if !ok {
|
||||
writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required"))
|
||||
return
|
||||
}
|
||||
|
||||
ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID)
|
||||
if err != nil {
|
||||
h.logger.Error("list agents failed", "error", err)
|
||||
writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to list agents"))
|
||||
return
|
||||
}
|
||||
|
||||
if len(ownedAgents) == 0 {
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"channels": []channelUnread{},
|
||||
"dms": []dmUnread{},
|
||||
"total_unread": 0,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// Aggregate channel summaries across all owned agents
|
||||
channelMap := make(map[string]*channelUnread)
|
||||
for _, agent := range ownedAgents {
|
||||
if h.channelService == nil {
|
||||
break
|
||||
}
|
||||
summaries, err := h.channelService.GetChannelSummaries(r.Context(), agent.Name)
|
||||
if err != nil {
|
||||
h.logger.Error("get channel summaries failed", "agent", agent.Name, "error", err)
|
||||
continue
|
||||
}
|
||||
for _, cs := range summaries {
|
||||
existing, ok := channelMap[cs.Name]
|
||||
if !ok {
|
||||
channelMap[cs.Name] = &channelUnread{
|
||||
Name: cs.Name,
|
||||
UnreadCount: cs.UnreadCount,
|
||||
LastMessageID: cs.LastMessageID,
|
||||
}
|
||||
} else {
|
||||
// Take the max unread count (different agents may see different counts)
|
||||
if cs.UnreadCount > existing.UnreadCount {
|
||||
existing.UnreadCount = cs.UnreadCount
|
||||
}
|
||||
if cs.LastMessageID > existing.LastMessageID {
|
||||
existing.LastMessageID = cs.LastMessageID
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
channelsList := make([]channelUnread, 0, len(channelMap))
|
||||
for _, cu := range channelMap {
|
||||
channelsList = append(channelsList, *cu)
|
||||
}
|
||||
|
||||
// Aggregate DM unread counts across all owned agents
|
||||
dmMap := make(map[string]*dmUnread)
|
||||
for _, agent := range ownedAgents {
|
||||
counts, err := h.msgService.GetDMUnreadCounts(r.Context(), agent.Name)
|
||||
if err != nil {
|
||||
h.logger.Error("get dm unread counts failed", "agent", agent.Name, "error", err)
|
||||
continue
|
||||
}
|
||||
for _, dc := range counts {
|
||||
existing, ok := dmMap[dc.Agent]
|
||||
if !ok {
|
||||
dmMap[dc.Agent] = &dmUnread{
|
||||
Agent: dc.Agent,
|
||||
UnreadCount: dc.UnreadCount,
|
||||
LastMessageID: dc.LastMessageID,
|
||||
}
|
||||
} else {
|
||||
existing.UnreadCount += dc.UnreadCount
|
||||
if dc.LastMessageID > existing.LastMessageID {
|
||||
existing.LastMessageID = dc.LastMessageID
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
dmsList := make([]dmUnread, 0, len(dmMap))
|
||||
for _, du := range dmMap {
|
||||
dmsList = append(dmsList, *du)
|
||||
}
|
||||
|
||||
totalUnread := 0
|
||||
for _, cu := range channelsList {
|
||||
totalUnread += cu.UnreadCount
|
||||
}
|
||||
for _, du := range dmsList {
|
||||
totalUnread += du.UnreadCount
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"channels": channelsList,
|
||||
"dms": dmsList,
|
||||
"total_unread": totalUnread,
|
||||
})
|
||||
}
|
||||
|
||||
// MarkRead handles POST /api/notifications/mark-read.
|
||||
func (h *NotificationsHandler) MarkRead(w http.ResponseWriter, r *http.Request) {
|
||||
ownerID, ok := OwnerIDFromContext(r.Context())
|
||||
if !ok {
|
||||
writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "Authentication required"))
|
||||
return
|
||||
}
|
||||
|
||||
var req struct {
|
||||
Type string `json:"type"`
|
||||
Target string `json:"target"`
|
||||
LastMessageID int64 `json:"last_message_id"`
|
||||
}
|
||||
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeJSON(w, http.StatusBadRequest, errorBody("invalid_request", "Invalid JSON body"))
|
||||
return
|
||||
}
|
||||
|
||||
if req.Type == "" || req.Target == "" || req.LastMessageID <= 0 {
|
||||
writeJSON(w, http.StatusBadRequest, errorBody("validation_error", "type, target, and last_message_id are required"))
|
||||
return
|
||||
}
|
||||
|
||||
if req.Type != "channel" && req.Type != "dm" {
|
||||
writeJSON(w, http.StatusBadRequest, errorBody("validation_error", "type must be 'channel' or 'dm'"))
|
||||
return
|
||||
}
|
||||
|
||||
ownedAgents, err := h.agentService.ListAgents(r.Context(), ownerID)
|
||||
if err != nil {
|
||||
h.logger.Error("list agents failed", "error", err)
|
||||
writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to list agents"))
|
||||
return
|
||||
}
|
||||
|
||||
if len(ownedAgents) == 0 {
|
||||
writeJSON(w, http.StatusBadRequest, errorBody("no_agents", "No agents registered"))
|
||||
return
|
||||
}
|
||||
|
||||
agentNames := make([]string, len(ownedAgents))
|
||||
for i, a := range ownedAgents {
|
||||
agentNames[i] = a.Name
|
||||
}
|
||||
|
||||
if req.Type == "channel" {
|
||||
if h.channelService == nil {
|
||||
writeJSON(w, http.StatusBadRequest, errorBody("not_available", "Channel service not available"))
|
||||
return
|
||||
}
|
||||
|
||||
ch, err := h.channelService.GetChannelByName(r.Context(), req.Target)
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusNotFound, errorBody("not_found", "Channel not found"))
|
||||
return
|
||||
}
|
||||
|
||||
// Get conversation IDs for messages in this channel up to the given message ID
|
||||
convIDs, err := h.msgService.GetConversationIDsForChannel(r.Context(), ch.ID, req.LastMessageID)
|
||||
if err != nil {
|
||||
h.logger.Error("get conversation ids failed", "error", err)
|
||||
writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to get conversations"))
|
||||
return
|
||||
}
|
||||
|
||||
// Update inbox state for all owned agents on all relevant conversations
|
||||
for _, agentName := range agentNames {
|
||||
for _, convID := range convIDs {
|
||||
if err := h.msgService.UpdateInboxState(r.Context(), agentName, convID, req.LastMessageID); err != nil {
|
||||
h.logger.Error("update inbox state failed",
|
||||
"agent", agentName,
|
||||
"conversation_id", convID,
|
||||
"error", err,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// DM mark-read
|
||||
convIDs, err := h.msgService.GetConversationIDsForDM(r.Context(), agentNames, req.Target, req.LastMessageID)
|
||||
if err != nil {
|
||||
h.logger.Error("get dm conversation ids failed", "error", err)
|
||||
writeJSON(w, http.StatusInternalServerError, errorBody("server_error", "Failed to get conversations"))
|
||||
return
|
||||
}
|
||||
|
||||
for _, agentName := range agentNames {
|
||||
for _, convID := range convIDs {
|
||||
if err := h.msgService.UpdateInboxState(r.Context(), agentName, convID, req.LastMessageID); err != nil {
|
||||
h.logger.Error("update inbox state failed",
|
||||
"agent", agentName,
|
||||
"conversation_id", convID,
|
||||
"error", err,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
||||
}
|
||||
@@ -0,0 +1,373 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
_ "modernc.org/sqlite"
|
||||
|
||||
"github.com/synapbus/synapbus/internal/agents"
|
||||
"github.com/synapbus/synapbus/internal/channels"
|
||||
"github.com/synapbus/synapbus/internal/messaging"
|
||||
)
|
||||
|
||||
func setupNotificationsRouter(t *testing.T) (chi.Router, *messaging.MessagingService, *agents.AgentService, *channels.Service) {
|
||||
t.Helper()
|
||||
db := newTestDBFull(t)
|
||||
|
||||
msgStore := messaging.NewSQLiteMessageStore(db)
|
||||
msgService := messaging.NewMessagingService(msgStore, nil)
|
||||
|
||||
agentStore := agents.NewSQLiteAgentStore(db)
|
||||
agentService := agents.NewAgentService(agentStore, nil)
|
||||
|
||||
channelStore := channels.NewSQLiteChannelStore(db)
|
||||
channelService := channels.NewService(channelStore, msgService, nil)
|
||||
|
||||
// Seed agents
|
||||
seedTestAgent(t, db, "human-agent", 1)
|
||||
seedTestAgent(t, db, "bot-alice", 2)
|
||||
seedTestAgent(t, db, "bot-bob", 2)
|
||||
|
||||
handler := NewNotificationsHandler(msgService, agentService, channelService)
|
||||
messagesHandler := NewMessagesHandler(msgService, agentService)
|
||||
channelsHandler := NewChannelsHandler(channelService, agentService, msgService)
|
||||
|
||||
router := chi.NewRouter()
|
||||
router.Group(func(r chi.Router) {
|
||||
r.Use(OwnerAuthMiddleware)
|
||||
r.Get("/api/notifications/unread", handler.UnreadCounts)
|
||||
r.Post("/api/notifications/mark-read", handler.MarkRead)
|
||||
r.Get("/api/channels/{name}/messages", channelsHandler.ChannelMessages)
|
||||
r.Get("/api/agents/{name}/messages", messagesHandler.DMMessages)
|
||||
})
|
||||
|
||||
return router, msgService, agentService, channelService
|
||||
}
|
||||
|
||||
func TestUnreadCounts_Unauthenticated(t *testing.T) {
|
||||
router, _, _, _ := setupNotificationsRouter(t)
|
||||
|
||||
req := httptest.NewRequest("GET", "/api/notifications/unread", nil)
|
||||
rr := httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
if rr.Code != http.StatusUnauthorized {
|
||||
t.Errorf("status = %d, want %d", rr.Code, http.StatusUnauthorized)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnreadCounts_NoAgents(t *testing.T) {
|
||||
router, _, _, _ := setupNotificationsRouter(t)
|
||||
|
||||
// Owner 99 has no agents
|
||||
req := httptest.NewRequest("GET", "/api/notifications/unread", nil)
|
||||
req.Header.Set("X-Owner-ID", "99")
|
||||
rr := httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want %d, body: %s", rr.Code, http.StatusOK, rr.Body.String())
|
||||
}
|
||||
|
||||
var resp struct {
|
||||
Channels []channelUnread `json:"channels"`
|
||||
DMs []dmUnread `json:"dms"`
|
||||
TotalUnread int `json:"total_unread"`
|
||||
}
|
||||
json.Unmarshal(rr.Body.Bytes(), &resp)
|
||||
|
||||
if len(resp.Channels) != 0 {
|
||||
t.Errorf("channels = %d, want 0", len(resp.Channels))
|
||||
}
|
||||
if len(resp.DMs) != 0 {
|
||||
t.Errorf("dms = %d, want 0", len(resp.DMs))
|
||||
}
|
||||
if resp.TotalUnread != 0 {
|
||||
t.Errorf("total_unread = %d, want 0", resp.TotalUnread)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnreadCounts_WithDMs(t *testing.T) {
|
||||
router, msgService, _, _ := setupNotificationsRouter(t)
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
// bot-alice sends DMs to human-agent (owner 1)
|
||||
_, err := msgService.SendMessage(ctx, "bot-alice", "human-agent", "Hello from Alice", messaging.SendOptions{Subject: "dm"})
|
||||
if err != nil {
|
||||
t.Fatalf("send message: %v", err)
|
||||
}
|
||||
_, err = msgService.SendMessage(ctx, "bot-alice", "human-agent", "Second message from Alice", messaging.SendOptions{Subject: "dm"})
|
||||
if err != nil {
|
||||
t.Fatalf("send message: %v", err)
|
||||
}
|
||||
|
||||
// bot-bob sends a DM to human-agent
|
||||
_, err = msgService.SendMessage(ctx, "bot-bob", "human-agent", "Hello from Bob", messaging.SendOptions{Subject: "dm"})
|
||||
if err != nil {
|
||||
t.Fatalf("send message: %v", err)
|
||||
}
|
||||
|
||||
req := httptest.NewRequest("GET", "/api/notifications/unread", nil)
|
||||
req.Header.Set("X-Owner-ID", "1")
|
||||
rr := httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want %d, body: %s", rr.Code, http.StatusOK, rr.Body.String())
|
||||
}
|
||||
|
||||
var resp struct {
|
||||
Channels []channelUnread `json:"channels"`
|
||||
DMs []dmUnread `json:"dms"`
|
||||
TotalUnread int `json:"total_unread"`
|
||||
}
|
||||
json.Unmarshal(rr.Body.Bytes(), &resp)
|
||||
|
||||
if len(resp.DMs) != 2 {
|
||||
t.Fatalf("dms = %d, want 2; body: %s", len(resp.DMs), rr.Body.String())
|
||||
}
|
||||
|
||||
// Find alice and bob counts
|
||||
dmsByAgent := make(map[string]dmUnread)
|
||||
for _, dm := range resp.DMs {
|
||||
dmsByAgent[dm.Agent] = dm
|
||||
}
|
||||
|
||||
if alice, ok := dmsByAgent["bot-alice"]; !ok {
|
||||
t.Error("expected DM from bot-alice")
|
||||
} else if alice.UnreadCount != 2 {
|
||||
t.Errorf("bot-alice unread = %d, want 2", alice.UnreadCount)
|
||||
}
|
||||
|
||||
if bob, ok := dmsByAgent["bot-bob"]; !ok {
|
||||
t.Error("expected DM from bot-bob")
|
||||
} else if bob.UnreadCount != 1 {
|
||||
t.Errorf("bot-bob unread = %d, want 1", bob.UnreadCount)
|
||||
}
|
||||
|
||||
if resp.TotalUnread != 3 {
|
||||
t.Errorf("total_unread = %d, want 3", resp.TotalUnread)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMarkRead_DM(t *testing.T) {
|
||||
router, msgService, _, _ := setupNotificationsRouter(t)
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
// bot-alice sends 3 DMs to human-agent
|
||||
msg1, _ := msgService.SendMessage(ctx, "bot-alice", "human-agent", "Message 1", messaging.SendOptions{Subject: "dm"})
|
||||
_, _ = msgService.SendMessage(ctx, "bot-alice", "human-agent", "Message 2", messaging.SendOptions{Subject: "dm"})
|
||||
msg3, _ := msgService.SendMessage(ctx, "bot-alice", "human-agent", "Message 3", messaging.SendOptions{Subject: "dm"})
|
||||
|
||||
// Mark read up to msg1
|
||||
body, _ := json.Marshal(map[string]any{
|
||||
"type": "dm",
|
||||
"target": "bot-alice",
|
||||
"last_message_id": msg1.ID,
|
||||
})
|
||||
req := httptest.NewRequest("POST", "/api/notifications/mark-read", bytes.NewReader(body))
|
||||
req.Header.Set("X-Owner-ID", "1")
|
||||
rr := httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Fatalf("mark-read status = %d, want %d, body: %s", rr.Code, http.StatusOK, rr.Body.String())
|
||||
}
|
||||
|
||||
// Check unread — should have 2 unread still
|
||||
req = httptest.NewRequest("GET", "/api/notifications/unread", nil)
|
||||
req.Header.Set("X-Owner-ID", "1")
|
||||
rr = httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
var resp struct {
|
||||
DMs []dmUnread `json:"dms"`
|
||||
TotalUnread int `json:"total_unread"`
|
||||
}
|
||||
json.Unmarshal(rr.Body.Bytes(), &resp)
|
||||
|
||||
if len(resp.DMs) != 1 {
|
||||
t.Fatalf("dms = %d, want 1; body: %s", len(resp.DMs), rr.Body.String())
|
||||
}
|
||||
if resp.DMs[0].UnreadCount != 2 {
|
||||
t.Errorf("unread after mark-read = %d, want 2", resp.DMs[0].UnreadCount)
|
||||
}
|
||||
|
||||
// Mark all read up to msg3
|
||||
body, _ = json.Marshal(map[string]any{
|
||||
"type": "dm",
|
||||
"target": "bot-alice",
|
||||
"last_message_id": msg3.ID,
|
||||
})
|
||||
req = httptest.NewRequest("POST", "/api/notifications/mark-read", bytes.NewReader(body))
|
||||
req.Header.Set("X-Owner-ID", "1")
|
||||
rr = httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Fatalf("mark-read status = %d, want %d", rr.Code, http.StatusOK)
|
||||
}
|
||||
|
||||
// Check unread — should be 0
|
||||
req = httptest.NewRequest("GET", "/api/notifications/unread", nil)
|
||||
req.Header.Set("X-Owner-ID", "1")
|
||||
rr = httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
json.Unmarshal(rr.Body.Bytes(), &resp)
|
||||
if resp.TotalUnread != 0 {
|
||||
t.Errorf("total_unread after marking all read = %d, want 0; body: %s", resp.TotalUnread, rr.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestMarkRead_Validation(t *testing.T) {
|
||||
router, _, _, _ := setupNotificationsRouter(t)
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
body map[string]any
|
||||
want int
|
||||
}{
|
||||
{
|
||||
name: "missing type",
|
||||
body: map[string]any{"target": "foo", "last_message_id": 1},
|
||||
want: http.StatusBadRequest,
|
||||
},
|
||||
{
|
||||
name: "missing target",
|
||||
body: map[string]any{"type": "dm", "last_message_id": 1},
|
||||
want: http.StatusBadRequest,
|
||||
},
|
||||
{
|
||||
name: "missing last_message_id",
|
||||
body: map[string]any{"type": "dm", "target": "foo"},
|
||||
want: http.StatusBadRequest,
|
||||
},
|
||||
{
|
||||
name: "invalid type",
|
||||
body: map[string]any{"type": "invalid", "target": "foo", "last_message_id": 1},
|
||||
want: http.StatusBadRequest,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
body, _ := json.Marshal(tt.body)
|
||||
req := httptest.NewRequest("POST", "/api/notifications/mark-read", bytes.NewReader(body))
|
||||
req.Header.Set("X-Owner-ID", "1")
|
||||
rr := httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
if rr.Code != tt.want {
|
||||
t.Errorf("status = %d, want %d, body: %s", rr.Code, tt.want, rr.Body.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDMMessages_IncludesLastRead(t *testing.T) {
|
||||
router, msgService, _, _ := setupNotificationsRouter(t)
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
// Send DMs from bot-alice to human-agent
|
||||
msg1, _ := msgService.SendMessage(ctx, "bot-alice", "human-agent", "Hello", messaging.SendOptions{Subject: "dm"})
|
||||
_, _ = msgService.SendMessage(ctx, "bot-alice", "human-agent", "World", messaging.SendOptions{Subject: "dm"})
|
||||
|
||||
// Mark read up to msg1
|
||||
body, _ := json.Marshal(map[string]any{
|
||||
"type": "dm",
|
||||
"target": "bot-alice",
|
||||
"last_message_id": msg1.ID,
|
||||
})
|
||||
req := httptest.NewRequest("POST", "/api/notifications/mark-read", bytes.NewReader(body))
|
||||
req.Header.Set("X-Owner-ID", "1")
|
||||
rr := httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Fatalf("mark-read status = %d", rr.Code)
|
||||
}
|
||||
|
||||
// GET DM messages should include last_read_message_id
|
||||
req = httptest.NewRequest("GET", "/api/agents/bot-alice/messages", nil)
|
||||
req.Header.Set("X-Owner-ID", "1")
|
||||
rr = httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Fatalf("dm messages status = %d, body: %s", rr.Code, rr.Body.String())
|
||||
}
|
||||
|
||||
var resp map[string]any
|
||||
json.Unmarshal(rr.Body.Bytes(), &resp)
|
||||
|
||||
lastRead, ok := resp["last_read_message_id"]
|
||||
if !ok {
|
||||
t.Fatal("response missing last_read_message_id")
|
||||
}
|
||||
if int64(lastRead.(float64)) != msg1.ID {
|
||||
t.Errorf("last_read_message_id = %v, want %d", lastRead, msg1.ID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelMessages_IncludesLastRead(t *testing.T) {
|
||||
router, _, _, channelService := setupNotificationsRouter(t)
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
// Create a channel and have human-agent join
|
||||
ch, err := channelService.CreateChannel(ctx, channels.CreateChannelRequest{
|
||||
Name: "test-channel",
|
||||
CreatedBy: "human-agent",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create channel: %v", err)
|
||||
}
|
||||
|
||||
// Broadcast messages
|
||||
msgs, err := channelService.BroadcastMessage(ctx, ch.ID, "human-agent", "Hello channel", 5, "")
|
||||
if err != nil {
|
||||
t.Fatalf("broadcast: %v", err)
|
||||
}
|
||||
|
||||
// Mark read up to the first channel message
|
||||
body, _ := json.Marshal(map[string]any{
|
||||
"type": "channel",
|
||||
"target": "test-channel",
|
||||
"last_message_id": msgs[0].ID,
|
||||
})
|
||||
req := httptest.NewRequest("POST", "/api/notifications/mark-read", bytes.NewReader(body))
|
||||
req.Header.Set("X-Owner-ID", "1")
|
||||
rr := httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Fatalf("mark-read status = %d, body: %s", rr.Code, rr.Body.String())
|
||||
}
|
||||
|
||||
// GET channel messages should include last_read_message_id
|
||||
req = httptest.NewRequest("GET", "/api/channels/test-channel/messages", nil)
|
||||
req.Header.Set("X-Owner-ID", "1")
|
||||
rr = httptest.NewRecorder()
|
||||
router.ServeHTTP(rr, req)
|
||||
|
||||
if rr.Code != http.StatusOK {
|
||||
t.Fatalf("channel messages status = %d, body: %s", rr.Code, rr.Body.String())
|
||||
}
|
||||
|
||||
var resp map[string]any
|
||||
json.Unmarshal(rr.Body.Bytes(), &resp)
|
||||
|
||||
_, ok := resp["last_read_message_id"]
|
||||
if !ok {
|
||||
t.Fatal("response missing last_read_message_id")
|
||||
}
|
||||
}
|
||||
@@ -85,6 +85,13 @@ func NewRouterWithConfig(cfg RouterConfig) chi.Router {
|
||||
if cfg.MsgService != nil && cfg.AgentService != nil {
|
||||
messagesHandler := NewMessagesHandler(cfg.MsgService, cfg.AgentService)
|
||||
agentsHandler := NewAgentsHandler(cfg.AgentService, cfg.TraceStore, cfg.ChannelService)
|
||||
notificationsHandler := NewNotificationsHandler(cfg.MsgService, cfg.AgentService, cfg.ChannelService)
|
||||
|
||||
// Wire up SSE broadcaster for real-time events
|
||||
if cfg.SSEHub != nil {
|
||||
broadcaster := NewSSEBroadcaster(cfg.SSEHub, cfg.AgentService, cfg.ChannelService)
|
||||
messagesHandler.SetBroadcaster(broadcaster)
|
||||
}
|
||||
|
||||
r.Group(func(r chi.Router) {
|
||||
r.Use(authMiddleware)
|
||||
@@ -109,6 +116,10 @@ func NewRouterWithConfig(cfg RouterConfig) chi.Router {
|
||||
r.Delete("/api/agents/{name}", agentsHandler.DeleteAgent)
|
||||
r.Post("/api/agents/{name}/revoke-key", agentsHandler.RevokeKey)
|
||||
r.Get("/api/agents/{name}/messages", messagesHandler.DMMessages)
|
||||
|
||||
// Notifications
|
||||
r.Get("/api/notifications/unread", notificationsHandler.UnreadCounts)
|
||||
r.Post("/api/notifications/mark-read", notificationsHandler.MarkRead)
|
||||
})
|
||||
|
||||
// API Keys
|
||||
|
||||
@@ -16,6 +16,7 @@ type ChannelSummary struct {
|
||||
ID int64 `json:"id"`
|
||||
Name string `json:"name"`
|
||||
UnreadCount int `json:"unread"`
|
||||
LastMessageID int64 `json:"last_message_id"`
|
||||
LastMessageAt *time.Time `json:"last_message_at"`
|
||||
}
|
||||
|
||||
|
||||
@@ -381,6 +381,7 @@ func (s *SQLiteChannelStore) GetChannelSummaries(ctx context.Context, agentName
|
||||
WHERE ist.agent_name = ? AND ist.conversation_id = m.conversation_id), 0)
|
||||
AND m.from_agent != ?
|
||||
) AS unread_count,
|
||||
COALESCE((SELECT MAX(m3.id) FROM messages m3 WHERE m3.channel_id = c.id), 0) AS last_message_id,
|
||||
(SELECT MAX(m2.created_at) FROM messages m2 WHERE m2.channel_id = c.id) AS last_message_at
|
||||
FROM channels c
|
||||
JOIN channel_members cm ON cm.channel_id = c.id AND cm.agent_name = ?
|
||||
@@ -396,7 +397,7 @@ func (s *SQLiteChannelStore) GetChannelSummaries(ctx context.Context, agentName
|
||||
for rows.Next() {
|
||||
var cs ChannelSummary
|
||||
var lastMsg sql.NullString
|
||||
if err := rows.Scan(&cs.ID, &cs.Name, &cs.UnreadCount, &lastMsg); err != nil {
|
||||
if err := rows.Scan(&cs.ID, &cs.Name, &cs.UnreadCount, &cs.LastMessageID, &lastMsg); err != nil {
|
||||
return nil, fmt.Errorf("scan channel summary: %w", err)
|
||||
}
|
||||
if lastMsg.Valid {
|
||||
|
||||
@@ -459,6 +459,36 @@ func (s *MessagingService) GetDMMessages(ctx context.Context, ownedAgents []stri
|
||||
return messages, nil
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
|
||||
// GetLastReadForChannel returns the last_read_message_id for an agent in a channel.
|
||||
func (s *MessagingService) GetLastReadForChannel(ctx context.Context, agentName string, channelID int64) (int64, error) {
|
||||
return s.store.GetLastReadForChannel(ctx, agentName, channelID)
|
||||
}
|
||||
|
||||
// GetLastReadForDM returns the last_read_message_id for owned agents in a DM with a peer.
|
||||
func (s *MessagingService) GetLastReadForDM(ctx context.Context, agentNames []string, peerAgent string) (int64, error) {
|
||||
return s.store.GetLastReadForDM(ctx, agentNames, peerAgent)
|
||||
}
|
||||
|
||||
// UpdateInboxState updates the read position for an agent in a conversation.
|
||||
func (s *MessagingService) UpdateInboxState(ctx context.Context, agentName string, conversationID int64, lastReadMsgID int64) error {
|
||||
return s.store.UpdateInboxState(ctx, agentName, conversationID, lastReadMsgID)
|
||||
}
|
||||
|
||||
// GetConversationIDsForChannel returns conversation IDs in a channel with messages up to lastMessageID.
|
||||
func (s *MessagingService) GetConversationIDsForChannel(ctx context.Context, channelID int64, lastMessageID int64) ([]int64, error) {
|
||||
return s.store.GetConversationIDsForChannel(ctx, channelID, lastMessageID)
|
||||
}
|
||||
|
||||
// GetConversationIDsForDM returns conversation IDs for DMs between owned agents and a peer.
|
||||
func (s *MessagingService) GetConversationIDsForDM(ctx context.Context, agentNames []string, peerAgent string, lastMessageID int64) ([]int64, error) {
|
||||
return s.store.GetConversationIDsForDM(ctx, agentNames, peerAgent, lastMessageID)
|
||||
}
|
||||
|
||||
// GetConversation returns a conversation and its messages.
|
||||
func (s *MessagingService) GetConversation(ctx context.Context, id int64) (*Conversation, []*Message, error) {
|
||||
conv, err := s.store.GetConversation(ctx, id)
|
||||
|
||||
@@ -34,6 +34,11 @@ type MessageStore interface {
|
||||
GetPendingDMs(ctx context.Context, agentName string, limit int) ([]*Message, error)
|
||||
GetRecentMentions(ctx context.Context, agentName string, limit int) ([]*Message, error)
|
||||
GetSystemNotifications(ctx context.Context, agentName string, limit int) ([]*Message, error)
|
||||
GetDMUnreadCounts(ctx context.Context, agentName string) ([]DMUnreadCount, error)
|
||||
GetLastReadForChannel(ctx context.Context, agentName string, channelID int64) (int64, error)
|
||||
GetLastReadForDM(ctx context.Context, agentNames []string, peerAgent string) (int64, error)
|
||||
GetConversationIDsForChannel(ctx context.Context, channelID int64, lastMessageID int64) ([]int64, error)
|
||||
GetConversationIDsForDM(ctx context.Context, agentNames []string, peerAgent string, lastMessageID int64) ([]int64, error)
|
||||
}
|
||||
|
||||
// SQLiteMessageStore implements MessageStore using SQLite.
|
||||
@@ -773,6 +778,184 @@ func scanMessageFromRows(rows *sql.Rows) (*Message, error) {
|
||||
return &msg, nil
|
||||
}
|
||||
|
||||
// GetDMUnreadCounts returns unread DM counts grouped by peer agent.
|
||||
// For each unique from_agent that has sent DMs to agentName, it computes
|
||||
// how many messages have id > last_read_message_id (from inbox_state).
|
||||
func (s *SQLiteMessageStore) GetDMUnreadCounts(ctx context.Context, agentName string) ([]DMUnreadCount, error) {
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
`SELECT
|
||||
m.from_agent,
|
||||
COUNT(CASE WHEN m.id > COALESCE(
|
||||
(SELECT ist.last_read_message_id FROM inbox_state ist
|
||||
WHERE ist.agent_name = ? AND ist.conversation_id = m.conversation_id), 0)
|
||||
THEN 1 END) AS unread_count,
|
||||
MAX(m.id) AS last_message_id
|
||||
FROM messages m
|
||||
WHERE m.to_agent = ?
|
||||
AND m.channel_id IS NULL
|
||||
AND m.from_agent != 'system'
|
||||
GROUP BY m.from_agent
|
||||
ORDER BY m.from_agent`,
|
||||
agentName, agentName,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get dm unread counts: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var results []DMUnreadCount
|
||||
for rows.Next() {
|
||||
var d DMUnreadCount
|
||||
if err := rows.Scan(&d.Agent, &d.UnreadCount, &d.LastMessageID); err != nil {
|
||||
return nil, fmt.Errorf("scan dm unread count: %w", err)
|
||||
}
|
||||
results = append(results, d)
|
||||
}
|
||||
if results == nil {
|
||||
results = []DMUnreadCount{}
|
||||
}
|
||||
return results, rows.Err()
|
||||
}
|
||||
|
||||
// GetLastReadForChannel returns the effective last_read_message_id for an agent
|
||||
// across all conversations in a given channel.
|
||||
func (s *SQLiteMessageStore) GetLastReadForChannel(ctx context.Context, agentName string, channelID int64) (int64, error) {
|
||||
var lastRead sql.NullInt64
|
||||
err := s.db.QueryRowContext(ctx,
|
||||
`SELECT MIN(COALESCE(ist.last_read_message_id, 0))
|
||||
FROM (SELECT DISTINCT conversation_id FROM messages WHERE channel_id = ?) conv
|
||||
LEFT JOIN inbox_state ist ON ist.conversation_id = conv.conversation_id AND ist.agent_name = ?`,
|
||||
channelID, agentName,
|
||||
).Scan(&lastRead)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("get last read for channel: %w", err)
|
||||
}
|
||||
if lastRead.Valid {
|
||||
return lastRead.Int64, nil
|
||||
}
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// GetLastReadForDM returns the effective last_read_message_id for an owned agent
|
||||
// in DM conversations with a specific peer agent.
|
||||
func (s *SQLiteMessageStore) GetLastReadForDM(ctx context.Context, agentNames []string, peerAgent string) (int64, error) {
|
||||
if len(agentNames) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
placeholders := make([]string, len(agentNames))
|
||||
args := make([]any, 0, len(agentNames)*2+1)
|
||||
for i, a := range agentNames {
|
||||
placeholders[i] = "?"
|
||||
args = append(args, a)
|
||||
}
|
||||
inClause := strings.Join(placeholders, ",")
|
||||
|
||||
// Find conversations between owned agents and peer agent (DMs only)
|
||||
// and get the max last_read_message_id
|
||||
query := fmt.Sprintf(
|
||||
`SELECT COALESCE(MAX(ist.last_read_message_id), 0)
|
||||
FROM inbox_state ist
|
||||
WHERE ist.agent_name IN (%s)
|
||||
AND ist.conversation_id IN (
|
||||
SELECT DISTINCT m.conversation_id FROM messages m
|
||||
WHERE m.channel_id IS NULL
|
||||
AND ((m.from_agent IN (%s) AND m.to_agent = ?)
|
||||
OR (m.from_agent = ? AND m.to_agent IN (%s)))
|
||||
)`,
|
||||
inClause, inClause, inClause,
|
||||
)
|
||||
for _, a := range agentNames {
|
||||
args = append(args, a)
|
||||
}
|
||||
args = append(args, peerAgent, peerAgent)
|
||||
for _, a := range agentNames {
|
||||
args = append(args, a)
|
||||
}
|
||||
|
||||
var lastRead int64
|
||||
err := s.db.QueryRowContext(ctx, query, args...).Scan(&lastRead)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("get last read for dm: %w", err)
|
||||
}
|
||||
return lastRead, nil
|
||||
}
|
||||
|
||||
// GetConversationIDsForChannel returns conversation IDs in a channel with messages up to lastMessageID.
|
||||
func (s *SQLiteMessageStore) GetConversationIDsForChannel(ctx context.Context, channelID int64, lastMessageID int64) ([]int64, error) {
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
`SELECT DISTINCT conversation_id FROM messages
|
||||
WHERE channel_id = ? AND id <= ?`,
|
||||
channelID, lastMessageID,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get conversation ids for channel: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var ids []int64
|
||||
for rows.Next() {
|
||||
var id int64
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return nil, fmt.Errorf("scan conversation id: %w", err)
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return ids, rows.Err()
|
||||
}
|
||||
|
||||
// GetConversationIDsForDM returns conversation IDs for DMs between owned agents and a peer agent,
|
||||
// with messages up to lastMessageID.
|
||||
func (s *SQLiteMessageStore) GetConversationIDsForDM(ctx context.Context, agentNames []string, peerAgent string, lastMessageID int64) ([]int64, error) {
|
||||
if len(agentNames) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
placeholders := make([]string, len(agentNames))
|
||||
args := make([]any, 0, len(agentNames)*2+2)
|
||||
for i, a := range agentNames {
|
||||
placeholders[i] = "?"
|
||||
args = append(args, a)
|
||||
}
|
||||
inClause := strings.Join(placeholders, ",")
|
||||
|
||||
query := fmt.Sprintf(
|
||||
`SELECT DISTINCT conversation_id FROM messages
|
||||
WHERE channel_id IS NULL
|
||||
AND id <= ?
|
||||
AND ((from_agent IN (%s) AND to_agent = ?)
|
||||
OR (from_agent = ? AND to_agent IN (%s)))`,
|
||||
inClause, inClause,
|
||||
)
|
||||
|
||||
// Reorder args: agentNames for first IN, lastMessageID, agentNames for second IN...
|
||||
finalArgs := make([]any, 0, len(agentNames)*2+3)
|
||||
finalArgs = append(finalArgs, lastMessageID)
|
||||
for _, a := range agentNames {
|
||||
finalArgs = append(finalArgs, a)
|
||||
}
|
||||
finalArgs = append(finalArgs, peerAgent, peerAgent)
|
||||
for _, a := range agentNames {
|
||||
finalArgs = append(finalArgs, a)
|
||||
}
|
||||
|
||||
rows, err := s.db.QueryContext(ctx, query, finalArgs...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get conversation ids for dm: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var ids []int64
|
||||
for rows.Next() {
|
||||
var id int64
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return nil, fmt.Errorf("scan conversation id: %w", err)
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return ids, rows.Err()
|
||||
}
|
||||
|
||||
// scanMessage scans a single message from sql.Row.
|
||||
func scanMessage(row *sql.Row) (*Message, error) {
|
||||
var msg Message
|
||||
|
||||
@@ -72,3 +72,10 @@ type InboxState struct {
|
||||
LastReadMessageID int64 `json:"last_read_message_id"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
// DMUnreadCount holds the unread DM count for a specific peer agent.
|
||||
type DMUnreadCount struct {
|
||||
Agent string `json:"agent"`
|
||||
UnreadCount int `json:"unread_count"`
|
||||
LastMessageID int64 `json:"last_message_id"`
|
||||
}
|
||||
|
||||
Vendored
+9
-9
@@ -8,29 +8,29 @@
|
||||
<link rel="preconnect" href="https://fonts.googleapis.com">
|
||||
<link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
|
||||
<link href="https://fonts.googleapis.com/css2?family=DM+Sans:wght@400;500;600;700&family=Instrument+Sans:wght@400;500;600;700&family=JetBrains+Mono:wght@400;500&display=swap" rel="stylesheet">
|
||||
<link href="/_app/immutable/entry/start.B4zIRc6l.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/chunks/k7nCSttu.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/chunks/DBeLgT1-.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/entry/start.ClBvSxrP.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/chunks/BOwfncGc.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/chunks/DqlV_t7d.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/chunks/SAcaBy3_.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/chunks/DL-Ee-iM.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/chunks/BCvik_Lu.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/chunks/BdrVqzRy.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/entry/app.UyBC-CXX.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/chunks/VWBPw6n7.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/chunks/BaoHVn2U.js" rel="modulepreload">
|
||||
<link href="/_app/immutable/entry/app.D00BGz8q.js" rel="modulepreload">
|
||||
|
||||
</head>
|
||||
<body data-sveltekit-preload-data="hover">
|
||||
<div style="display: contents">
|
||||
<script>
|
||||
{
|
||||
__sveltekit_ymf88 = {
|
||||
__sveltekit_ra3pf6 = {
|
||||
base: ""
|
||||
};
|
||||
|
||||
const element = document.currentScript.parentElement;
|
||||
|
||||
Promise.all([
|
||||
import("/_app/immutable/entry/start.B4zIRc6l.js"),
|
||||
import("/_app/immutable/entry/app.UyBC-CXX.js")
|
||||
import("/_app/immutable/entry/start.ClBvSxrP.js"),
|
||||
import("/_app/immutable/entry/app.D00BGz8q.js")
|
||||
]).then(([kit, app]) => {
|
||||
kit.start(app, element);
|
||||
});
|
||||
|
||||
@@ -188,4 +188,12 @@ export const apiKeys = {
|
||||
get: (id: number) => request<any>('GET', `/api/keys/${id}`)
|
||||
};
|
||||
|
||||
// Notifications
|
||||
export const notificationsApi = {
|
||||
unread: () =>
|
||||
request<{ channels: Record<string, number>; dms: Record<string, number> }>('GET', '/api/notifications/unread'),
|
||||
markRead: (type: 'channel' | 'dm', target: string, lastMessageId?: number) =>
|
||||
request<{ status: string }>('POST', '/api/notifications/mark-read', { type, target, last_message_id: lastMessageId })
|
||||
};
|
||||
|
||||
export { ApiError };
|
||||
|
||||
@@ -26,7 +26,7 @@ export class SSEClient {
|
||||
};
|
||||
|
||||
// Listen for typed events
|
||||
const eventTypes = ['connected', 'new_message', 'message_updated', 'agent_connected', 'agent_disconnected', 'heartbeat'];
|
||||
const eventTypes = ['connected', 'new_message', 'message_updated', 'agent_connected', 'agent_disconnected', 'heartbeat', 'unread_update'];
|
||||
for (const type of eventTypes) {
|
||||
this.eventSource.addEventListener(type, (e: MessageEvent) => {
|
||||
try {
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
import { page } from '$app/stores';
|
||||
import { goto } from '$app/navigation';
|
||||
import { user, logout } from '$lib/stores/auth';
|
||||
import { notifications } from '$lib/stores/notifications';
|
||||
import { channels as channelsApi, agents as agentsApi, deadLetters as deadLettersApi } from '$lib/api/client';
|
||||
|
||||
let channelList = $state<any[]>([]);
|
||||
@@ -44,6 +45,10 @@
|
||||
goto('/login');
|
||||
}
|
||||
|
||||
function badgeText(count: number): string {
|
||||
return count > 99 ? '99+' : String(count);
|
||||
}
|
||||
|
||||
const adminLinks = [
|
||||
{ href: '/agents', label: 'Agents' },
|
||||
{ href: '/settings', label: 'Settings' }
|
||||
@@ -152,6 +157,7 @@
|
||||
<p class="px-3 py-1 text-xs text-text-secondary italic">No channels</p>
|
||||
{:else}
|
||||
{#each channelList as ch}
|
||||
{@const chUnread = $notifications.channels.get(ch.name) ?? 0}
|
||||
<a
|
||||
href="/channels/{ch.name}"
|
||||
class="sidebar-item {isActive('/channels/' + ch.name) ? 'sidebar-item-active' : ''}"
|
||||
@@ -160,16 +166,19 @@
|
||||
<svg class="w-4 h-4 flex-shrink-0 text-accent-purple" fill="none" stroke="currentColor" viewBox="0 0 24 24" stroke-width="1.5">
|
||||
<path stroke-linecap="round" stroke-linejoin="round" d="M9.75 17L9 20l-1 1h8l-1-1-.75-3M3 13h18M5 17h14a2 2 0 002-2V5a2 2 0 00-2-2H5a2 2 0 00-2 2v10a2 2 0 002 2z" />
|
||||
</svg>
|
||||
<span class="truncate">My Agents</span>
|
||||
<span class="truncate {chUnread > 0 ? 'font-bold text-text-primary' : ''}">My Agents</span>
|
||||
{:else}
|
||||
<span class="text-text-secondary font-mono text-xs">#</span>
|
||||
<span class="truncate">{ch.name}</span>
|
||||
<span class="truncate {chUnread > 0 ? 'font-bold text-text-primary' : ''}">{ch.name}</span>
|
||||
{/if}
|
||||
{#if ch.is_private && !ch.name.startsWith('my-agents-')}
|
||||
<svg class="w-3 h-3 text-text-secondary ml-auto flex-shrink-0" fill="currentColor" viewBox="0 0 20 20">
|
||||
<svg class="w-3 h-3 text-text-secondary {chUnread > 0 ? '' : 'ml-auto'} flex-shrink-0" fill="currentColor" viewBox="0 0 20 20">
|
||||
<path fill-rule="evenodd" d="M5 9V7a5 5 0 0110 0v2a2 2 0 012 2v5a2 2 0 01-2 2H5a2 2 0 01-2-2v-5a2 2 0 012-2zm8-2v2H7V7a3 3 0 016 0z" clip-rule="evenodd" />
|
||||
</svg>
|
||||
{/if}
|
||||
{#if chUnread > 0}
|
||||
<span class="ml-auto text-[10px] font-bold text-white bg-accent-red px-1.5 py-0.5 rounded-full min-w-[18px] text-center flex-shrink-0">{badgeText(chUnread)}</span>
|
||||
{/if}
|
||||
</a>
|
||||
{/each}
|
||||
{/if}
|
||||
@@ -196,6 +205,7 @@
|
||||
<p class="px-3 py-1 text-xs text-text-secondary italic">No agents</p>
|
||||
{:else}
|
||||
{#each agentList as agent}
|
||||
{@const dmUnread = $notifications.dms.get(agent.name) ?? 0}
|
||||
<a
|
||||
href="/dm/{agent.name}"
|
||||
class="sidebar-item {isActive('/dm/' + agent.name) ? 'sidebar-item-active' : ''}"
|
||||
@@ -208,9 +218,11 @@
|
||||
class="absolute -bottom-0.5 -right-0.5 w-2 h-2 rounded-full border border-bg-secondary {agent.status === 'active' ? 'bg-accent-green' : 'bg-text-secondary'}"
|
||||
></span>
|
||||
</span>
|
||||
<span class="truncate">{agent.display_name || agent.name}</span>
|
||||
<span class="truncate {dmUnread > 0 ? 'font-bold text-text-primary' : ''}">{agent.display_name || agent.name}</span>
|
||||
<span class="text-[9px] text-text-secondary flex-shrink-0">(you)</span>
|
||||
{#if agent.type === 'ai'}
|
||||
{#if dmUnread > 0}
|
||||
<span class="ml-auto text-[10px] font-bold text-white bg-accent-red px-1.5 py-0.5 rounded-full min-w-[18px] text-center flex-shrink-0">{badgeText(dmUnread)}</span>
|
||||
{:else if agent.type === 'ai'}
|
||||
<span class="ml-auto text-[9px] font-mono text-accent-purple bg-accent-purple/10 px-1 rounded flex-shrink-0">AI</span>
|
||||
{/if}
|
||||
</a>
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
import { writable, get } from 'svelte/store';
|
||||
|
||||
export interface UnreadCounts {
|
||||
channels: Map<string, number>;
|
||||
dms: Map<string, number>;
|
||||
totalUnread: number;
|
||||
}
|
||||
|
||||
function createNotificationStore() {
|
||||
const { subscribe, set, update } = writable<UnreadCounts>({
|
||||
channels: new Map(),
|
||||
dms: new Map(),
|
||||
totalUnread: 0
|
||||
});
|
||||
|
||||
function recalcTotal(counts: UnreadCounts): number {
|
||||
let total = 0;
|
||||
for (const v of counts.channels.values()) total += v;
|
||||
for (const v of counts.dms.values()) total += v;
|
||||
return total;
|
||||
}
|
||||
|
||||
return {
|
||||
subscribe,
|
||||
|
||||
/** Initialize from the backend GET /api/notifications/unread */
|
||||
async initialize() {
|
||||
try {
|
||||
const res = await fetch('/api/notifications/unread', { credentials: 'same-origin' });
|
||||
if (!res.ok) return;
|
||||
const data: { channels?: Record<string, number>; dms?: Record<string, number> } = await res.json();
|
||||
const channels = new Map(Object.entries(data.channels ?? {}));
|
||||
const dms = new Map(Object.entries(data.dms ?? {}));
|
||||
const counts: UnreadCounts = { channels, dms, totalUnread: 0 };
|
||||
counts.totalUnread = recalcTotal(counts);
|
||||
set(counts);
|
||||
} catch {
|
||||
// API may not be available yet — silently ignore
|
||||
}
|
||||
},
|
||||
|
||||
/** Get unread count for a channel */
|
||||
channelUnread(name: string): number {
|
||||
return get({ subscribe }).channels.get(name) ?? 0;
|
||||
},
|
||||
|
||||
/** Get unread count for a DM agent */
|
||||
dmUnread(name: string): number {
|
||||
return get({ subscribe }).dms.get(name) ?? 0;
|
||||
},
|
||||
|
||||
/** Get total unread count */
|
||||
get totalUnread(): number {
|
||||
return get({ subscribe }).totalUnread;
|
||||
},
|
||||
|
||||
/** Increment unread count for a channel or DM */
|
||||
incrementUnread(type: 'channel' | 'dm', target: string) {
|
||||
update((counts) => {
|
||||
const map = type === 'channel' ? counts.channels : counts.dms;
|
||||
map.set(target, (map.get(target) ?? 0) + 1);
|
||||
counts.totalUnread = recalcTotal(counts);
|
||||
return counts;
|
||||
});
|
||||
},
|
||||
|
||||
/** Set exact unread count for a channel or DM */
|
||||
setUnread(type: 'channel' | 'dm', target: string, count: number) {
|
||||
update((counts) => {
|
||||
const map = type === 'channel' ? counts.channels : counts.dms;
|
||||
if (count > 0) {
|
||||
map.set(target, count);
|
||||
} else {
|
||||
map.delete(target);
|
||||
}
|
||||
counts.totalUnread = recalcTotal(counts);
|
||||
return counts;
|
||||
});
|
||||
},
|
||||
|
||||
/** Mark a channel or DM as read — POST to backend and clear local count */
|
||||
async markAsRead(type: 'channel' | 'dm', target: string, lastMessageId?: number) {
|
||||
try {
|
||||
await fetch('/api/notifications/mark-read', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
credentials: 'same-origin',
|
||||
body: JSON.stringify({ type, target, last_message_id: lastMessageId })
|
||||
});
|
||||
} catch {
|
||||
// Best effort — clear locally regardless
|
||||
}
|
||||
update((counts) => {
|
||||
const map = type === 'channel' ? counts.channels : counts.dms;
|
||||
map.delete(target);
|
||||
counts.totalUnread = recalcTotal(counts);
|
||||
return counts;
|
||||
});
|
||||
},
|
||||
|
||||
/** Reset all counts (e.g. on logout) */
|
||||
reset() {
|
||||
set({ channels: new Map(), dms: new Map(), totalUnread: 0 });
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
export const notifications = createNotificationStore();
|
||||
@@ -4,16 +4,36 @@
|
||||
import { goto } from '$app/navigation';
|
||||
import { checkAuth, user, loading } from '$lib/stores/auth';
|
||||
import { SSEClient } from '$lib/api/sse';
|
||||
import { notifications } from '$lib/stores/notifications';
|
||||
import Sidebar from '$lib/components/Sidebar.svelte';
|
||||
import Header from '$lib/components/Header.svelte';
|
||||
import ThreadPanel from '$lib/components/ThreadPanel.svelte';
|
||||
|
||||
let { children } = $props();
|
||||
let sseClient: SSEClient | null = $state(null);
|
||||
let sseUnsubscribe: (() => void) | null = $state(null);
|
||||
let initialized = $state(false);
|
||||
|
||||
let isLoginPage = $derived($page.url.pathname === '/login');
|
||||
|
||||
function setupNotifications(client: SSEClient) {
|
||||
notifications.initialize();
|
||||
return client.onEvent((event) => {
|
||||
if (event.type === 'new_message') {
|
||||
const d = event.data;
|
||||
if (d.channel_name) {
|
||||
notifications.incrementUnread('channel', d.channel_name);
|
||||
} else if (d.from_agent) {
|
||||
notifications.incrementUnread('dm', d.from_agent);
|
||||
}
|
||||
} else if (event.type === 'unread_update') {
|
||||
const d = event.data;
|
||||
const type = d.type === 'channel' ? 'channel' : 'dm';
|
||||
notifications.setUnread(type as 'channel' | 'dm', d.target, d.count ?? 0);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
$effect(() => {
|
||||
if (!initialized) {
|
||||
initialized = true;
|
||||
@@ -23,6 +43,7 @@
|
||||
} else if (authenticated) {
|
||||
sseClient = new SSEClient();
|
||||
sseClient.connect();
|
||||
sseUnsubscribe = setupNotifications(sseClient);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -36,7 +57,9 @@
|
||||
|
||||
$effect(() => {
|
||||
return () => {
|
||||
sseUnsubscribe?.();
|
||||
sseClient?.disconnect();
|
||||
notifications.reset();
|
||||
};
|
||||
});
|
||||
</script>
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
<script lang="ts">
|
||||
import { page } from '$app/stores';
|
||||
import { onDestroy } from 'svelte';
|
||||
import { channels as channelsApi, messages as messagesApi, agents as agentsApi } from '$lib/api/client';
|
||||
import { openThread, closeThread } from '$lib/stores/thread';
|
||||
import { notifications } from '$lib/stores/notifications';
|
||||
|
||||
let channel = $state<any>(null);
|
||||
let members = $state<any[]>([]);
|
||||
@@ -12,12 +14,16 @@
|
||||
let joining = $state(false);
|
||||
let showInfo = $state(false);
|
||||
let leaveError = $state('');
|
||||
let lastReadMessageId = $state<number | null>(null);
|
||||
|
||||
// Compose state
|
||||
let body = $state('');
|
||||
let sending = $state(false);
|
||||
let sendError = $state('');
|
||||
|
||||
// Mark-as-read timer
|
||||
let markReadTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
|
||||
let channelName = $derived($page.params.name);
|
||||
|
||||
let messagesContainer: HTMLDivElement;
|
||||
@@ -43,14 +49,38 @@
|
||||
async function loadMessages() {
|
||||
if (!channel) return;
|
||||
try {
|
||||
const res = await channelsApi.messages(channelName);
|
||||
const res = await channelsApi.messages(channelName) as any;
|
||||
messageList = res.messages;
|
||||
lastReadMessageId = res.last_read_message_id ?? null;
|
||||
scrollToBottom();
|
||||
startMarkReadTimer();
|
||||
} catch {
|
||||
// handled
|
||||
}
|
||||
}
|
||||
|
||||
function startMarkReadTimer() {
|
||||
clearMarkReadTimer();
|
||||
const unread = notifications.channelUnread(channelName);
|
||||
if (unread > 0 && messageList.length > 0) {
|
||||
const lastMsgId = messageList[messageList.length - 1]?.id;
|
||||
markReadTimer = setTimeout(() => {
|
||||
notifications.markAsRead('channel', channelName, lastMsgId);
|
||||
}, 2000);
|
||||
}
|
||||
}
|
||||
|
||||
function clearMarkReadTimer() {
|
||||
if (markReadTimer !== null) {
|
||||
clearTimeout(markReadTimer);
|
||||
markReadTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
onDestroy(() => {
|
||||
clearMarkReadTimer();
|
||||
});
|
||||
|
||||
function scrollToBottom() {
|
||||
requestAnimationFrame(() => {
|
||||
if (messagesContainer) {
|
||||
@@ -62,7 +92,9 @@
|
||||
let _prevChannel = $state('');
|
||||
$effect(() => {
|
||||
if (channelName !== _prevChannel) {
|
||||
clearMarkReadTimer();
|
||||
_prevChannel = channelName;
|
||||
lastReadMessageId = null;
|
||||
closeThread();
|
||||
loadChannel();
|
||||
}
|
||||
@@ -212,7 +244,14 @@
|
||||
</div>
|
||||
{:else}
|
||||
<div class="py-2">
|
||||
{#each messageList as msg (msg.id)}
|
||||
{#each messageList as msg, i (msg.id)}
|
||||
{#if lastReadMessageId !== null && msg.id > lastReadMessageId && (i === 0 || messageList[i - 1].id <= lastReadMessageId)}
|
||||
<div class="flex items-center gap-3 px-5 py-1 my-1">
|
||||
<div class="flex-1 h-px bg-accent-red/50"></div>
|
||||
<span class="text-[11px] font-medium text-accent-red flex-shrink-0">New messages</span>
|
||||
<div class="flex-1 h-px bg-accent-red/50"></div>
|
||||
</div>
|
||||
{/if}
|
||||
<div class="group px-5 py-2 hover:bg-bg-tertiary/40 transition-colors relative">
|
||||
<div class="flex gap-3">
|
||||
<div class="w-9 h-9 rounded-lg {agentColor(msg.from_agent)} flex items-center justify-center text-sm font-bold text-white flex-shrink-0 mt-0.5">
|
||||
|
||||
@@ -1,19 +1,25 @@
|
||||
<script lang="ts">
|
||||
import { page } from '$app/stores';
|
||||
import { onDestroy } from 'svelte';
|
||||
import { agents as agentsApi, messages as messagesApi } from '$lib/api/client';
|
||||
import { openThread, closeThread } from '$lib/stores/thread';
|
||||
import { notifications } from '$lib/stores/notifications';
|
||||
|
||||
let peerAgent = $derived($page.params.name);
|
||||
let peer = $state<any>(null);
|
||||
let messageList = $state<any[]>([]);
|
||||
let ownAgents = $state<any[]>([]);
|
||||
let loadingData = $state(true);
|
||||
let lastReadMessageId = $state<number | null>(null);
|
||||
|
||||
// Compose state
|
||||
let body = $state('');
|
||||
let sending = $state(false);
|
||||
let sendError = $state('');
|
||||
|
||||
// Mark-as-read timer
|
||||
let markReadTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
|
||||
let messagesContainer: HTMLDivElement;
|
||||
|
||||
async function loadData() {
|
||||
@@ -35,14 +41,38 @@
|
||||
|
||||
async function loadMessages() {
|
||||
try {
|
||||
const res = await agentsApi.messages(peerAgent);
|
||||
const res = await agentsApi.messages(peerAgent) as any;
|
||||
messageList = res.messages;
|
||||
lastReadMessageId = res.last_read_message_id ?? null;
|
||||
scrollToBottom();
|
||||
startMarkReadTimer();
|
||||
} catch {
|
||||
// handled
|
||||
}
|
||||
}
|
||||
|
||||
function startMarkReadTimer() {
|
||||
clearMarkReadTimer();
|
||||
const unread = notifications.dmUnread(peerAgent);
|
||||
if (unread > 0 && messageList.length > 0) {
|
||||
const lastMsgId = messageList[messageList.length - 1]?.id;
|
||||
markReadTimer = setTimeout(() => {
|
||||
notifications.markAsRead('dm', peerAgent, lastMsgId);
|
||||
}, 2000);
|
||||
}
|
||||
}
|
||||
|
||||
function clearMarkReadTimer() {
|
||||
if (markReadTimer !== null) {
|
||||
clearTimeout(markReadTimer);
|
||||
markReadTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
onDestroy(() => {
|
||||
clearMarkReadTimer();
|
||||
});
|
||||
|
||||
function scrollToBottom() {
|
||||
requestAnimationFrame(() => {
|
||||
if (messagesContainer) {
|
||||
@@ -54,7 +84,9 @@
|
||||
let _prevPeer = $state('');
|
||||
$effect(() => {
|
||||
if (peerAgent !== _prevPeer) {
|
||||
clearMarkReadTimer();
|
||||
_prevPeer = peerAgent;
|
||||
lastReadMessageId = null;
|
||||
closeThread();
|
||||
loadData();
|
||||
}
|
||||
@@ -180,7 +212,14 @@
|
||||
</div>
|
||||
{:else}
|
||||
<div class="py-2">
|
||||
{#each messageList as msg (msg.id)}
|
||||
{#each messageList as msg, i (msg.id)}
|
||||
{#if lastReadMessageId !== null && msg.id > lastReadMessageId && (i === 0 || messageList[i - 1].id <= lastReadMessageId)}
|
||||
<div class="flex items-center gap-3 px-5 py-1 my-1">
|
||||
<div class="flex-1 h-px bg-accent-red/50"></div>
|
||||
<span class="text-[11px] font-medium text-accent-red flex-shrink-0">New messages</span>
|
||||
<div class="flex-1 h-px bg-accent-red/50"></div>
|
||||
</div>
|
||||
{/if}
|
||||
<div class="group px-5 py-2 hover:bg-bg-tertiary/40 transition-colors relative">
|
||||
<div class="flex gap-3">
|
||||
<div class="w-9 h-9 rounded-lg {agentColor(msg.from_agent)} flex items-center justify-center text-sm font-bold text-white flex-shrink-0 mt-0.5">
|
||||
|
||||
Reference in New Issue
Block a user