From 23f05660b2c5bcbc4f1f4962fd3c591238b4bb02 Mon Sep 17 00:00:00 2001 From: Mateus Bellozupko Date: Mon, 28 Sep 2026 15:43:07 -0300 Subject: [PATCH 1/6] feat(debounce): add RenewTimer for presence-driven timer extension --- pkg/debounce/service/debounce_engine.go | 33 +++++++++------ pkg/debounce/service/debounce_engine_test.go | 44 ++++++++++++++++++++ 2 files changed, 65 insertions(+), 12 deletions(-) diff --git a/pkg/debounce/service/debounce_engine.go b/pkg/debounce/service/debounce_engine.go index 1ab9fa2..079d217 100644 --- a/pkg/debounce/service/debounce_engine.go +++ b/pkg/debounce/service/debounce_engine.go @@ -14,6 +14,10 @@ import ( type DebounceEngine interface { Start(ctx context.Context, contactID, conversationID int64, content string, atts []model.Attachment, cfg model.BotConfig) error Reset(ctx context.Context, contactID, conversationID int64, content string, atts []model.Attachment, cfg model.BotConfig) error + // RenewTimer extends the debounce timer TTL without touching the buffer or + // attachments — used when a presence ("customer is typing") signal arrives + // mid-debounce, as opposed to Reset which is for an actual new message. + RenewTimer(ctx context.Context, contactID, conversationID int64, cfg model.BotConfig) error GetBuffer(ctx context.Context, contactID, conversationID int64) (string, error) GetAttachments(ctx context.Context, contactID, conversationID int64) ([]model.Attachment, error) TimerExists(ctx context.Context, contactID, conversationID int64) (bool, error) @@ -35,13 +39,7 @@ func (d *debounceEngine) Start(ctx context.Context, contactID, conversationID in if err := d.repo.AppendAttachments(ctx, contactID, conversationID, atts); err != nil { return fmt.Errorf("debounce.start.append_attachments: %w", err) } - if cfg.DebounceTime > 0 { - ttl := time.Duration(cfg.DebounceTime) * time.Second - if err := d.repo.SetTimer(ctx, contactID, conversationID, ttl); err != nil { - return fmt.Errorf("debounce.start.set_timer: %w", err) - } - } - return nil + return d.setTimerIfConfigured(ctx, contactID, conversationID, cfg, "start") } func (d *debounceEngine) Reset(ctx context.Context, contactID, conversationID int64, content string, atts []model.Attachment, cfg model.BotConfig) error { @@ -51,11 +49,22 @@ func (d *debounceEngine) Reset(ctx context.Context, contactID, conversationID in if err := d.repo.AppendAttachments(ctx, contactID, conversationID, atts); err != nil { return fmt.Errorf("debounce.reset.append_attachments: %w", err) } - if cfg.DebounceTime > 0 { - ttl := time.Duration(cfg.DebounceTime) * time.Second - if err := d.repo.SetTimer(ctx, contactID, conversationID, ttl); err != nil { - return fmt.Errorf("debounce.reset.set_timer: %w", err) - } + return d.setTimerIfConfigured(ctx, contactID, conversationID, cfg, "reset") +} + +func (d *debounceEngine) RenewTimer(ctx context.Context, contactID, conversationID int64, cfg model.BotConfig) error { + return d.setTimerIfConfigured(ctx, contactID, conversationID, cfg, "renew_timer") +} + +// setTimerIfConfigured is the timer-setting logic shared by Start, Reset and +// RenewTimer — a no-op when DebounceTime <= 0 (pass-through mode). +func (d *debounceEngine) setTimerIfConfigured(ctx context.Context, contactID, conversationID int64, cfg model.BotConfig, op string) error { + if cfg.DebounceTime <= 0 { + return nil + } + ttl := time.Duration(cfg.DebounceTime) * time.Second + if err := d.repo.SetTimer(ctx, contactID, conversationID, ttl); err != nil { + return fmt.Errorf("debounce.%s.set_timer: %w", op, err) } return nil } diff --git a/pkg/debounce/service/debounce_engine_test.go b/pkg/debounce/service/debounce_engine_test.go index 162ada8..53f0aab 100644 --- a/pkg/debounce/service/debounce_engine_test.go +++ b/pkg/debounce/service/debounce_engine_test.go @@ -152,3 +152,47 @@ func TestTimerExists_ReturnsFalseAfterExpiry(t *testing.T) { t.Error("timer must be expired after TTL") } } + +func TestRenewTimer_ExtendsTTLWithoutTouchingBuffer(t *testing.T) { + eng, rdb := setupEngine(t) + ctx := context.Background() + cfg := model.BotConfig{DebounceTime: 5} + + if err := eng.Start(ctx, 6, 6, "first message", nil, cfg); err != nil { + t.Fatalf("Start returned error: %v", err) + } + + // Let some of the TTL elapse so renewal is observable. + time.Sleep(200 * time.Millisecond) + + if err := eng.RenewTimer(ctx, 6, 6, cfg); err != nil { + t.Fatalf("RenewTimer returned error: %v", err) + } + + ttl := rdb.TTL(ctx, "bot_runtime:timer:6:6").Val() + if ttl <= 4*time.Second || ttl > 5*time.Second { + t.Errorf("timer TTL should be renewed to ~5s, got %v", ttl) + } + + buffer, err := eng.GetBuffer(ctx, 6, 6) + if err != nil { + t.Fatalf("GetBuffer returned error: %v", err) + } + if buffer != "first message" { + t.Errorf("buffer should be unchanged by RenewTimer, got %q", buffer) + } +} + +func TestRenewTimer_NoopWhenDebounceTimeIsZero(t *testing.T) { + eng, rdb := setupEngine(t) + ctx := context.Background() + + if err := eng.RenewTimer(ctx, 7, 7, model.BotConfig{DebounceTime: 0}); err != nil { + t.Fatalf("RenewTimer returned error: %v", err) + } + + exists := rdb.Exists(ctx, "bot_runtime:timer:7:7").Val() + if exists != 0 { + t.Errorf("no timer key should have been created for DebounceTime=0") + } +} From 6d696fb632df019ef8d5cbef462aa8ad3af71a64 Mon Sep 17 00:00:00 2001 From: Mateus Bellozupko Date: Mon, 28 Sep 2026 15:45:18 -0300 Subject: [PATCH 2/6] feat(model): add PresenceEvent --- pkg/pipeline/model/pipeline.go | 30 ++++++++++++++++++++--- pkg/pipeline/model/presence_event_test.go | 28 +++++++++++++++++++++ 2 files changed, 54 insertions(+), 4 deletions(-) create mode 100644 pkg/pipeline/model/presence_event_test.go diff --git a/pkg/pipeline/model/pipeline.go b/pkg/pipeline/model/pipeline.go index cf39048..b5a9196 100644 --- a/pkg/pipeline/model/pipeline.go +++ b/pkg/pipeline/model/pipeline.go @@ -73,14 +73,23 @@ type BotConfig struct { DelayPerCharacter float64 `json:"delay_per_character"` // ms per char between parts } -// Validate checks semantic constraints on a MessageEvent after JSON binding. -func (e *MessageEvent) Validate() error { - if e.ContactID <= 0 { +// validatePairID checks the ContactID/ConversationID constraints shared by +// every event type keyed on a (contact, conversation) pair. +func validatePairID(contactID, conversationID int64) error { + if contactID <= 0 { return errors.New("contact_id must be > 0") } - if e.ConversationID <= 0 { + if conversationID <= 0 { return errors.New("conversation_id must be > 0") } + return nil +} + +// Validate checks semantic constraints on a MessageEvent after JSON binding. +func (e *MessageEvent) Validate() error { + if err := validatePairID(e.ContactID, e.ConversationID); err != nil { + return err + } if e.PostbackURL == "" { return errors.New("postback_url is required") } @@ -95,3 +104,16 @@ func (e *MessageEvent) Validate() error { } return nil } + +// PresenceEvent is the inbound "customer is typing" signal from evo-ai-crm's +// presence webhook ingestion. Unlike MessageEvent, it carries no content — +// it only ever renews an already-running debounce timer. +type PresenceEvent struct { + ContactID int64 `json:"contact_id"` + ConversationID int64 `json:"conversation_id"` +} + +// Validate checks semantic constraints on a PresenceEvent after JSON binding. +func (e *PresenceEvent) Validate() error { + return validatePairID(e.ContactID, e.ConversationID) +} diff --git a/pkg/pipeline/model/presence_event_test.go b/pkg/pipeline/model/presence_event_test.go new file mode 100644 index 0000000..2763221 --- /dev/null +++ b/pkg/pipeline/model/presence_event_test.go @@ -0,0 +1,28 @@ +package model_test + +import ( + "testing" + + "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model" +) + +func TestPresenceEventValidate_ValidEvent(t *testing.T) { + e := model.PresenceEvent{ContactID: 1, ConversationID: 1} + if err := e.Validate(); err != nil { + t.Errorf("expected no error, got %v", err) + } +} + +func TestPresenceEventValidate_RejectsNonPositiveContactID(t *testing.T) { + e := model.PresenceEvent{ContactID: 0, ConversationID: 1} + if err := e.Validate(); err == nil { + t.Error("expected error for contact_id <= 0, got nil") + } +} + +func TestPresenceEventValidate_RejectsNonPositiveConversationID(t *testing.T) { + e := model.PresenceEvent{ContactID: 1, ConversationID: 0} + if err := e.Validate(); err == nil { + t.Error("expected error for conversation_id <= 0, got nil") + } +} From 34e92beb5d9793f01d4327057a99ceb17c24fa65 Mon Sep 17 00:00:00 2001 From: Mateus Bellozupko Date: Mon, 28 Sep 2026 15:50:22 -0300 Subject: [PATCH 3/6] feat(pipeline): add Presence to renew debounce timer on typing signal --- pkg/pipeline/service/pipeline_service.go | 34 ++++++++ pkg/pipeline/service/pipeline_service_test.go | 87 +++++++++++++++++++ 2 files changed, 121 insertions(+) diff --git a/pkg/pipeline/service/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index c2983d2..9558927 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -24,6 +24,10 @@ import ( // PipelineService orchestrates the per-pair pipeline state machine. type PipelineService interface { Process(ctx context.Context, event *model.MessageEvent) error + // Presence renews an already-running debounce timer when the customer is + // composing/recording — a no-op if no StageDebounce is active for this pair + // (nothing buffered yet, or the pipeline already advanced past debounce). + Presence(ctx context.Context, contactID, conversationID int64) error Cancel(contactID, conversationID int64) error Start() error // Shutdown stops the polling goroutine and cancels all in-flight pipeline @@ -268,6 +272,36 @@ func (s *pipelineService) resetDebounce(ctx context.Context, event *model.Messag return nil } +// Presence renews the debounce timer for an active StageDebounce pair. Unlike +// Process/resetDebounce, it never appends to the buffer, never starts a new +// debounce window, and never interrupts StageAI/StageDispatch — a presence +// pulse alone must not cancel a reply already being generated or sent. +func (s *pipelineService) Presence(ctx context.Context, contactID, conversationID int64) error { + mu, err := s.repo.AcquireLock(ctx, contactID, conversationID) + if err != nil { + return brtErrors.ErrLockFailed + } + defer mu.Unlock() + + state, err := s.repo.GetState(ctx, contactID, conversationID) + if err != nil { + return fmt.Errorf("pipeline.presence.get_state: %w", err) + } + if state == nil || state.Stage != model.StageDebounce { + return nil + } + + if err := s.debounce.RenewTimer(ctx, contactID, conversationID, state.BotConfig); err != nil { + return fmt.Errorf("pipeline.presence.renew_timer: %w", err) + } + + slog.Info("pipeline.debounce.renewed_by_presence", + "contact_id", contactID, + "conversation_id", conversationID, + ) + return nil +} + // advanceToAI acquires the lock and transitions the pair from StageDebounce to StageAI. // Double-check under lock prevents races with concurrent Process calls (NFR-02). func (s *pipelineService) advanceToAI(contactID, conversationID int64) { diff --git a/pkg/pipeline/service/pipeline_service_test.go b/pkg/pipeline/service/pipeline_service_test.go index a19197f..8f310ef 100644 --- a/pkg/pipeline/service/pipeline_service_test.go +++ b/pkg/pipeline/service/pipeline_service_test.go @@ -55,6 +55,7 @@ var _ dispatchIface.DispatchEngine = (*mockDispatchEngine)(nil) // compile-time type mockDebounce struct { startCalled bool resetCalled bool + renewCalled bool startErr error } @@ -69,6 +70,10 @@ func (m *mockDebounce) Reset(_ context.Context, _, _ int64, _ string, _ []model. m.resetCalled = true return nil } +func (m *mockDebounce) RenewTimer(_ context.Context, _, _ int64, _ model.BotConfig) error { + m.renewCalled = true + return nil +} func (m *mockDebounce) GetBuffer(_ context.Context, _, _ int64) (string, error) { return "", nil } func (m *mockDebounce) TimerExists(_ context.Context, _, _ int64) (bool, error) { return false, nil } @@ -816,3 +821,85 @@ func clearPair(t *testing.T, svc *pipelineService, id int64) { _ = svc.repo.ClearState(context.Background(), id, id) }) } + +func TestPresence_RenewsTimerWhenDebounceActive(t *testing.T) { + svc, rdb := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, &mockDispatchEngine{}) + clearPair(t, svc, 30000) + ctx := context.Background() + cfg := model.BotConfig{DebounceTime: 5} + + event := &model.MessageEvent{ + ContactID: 30000, ConversationID: 30000, MessageContent: "oi", + BotConfig: cfg, PostbackURL: "http://example.com/postback", + } + if err := svc.Process(ctx, event); err != nil { + t.Fatalf("Process (start debounce) returned error: %v", err) + } + + time.Sleep(200 * time.Millisecond) + + if err := svc.Presence(ctx, 30000, 30000); err != nil { + t.Fatalf("Presence returned error: %v", err) + } + + ttl := rdb.TTL(ctx, "bot_runtime:timer:30000:30000").Val() + if ttl <= 4*time.Second || ttl > 5*time.Second { + t.Errorf("timer TTL should be renewed to ~5s, got %v", ttl) + } + + buffer, err := rdb.LRange(ctx, "bot_runtime:buffer:30000:30000", 0, -1).Result() + if err != nil { + t.Fatalf("LRange returned error: %v", err) + } + if len(buffer) != 1 || buffer[0] != "oi" { + t.Errorf("buffer should be unchanged by Presence, got %v", buffer) + } +} + +func TestPresence_NoopWhenNoActiveDebounce(t *testing.T) { + svc, _ := setupSvcWithAIAndDispatch(t, &mockAIAdapter{}, &mockDispatchEngine{}) + clearPair(t, svc, 30001) + ctx := context.Background() + + // No Process call was ever made for this pair — no state exists. + if err := svc.Presence(ctx, 30001, 30001); err != nil { + t.Fatalf("Presence returned error for absent state: %v", err) + } +} + +func TestPresence_NoopDuringAIStage(t *testing.T) { + aiStarted := make(chan struct{}) + aiRelease := make(chan struct{}) + ai := &mockAIAdapter{callFn: func(ctx context.Context, req *aiModel.A2ARequest) (*aiModel.NormalizedResponse, error) { + close(aiStarted) + <-aiRelease + return &aiModel.NormalizedResponse{Content: "resposta"}, nil + }} + svc, rdb := setupSvcWithAIAndDispatch(t, ai, &mockDispatchEngine{dispatchFn: func(ctx context.Context, contactID, conversationID int64, content string, cfg model.BotConfig, postbackURL string) error { + return nil + }}) + clearPair(t, svc, 30002) + ctx := context.Background() + + event := &model.MessageEvent{ + ContactID: 30002, ConversationID: 30002, MessageContent: "oi", + BotConfig: model.BotConfig{DebounceTime: 0}, PostbackURL: "http://example.com/postback", + } + if err := svc.Process(ctx, event); err != nil { + t.Fatalf("Process (skip debounce, DebounceTime=0) returned error: %v", err) + } + <-aiStarted // now in StageAI + + if err := svc.Presence(ctx, 30002, 30002); err != nil { + t.Fatalf("Presence returned error during StageAI: %v", err) + } + + // No debounce timer should have been created — StageAI is not StageDebounce. + exists := rdb.Exists(ctx, "bot_runtime:timer:30002:30002").Val() + if exists != 0 { + t.Errorf("Presence must not create a debounce timer during StageAI") + } + + close(aiRelease) // let the AI stage finish so the goroutine doesn't leak past the test + time.Sleep(50 * time.Millisecond) +} From bc0bf2f6ce5e01db65e51224e270e1fe4b6d16a7 Mon Sep 17 00:00:00 2001 From: Mateus Bellozupko Date: Mon, 28 Sep 2026 15:54:09 -0300 Subject: [PATCH 4/6] feat(handler): add POST /events/presence --- pkg/pipeline/handler/handler.go | 47 +++++++++++++++-- pkg/pipeline/handler/handler_test.go | 76 ++++++++++++++++++++++++++-- 2 files changed, 115 insertions(+), 8 deletions(-) diff --git a/pkg/pipeline/handler/handler.go b/pkg/pipeline/handler/handler.go index e587d8a..c5945c7 100644 --- a/pkg/pipeline/handler/handler.go +++ b/pkg/pipeline/handler/handler.go @@ -25,6 +25,7 @@ func NewHandler(repo repository.PipelineRepository, svc pipelineService.Pipeline func (h *Handler) RegisterRoutes(r *gin.Engine) { r.POST("/events", SecretMiddleware(h.secret), h.handleEvent) + r.POST("/events/presence", SecretMiddleware(h.secret), h.handlePresence) r.GET("/health", h.handleHealth) } @@ -39,20 +40,35 @@ func (h *Handler) handleHealth(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"status": "ok"}) } -func (h *Handler) handleEvent(c *gin.Context) { - var event model.MessageEvent - if err := c.ShouldBindJSON(&event); err != nil { +// validatable is any inbound event with post-JSON-bind semantic checks. +type validatable interface { + Validate() error +} + +// bindAndValidate binds the request body into event and runs its Validate. +// On failure it writes the 400 response itself and returns false — the +// caller returns immediately without doing any further work. +func bindAndValidate[T validatable](c *gin.Context, event T) bool { + if err := c.ShouldBindJSON(event); err != nil { c.JSON(http.StatusBadRequest, gin.H{ "error": "invalid payload", "code": "ERR_INVALID_EVENT", }) - return + return false } if err := event.Validate(); err != nil { c.JSON(http.StatusBadRequest, gin.H{ "error": err.Error(), "code": "ERR_INVALID_EVENT", }) + return false + } + return true +} + +func (h *Handler) handleEvent(c *gin.Context) { + var event model.MessageEvent + if !bindAndValidate(c, &event) { return } @@ -77,6 +93,29 @@ func (h *Handler) handleEvent(c *gin.Context) { c.JSON(http.StatusAccepted, gin.H{"status": "accepted"}) } +// handlePresence renews an in-flight debounce timer when the customer is +// composing/recording. Unlike handleEvent, it does NOT persist StageIncoming +// before processing — a presence pulse must never overwrite an active +// StageAI/StageDispatch state, only ever renew StageDebounce. It always +// responds 202 on a structurally valid payload: "accepted and renewed" and +// "accepted and no-op" are not distinguished to the caller. +func (h *Handler) handlePresence(c *gin.Context) { + var event model.PresenceEvent + if !bindAndValidate(c, &event) { + return + } + + if err := h.svc.Presence(c.Request.Context(), event.ContactID, event.ConversationID); err != nil { + slog.Warn("pipeline.presence.failed", + "contact_id", event.ContactID, + "conversation_id", event.ConversationID, + "error", err, + ) + } + + c.JSON(http.StatusAccepted, gin.H{"status": "accepted"}) +} + func (h *Handler) runProcess(event model.MessageEvent) { if err := h.svc.Process(context.Background(), &event); err != nil { slog.Error("pipeline.process.failed", diff --git a/pkg/pipeline/handler/handler_test.go b/pkg/pipeline/handler/handler_test.go index eb4f75a..6288d61 100644 --- a/pkg/pipeline/handler/handler_test.go +++ b/pkg/pipeline/handler/handler_test.go @@ -57,12 +57,20 @@ func (m *mockRepo) ScanStates(_ context.Context, _ int) ([]model.PairID, error) func (m *mockRepo) Ping(_ context.Context) error { return m.pingErr } // mockSvc satisfies pipelineService.PipelineService for handler tests. -type mockSvc struct{ processErr error } +type mockSvc struct { + processErr error + presenceErr error + presenceCalls [][2]int64 +} func (m *mockSvc) Process(_ context.Context, _ *model.MessageEvent) error { return m.processErr } -func (m *mockSvc) Cancel(_, _ int64) error { return nil } -func (m *mockSvc) Start() error { return nil } -func (m *mockSvc) Shutdown(_ context.Context) {} +func (m *mockSvc) Presence(_ context.Context, contactID, conversationID int64) error { + m.presenceCalls = append(m.presenceCalls, [2]int64{contactID, conversationID}) + return m.presenceErr +} +func (m *mockSvc) Cancel(_, _ int64) error { return nil } +func (m *mockSvc) Start() error { return nil } +func (m *mockSvc) Shutdown(_ context.Context) {} var _ pipelineService.PipelineService = (*mockSvc)(nil) @@ -318,3 +326,63 @@ func TestHealth_503_WhenRedisUnreachable(t *testing.T) { t.Errorf("body.detail: got %q, want %q", body["detail"], "redis unreachable") } } + +func TestHandlePresence_202_OnValidRequest(t *testing.T) { + svc := &mockSvc{} + r := setupRouter(&mockRepo{}, svc) + + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/events/presence", bytes.NewReader([]byte(`{"contact_id":1,"conversation_id":1}`))) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Bot-Runtime-Secret", testSecret) + r.ServeHTTP(w, req) + + if w.Code != http.StatusAccepted { + t.Errorf("status: got %d, want %d", w.Code, http.StatusAccepted) + } + if len(svc.presenceCalls) != 1 || svc.presenceCalls[0] != [2]int64{1, 1} { + t.Errorf("expected Presence(1, 1) to have been called, got %v", svc.presenceCalls) + } +} + +func TestHandlePresence_202_EvenWhenPresenceFails(t *testing.T) { + // Presence failures are best-effort (see Global Constraints) — must never + // surface as a non-2xx to the caller. + svc := &mockSvc{presenceErr: errors.New("redis unavailable")} + r := setupRouter(&mockRepo{}, svc) + + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/events/presence", bytes.NewReader([]byte(`{"contact_id":1,"conversation_id":1}`))) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Bot-Runtime-Secret", testSecret) + r.ServeHTTP(w, req) + + if w.Code != http.StatusAccepted { + t.Errorf("status: got %d, want %d", w.Code, http.StatusAccepted) + } +} + +func TestHandlePresence_400_OnMissingRequiredFields(t *testing.T) { + r := setupRouter(&mockRepo{}, &mockSvc{}) + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/events/presence", bytes.NewReader([]byte(`{"contact_id":0,"conversation_id":1}`))) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Bot-Runtime-Secret", testSecret) + r.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("status: got %d, want %d", w.Code, http.StatusBadRequest) + } +} + +func TestHandlePresence_401_OnMissingSecret(t *testing.T) { + r := setupRouter(&mockRepo{}, &mockSvc{}) + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/events/presence", bytes.NewReader([]byte(`{"contact_id":1,"conversation_id":1}`))) + req.Header.Set("Content-Type", "application/json") + r.ServeHTTP(w, req) + + if w.Code != http.StatusUnauthorized { + t.Errorf("status: got %d, want %d", w.Code, http.StatusUnauthorized) + } +} From 461f1b39a88601c0a77d96acdcf794082f5ebc75 Mon Sep 17 00:00:00 2001 From: Mateus Bellozupko Date: Mon, 28 Sep 2026 16:57:41 -0300 Subject: [PATCH 5/6] fix(pipeline): drop the per-pair lock from Presence, fix vacuous test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Presence competing for the same WithTries(1) lock Process uses risked silently dropping a real message that lost the race against a purely cosmetic presence pulse. The operation is idempotent (GetState + SETEX), so no lock is needed; a race against a stage transition leaves at worst a stale timer key, already rendered harmless by advanceToAI's own TimerExists/Stage re-check under its own lock. TestPresence_NoopDuringAIStage previously drove the pair through a zero-DebounceTime config, so the real DebounceEngine's own no-op-at-zero behavior masked whether Presence's StageDebounce guard ran at all — deleting the guard still passed. Rewritten with a mock debounce engine (provable renew-call assertion) and a dispatch-completion assertion proving the in-flight AI/dispatch stage was not cancelled. --- pkg/pipeline/service/pipeline_service.go | 13 +++---- pkg/pipeline/service/pipeline_service_test.go | 34 ++++++++++++++----- 2 files changed, 33 insertions(+), 14 deletions(-) diff --git a/pkg/pipeline/service/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index 9558927..0c7f452 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -276,13 +276,14 @@ func (s *pipelineService) resetDebounce(ctx context.Context, event *model.Messag // Process/resetDebounce, it never appends to the buffer, never starts a new // debounce window, and never interrupts StageAI/StageDispatch — a presence // pulse alone must not cancel a reply already being generated or sent. +// Deliberately does NOT take the per-pair lock Process/resetDebounce use: +// this is a GetState + idempotent SETEX, and losing a race against a real +// message would otherwise risk that message being dropped (AcquireLock uses +// WithTries(1), so a lock miss is not retried). Worst case if this races a +// stage transition is a stale timer key on a pair that already advanced — +// harmless, since advanceToAI re-checks TimerExists/Stage under its own lock +// before doing anything with it. func (s *pipelineService) Presence(ctx context.Context, contactID, conversationID int64) error { - mu, err := s.repo.AcquireLock(ctx, contactID, conversationID) - if err != nil { - return brtErrors.ErrLockFailed - } - defer mu.Unlock() - state, err := s.repo.GetState(ctx, contactID, conversationID) if err != nil { return fmt.Errorf("pipeline.presence.get_state: %w", err) diff --git a/pkg/pipeline/service/pipeline_service_test.go b/pkg/pipeline/service/pipeline_service_test.go index 8f310ef..bc7b85b 100644 --- a/pkg/pipeline/service/pipeline_service_test.go +++ b/pkg/pipeline/service/pipeline_service_test.go @@ -867,7 +867,16 @@ func TestPresence_NoopWhenNoActiveDebounce(t *testing.T) { } } +// Uses mockDebounce (not the real engine) so "no timer created" is provable +// even for DebounceTime=0 configs: the real engine's setTimerIfConfigured +// would no-op on DebounceTime=0 regardless of whether Presence's own +// StageDebounce guard ran, which made the previous version of this test pass +// even with that guard deleted. Also asserts the AI stage actually completed +// and dispatched — proving Presence did not cancel the in-flight stage, +// the other half of the requirement (renew nothing AND don't interrupt). func TestPresence_NoopDuringAIStage(t *testing.T) { + repo := setupRealRepo(t) + mockDeb := &mockDebounce{} aiStarted := make(chan struct{}) aiRelease := make(chan struct{}) ai := &mockAIAdapter{callFn: func(ctx context.Context, req *aiModel.A2ARequest) (*aiModel.NormalizedResponse, error) { @@ -875,9 +884,12 @@ func TestPresence_NoopDuringAIStage(t *testing.T) { <-aiRelease return &aiModel.NormalizedResponse{Content: "resposta"}, nil }} - svc, rdb := setupSvcWithAIAndDispatch(t, ai, &mockDispatchEngine{dispatchFn: func(ctx context.Context, contactID, conversationID int64, content string, cfg model.BotConfig, postbackURL string) error { + dispatched := make(chan string, 1) + dispatch := &mockDispatchEngine{dispatchFn: func(ctx context.Context, contactID, conversationID int64, content string, cfg model.BotConfig, postbackURL string) error { + dispatched <- content return nil - }}) + }} + svc := &pipelineService{repo: repo, debounce: mockDeb, aiAdapter: ai, dispatchEng: dispatch} clearPair(t, svc, 30002) ctx := context.Background() @@ -894,12 +906,18 @@ func TestPresence_NoopDuringAIStage(t *testing.T) { t.Fatalf("Presence returned error during StageAI: %v", err) } - // No debounce timer should have been created — StageAI is not StageDebounce. - exists := rdb.Exists(ctx, "bot_runtime:timer:30002:30002").Val() - if exists != 0 { - t.Errorf("Presence must not create a debounce timer during StageAI") + if mockDeb.renewCalled { + t.Error("Presence must not renew the debounce timer during StageAI (nothing to renew)") } - close(aiRelease) // let the AI stage finish so the goroutine doesn't leak past the test - time.Sleep(50 * time.Millisecond) + close(aiRelease) // let the AI stage finish and reach dispatch + + select { + case content := <-dispatched: + if content != "resposta" { + t.Errorf("dispatched content = %q, want %q", content, "resposta") + } + case <-time.After(2 * time.Second): + t.Fatal("AI stage never reached dispatch — Presence may have cancelled the in-flight stage") + } } From 7e2482a4bb0e47e1ff89dec15c3a29485baf30f2 Mon Sep 17 00:00:00 2001 From: Mateus Bellozupko Date: Tue, 29 Sep 2026 05:58:34 -0300 Subject: [PATCH 6/6] fix(debounce): RenewTimer must never resurrect an expired timer key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sourcery review on evolution-foundation/evo-bot-runtime#12: RenewTimer used an unconditional SET, so a presence pulse landing in the ~100ms window after pollDebounceExpiry sees the timer key gone (but before it advances the pair to StageAI) would recreate the key and indefinitely postpone the reply. Added PipelineRepository.RenewTimerIfExists (Redis SET ... XX — only sets if the key already exists) and wired DebounceEngine.RenewTimer through it instead of the plain SetTimer used by Start/Reset, which legitimately need to create a fresh key. --- pkg/debounce/service/debounce_engine.go | 28 +++++++++-- pkg/debounce/service/debounce_engine_test.go | 21 ++++++++ pkg/pipeline/handler/handler_test.go | 3 ++ .../repository/pipeline_repository.go | 5 ++ .../repository/redis_pipeline_repository.go | 8 ++++ .../redis_pipeline_repository_test.go | 48 +++++++++++++++++++ 6 files changed, 110 insertions(+), 3 deletions(-) diff --git a/pkg/debounce/service/debounce_engine.go b/pkg/debounce/service/debounce_engine.go index 079d217..30d3059 100644 --- a/pkg/debounce/service/debounce_engine.go +++ b/pkg/debounce/service/debounce_engine.go @@ -52,23 +52,45 @@ func (d *debounceEngine) Reset(ctx context.Context, contactID, conversationID in return d.setTimerIfConfigured(ctx, contactID, conversationID, cfg, "reset") } +// RenewTimer extends an already-running timer only — unlike Start/Reset, it +// must never create one. A presence pulse landing in the ~100ms window after +// pollDebounceExpiry sees the key gone (but before it advances the pair to +// StageAI) must not resurrect that window and indefinitely postpone the reply +// (Sourcery finding on evolution-foundation/evo-bot-runtime#12). func (d *debounceEngine) RenewTimer(ctx context.Context, contactID, conversationID int64, cfg model.BotConfig) error { - return d.setTimerIfConfigured(ctx, contactID, conversationID, cfg, "renew_timer") + ttl, ok := d.timerTTLIfConfigured(cfg) + if !ok { + return nil + } + if _, err := d.repo.RenewTimerIfExists(ctx, contactID, conversationID, ttl); err != nil { + return fmt.Errorf("debounce.renew_timer.renew_timer_if_exists: %w", err) + } + return nil } // setTimerIfConfigured is the timer-setting logic shared by Start, Reset and // RenewTimer — a no-op when DebounceTime <= 0 (pass-through mode). func (d *debounceEngine) setTimerIfConfigured(ctx context.Context, contactID, conversationID int64, cfg model.BotConfig, op string) error { - if cfg.DebounceTime <= 0 { + ttl, ok := d.timerTTLIfConfigured(cfg) + if !ok { return nil } - ttl := time.Duration(cfg.DebounceTime) * time.Second if err := d.repo.SetTimer(ctx, contactID, conversationID, ttl); err != nil { return fmt.Errorf("debounce.%s.set_timer: %w", op, err) } return nil } +// timerTTLIfConfigured is the DebounceTime<=0 pass-through check shared by +// every timer-setting path (Start, Reset, RenewTimer) — ok is false when +// there's no timer to set/renew at all. +func (d *debounceEngine) timerTTLIfConfigured(cfg model.BotConfig) (ttl time.Duration, ok bool) { + if cfg.DebounceTime <= 0 { + return 0, false + } + return time.Duration(cfg.DebounceTime) * time.Second, true +} + func (d *debounceEngine) GetBuffer(ctx context.Context, contactID, conversationID int64) (string, error) { entries, err := d.repo.GetBuffer(ctx, contactID, conversationID) if err != nil { diff --git a/pkg/debounce/service/debounce_engine_test.go b/pkg/debounce/service/debounce_engine_test.go index 53f0aab..801b9c0 100644 --- a/pkg/debounce/service/debounce_engine_test.go +++ b/pkg/debounce/service/debounce_engine_test.go @@ -196,3 +196,24 @@ func TestRenewTimer_NoopWhenDebounceTimeIsZero(t *testing.T) { t.Errorf("no timer key should have been created for DebounceTime=0") } } + +// Sourcery finding on PR #12: RenewTimer must never resurrect a timer key +// that has already expired (or was never started) — otherwise a presence +// pulse landing in the poller's detection window could indefinitely +// postpone a reply the pipeline had already decided to send. +func TestRenewTimer_DoesNotResurrectAnExpiredTimer(t *testing.T) { + eng, rdb := setupEngine(t) + ctx := context.Background() + cfg := model.BotConfig{DebounceTime: 5} + + // No Start() call for this pair — the timer key was never created + // (equivalent to: it already expired and nothing has recreated it yet). + if err := eng.RenewTimer(ctx, 8, 8, cfg); err != nil { + t.Fatalf("RenewTimer returned error: %v", err) + } + + exists := rdb.Exists(ctx, "bot_runtime:timer:8:8").Val() + if exists != 0 { + t.Error("RenewTimer must not create a timer key that didn't already exist") + } +} diff --git a/pkg/pipeline/handler/handler_test.go b/pkg/pipeline/handler/handler_test.go index 6288d61..debb4c9 100644 --- a/pkg/pipeline/handler/handler_test.go +++ b/pkg/pipeline/handler/handler_test.go @@ -46,6 +46,9 @@ func (m *mockRepo) GetAttachments(_ context.Context, _, _ int64) ([]model.Attach return nil, nil } func (m *mockRepo) SetTimer(_ context.Context, _, _ int64, _ time.Duration) error { return nil } +func (m *mockRepo) RenewTimerIfExists(_ context.Context, _, _ int64, _ time.Duration) (bool, error) { + return true, nil +} func (m *mockRepo) DeleteTimer(_ context.Context, _, _ int64) error { return nil } func (m *mockRepo) TimerExists(_ context.Context, _, _ int64) (bool, error) { return false, nil } func (m *mockRepo) AcquireLock(_ context.Context, _, _ int64) (repository.Mutex, error) { diff --git a/pkg/pipeline/repository/pipeline_repository.go b/pkg/pipeline/repository/pipeline_repository.go index 97670da..f15d8a2 100644 --- a/pkg/pipeline/repository/pipeline_repository.go +++ b/pkg/pipeline/repository/pipeline_repository.go @@ -28,6 +28,11 @@ type PipelineRepository interface { GetAttachments(ctx context.Context, contactID, conversationID int64) ([]model.Attachment, error) SetTimer(ctx context.Context, contactID, conversationID int64, ttl time.Duration) error + // RenewTimerIfExists extends the timer TTL only if the key is already + // present — it never creates one. Returns whether it actually renewed, + // so a caller can tell "renewed" apart from "nothing to renew" (the + // timer already expired, or the debounce window never started). + RenewTimerIfExists(ctx context.Context, contactID, conversationID int64, ttl time.Duration) (bool, error) DeleteTimer(ctx context.Context, contactID, conversationID int64) error TimerExists(ctx context.Context, contactID, conversationID int64) (bool, error) diff --git a/pkg/pipeline/repository/redis_pipeline_repository.go b/pkg/pipeline/repository/redis_pipeline_repository.go index 48ca9d5..7fce02e 100644 --- a/pkg/pipeline/repository/redis_pipeline_repository.go +++ b/pkg/pipeline/repository/redis_pipeline_repository.go @@ -112,6 +112,14 @@ func (r *redisPipelineRepository) SetTimer(ctx context.Context, contactID, conve return r.rdb.Set(ctx, timerKey(contactID, conversationID), "1", ttl).Err() } +func (r *redisPipelineRepository) RenewTimerIfExists(ctx context.Context, contactID, conversationID int64, ttl time.Duration) (bool, error) { + ok, err := r.rdb.SetXX(ctx, timerKey(contactID, conversationID), "1", ttl).Result() + if err != nil { + return false, err + } + return ok, nil +} + func (r *redisPipelineRepository) DeleteTimer(ctx context.Context, contactID, conversationID int64) error { return r.rdb.Del(ctx, timerKey(contactID, conversationID)).Err() } diff --git a/pkg/pipeline/repository/redis_pipeline_repository_test.go b/pkg/pipeline/repository/redis_pipeline_repository_test.go index 13f9460..1dc4987 100644 --- a/pkg/pipeline/repository/redis_pipeline_repository_test.go +++ b/pkg/pipeline/repository/redis_pipeline_repository_test.go @@ -298,3 +298,51 @@ func TestAcquireLock_Contention(t *testing.T) { t.Errorf("expected exactly 1 goroutine to fail with ErrLockFailed, got %d failures (results: %v)", failures, results) } } + +// Sourcery finding on PR #12: RenewTimer must never resurrect an expired/ +// deleted timer key. If the poller already saw the key gone and is about to +// advance the pair to StageAI, a presence pulse landing in that window must +// not recreate the key and indefinitely postpone the reply. +func TestRenewTimerIfExists_NoopWhenKeyAbsent(t *testing.T) { + repo, cleanup := setupTestRepo(t) + defer cleanup() + + ctx := context.Background() + var contactID, convID int64 = 103, 203 + + renewed, err := repo.RenewTimerIfExists(ctx, contactID, convID, 5*time.Second) + if err != nil { + t.Fatalf("RenewTimerIfExists: %v", err) + } + if renewed { + t.Error("expected renewed=false when no timer key exists") + } + + exists, err := repo.TimerExists(ctx, contactID, convID) + if err != nil { + t.Fatalf("TimerExists: %v", err) + } + if exists { + t.Error("RenewTimerIfExists must not create a timer key that didn't already exist") + } +} + +func TestRenewTimerIfExists_ExtendsExistingKey(t *testing.T) { + repo, cleanup := setupTestRepo(t) + defer cleanup() + + ctx := context.Background() + var contactID, convID int64 = 104, 204 + + if err := repo.SetTimer(ctx, contactID, convID, 1*time.Second); err != nil { + t.Fatalf("SetTimer: %v", err) + } + + renewed, err := repo.RenewTimerIfExists(ctx, contactID, convID, 5*time.Second) + if err != nil { + t.Fatalf("RenewTimerIfExists: %v", err) + } + if !renewed { + t.Error("expected renewed=true when the timer key already exists") + } +}