Merge pull request #9 from dj-oyu/fix/status-message-dedup
fix: deduplicate status/task messages by editing in-place
This commit is contained in:
dj-oyu
GitHub
commit
eda12fdae4
6 changed files with 611 additions and 5 deletions
|
|
@ -219,6 +219,42 @@ func (c *DiscordChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMes
|
|||
}
|
||||
}
|
||||
|
||||
// SendWithID implements channels.MessageSenderWithID.
|
||||
// It sends a message and returns the platform message ID.
|
||||
func (c *DiscordChannel) SendWithID(ctx context.Context, chatID string, content string) (string, error) {
|
||||
if !c.IsRunning() {
|
||||
return "", channels.ErrNotRunning
|
||||
}
|
||||
|
||||
if chatID == "" {
|
||||
return "", fmt.Errorf("channel ID is empty")
|
||||
}
|
||||
|
||||
sendCtx, cancel := context.WithTimeout(ctx, sendTimeout)
|
||||
defer cancel()
|
||||
|
||||
type result struct {
|
||||
id string
|
||||
err error
|
||||
}
|
||||
done := make(chan result, 1)
|
||||
go func() {
|
||||
msg, err := c.session.ChannelMessageSend(chatID, content)
|
||||
if err != nil {
|
||||
done <- result{"", fmt.Errorf("discord send: %w", channels.ErrTemporary)}
|
||||
} else {
|
||||
done <- result{msg.ID, nil}
|
||||
}
|
||||
}()
|
||||
|
||||
select {
|
||||
case r := <-done:
|
||||
return r.id, r.err
|
||||
case <-sendCtx.Done():
|
||||
return "", sendCtx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// EditMessage implements channels.MessageEditor.
|
||||
func (c *DiscordChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error {
|
||||
_, err := c.session.ChannelMessageEdit(chatID, messageID, content)
|
||||
|
|
|
|||
|
|
@ -22,6 +22,12 @@ type ReactionCapable interface {
|
|||
ReactToMessage(ctx context.Context, chatID, messageID string) (undo func(), err error)
|
||||
}
|
||||
|
||||
// MessageSenderWithID — channels that can send a message and return its platform-specific ID.
|
||||
// Used by Manager to track status/task messages for later editing.
|
||||
type MessageSenderWithID interface {
|
||||
SendWithID(ctx context.Context, chatID string, content string) (messageID string, err error)
|
||||
}
|
||||
|
||||
// PlaceholderCapable — channels that can send a placeholder message
|
||||
// (e.g. "Thinking... 💭") that will later be edited to the actual response.
|
||||
// The channel MUST also implement MessageEditor for the placeholder to be useful.
|
||||
|
|
|
|||
|
|
@ -35,6 +35,8 @@ const (
|
|||
janitorInterval = 10 * time.Second
|
||||
typingStopTTL = 5 * time.Minute
|
||||
placeholderTTL = 10 * time.Minute
|
||||
statusMsgTTL = 5 * time.Minute
|
||||
taskMsgTTL = 30 * time.Minute
|
||||
)
|
||||
|
||||
// typingEntry wraps a typing stop function with a creation timestamp for TTL eviction.
|
||||
|
|
@ -55,6 +57,12 @@ type placeholderEntry struct {
|
|||
createdAt time.Time
|
||||
}
|
||||
|
||||
// statusMsgEntry tracks a status or task message ID for later editing.
|
||||
type statusMsgEntry struct {
|
||||
messageID string
|
||||
createdAt time.Time
|
||||
}
|
||||
|
||||
// channelRateConfig maps channel name to per-second rate limit.
|
||||
var channelRateConfig = map[string]float64{
|
||||
"telegram": 20,
|
||||
|
|
@ -80,9 +88,11 @@ type Manager struct {
|
|||
mediaStore media.MediaStore
|
||||
dispatchTask *asyncTask
|
||||
mu sync.RWMutex
|
||||
placeholders sync.Map // "channel:chatID" → placeholderID (string)
|
||||
typingStops sync.Map // "channel:chatID" → func()
|
||||
placeholders sync.Map // "channel:chatID" → placeholderEntry
|
||||
typingStops sync.Map // "channel:chatID" → typingEntry
|
||||
reactionUndos sync.Map // "channel:chatID" → reactionEntry
|
||||
statusMsgIDs sync.Map // "channel:chatID" → statusMsgEntry (streaming preview)
|
||||
taskMsgIDs sync.Map // taskID → statusMsgEntry (background task status)
|
||||
}
|
||||
|
||||
type asyncTask struct {
|
||||
|
|
@ -110,8 +120,8 @@ func (m *Manager) RecordReactionUndo(channel, chatID string, undo func()) {
|
|||
m.reactionUndos.Store(key, reactionEntry{undo: undo, createdAt: time.Now()})
|
||||
}
|
||||
|
||||
// preSend handles typing stop, reaction undo, and placeholder editing before sending a message.
|
||||
// Returns true if the message was edited into a placeholder (skip Send).
|
||||
// preSend handles typing stop, reaction undo, and placeholder/status editing before sending a message.
|
||||
// Returns true if the message was edited into an existing message (skip Send).
|
||||
func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMessage, ch Channel) bool {
|
||||
key := name + ":" + msg.ChatID
|
||||
|
||||
|
|
@ -129,7 +139,18 @@ func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMess
|
|||
}
|
||||
}
|
||||
|
||||
// 3. Try editing placeholder
|
||||
// 3. Try editing a tracked status message (from streaming preview)
|
||||
if v, loaded := m.statusMsgIDs.LoadAndDelete(key); loaded {
|
||||
if entry, ok := v.(statusMsgEntry); ok && entry.messageID != "" {
|
||||
if editor, ok := ch.(MessageEditor); ok {
|
||||
if err := editor.EditMessage(ctx, msg.ChatID, entry.messageID, msg.Content); err == nil {
|
||||
return true // edited successfully, skip Send
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 4. Try editing placeholder
|
||||
if v, loaded := m.placeholders.LoadAndDelete(key); loaded {
|
||||
if entry, ok := v.(placeholderEntry); ok && entry.id != "" {
|
||||
if editor, ok := ch.(MessageEditor); ok {
|
||||
|
|
@ -419,6 +440,17 @@ func (m *Manager) runWorker(ctx context.Context, name string, w *channelWorker)
|
|||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
// Route status/task messages to dedicated handlers
|
||||
if msg.IsStatus {
|
||||
m.handleStatusSend(ctx, name, w, msg)
|
||||
continue
|
||||
}
|
||||
if msg.IsTaskStatus {
|
||||
m.handleTaskStatusSend(ctx, name, w, msg)
|
||||
continue
|
||||
}
|
||||
|
||||
maxLen := 0
|
||||
if mlp, ok := w.ch.(MessageLengthProvider); ok {
|
||||
maxLen = mlp.MaxMessageLength()
|
||||
|
|
@ -439,6 +471,93 @@ func (m *Manager) runWorker(ctx context.Context, name string, w *channelWorker)
|
|||
}
|
||||
}
|
||||
|
||||
// handleStatusSend processes IsStatus messages (streaming previews).
|
||||
// It reuses an existing placeholder or tracked status message, or sends a new
|
||||
// one via SendWithID so subsequent status updates edit the same bubble.
|
||||
// If the channel doesn't support editing, the message is silently dropped.
|
||||
func (m *Manager) handleStatusSend(ctx context.Context, name string, w *channelWorker, msg bus.OutboundMessage) {
|
||||
if err := w.limiter.Wait(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
key := name + ":" + msg.ChatID
|
||||
|
||||
// 1. Try editing an existing placeholder
|
||||
if v, loaded := m.placeholders.Load(key); loaded {
|
||||
if entry, ok := v.(placeholderEntry); ok && entry.id != "" {
|
||||
if editor, ok := w.ch.(MessageEditor); ok {
|
||||
if err := editor.EditMessage(ctx, msg.ChatID, entry.id, msg.Content); err == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Try editing a previously tracked status message
|
||||
if v, loaded := m.statusMsgIDs.Load(key); loaded {
|
||||
if entry, ok := v.(statusMsgEntry); ok && entry.messageID != "" {
|
||||
if editor, ok := w.ch.(MessageEditor); ok {
|
||||
if err := editor.EditMessage(ctx, msg.ChatID, entry.messageID, msg.Content); err == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 3. Send new message via SendWithID and track it
|
||||
if sender, ok := w.ch.(MessageSenderWithID); ok {
|
||||
if msgID, err := sender.SendWithID(ctx, msg.ChatID, msg.Content); err == nil && msgID != "" {
|
||||
m.statusMsgIDs.Store(key, statusMsgEntry{
|
||||
messageID: msgID,
|
||||
createdAt: time.Now(),
|
||||
})
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// 4. Channel doesn't support SendWithID or editing — drop silently
|
||||
}
|
||||
|
||||
// handleTaskStatusSend processes IsTaskStatus messages (background task status).
|
||||
// It reuses a previously tracked task message, or sends a new one via SendWithID.
|
||||
// If the channel doesn't support editing, falls back to regular Send.
|
||||
func (m *Manager) handleTaskStatusSend(ctx context.Context, name string, w *channelWorker, msg bus.OutboundMessage) {
|
||||
if err := w.limiter.Wait(ctx); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
taskKey := msg.TaskID
|
||||
|
||||
// 1. Try editing an existing task message
|
||||
if taskKey != "" {
|
||||
if v, loaded := m.taskMsgIDs.Load(taskKey); loaded {
|
||||
if entry, ok := v.(statusMsgEntry); ok && entry.messageID != "" {
|
||||
if editor, ok := w.ch.(MessageEditor); ok {
|
||||
if err := editor.EditMessage(ctx, msg.ChatID, entry.messageID, msg.Content); err == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Send new message via SendWithID and track it
|
||||
if sender, ok := w.ch.(MessageSenderWithID); ok {
|
||||
if msgID, err := sender.SendWithID(ctx, msg.ChatID, msg.Content); err == nil && msgID != "" {
|
||||
if taskKey != "" {
|
||||
m.taskMsgIDs.Store(taskKey, statusMsgEntry{
|
||||
messageID: msgID,
|
||||
createdAt: time.Now(),
|
||||
})
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// 3. Fallback: regular Send (for channels without SendWithID)
|
||||
_ = w.ch.Send(ctx, msg)
|
||||
}
|
||||
|
||||
// sendWithRetry sends a message through the channel with rate limiting and
|
||||
// retry logic. It classifies errors to determine the retry strategy:
|
||||
// - ErrNotRunning / ErrSendFailed: permanent, no retry
|
||||
|
|
@ -700,6 +819,22 @@ func (m *Manager) runTTLJanitor(ctx context.Context) {
|
|||
}
|
||||
return true
|
||||
})
|
||||
m.statusMsgIDs.Range(func(key, value any) bool {
|
||||
if entry, ok := value.(statusMsgEntry); ok {
|
||||
if now.Sub(entry.createdAt) > statusMsgTTL {
|
||||
m.statusMsgIDs.Delete(key)
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
m.taskMsgIDs.Range(func(key, value any) bool {
|
||||
if entry, ok := value.(statusMsgEntry); ok {
|
||||
if now.Sub(entry.createdAt) > taskMsgTTL {
|
||||
m.taskMsgIDs.Delete(key)
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -862,3 +862,383 @@ func TestBuildMediaScope_WithMessageID(t *testing.T) {
|
|||
t.Fatalf("expected %s, got %s", expected, scope)
|
||||
}
|
||||
}
|
||||
|
||||
// --- Status / TaskStatus message handling tests ---
|
||||
|
||||
// mockEditorWithSendID implements MessageEditor and MessageSenderWithID.
|
||||
type mockEditorWithSendID struct {
|
||||
mockChannel
|
||||
editFn func(ctx context.Context, chatID, messageID, content string) error
|
||||
sendWithID func(ctx context.Context, chatID, content string) (string, error)
|
||||
}
|
||||
|
||||
func (m *mockEditorWithSendID) EditMessage(ctx context.Context, chatID, messageID, content string) error {
|
||||
return m.editFn(ctx, chatID, messageID, content)
|
||||
}
|
||||
|
||||
func (m *mockEditorWithSendID) SendWithID(ctx context.Context, chatID, content string) (string, error) {
|
||||
return m.sendWithID(ctx, chatID, content)
|
||||
}
|
||||
|
||||
func TestHandleStatusSend_EditsPlaceholder(t *testing.T) {
|
||||
m := newTestManager()
|
||||
var editCalled bool
|
||||
var editedContent string
|
||||
|
||||
ch := &mockEditorWithSendID{
|
||||
mockChannel: mockChannel{
|
||||
sendFn: func(_ context.Context, _ bus.OutboundMessage) error { return nil },
|
||||
},
|
||||
editFn: func(_ context.Context, _, messageID, content string) error {
|
||||
editCalled = true
|
||||
editedContent = content
|
||||
if messageID != "ph-42" {
|
||||
t.Fatalf("expected messageID ph-42, got %s", messageID)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
sendWithID: func(_ context.Context, _, _ string) (string, error) {
|
||||
t.Fatal("SendWithID should not be called when placeholder exists")
|
||||
return "", nil
|
||||
},
|
||||
}
|
||||
|
||||
w := &channelWorker{ch: ch, limiter: rate.NewLimiter(rate.Inf, 1)}
|
||||
|
||||
// Register a placeholder
|
||||
m.RecordPlaceholder("test", "123", "ph-42")
|
||||
|
||||
msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "status update 1", IsStatus: true}
|
||||
m.handleStatusSend(context.Background(), "test", w, msg)
|
||||
|
||||
if !editCalled {
|
||||
t.Fatal("expected EditMessage to be called on placeholder")
|
||||
}
|
||||
if editedContent != "status update 1" {
|
||||
t.Fatalf("expected content 'status update 1', got %s", editedContent)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleStatusSend_EditsTrackedStatus(t *testing.T) {
|
||||
m := newTestManager()
|
||||
var editCalled bool
|
||||
|
||||
ch := &mockEditorWithSendID{
|
||||
mockChannel: mockChannel{
|
||||
sendFn: func(_ context.Context, _ bus.OutboundMessage) error { return nil },
|
||||
},
|
||||
editFn: func(_ context.Context, _, messageID, _ string) error {
|
||||
editCalled = true
|
||||
if messageID != "status-99" {
|
||||
t.Fatalf("expected messageID status-99, got %s", messageID)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
sendWithID: func(_ context.Context, _, _ string) (string, error) {
|
||||
t.Fatal("SendWithID should not be called when statusMsgID exists")
|
||||
return "", nil
|
||||
},
|
||||
}
|
||||
|
||||
w := &channelWorker{ch: ch, limiter: rate.NewLimiter(rate.Inf, 1)}
|
||||
|
||||
// Pre-store a tracked status message
|
||||
m.statusMsgIDs.Store("test:123", statusMsgEntry{messageID: "status-99", createdAt: time.Now()})
|
||||
|
||||
msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "update 2", IsStatus: true}
|
||||
m.handleStatusSend(context.Background(), "test", w, msg)
|
||||
|
||||
if !editCalled {
|
||||
t.Fatal("expected EditMessage to be called on tracked status message")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleStatusSend_SendsNewAndTracks(t *testing.T) {
|
||||
m := newTestManager()
|
||||
var sendWithIDCalled bool
|
||||
|
||||
ch := &mockEditorWithSendID{
|
||||
mockChannel: mockChannel{
|
||||
sendFn: func(_ context.Context, _ bus.OutboundMessage) error { return nil },
|
||||
},
|
||||
editFn: func(_ context.Context, _, _, _ string) error {
|
||||
return nil
|
||||
},
|
||||
sendWithID: func(_ context.Context, chatID, content string) (string, error) {
|
||||
sendWithIDCalled = true
|
||||
if chatID != "123" {
|
||||
t.Fatalf("expected chatID 123, got %s", chatID)
|
||||
}
|
||||
return "new-msg-1", nil
|
||||
},
|
||||
}
|
||||
|
||||
w := &channelWorker{ch: ch, limiter: rate.NewLimiter(rate.Inf, 1)}
|
||||
|
||||
// No placeholder, no tracked status → should use SendWithID
|
||||
msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "first status", IsStatus: true}
|
||||
m.handleStatusSend(context.Background(), "test", w, msg)
|
||||
|
||||
if !sendWithIDCalled {
|
||||
t.Fatal("expected SendWithID to be called")
|
||||
}
|
||||
|
||||
// Verify tracked
|
||||
v, ok := m.statusMsgIDs.Load("test:123")
|
||||
if !ok {
|
||||
t.Fatal("expected statusMsgIDs to contain tracked entry")
|
||||
}
|
||||
entry := v.(statusMsgEntry)
|
||||
if entry.messageID != "new-msg-1" {
|
||||
t.Fatalf("expected messageID new-msg-1, got %s", entry.messageID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleTaskStatusSend_EditsExisting(t *testing.T) {
|
||||
m := newTestManager()
|
||||
var editCalled bool
|
||||
|
||||
ch := &mockEditorWithSendID{
|
||||
mockChannel: mockChannel{
|
||||
sendFn: func(_ context.Context, _ bus.OutboundMessage) error { return nil },
|
||||
},
|
||||
editFn: func(_ context.Context, _, messageID, content string) error {
|
||||
editCalled = true
|
||||
if messageID != "task-msg-1" {
|
||||
t.Fatalf("expected messageID task-msg-1, got %s", messageID)
|
||||
}
|
||||
if content != "task progress 50%" {
|
||||
t.Fatalf("expected content 'task progress 50%%', got %s", content)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
sendWithID: func(_ context.Context, _, _ string) (string, error) {
|
||||
t.Fatal("SendWithID should not be called when task message exists")
|
||||
return "", nil
|
||||
},
|
||||
}
|
||||
|
||||
w := &channelWorker{ch: ch, limiter: rate.NewLimiter(rate.Inf, 1)}
|
||||
|
||||
// Pre-store task message
|
||||
m.taskMsgIDs.Store("task-abc", statusMsgEntry{messageID: "task-msg-1", createdAt: time.Now()})
|
||||
|
||||
msg := bus.OutboundMessage{
|
||||
Channel: "test",
|
||||
ChatID: "123",
|
||||
Content: "task progress 50%",
|
||||
IsTaskStatus: true,
|
||||
TaskID: "task-abc",
|
||||
}
|
||||
m.handleTaskStatusSend(context.Background(), "test", w, msg)
|
||||
|
||||
if !editCalled {
|
||||
t.Fatal("expected EditMessage to be called")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleTaskStatusSend_SendsNewAndTracks(t *testing.T) {
|
||||
m := newTestManager()
|
||||
var sendWithIDCalled bool
|
||||
|
||||
ch := &mockEditorWithSendID{
|
||||
mockChannel: mockChannel{
|
||||
sendFn: func(_ context.Context, _ bus.OutboundMessage) error { return nil },
|
||||
},
|
||||
editFn: func(_ context.Context, _, _, _ string) error { return nil },
|
||||
sendWithID: func(_ context.Context, _, _ string) (string, error) {
|
||||
sendWithIDCalled = true
|
||||
return "new-task-msg", nil
|
||||
},
|
||||
}
|
||||
|
||||
w := &channelWorker{ch: ch, limiter: rate.NewLimiter(rate.Inf, 1)}
|
||||
|
||||
msg := bus.OutboundMessage{
|
||||
Channel: "test",
|
||||
ChatID: "123",
|
||||
Content: "task started",
|
||||
IsTaskStatus: true,
|
||||
TaskID: "task-xyz",
|
||||
}
|
||||
m.handleTaskStatusSend(context.Background(), "test", w, msg)
|
||||
|
||||
if !sendWithIDCalled {
|
||||
t.Fatal("expected SendWithID to be called")
|
||||
}
|
||||
|
||||
v, ok := m.taskMsgIDs.Load("task-xyz")
|
||||
if !ok {
|
||||
t.Fatal("expected taskMsgIDs to contain tracked entry")
|
||||
}
|
||||
entry := v.(statusMsgEntry)
|
||||
if entry.messageID != "new-task-msg" {
|
||||
t.Fatalf("expected messageID new-task-msg, got %s", entry.messageID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleTaskStatusSend_FallbackToSend(t *testing.T) {
|
||||
m := newTestManager()
|
||||
var sendCalled bool
|
||||
|
||||
// Channel without SendWithID — only has Send
|
||||
ch := &mockChannel{
|
||||
sendFn: func(_ context.Context, msg bus.OutboundMessage) error {
|
||||
sendCalled = true
|
||||
if msg.Content != "task status" {
|
||||
t.Fatalf("expected content 'task status', got %s", msg.Content)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
w := &channelWorker{ch: ch, limiter: rate.NewLimiter(rate.Inf, 1)}
|
||||
|
||||
msg := bus.OutboundMessage{
|
||||
Channel: "test",
|
||||
ChatID: "123",
|
||||
Content: "task status",
|
||||
IsTaskStatus: true,
|
||||
TaskID: "task-fallback",
|
||||
}
|
||||
m.handleTaskStatusSend(context.Background(), "test", w, msg)
|
||||
|
||||
if !sendCalled {
|
||||
t.Fatal("expected fallback Send to be called")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPreSend_EditsStatusMessage(t *testing.T) {
|
||||
m := newTestManager()
|
||||
var editCalled bool
|
||||
|
||||
ch := &mockMessageEditor{
|
||||
mockChannel: mockChannel{
|
||||
sendFn: func(_ context.Context, _ bus.OutboundMessage) error { return nil },
|
||||
},
|
||||
editFn: func(_ context.Context, _, messageID, _ string) error {
|
||||
editCalled = true
|
||||
if messageID != "status-msg-77" {
|
||||
t.Fatalf("expected messageID status-msg-77, got %s", messageID)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
// Store a tracked status message
|
||||
m.statusMsgIDs.Store("test:123", statusMsgEntry{messageID: "status-msg-77", createdAt: time.Now()})
|
||||
|
||||
msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "final response"}
|
||||
edited := m.preSend(context.Background(), "test", msg, ch)
|
||||
|
||||
if !edited {
|
||||
t.Fatal("expected preSend to return true (status message edited)")
|
||||
}
|
||||
if !editCalled {
|
||||
t.Fatal("expected EditMessage to be called")
|
||||
}
|
||||
|
||||
// Verify status message was consumed (LoadAndDelete)
|
||||
if _, loaded := m.statusMsgIDs.Load("test:123"); loaded {
|
||||
t.Fatal("expected statusMsgIDs entry to be deleted after preSend")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunWorker_RoutesStatusMessages(t *testing.T) {
|
||||
m := newTestManager()
|
||||
|
||||
var regularSendCount atomic.Int32
|
||||
var sendWithIDCount atomic.Int32
|
||||
|
||||
ch := &mockEditorWithSendID{
|
||||
mockChannel: mockChannel{
|
||||
sendFn: func(_ context.Context, _ bus.OutboundMessage) error {
|
||||
regularSendCount.Add(1)
|
||||
return nil
|
||||
},
|
||||
},
|
||||
editFn: func(_ context.Context, _, _, _ string) error {
|
||||
return nil
|
||||
},
|
||||
sendWithID: func(_ context.Context, _, _ string) (string, error) {
|
||||
sendWithIDCount.Add(1)
|
||||
return "tracked-1", nil
|
||||
},
|
||||
}
|
||||
|
||||
w := &channelWorker{
|
||||
ch: ch,
|
||||
queue: make(chan bus.OutboundMessage, 10),
|
||||
done: make(chan struct{}),
|
||||
limiter: rate.NewLimiter(rate.Inf, 1),
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
go m.runWorker(ctx, "test", w)
|
||||
|
||||
// Send a status message for chatID "1" (routed to handleStatusSend → SendWithID)
|
||||
w.queue <- bus.OutboundMessage{Channel: "test", ChatID: "1", Content: "status", IsStatus: true}
|
||||
// Send a task status message for chatID "2" (routed to handleTaskStatusSend → SendWithID)
|
||||
w.queue <- bus.OutboundMessage{Channel: "test", ChatID: "2", Content: "task", IsTaskStatus: true, TaskID: "t1"}
|
||||
// Send a regular message for chatID "3" (no tracked status → regular Send)
|
||||
w.queue <- bus.OutboundMessage{Channel: "test", ChatID: "3", Content: "hello"}
|
||||
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
|
||||
if regularSendCount.Load() != 1 {
|
||||
t.Fatalf("expected 1 regular Send call, got %d", regularSendCount.Load())
|
||||
}
|
||||
if sendWithIDCount.Load() != 2 {
|
||||
t.Fatalf("expected 2 SendWithID calls (status + task), got %d", sendWithIDCount.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestStatusMsgTTLJanitor(t *testing.T) {
|
||||
m := newTestManager()
|
||||
|
||||
// Store entries with timestamps in the past
|
||||
m.statusMsgIDs.Store("test:old", statusMsgEntry{
|
||||
messageID: "old-status",
|
||||
createdAt: time.Now().Add(-10 * time.Minute),
|
||||
})
|
||||
m.taskMsgIDs.Store("task-old", statusMsgEntry{
|
||||
messageID: "old-task",
|
||||
createdAt: time.Now().Add(-60 * time.Minute),
|
||||
})
|
||||
// Store a fresh entry that should survive
|
||||
m.statusMsgIDs.Store("test:fresh", statusMsgEntry{
|
||||
messageID: "fresh-status",
|
||||
createdAt: time.Now(),
|
||||
})
|
||||
|
||||
// Simulate janitor logic
|
||||
now := time.Now()
|
||||
m.statusMsgIDs.Range(func(key, value any) bool {
|
||||
if entry, ok := value.(statusMsgEntry); ok {
|
||||
if now.Sub(entry.createdAt) > statusMsgTTL {
|
||||
m.statusMsgIDs.Delete(key)
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
m.taskMsgIDs.Range(func(key, value any) bool {
|
||||
if entry, ok := value.(statusMsgEntry); ok {
|
||||
if now.Sub(entry.createdAt) > taskMsgTTL {
|
||||
m.taskMsgIDs.Delete(key)
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
|
||||
if _, loaded := m.statusMsgIDs.Load("test:old"); loaded {
|
||||
t.Fatal("expected old status entry to be evicted")
|
||||
}
|
||||
if _, loaded := m.taskMsgIDs.Load("task-old"); loaded {
|
||||
t.Fatal("expected old task entry to be evicted")
|
||||
}
|
||||
if _, loaded := m.statusMsgIDs.Load("test:fresh"); !loaded {
|
||||
t.Fatal("expected fresh status entry to survive")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -150,6 +150,26 @@ func (c *PicoChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
|
|||
return c.broadcastToSession(msg.ChatID, outMsg)
|
||||
}
|
||||
|
||||
// SendWithID implements channels.MessageSenderWithID.
|
||||
// It sends a message and returns a generated message ID.
|
||||
func (c *PicoChannel) SendWithID(ctx context.Context, chatID string, content string) (string, error) {
|
||||
if !c.IsRunning() {
|
||||
return "", channels.ErrNotRunning
|
||||
}
|
||||
|
||||
msgID := uuid.New().String()
|
||||
outMsg := newMessage(TypeMessageCreate, map[string]any{
|
||||
"content": content,
|
||||
"message_id": msgID,
|
||||
})
|
||||
|
||||
if err := c.broadcastToSession(chatID, outMsg); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return msgID, nil
|
||||
}
|
||||
|
||||
// EditMessage implements channels.MessageEditor.
|
||||
func (c *PicoChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error {
|
||||
outMsg := newMessage(TypeMessageUpdate, map[string]any{
|
||||
|
|
|
|||
|
|
@ -192,6 +192,35 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
|
|||
return nil
|
||||
}
|
||||
|
||||
// SendWithID implements channels.MessageSenderWithID.
|
||||
// It sends a message and returns the platform message ID.
|
||||
func (c *TelegramChannel) SendWithID(ctx context.Context, chatID string, content string) (string, error) {
|
||||
if !c.IsRunning() {
|
||||
return "", channels.ErrNotRunning
|
||||
}
|
||||
|
||||
cid, err := parseChatID(chatID)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("invalid chat ID %s: %w", chatID, channels.ErrSendFailed)
|
||||
}
|
||||
|
||||
htmlContent := markdownToTelegramHTML(content)
|
||||
tgMsg := tu.Message(tu.ID(cid), htmlContent)
|
||||
tgMsg.ParseMode = telego.ModeHTML
|
||||
|
||||
sent, err := c.bot.SendMessage(ctx, tgMsg)
|
||||
if err != nil {
|
||||
// Fallback to plain text
|
||||
tgMsg.ParseMode = ""
|
||||
sent, err = c.bot.SendMessage(ctx, tgMsg)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("telegram send: %w", channels.ErrTemporary)
|
||||
}
|
||||
}
|
||||
|
||||
return fmt.Sprintf("%d", sent.MessageID), nil
|
||||
}
|
||||
|
||||
// StartTyping implements channels.TypingCapable.
|
||||
// It sends ChatAction(typing) immediately and then repeats every 4 seconds
|
||||
// (Telegram's typing indicator expires after ~5s) in a background goroutine.
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue