From 785dae12cf91cfd1843aa84c2e08554ae5cd264e Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Fri, 13 Mar 2026 16:06:58 +0900 Subject: [PATCH] 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 --- pkg/channels/manager.go | 291 ++---------------------------------- pkg/channels/manager_ext.go | 286 +++++++++++++++++++++++++++++++++++ 2 files changed, 298 insertions(+), 279 deletions(-) create mode 100644 pkg/channels/manager_ext.go diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 631f0aa1c..8e663ff86 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -10,9 +10,7 @@ import ( "context" "errors" "fmt" - "hash/fnv" "math" - "strings" "sync" "time" @@ -37,14 +35,6 @@ const ( janitorInterval = 10 * time.Second typingStopTTL = 5 * 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. @@ -65,13 +55,6 @@ type placeholderEntry struct { 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. var channelRateConfig = map[string]float64{ "telegram": 20, @@ -93,19 +76,18 @@ type channelWorker struct { } type Manager struct { - channels map[string]Channel - workers map[string]*channelWorker - bus *bus.MessageBus - config *config.Config - mediaStore media.MediaStore - dispatchTask *asyncTask - mu sync.RWMutex - 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 // "channel:chatID:taskID" → statusMsgEntry (background task status) - statusEditTimes sync.Map // key → time.Time — last EditMessage time for throttling + managerExt // fork-specific fields (see manager_ext.go) + + channels map[string]Channel + workers map[string]*channelWorker + bus *bus.MessageBus + config *config.Config + mediaStore media.MediaStore + dispatchTask *asyncTask + mu sync.RWMutex + placeholders sync.Map // "channel:chatID" → placeholderEntry + typingStops sync.Map // "channel:chatID" → typingEntry + reactionUndos sync.Map // "channel:chatID" → reactionEntry } 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 // retry logic. It classifies errors to determine the retry strategy: // - 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) { m.mu.RLock() defer m.mu.RUnlock() diff --git a/pkg/channels/manager_ext.go b/pkg/channels/manager_ext.go new file mode 100644 index 000000000..0c84cd07f --- /dev/null +++ b/pkg/channels/manager_ext.go @@ -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 +}