| 1 | package bot |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "strings" |
| 7 | |
| 8 | "reasonix/internal/sessioninbox" |
| 9 | ) |
| 10 | |
| 11 | func (gw *BotGateway) dispatchQueueResult(ctx context.Context, adapter Adapter, key string, msg InboundMessage, cleanup func(), result QueueResult) { |
| 12 | if result.Queued { |
| 13 | // Unexpected with Cap=max and Drop=new while idle; persist as fallback. |
| 14 | if rec, err := gw.followupActiveSessionDurable(ctx, adapter, key, msg); err == nil { |
| 15 | gw.storeReactionCleanup(key, cleanup) |
| 16 | _ = gw.sendText(ctx, adapter, msg, formatQueuedReceipt(rec)) |
| 17 | } else { |
| 18 | gw.storeReactionCleanup(key, cleanup) |
| 19 | } |
| 20 | return |
| 21 | } |
| 22 | if !result.Acquired { |
| 23 | gw.logger.Debug("session busy without queue action", "session", key[:8]) |
| 24 | gw.storeReactionCleanup(key, cleanup) |
| 25 | return |
| 26 | } |
| 27 | // Keep the dispatch loop free to deliver approval/answer replies while the |
| 28 | // active turn blocks. The per-session lock still serializes all turns. |
| 29 | gw.turnWG.Go(func() { gw.runTurn(ctx, adapter, key, msg, cleanup) }) |
| 30 | } |
| 31 | |
| 32 | func (gw *BotGateway) finishTurnItem(ctx context.Context, adapter Adapter, key string, fallback InboundMessage, cleanup func()) { |
| 33 | // Legacy in-memory pending first (compat), then durable session inbox. |
| 34 | next := gw.sessions.Release(key) |
| 35 | nextInboxID := "" |
| 36 | if next == nil { |
| 37 | if queued := gw.nextInboxTurn(key, fallback); queued != nil { |
| 38 | next = &queued.msg |
| 39 | nextInboxID = queued.itemID |
| 40 | // Release made the session idle. If another inbound message wins the |
| 41 | // lock, leave this disk-backed item queued for that turn's drain. |
| 42 | if !gw.sessions.TryAcquireIdle(key) { |
| 43 | if cleanup != nil { |
| 44 | cleanup() |
| 45 | } |
| 46 | return |
| 47 | } |
| 48 | } |
| 49 | } |
| 50 | if next == nil { |
| 51 | gw.flushReactionCleanups(key, cleanup) |
| 52 | return |
| 53 | } |
| 54 | if cleanup != nil { |
| 55 | cleanup() |
| 56 | } |
| 57 | nextCleanup := makeReactionCleanup(gw.takeReactionCleanups(key)) |
| 58 | gw.logger.Info("bot pending message released", "platform", next.Platform, "chat_type", next.ChatType, "chat", hashID(next.ChatID), "session", key[:8]) |
| 59 | gw.runTurnItem(ctx, adapter, key, *next, nextInboxID, nextCleanup) |
| 60 | } |
| 61 | |
| 62 | func (gw *BotGateway) followupActiveSessionDurable(ctx context.Context, adapter Adapter, key string, msg InboundMessage) (sessioninbox.InboxReceipt, error) { |
| 63 | api := gw.sessionAPI(key) |
| 64 | if api == nil { |
| 65 | return sessioninbox.InboxReceipt{}, fmt.Errorf("no session controller") |
| 66 | } |
| 67 | gw.mu.Lock() |
| 68 | state := gw.controllers[key] |
| 69 | gw.mu.Unlock() |
| 70 | msg = gw.prepareDurableInboxMessage(ctx, adapter, msg, state) |
| 71 | return enqueueViaInbox(api, msg, sessioninbox.IntentFollowup) |
| 72 | } |
| 73 | |
| 74 | func (gw *BotGateway) collectActiveSessionDurable(ctx context.Context, adapter Adapter, key string, msg InboundMessage) (sessioninbox.InboxReceipt, error) { |
| 75 | api := gw.sessionAPI(key) |
| 76 | if api == nil { |
| 77 | return sessioninbox.InboxReceipt{}, fmt.Errorf("no session controller") |
| 78 | } |
| 79 | gw.mu.Lock() |
| 80 | state := gw.controllers[key] |
| 81 | gw.mu.Unlock() |
| 82 | msg = gw.prepareDurableInboxMessage(ctx, adapter, msg, state) |
| 83 | return collectAppend(api, msg, gw.sessions.Debounce()) |
| 84 | } |
| 85 | |
| 86 | func (gw *BotGateway) interruptActiveSessionDurable(ctx context.Context, adapter Adapter, key string, msg InboundMessage) (sessioninbox.InboxReceipt, error) { |
| 87 | api := gw.sessionAPI(key) |
| 88 | if api == nil { |
| 89 | return sessioninbox.InboxReceipt{}, fmt.Errorf("no session controller") |
| 90 | } |
| 91 | gw.mu.Lock() |
| 92 | state := gw.controllers[key] |
| 93 | gw.mu.Unlock() |
| 94 | msg = gw.prepareDurableInboxMessage(ctx, adapter, msg, state) |
| 95 | return interruptEnqueue(api, msg) |
| 96 | } |
| 97 | |
| 98 | func (gw *BotGateway) prepareDurableInboxMessage(ctx context.Context, adapter Adapter, msg InboundMessage, state *sessionState) InboundMessage { |
| 99 | msg.Text = gw.inputTextWithMedia(ctx, adapter, msg, state) |
| 100 | if msg.ChatType == ChatGroup { |
| 101 | userName := strings.TrimSpace(msg.UserName) |
| 102 | if msg.ResolveUserName != nil { |
| 103 | if resolved := strings.TrimSpace(msg.ResolveUserName(ctx)); resolved != "" { |
| 104 | userName = resolved |
| 105 | } |
| 106 | } |
| 107 | msg.Text = fmt.Sprintf("[%s] %s", userName, msg.Text) |
| 108 | msg.UserName = userName |
| 109 | } |
| 110 | msg.Media = nil |
| 111 | msg.MediaURLs = nil |
| 112 | msg.ResolveUserName = nil |
| 113 | msg.Raw = nil |
| 114 | return msg |
| 115 | } |
| 116 | |
| 117 | type botInboxTurn struct { |
| 118 | itemID string |
| 119 | msg InboundMessage |
| 120 | } |
| 121 | |
| 122 | // nextInboxTurn loads the next durable FIFO follow-up with its original routing |
| 123 | // metadata. RunInboxTurn performs the atomic queued -> running claim. |
| 124 | func (gw *BotGateway) nextInboxTurn(key string, fallback InboundMessage) *botInboxTurn { |
| 125 | api := gw.sessionAPI(key) |
| 126 | if api == nil { |
| 127 | return nil |
| 128 | } |
| 129 | snap := api.InboxSnapshot() |
| 130 | if snap.Paused { |
| 131 | return nil |
| 132 | } |
| 133 | for _, it := range snap.Items { |
| 134 | if it.State != sessioninbox.StateQueued { |
| 135 | continue |
| 136 | } |
| 137 | _, env, err := api.ReadInboxItem(it.ID) |
| 138 | if err != nil { |
| 139 | continue |
| 140 | } |
| 141 | msg := botMessageFromEnvelope(env, fallback) |
| 142 | if _, hasStoredRoute := env.Extra[botInboxMessageExtraKey]; !hasStoredRoute && it.Idempotency != "" { |
| 143 | msg.MessageID = it.Idempotency |
| 144 | } |
| 145 | return &botInboxTurn{itemID: it.ID, msg: msg} |
| 146 | } |
| 147 | return nil |
| 148 | } |
| 149 | |
| 150 | // nextInboxMessage is retained for focused queue inspection tests. |
| 151 | func (gw *BotGateway) nextInboxMessage(key string) *InboundMessage { |
| 152 | next := gw.nextInboxTurn(key, InboundMessage{ChatID: key}) |
| 153 | if next == nil { |
| 154 | return nil |
| 155 | } |
| 156 | return &next.msg |
| 157 | } |
| 158 |