refactor: extract manager.go fork additions into manager_ext.go
Move fork-specific status/task message handling to reduce upstream conflicts: - managerExt embedded struct with statusMsgIDs, taskMsgIDs, statusEditTimes - statusMsgEntry type, fork-specific constants - handleStatusSend, handleTaskStatusSend, generateDraftID, PromoteStatusToTask Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
dj-oyu
parent
e99ff8b96b
commit
785dae12cf
2 changed files with 298 additions and 279 deletions
|
|
@ -10,9 +10,7 @@ import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"hash/fnv"
|
|
||||||
"math"
|
"math"
|
||||||
"strings"
|
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
|
@ -37,14 +35,6 @@ const (
|
||||||
janitorInterval = 10 * time.Second
|
janitorInterval = 10 * time.Second
|
||||||
typingStopTTL = 5 * time.Minute
|
typingStopTTL = 5 * time.Minute
|
||||||
placeholderTTL = 10 * time.Minute
|
placeholderTTL = 10 * time.Minute
|
||||||
statusMsgTTL = 5 * time.Minute
|
|
||||||
taskMsgTTL = 30 * time.Minute
|
|
||||||
|
|
||||||
// statusEditInterval is the minimum interval between EditMessage calls
|
|
||||||
// for the same status/task bubble. EditMessage APIs are more rate-sensitive
|
|
||||||
// than SendMessageDraft, so we throttle edits to avoid "(edited)" flicker
|
|
||||||
// and API rate limit errors. Draft-based channels bypass this throttle.
|
|
||||||
statusEditInterval = 500 * time.Millisecond
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// typingEntry wraps a typing stop function with a creation timestamp for TTL eviction.
|
// typingEntry wraps a typing stop function with a creation timestamp for TTL eviction.
|
||||||
|
|
@ -65,13 +55,6 @@ type placeholderEntry struct {
|
||||||
createdAt time.Time
|
createdAt time.Time
|
||||||
}
|
}
|
||||||
|
|
||||||
// statusMsgEntry tracks a status or task message ID for later editing.
|
|
||||||
type statusMsgEntry struct {
|
|
||||||
messageID string
|
|
||||||
draftID int // non-zero when using draft-based streaming
|
|
||||||
createdAt time.Time
|
|
||||||
}
|
|
||||||
|
|
||||||
// channelRateConfig maps channel name to per-second rate limit.
|
// channelRateConfig maps channel name to per-second rate limit.
|
||||||
var channelRateConfig = map[string]float64{
|
var channelRateConfig = map[string]float64{
|
||||||
"telegram": 20,
|
"telegram": 20,
|
||||||
|
|
@ -93,6 +76,8 @@ type channelWorker struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
type Manager struct {
|
type Manager struct {
|
||||||
|
managerExt // fork-specific fields (see manager_ext.go)
|
||||||
|
|
||||||
channels map[string]Channel
|
channels map[string]Channel
|
||||||
workers map[string]*channelWorker
|
workers map[string]*channelWorker
|
||||||
bus *bus.MessageBus
|
bus *bus.MessageBus
|
||||||
|
|
@ -103,9 +88,6 @@ type Manager struct {
|
||||||
placeholders sync.Map // "channel:chatID" → placeholderEntry
|
placeholders sync.Map // "channel:chatID" → placeholderEntry
|
||||||
typingStops sync.Map // "channel:chatID" → typingEntry
|
typingStops sync.Map // "channel:chatID" → typingEntry
|
||||||
reactionUndos sync.Map // "channel:chatID" → reactionEntry
|
reactionUndos sync.Map // "channel:chatID" → reactionEntry
|
||||||
statusMsgIDs sync.Map // "channel:chatID" → statusMsgEntry (streaming preview)
|
|
||||||
taskMsgIDs sync.Map // "channel:chatID:taskID" → statusMsgEntry (background task status)
|
|
||||||
statusEditTimes sync.Map // key → time.Time — last EditMessage time for throttling
|
|
||||||
}
|
}
|
||||||
|
|
||||||
type asyncTask struct {
|
type asyncTask struct {
|
||||||
|
|
@ -551,235 +533,6 @@ 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.
|
|
||||||
// For channels implementing DraftSender (e.g. Telegram private chats),
|
|
||||||
// sendMessageDraft is preferred as it avoids the "(edited)" indicator.
|
|
||||||
// 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
|
|
||||||
|
|
||||||
// 0. Draft-based streaming (preferred for supported channels)
|
|
||||||
if drafter, ok := w.ch.(DraftSender); ok {
|
|
||||||
var did int
|
|
||||||
if v, loaded := m.statusMsgIDs.Load(key); loaded {
|
|
||||||
if entry, ok := v.(statusMsgEntry); ok && entry.draftID != 0 {
|
|
||||||
did = entry.draftID
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if did == 0 {
|
|
||||||
did = generateDraftID(key)
|
|
||||||
}
|
|
||||||
if err := drafter.SendDraft(ctx, msg.ChatID, did, msg.Content); err == nil {
|
|
||||||
// Track draft only after successful send. If draft fails (e.g. group
|
|
||||||
// main thread), keep existing messageID entry so fallback edits can
|
|
||||||
// reuse the same status bubble instead of creating duplicates.
|
|
||||||
m.statusMsgIDs.Store(key, statusMsgEntry{
|
|
||||||
draftID: did,
|
|
||||||
createdAt: time.Now(),
|
|
||||||
})
|
|
||||||
return
|
|
||||||
}
|
|
||||||
// Draft failed — fall through to edit-based approach
|
|
||||||
}
|
|
||||||
|
|
||||||
// Edit-based path: throttle to statusEditInterval per key to avoid
|
|
||||||
// API rate limit errors and "(edited)" flicker.
|
|
||||||
if v, loaded := m.statusEditTimes.Load(key); loaded {
|
|
||||||
if t, ok := v.(time.Time); ok && time.Since(t) < statusEditInterval {
|
|
||||||
return // too recent, skip this update
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// 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 {
|
|
||||||
m.statusEditTimes.Store(key, time.Now())
|
|
||||||
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 {
|
|
||||||
m.statusEditTimes.Store(key, time.Now())
|
|
||||||
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
|
|
||||||
}
|
|
||||||
|
|
||||||
func taskStatusKey(channel, chatID, taskID string) string {
|
|
||||||
if taskID == "" {
|
|
||||||
return ""
|
|
||||||
}
|
|
||||||
if channel == "" || chatID == "" {
|
|
||||||
return taskID
|
|
||||||
}
|
|
||||||
return channel + ":" + chatID + ":" + taskID
|
|
||||||
}
|
|
||||||
|
|
||||||
// handleTaskStatusSend processes IsTaskStatus messages (background task status).
|
|
||||||
// It reuses a previously tracked task message, or sends a new one via SendWithID.
|
|
||||||
// For channels implementing DraftSender, sendMessageDraft is used to avoid "(edited)".
|
|
||||||
// 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 := taskStatusKey(name, msg.ChatID, msg.TaskID)
|
|
||||||
|
|
||||||
// Final message: reuse the existing bubble when possible to avoid
|
|
||||||
// duplicate messages. If a permanent message (messageID) is tracked,
|
|
||||||
// edit it in-place. If a draft (draftID) is tracked, update it with
|
|
||||||
// the completion content (the draft persists in Telegram and serves
|
|
||||||
// as the visible message; sending a separate permanent message would
|
|
||||||
// create a duplicate).
|
|
||||||
if msg.Final {
|
|
||||||
v, loaded := m.taskMsgIDs.LoadAndDelete(taskKey)
|
|
||||||
m.statusEditTimes.Delete(taskKey)
|
|
||||||
|
|
||||||
if loaded {
|
|
||||||
if entry, ok := v.(statusMsgEntry); ok {
|
|
||||||
// Path A: a permanent message exists — edit it in-place.
|
|
||||||
if entry.messageID != "" {
|
|
||||||
if editor, ok := w.ch.(MessageEditor); ok {
|
|
||||||
if err := editor.EditMessage(ctx, msg.ChatID, entry.messageID, msg.Content); err == nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// Edit failed — fall through to send a new message.
|
|
||||||
}
|
|
||||||
|
|
||||||
// Path B: a draft exists — update it with the final
|
|
||||||
// content. Drafts persist visibly in Telegram, so do NOT
|
|
||||||
// send a separate permanent message (that causes duplicates).
|
|
||||||
if entry.draftID != 0 {
|
|
||||||
if drafter, ok := w.ch.(DraftSender); ok {
|
|
||||||
if err := drafter.SendDraft(ctx, msg.ChatID, entry.draftID, msg.Content); err == nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// Draft update failed — fall through to send permanent.
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// No existing bubble to reuse — send a new permanent message.
|
|
||||||
if sender, ok := w.ch.(MessageSenderWithID); ok {
|
|
||||||
if msgID, err := sender.SendWithID(ctx, msg.ChatID, msg.Content); err == nil && msgID != "" {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
_ = w.ch.Send(ctx, msg)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// 0. Draft-based streaming (preferred for supported channels)
|
|
||||||
if drafter, ok := w.ch.(DraftSender); ok && taskKey != "" {
|
|
||||||
var did int
|
|
||||||
if v, loaded := m.taskMsgIDs.Load(taskKey); loaded {
|
|
||||||
if entry, ok := v.(statusMsgEntry); ok && entry.draftID != 0 {
|
|
||||||
did = entry.draftID
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if did == 0 {
|
|
||||||
did = generateDraftID(taskKey)
|
|
||||||
}
|
|
||||||
if err := drafter.SendDraft(ctx, msg.ChatID, did, msg.Content); err == nil {
|
|
||||||
// Track draft only after successful send to avoid clobbering an
|
|
||||||
// existing messageID entry when drafts are unsupported.
|
|
||||||
m.taskMsgIDs.Store(taskKey, statusMsgEntry{
|
|
||||||
draftID: did,
|
|
||||||
createdAt: time.Now(),
|
|
||||||
})
|
|
||||||
return
|
|
||||||
}
|
|
||||||
// Draft failed — fall through to edit-based approach
|
|
||||||
}
|
|
||||||
|
|
||||||
// Edit-based path: throttle to statusEditInterval per task key.
|
|
||||||
if taskKey != "" {
|
|
||||||
if v, loaded := m.statusEditTimes.Load(taskKey); loaded {
|
|
||||||
if t, ok := v.(time.Time); ok && time.Since(t) < statusEditInterval {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// 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 {
|
|
||||||
m.statusEditTimes.Store(taskKey, time.Now())
|
|
||||||
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)
|
|
||||||
}
|
|
||||||
|
|
||||||
// generateDraftID produces a stable non-zero int from a key string.
|
|
||||||
// The same key always maps to the same draft ID so successive calls
|
|
||||||
// animate the same Telegram draft bubble.
|
|
||||||
func generateDraftID(key string) int {
|
|
||||||
h := fnv.New32a()
|
|
||||||
h.Write([]byte(key))
|
|
||||||
v := int(h.Sum32())
|
|
||||||
if v == 0 {
|
|
||||||
v = 1 // draftID must be non-zero
|
|
||||||
}
|
|
||||||
if v < 0 {
|
|
||||||
v = -v
|
|
||||||
}
|
|
||||||
return v
|
|
||||||
}
|
|
||||||
|
|
||||||
// sendWithRetry sends a message through the channel with rate limiting and
|
// sendWithRetry sends a message through the channel with rate limiting and
|
||||||
// retry logic. It classifies errors to determine the retry strategy:
|
// retry logic. It classifies errors to determine the retry strategy:
|
||||||
// - ErrNotRunning / ErrSendFailed: permanent, no retry
|
// - ErrNotRunning / ErrSendFailed: permanent, no retry
|
||||||
|
|
@ -1073,26 +826,6 @@ func (m *Manager) runTTLJanitor(ctx context.Context) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// PromoteStatusToTask moves the tracked streaming status message for the given
|
|
||||||
// channel:chatID key into the task message map under channel:chatID:taskID. This allows the
|
|
||||||
// next IsTaskStatus publish to edit the streaming bubble instead of creating a
|
|
||||||
// new message. Returns true if a status message was found and promoted.
|
|
||||||
func (m *Manager) PromoteStatusToTask(statusKey, taskID string) bool {
|
|
||||||
v, loaded := m.statusMsgIDs.LoadAndDelete(statusKey)
|
|
||||||
if !loaded {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
parts := strings.SplitN(statusKey, ":", 2)
|
|
||||||
if len(parts) == 2 {
|
|
||||||
m.taskMsgIDs.Store(taskStatusKey(parts[0], parts[1], taskID), v)
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
|
|
||||||
m.taskMsgIDs.Store(taskID, v)
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
|
|
||||||
func (m *Manager) GetChannel(name string) (Channel, bool) {
|
func (m *Manager) GetChannel(name string) (Channel, bool) {
|
||||||
m.mu.RLock()
|
m.mu.RLock()
|
||||||
defer m.mu.RUnlock()
|
defer m.mu.RUnlock()
|
||||||
|
|
|
||||||
286
pkg/channels/manager_ext.go
Normal file
286
pkg/channels/manager_ext.go
Normal file
|
|
@ -0,0 +1,286 @@
|
||||||
|
package channels
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"hash/fnv"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/sipeed/picoclaw/pkg/bus"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
statusMsgTTL = 5 * time.Minute
|
||||||
|
taskMsgTTL = 30 * time.Minute
|
||||||
|
|
||||||
|
// statusEditInterval is the minimum interval between EditMessage calls
|
||||||
|
// for the same status/task bubble. EditMessage APIs are more rate-sensitive
|
||||||
|
// than SendMessageDraft, so we throttle edits to avoid "(edited)" flicker
|
||||||
|
// and API rate limit errors. Draft-based channels bypass this throttle.
|
||||||
|
statusEditInterval = 500 * time.Millisecond
|
||||||
|
)
|
||||||
|
|
||||||
|
// statusMsgEntry tracks a status or task message ID for later editing.
|
||||||
|
type statusMsgEntry struct {
|
||||||
|
messageID string
|
||||||
|
draftID int // non-zero when using draft-based streaming
|
||||||
|
createdAt time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
// managerExt holds fork-specific fields for Manager.
|
||||||
|
// Embedded in Manager so existing field access continues to work.
|
||||||
|
type managerExt struct {
|
||||||
|
statusMsgIDs sync.Map // "channel:chatID" → statusMsgEntry (streaming preview)
|
||||||
|
taskMsgIDs sync.Map // "channel:chatID:taskID" → statusMsgEntry (background task status)
|
||||||
|
statusEditTimes sync.Map // key → time.Time — last EditMessage time for throttling
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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.
|
||||||
|
// For channels implementing DraftSender (e.g. Telegram private chats),
|
||||||
|
// sendMessageDraft is preferred as it avoids the "(edited)" indicator.
|
||||||
|
// 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
|
||||||
|
|
||||||
|
// 0. Draft-based streaming (preferred for supported channels)
|
||||||
|
if drafter, ok := w.ch.(DraftSender); ok {
|
||||||
|
var did int
|
||||||
|
if v, loaded := m.statusMsgIDs.Load(key); loaded {
|
||||||
|
if entry, ok := v.(statusMsgEntry); ok && entry.draftID != 0 {
|
||||||
|
did = entry.draftID
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if did == 0 {
|
||||||
|
did = generateDraftID(key)
|
||||||
|
}
|
||||||
|
if err := drafter.SendDraft(ctx, msg.ChatID, did, msg.Content); err == nil {
|
||||||
|
// Track draft only after successful send. If draft fails (e.g. group
|
||||||
|
// main thread), keep existing messageID entry so fallback edits can
|
||||||
|
// reuse the same status bubble instead of creating duplicates.
|
||||||
|
m.statusMsgIDs.Store(key, statusMsgEntry{
|
||||||
|
draftID: did,
|
||||||
|
createdAt: time.Now(),
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// Draft failed — fall through to edit-based approach
|
||||||
|
}
|
||||||
|
|
||||||
|
// Edit-based path: throttle to statusEditInterval per key to avoid
|
||||||
|
// API rate limit errors and "(edited)" flicker.
|
||||||
|
if v, loaded := m.statusEditTimes.Load(key); loaded {
|
||||||
|
if t, ok := v.(time.Time); ok && time.Since(t) < statusEditInterval {
|
||||||
|
return // too recent, skip this update
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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 {
|
||||||
|
m.statusEditTimes.Store(key, time.Now())
|
||||||
|
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 {
|
||||||
|
m.statusEditTimes.Store(key, time.Now())
|
||||||
|
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
|
||||||
|
}
|
||||||
|
|
||||||
|
func taskStatusKey(channel, chatID, taskID string) string {
|
||||||
|
if taskID == "" {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
if channel == "" || chatID == "" {
|
||||||
|
return taskID
|
||||||
|
}
|
||||||
|
return channel + ":" + chatID + ":" + taskID
|
||||||
|
}
|
||||||
|
|
||||||
|
// handleTaskStatusSend processes IsTaskStatus messages (background task status).
|
||||||
|
// It reuses a previously tracked task message, or sends a new one via SendWithID.
|
||||||
|
// For channels implementing DraftSender, sendMessageDraft is used to avoid "(edited)".
|
||||||
|
// 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 := taskStatusKey(name, msg.ChatID, msg.TaskID)
|
||||||
|
|
||||||
|
// Final message: reuse the existing bubble when possible to avoid
|
||||||
|
// duplicate messages. If a permanent message (messageID) is tracked,
|
||||||
|
// edit it in-place. If a draft (draftID) is tracked, update it with
|
||||||
|
// the completion content (the draft persists in Telegram and serves
|
||||||
|
// as the visible message; sending a separate permanent message would
|
||||||
|
// create a duplicate).
|
||||||
|
if msg.Final {
|
||||||
|
v, loaded := m.taskMsgIDs.LoadAndDelete(taskKey)
|
||||||
|
m.statusEditTimes.Delete(taskKey)
|
||||||
|
|
||||||
|
if loaded {
|
||||||
|
if entry, ok := v.(statusMsgEntry); ok {
|
||||||
|
// Path A: a permanent message exists — edit it in-place.
|
||||||
|
if entry.messageID != "" {
|
||||||
|
if editor, ok := w.ch.(MessageEditor); ok {
|
||||||
|
if err := editor.EditMessage(ctx, msg.ChatID, entry.messageID, msg.Content); err == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Edit failed — fall through to send a new message.
|
||||||
|
}
|
||||||
|
|
||||||
|
// Path B: a draft exists — update it with the final
|
||||||
|
// content. Drafts persist visibly in Telegram, so do NOT
|
||||||
|
// send a separate permanent message (that causes duplicates).
|
||||||
|
if entry.draftID != 0 {
|
||||||
|
if drafter, ok := w.ch.(DraftSender); ok {
|
||||||
|
if err := drafter.SendDraft(ctx, msg.ChatID, entry.draftID, msg.Content); err == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Draft update failed — fall through to send permanent.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// No existing bubble to reuse — send a new permanent message.
|
||||||
|
if sender, ok := w.ch.(MessageSenderWithID); ok {
|
||||||
|
if msgID, err := sender.SendWithID(ctx, msg.ChatID, msg.Content); err == nil && msgID != "" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_ = w.ch.Send(ctx, msg)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// 0. Draft-based streaming (preferred for supported channels)
|
||||||
|
if drafter, ok := w.ch.(DraftSender); ok && taskKey != "" {
|
||||||
|
var did int
|
||||||
|
if v, loaded := m.taskMsgIDs.Load(taskKey); loaded {
|
||||||
|
if entry, ok := v.(statusMsgEntry); ok && entry.draftID != 0 {
|
||||||
|
did = entry.draftID
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if did == 0 {
|
||||||
|
did = generateDraftID(taskKey)
|
||||||
|
}
|
||||||
|
if err := drafter.SendDraft(ctx, msg.ChatID, did, msg.Content); err == nil {
|
||||||
|
// Track draft only after successful send to avoid clobbering an
|
||||||
|
// existing messageID entry when drafts are unsupported.
|
||||||
|
m.taskMsgIDs.Store(taskKey, statusMsgEntry{
|
||||||
|
draftID: did,
|
||||||
|
createdAt: time.Now(),
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// Draft failed — fall through to edit-based approach
|
||||||
|
}
|
||||||
|
|
||||||
|
// Edit-based path: throttle to statusEditInterval per task key.
|
||||||
|
if taskKey != "" {
|
||||||
|
if v, loaded := m.statusEditTimes.Load(taskKey); loaded {
|
||||||
|
if t, ok := v.(time.Time); ok && time.Since(t) < statusEditInterval {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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 {
|
||||||
|
m.statusEditTimes.Store(taskKey, time.Now())
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
|
||||||
|
// generateDraftID produces a stable non-zero int from a key string.
|
||||||
|
// The same key always maps to the same draft ID so successive calls
|
||||||
|
// animate the same Telegram draft bubble.
|
||||||
|
func generateDraftID(key string) int {
|
||||||
|
h := fnv.New32a()
|
||||||
|
h.Write([]byte(key))
|
||||||
|
v := int(h.Sum32())
|
||||||
|
if v == 0 {
|
||||||
|
v = 1 // draftID must be non-zero
|
||||||
|
}
|
||||||
|
if v < 0 {
|
||||||
|
v = -v
|
||||||
|
}
|
||||||
|
return v
|
||||||
|
}
|
||||||
|
|
||||||
|
// PromoteStatusToTask moves the tracked streaming status message for the given
|
||||||
|
// channel:chatID key into the task message map under channel:chatID:taskID. This allows the
|
||||||
|
// next IsTaskStatus publish to edit the streaming bubble instead of creating a
|
||||||
|
// new message. Returns true if a status message was found and promoted.
|
||||||
|
func (m *Manager) PromoteStatusToTask(statusKey, taskID string) bool {
|
||||||
|
v, loaded := m.statusMsgIDs.LoadAndDelete(statusKey)
|
||||||
|
if !loaded {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
parts := strings.SplitN(statusKey, ":", 2)
|
||||||
|
if len(parts) == 2 {
|
||||||
|
m.taskMsgIDs.Store(taskStatusKey(parts[0], parts[1], taskID), v)
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
m.taskMsgIDs.Store(taskID, v)
|
||||||
|
return true
|
||||||
|
}
|
||||||
Loading…
Add table
Reference in a new issue