diff --git a/CHANGELOG.md b/CHANGELOG.md index c0006de..5f87692 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,13 @@ - `MsgContext` normalization now also carries `Chat` and `ChatID` for more Telegram update kinds, allowing policy and update handlers to rely on normalized chat identity outside message-only flows. - `MsgContext.Error(...)` and returned handler errors now suppress the automatic user reply when the error is explicitly marked with `AsInternalError(...)`, while keeping the previous user-visible default for unclassified errors. - Godoc, README examples, and regression-test naming now consistently describe the shared generic dependency model as app data, including `NoData` and `SetAppData(...)`. +- Observer configuration now treats `SetObserver(nil)` as clearing instrumentation instead of leaving the previous observer attached. +- Observer lifecycle events now cover generic update handlers and scene command, step, and message-fallback handlers with logical handler names and durations. +- `RequirePolicy(...)` now emits `PolicyCheckedEvent` for both passed and denied policy decisions. +- Scene command, step, and message-fallback flows now emit observer `ErrorEvent`s with scene-specific handler kinds and logical handler names. +- Scene transition observer events now use the same transition payload for scene command, step, and message-fallback flows. +- Observer error emission now also covers generic update handlers, callback payload decode failures, runner failures, and polling retries, including dedicated runner and polling handler kinds in `ErrorEvent`. +- `TODO.md` and the framework backlog pages now mark the observability model as completed for `v1.0.0-rc.13`. - `tgapi.Chat.Type` now uses the typed `tgapi.ChatType` enum in public DTOs and tests instead of raw string casts. ### Tests @@ -21,6 +28,9 @@ - Added table-driven update-contract coverage for `prepareUpdateCtx(...)`, including message-backed, callback-backed, user-backed, and no-user update kinds. - Added regression tests for policy middleware blocking, built-in private-chat policy decisions, normalized chat identity, and admin checks that use normalized `ChatID` and `FromID`. - Added regression tests for policy composition semantics, including all-of, any-of, and deny inversion with preserved internal failures. +- Added regression tests for `SetObserver(...)`, `GetObserver()`, and clearing the observer with `SetObserver(nil)`. +- Added observer regression tests for generic update-handler errors, callback payload decode failures, runner failure events, and polling retry emission. +- Added observer regression tests for update and scene handler lifecycle events and `PolicyCheckedEvent` emission. - Added regression tests proving that `edited_message` and `edited_channel_post` stay out of command routing and continue through generic update handlers. - Added callback-routing regression tests for both chat-message and inline-message callback targets, including `CallbackQueryId`, `CallbackMsgId`, `InlineMsgId`, and payload-argument guarantees. - Added regression tests for the new error-visibility model in both message and callback flows, including silent internal-only errors and explicit user-visible callback replies. diff --git a/TODO.md b/TODO.md index 4057971..ec11cf8 100644 --- a/TODO.md +++ b/TODO.md @@ -12,16 +12,16 @@ Russian page: Current priority split: -- `Priority 1`: observability model. - `Priority 2`: service layer and dependency graph model. - `Partial`: webhook runtime model, plugin composition contract. Completed former high-priority items: +- `[v1.0.0-rc.13] Observability model`: added first-class `Observer` events for update, command, payload, scene, policy, runner, polling, and centralized error flows, with safe event dispatch and regression coverage for the new runtime hooks. - `[v1.0.0-rc.13] Authorization and policy model`: added first-class `Policy[T]`, middleware integration through `RequirePolicy(...)`, plugin and bot policy registration helpers, built-in Telegram-aware policies, and composable `AllPolicies(...)`, `AnyPolicy(...)`, and `NotPolicy(...)` helpers with regression coverage. - `[v1.0.0-rc.13] Update schema contract`: documented and tested the normalized `MsgContext` update-routing contract, including routing categories and per-update field guarantees. - `[v1.0.0-rc.13] User-facing vs internal error model`: added explicit user-visible vs internal-only error markers and updated centralized handler error routing accordingly. - `[v1.0.0-rc.13] Configuration freeze model`: formalized bot configuration freeze after first run, documented lifecycle commit points, and added regression coverage for ignored late mutations. -- `1. Conversation / Scene Model`: completed in `v1.0.0-rc.12`. -- `2. Typed Handler Input Model`: completed in `v1.0.0-rc.12`. -- `3. Request Context / Cancellation Model`: completed in `v1.0.0-rc.12`. +- `[v1.0.0-rc.12] Conversation / Scene Model`. +- `[v1.0.0-rc.12] Typed Handler Input Model`. +- `[v1.0.0-rc.12] Request Context / Cancellation Model`. diff --git a/bot.go b/bot.go index a437b92..becf416 100644 --- a/bot.go +++ b/bot.go @@ -105,6 +105,7 @@ type Bot[T AppData] struct { uploader *tgapi.Uploader // File uploader l10n *L10n // Localization manager draftProvider *DraftProvider // Draft message builder + observer Observer // Optional event observer for instrumentation appData T // Injected application data hasAppData bool @@ -356,6 +357,7 @@ func (bot *Bot[T]) RunWithContext(ctx context.Context) error { close(bot.updateQueue) }() retryDelay := time.Duration(0) + retryCount := 0 for { select { case <-ctx.Done(): @@ -368,6 +370,19 @@ func (bot *Bot[T]) RunWithContext(ctx context.Context) error { } bot.logger.Errorln("failed to fetch updates:", err) retryDelay = nextPollRetryDelay(retryDelay) + retryCount++ + bot.safeEmitEvent(ctx, PollingRetryEvent{ + Attempt: retryCount, + Delay: retryDelay, + Err: err, + }) + bot.safeEmitEvent(ctx, ErrorEvent{ + Plugin: "bot", + HandlerKind: HandlerPollingKind, + HandlerName: "getUpdates", + Err: err, + UserFacing: false, + }) timer := time.NewTimer(retryDelay) select { case <-ctx.Done(): @@ -380,6 +395,7 @@ func (bot *Bot[T]) RunWithContext(ctx context.Context) error { continue } retryDelay = 0 + retryCount = 0 for _, update := range updates { u := update // copy loop variable to avoid race condition diff --git a/bot_config.go b/bot_config.go index a2d8cd1..2be3f4c 100644 --- a/bot_config.go +++ b/bot_config.go @@ -33,6 +33,27 @@ func (bot *Bot[T]) GetDraftProvider() *DraftProvider { return bot.draftProvider } +// SetObserver sets an event observer for instrumentation. +func (bot *Bot[T]) SetObserver(observer Observer) *Bot[T] { + if !bot.configMutable("SetObserver") { + return bot + } + if observer == nil { + if bot.logger != nil { + bot.logger.Warn("SetObserver called with nil observer; instrumentation will be disabled") + } + bot.observer = nil + return bot + } + bot.observer = observer + return bot +} + +// GetObserver returns the bot's event observer, or nil if no observer is set. +func (bot *Bot[T]) GetObserver() Observer { + return bot.observer +} + // SetSessionStore replaces the session store used for scene management. func (bot *Bot[T]) SetSessionStore(store SessionStore) *Bot[T] { if !bot.configMutable("SetSessionStore") { diff --git a/bot_test.go b/bot_test.go index 69f5832..e85c156 100644 --- a/bot_test.go +++ b/bot_test.go @@ -3,8 +3,11 @@ package laniakea import ( "context" "errors" + "io" + "net/http" "path/filepath" "reflect" + "strings" "testing" "time" @@ -12,6 +15,37 @@ import ( "git.scuroneko.dev/scuroneko/slog" ) +type pollingRoundTripFunc func(*http.Request) (*http.Response, error) + +func (f pollingRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} + +type pollingRetryObserver struct { + recordingObserver + cancel context.CancelFunc +} + +func (o *pollingRetryObserver) OnPollingRetry(ctx context.Context, ev PollingRetryEvent) { + o.recordingObserver.OnPollingRetry(ctx, ev) + if o.cancel != nil { + o.cancel() + } +} + +type testObserver struct{} + +func (testObserver) OnReceiveUpdate(context.Context, UpdateReceivedEvent) {} +func (testObserver) OnHandledUpdate(context.Context, UpdateHandledEvent) {} +func (testObserver) OnHandlerStarted(context.Context, HandlerStartedEvent) {} +func (testObserver) OnHandlerFinished(context.Context, HandlerFinishedEvent) { +} +func (testObserver) OnSceneTransition(context.Context, SceneTransitionEvent) {} +func (testObserver) OnPolicyChecked(context.Context, PolicyCheckedEvent) {} +func (testObserver) OnRunnerFinished(context.Context, RunnerFinishedEvent) {} +func (testObserver) OnPollingRetry(context.Context, PollingRetryEvent) {} +func (testObserver) OnError(context.Context, ErrorEvent) {} + func TestGetUpdateTypesReturnsCopy(t *testing.T) { bot := &Bot[NoData]{updateTypes: []tgapi.UpdateType{tgapi.UpdateTypeMessage}} @@ -196,6 +230,34 @@ func TestSetAppDataMarksValueWarningOnce(t *testing.T) { } } +func TestSetObserverAndGetObserver(t *testing.T) { + bot := &Bot[NoData]{logger: slog.CreateLogger()} + observer := testObserver{} + + if got := bot.GetObserver(); got != nil { + t.Fatalf("expected nil observer by default, got %#v", got) + } + + bot.SetObserver(observer) + if got := bot.GetObserver(); got == nil { + t.Fatal("expected observer to be stored") + } +} + +func TestSetObserverNilClearsObserver(t *testing.T) { + bot := &Bot[NoData]{logger: slog.CreateLogger()} + bot.SetObserver(testObserver{}) + + if bot.GetObserver() == nil { + t.Fatal("expected observer to be set") + } + + bot.SetObserver(nil) + if got := bot.GetObserver(); got != nil { + t.Fatalf("expected nil observer after clearing, got %#v", got) + } +} + func TestRunWithContextRejectsSecondRun(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() @@ -216,6 +278,58 @@ func TestRunWithContextRejectsSecondRun(t *testing.T) { } } +func TestRunWithContextEmitsPollingRetryAndErrorEvents(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + observer := &pollingRetryObserver{cancel: cancel} + + client := &http.Client{ + Transport: pollingRoundTripFunc(func(r *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"ok":false,"error_code":500,"description":"boom"}`)), + }, nil + }), + } + + api := tgapi.NewAPI( + tgapi.NewAPIOpts("token"). + SetAPIUrl("http://example.invalid"). + SetHTTPClient(client), + ) + defer func() { + _ = api.Close() + }() + + bot := &Bot[NoData]{ + logger: slog.CreateLogger(), + api: api, + prefixes: []string{"/"}, + plugins: []Plugin[NoData]{{name: "demo"}}, + updateQueue: make(chan *tgapi.Update, 1), + maxWorkers: 1, + observer: observer, + } + + if err := bot.RunWithContext(ctx); err != nil { + t.Fatalf("RunWithContext returned error: %v", err) + } + + if len(observer.retries) != 1 { + t.Fatalf("expected one polling retry event, got %d", len(observer.retries)) + } + if got := observer.retries[0]; got.Attempt != 1 || got.Delay <= 0 || got.Err == nil { + t.Fatalf("unexpected polling retry event: %#v", got) + } + if len(observer.errors) != 1 { + t.Fatalf("expected one polling error event, got %d", len(observer.errors)) + } + if got := observer.errors[0]; got.HandlerKind != HandlerPollingKind || got.HandlerName != "getUpdates" || got.Plugin != "bot" || got.Err == nil || got.UserFacing { + t.Fatalf("unexpected polling error event: %#v", got) + } +} + func TestBotConfigurationFreezesAfterRunStarts(t *testing.T) { type testDB struct{ Name string } diff --git a/handler.go b/handler.go index 89797e7..16046f2 100644 --- a/handler.go +++ b/handler.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "time" "git.scuroneko.dev/scuroneko/laniakea/tgapi" ) @@ -19,6 +20,7 @@ func (bot *Bot[T]) handle(parentCtx context.Context, u *tgapi.Update) { bot.logger.Errorln(fmt.Sprintf("panic in handle: %v", r)) } }() + startTime := time.Now() ctx, cancel := context.WithCancel(parentCtx) defer cancel() @@ -30,10 +32,17 @@ func (bot *Bot[T]) handle(parentCtx context.Context, u *tgapi.Update) { l10n: bot.l10n, draftProvider: bot.draftProvider, sceneRuntime: bot, + observer: bot.observer, payloadType: bot.payloadType, ctx: ctx, } bot.prepareUpdateCtx(u, msgCtx) + bot.safeEmitEvent(ctx, UpdateReceivedEvent{ + UpdateID: u.UpdateID, + UpdateType: u.Type, + FromID: msgCtx.FromID, + ChatID: msgCtx.ChatID, + }) for _, middleware := range bot.middlewares { if !middleware.Execute(msgCtx, bot.appData) { @@ -44,20 +53,56 @@ func (bot *Bot[T]) handle(parentCtx context.Context, u *tgapi.Update) { sceneHandled, err := bot.tryHandleScene(msgCtx) if err != nil { bot.logger.Errorln(err) + bot.safeEmitEvent(ctx, UpdateHandledEvent{ + UpdateID: u.UpdateID, + UpdateType: u.Type, + FromID: msgCtx.FromID, + ChatID: msgCtx.ChatID, + Duration: time.Since(startTime), + Handled: false, + }) + bot.safeEmitEvent(ctx, ErrorEvent{ + UpdateID: u.UpdateID, + UpdateType: u.Type, + Plugin: "bot", + HandlerKind: HandlerSceneKind, + HandlerName: "tryHandleScene", + FromID: msgCtx.FromID, + ChatID: msgCtx.ChatID, + Err: err, + UserFacing: false, + }) return } if sceneHandled { + bot.safeEmitEvent(ctx, UpdateHandledEvent{ + UpdateID: u.UpdateID, + UpdateType: u.Type, + FromID: msgCtx.FromID, + ChatID: msgCtx.ChatID, + Duration: time.Since(startTime), + Handled: true, + }) return } + handled := false switch u.Type { case tgapi.UpdateTypeMessage, tgapi.UpdateTypeChannelPost: - bot.handleMessage(u, msgCtx) + handled = bot.handleMessage(u, msgCtx) case tgapi.UpdateTypeCallbackQuery: - bot.handleCallback(u, msgCtx) + handled = bot.handleCallback(u, msgCtx) default: - bot.handleUpdate(u, msgCtx) + handled = bot.handleUpdate(u, msgCtx) } + bot.safeEmitEvent(ctx, UpdateHandledEvent{ + UpdateID: u.UpdateID, + UpdateType: u.Type, + FromID: msgCtx.FromID, + ChatID: msgCtx.ChatID, + Duration: time.Since(startTime), + Handled: handled, + }) } func cloneMsgContext(src *MsgContext) *MsgContext { diff --git a/handler_test.go b/handler_test.go index 3b0674c..757f14d 100644 --- a/handler_test.go +++ b/handler_test.go @@ -2,6 +2,7 @@ package laniakea import ( "context" + "errors" "testing" "git.scuroneko.dev/scuroneko/laniakea/tgapi" @@ -12,6 +13,37 @@ func ptr[T any](v T) *T { return &v } +type recordingObserver struct { + started []HandlerStartedEvent + finished []HandlerFinishedEvent + errors []ErrorEvent + policies []PolicyCheckedEvent + runners []RunnerFinishedEvent + retries []PollingRetryEvent +} + +func (*recordingObserver) OnReceiveUpdate(context.Context, UpdateReceivedEvent) {} +func (*recordingObserver) OnHandledUpdate(context.Context, UpdateHandledEvent) {} +func (o *recordingObserver) OnHandlerStarted(_ context.Context, ev HandlerStartedEvent) { + o.started = append(o.started, ev) +} +func (o *recordingObserver) OnHandlerFinished(_ context.Context, ev HandlerFinishedEvent) { + o.finished = append(o.finished, ev) +} +func (*recordingObserver) OnSceneTransition(context.Context, SceneTransitionEvent) {} +func (o *recordingObserver) OnPolicyChecked(_ context.Context, ev PolicyCheckedEvent) { + o.policies = append(o.policies, ev) +} +func (o *recordingObserver) OnRunnerFinished(_ context.Context, ev RunnerFinishedEvent) { + o.runners = append(o.runners, ev) +} +func (o *recordingObserver) OnPollingRetry(_ context.Context, ev PollingRetryEvent) { + o.retries = append(o.retries, ev) +} +func (o *recordingObserver) OnError(_ context.Context, ev ErrorEvent) { + o.errors = append(o.errors, ev) +} + func TestCheckPrefixesSkipsEmptyPrefixes(t *testing.T) { bot := &Bot[NoData]{prefixes: []string{"", "/"}} @@ -505,6 +537,57 @@ func TestHandleUpdateHandlersReceiveIsolatedContexts(t *testing.T) { } } +func TestHandleUpdateObserverEmitsUpdateErrors(t *testing.T) { + observer := &recordingObserver{} + plugin := NewPlugin[NoData]("test").AddUpdateHandler(tgapi.UpdateTypeInlineQuery, func(ctx *MsgContext, db NoData) error { + return AsUserError(errors.New("update failed")) + }) + + bot := &Bot[NoData]{ + logger: slog.CreateLogger(), + plugins: []Plugin[NoData]{clonePlugin(plugin)}, + observer: observer, + } + + bot.handle(context.Background(), &tgapi.Update{ + UpdateID: 4, + Type: tgapi.UpdateTypeInlineQuery, + InlineQuery: &tgapi.InlineQuery{ + ID: "iq", + From: tgapi.User{ID: 41}, + }, + }) + + if len(observer.errors) != 1 { + t.Fatalf("expected one observer error event, got %d", len(observer.errors)) + } + ev := observer.errors[0] + if ev.Plugin != "test" { + t.Fatalf("unexpected plugin: %q", ev.Plugin) + } + if ev.HandlerKind != HandlerUpdateKind { + t.Fatalf("unexpected handler kind: %q", ev.HandlerKind) + } + if ev.HandlerName != string(tgapi.UpdateTypeInlineQuery) { + t.Fatalf("unexpected handler name: %q", ev.HandlerName) + } + if !ev.UserFacing { + t.Fatal("expected update error to be marked user-facing") + } + if len(observer.started) != 1 { + t.Fatalf("expected one handler started event, got %d", len(observer.started)) + } + if got := observer.started[0]; got.HandlerKind != HandlerUpdateKind || got.HandlerName != string(tgapi.UpdateTypeInlineQuery) || got.Plugin != "test" { + t.Fatalf("unexpected started event: %#v", got) + } + if len(observer.finished) != 1 { + t.Fatalf("expected one handler finished event, got %d", len(observer.finished)) + } + if got := observer.finished[0]; got.HandlerKind != HandlerUpdateKind || got.HandlerName != string(tgapi.UpdateTypeInlineQuery) || got.Plugin != "test" || got.Err == nil || !got.UserFacing { + t.Fatalf("unexpected finished event: %#v", got) + } +} + func TestHandleChannelPostCommandWithSenderChat(t *testing.T) { called := false plugin := NewPlugin[NoData]("test") @@ -832,3 +915,157 @@ func TestHandleCallbackPopulatesInlineTargets(t *testing.T) { t.Fatal("expected inline payload handler to be called") } } + +func TestHandleCallbackObserverEmitsPayloadEvents(t *testing.T) { + observer := &recordingObserver{} + plugin := NewPlugin[NoData]("test") + plugin.NewPayload(func(ctx *MsgContext, db NoData) error { + return nil + }, "approve") + + bot := &Bot[NoData]{ + logger: slog.CreateLogger(), + payloadType: BotPayloadJson, + plugins: []Plugin[NoData]{clonePlugin(plugin)}, + observer: observer, + } + + data, err := encodeJsonPayload(CallbackData{Command: "approve", Args: []string{"7"}}) + if err != nil { + t.Fatalf("encodeJsonPayload returned error: %v", err) + } + + bot.handle(context.Background(), &tgapi.Update{ + UpdateID: 32, + Type: tgapi.UpdateTypeCallbackQuery, + CallbackQuery: &tgapi.CallbackQuery{ + ID: "cb-observer", + Data: data, + From: tgapi.User{ID: 7}, + Message: &tgapi.Message{ + MessageID: 56, + Chat: &tgapi.Chat{ID: 78}, + }, + }, + }) + + if len(observer.started) != 1 { + t.Fatalf("expected one started event, got %d", len(observer.started)) + } + if got := observer.started[0]; got.HandlerKind != HandlerPayloadKind || got.HandlerName != "approve" || got.Plugin != "test" { + t.Fatalf("unexpected started event: %#v", got) + } + if len(observer.finished) != 1 { + t.Fatalf("expected one finished event, got %d", len(observer.finished)) + } + if got := observer.finished[0]; got.HandlerKind != HandlerPayloadKind || got.HandlerName != "approve" || got.Plugin != "test" || got.Err != nil || got.UserFacing { + t.Fatalf("unexpected finished event: %#v", got) + } + if len(observer.errors) != 0 { + t.Fatalf("did not expect error events, got %#v", observer.errors) + } +} + +func TestHandleCallbackObserverEmitsPayloadErrors(t *testing.T) { + observer := &recordingObserver{} + plugin := NewPlugin[NoData]("test") + wantErr := AsInternalError(errors.New("boom")) + plugin.NewPayload(func(ctx *MsgContext, db NoData) error { + return wantErr + }, "approve") + + bot := &Bot[NoData]{ + logger: slog.CreateLogger(), + payloadType: BotPayloadJson, + plugins: []Plugin[NoData]{clonePlugin(plugin)}, + observer: observer, + } + + data, err := encodeJsonPayload(CallbackData{Command: "approve", Args: []string{"7"}}) + if err != nil { + t.Fatalf("encodeJsonPayload returned error: %v", err) + } + + bot.handle(context.Background(), &tgapi.Update{ + UpdateID: 33, + Type: tgapi.UpdateTypeCallbackQuery, + CallbackQuery: &tgapi.CallbackQuery{ + ID: "cb-observer-err", + Data: data, + From: tgapi.User{ID: 7}, + Message: &tgapi.Message{ + MessageID: 57, + Chat: &tgapi.Chat{ID: 79}, + }, + }, + }) + + if len(observer.started) != 1 { + t.Fatalf("expected one started event, got %d", len(observer.started)) + } + if len(observer.finished) != 1 { + t.Fatalf("expected one finished event, got %d", len(observer.finished)) + } + if got := observer.finished[0]; !errors.Is(got.Err, wantErr) || got.UserFacing { + t.Fatalf("unexpected finished event: %#v", got) + } + if len(observer.errors) != 1 { + t.Fatalf("expected one error event, got %d", len(observer.errors)) + } + if got := observer.errors[0]; !errors.Is(got.Err, wantErr) || got.HandlerKind != HandlerPayloadKind || got.HandlerName != "approve" || got.Plugin != "test" || got.UserFacing { + t.Fatalf("unexpected error event: %#v", got) + } +} + +func TestHandleCallbackObserverEmitsDecodeErrors(t *testing.T) { + observer := &recordingObserver{} + bot := &Bot[NoData]{ + logger: slog.CreateLogger(), + payloadType: BotPayloadJson, + observer: observer, + } + + handled := bot.handleCallback(&tgapi.Update{ + UpdateID: 34, + Type: tgapi.UpdateTypeCallbackQuery, + CallbackQuery: &tgapi.CallbackQuery{ + ID: "cb-bad", + Data: "{not-json", + From: tgapi.User{ID: 7}, + }, + }, &MsgContext{ + Update: tgapi.Update{ + UpdateID: 34, + Type: tgapi.UpdateTypeCallbackQuery, + }, + Logger: bot.logger, + ctx: context.Background(), + CallbackQueryId: "cb-bad", + From: &tgapi.User{ID: 7}, + FromID: 7, + sceneRuntime: bot, + }) + + if handled { + t.Fatal("expected invalid callback payload to stay unhandled") + } + if len(observer.started) != 0 || len(observer.finished) != 0 { + t.Fatalf("expected no handler lifecycle events for decode failure, got started=%d finished=%d", len(observer.started), len(observer.finished)) + } + if len(observer.errors) != 1 { + t.Fatalf("expected one observer error event, got %d", len(observer.errors)) + } + ev := observer.errors[0] + if ev.Plugin != "bot" { + t.Fatalf("unexpected plugin: %q", ev.Plugin) + } + if ev.HandlerKind != HandlerPayloadKind { + t.Fatalf("unexpected handler kind: %q", ev.HandlerKind) + } + if ev.HandlerName != "decodePayload" { + t.Fatalf("unexpected handler name: %q", ev.HandlerName) + } + if ev.UserFacing { + t.Fatal("expected decode failure to stay internal") + } +} diff --git a/msg_context.go b/msg_context.go index b749013..9edb8b2 100644 --- a/msg_context.go +++ b/msg_context.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "log" "reflect" "strconv" "strings" @@ -79,6 +80,7 @@ type MsgContext struct { draftProvider *DraftProvider payloadType BotPayloadType sceneRuntime sceneRuntime + observer Observer ctx context.Context } @@ -689,6 +691,22 @@ func (ctx *MsgContext) Context() context.Context { return ctx.ctx } +func (ctx *MsgContext) emitPolicyChecked(event PolicyCheckedEvent) { + if ctx == nil || ctx.observer == nil { + return + } + defer func() { + if r := recover(); r != nil { + if ctx.Logger != nil { + ctx.Logger.Errorln(fmt.Sprintf("panic in observer policy event: %v", r)) + return + } + log.Printf("panic in observer policy event: %v", r) + } + }() + ctx.observer.OnPolicyChecked(ctx.Context(), event) +} + // EnterScene enters the named scene at its configured entry step. func (ctx *MsgContext) EnterScene(name string) error { if ctx.sceneRuntime == nil { diff --git a/msg_handler.go b/msg_handler.go index 39dba5b..8d355d2 100644 --- a/msg_handler.go +++ b/msg_handler.go @@ -2,18 +2,19 @@ package laniakea import ( "strings" + "time" "git.scuroneko.dev/scuroneko/laniakea/tgapi" ) -func (bot *Bot[T]) handleMessage(update *tgapi.Update, ctx *MsgContext) { +func (bot *Bot[T]) handleMessage(update *tgapi.Update, ctx *MsgContext) bool { var msg *tgapi.Message if update.Message != nil { msg = update.Message } else if update.ChannelPost != nil { msg = update.ChannelPost } else { - return + return false } var text string @@ -22,12 +23,12 @@ func (bot *Bot[T]) handleMessage(update *tgapi.Update, ctx *MsgContext) { } else if len(msg.Caption) > 0 { text = msg.Caption } else { - return + return false } prefix, cmd, args := bot.parseCommand(text) if cmd == "" { - return + return false } ctx.Prefix = prefix @@ -37,10 +38,10 @@ func (bot *Bot[T]) handleMessage(update *tgapi.Update, ctx *MsgContext) { cmd = cmd[:len(cmd)-len("@"+botUsername)] // убираем @botname } } - // Ищем команду по точному совпадению for _, plugin := range bot.plugins { if _, exists := plugin.commands[cmd]; exists { + ctx.Text = args ctx.Args = strings.Fields(args) // Убирает лишние пробелы @@ -48,19 +49,75 @@ func (bot *Bot[T]) handleMessage(update *tgapi.Update, ctx *MsgContext) { ctx.Logger = plugin.logger } if !plugin.executeMiddlewares(ctx, bot.appData) { - return + return false } - plugin.executeCmd(cmd, ctx, bot.appData) - return + + startTime := time.Now() + bot.safeEmitEvent(ctx.Context(), HandlerStartedEvent{ + UpdateID: update.UpdateID, + UpdateType: update.Type, + Plugin: plugin.name, + HandlerKind: HandlerCommandKind, + HandlerName: cmd, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + }) + + err := plugin.executeCmd(cmd, ctx, bot.appData) + handlerEndEvent := HandlerFinishedEvent{ + UpdateID: update.UpdateID, + UpdateType: update.Type, + Plugin: plugin.name, + HandlerKind: HandlerCommandKind, + HandlerName: cmd, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + Duration: time.Since(startTime), + } + + var errorEvent *ErrorEvent = nil + if err != nil { + ctx.error(err) + handlerEndEvent.Err = err + handlerEndEvent.UserFacing = IsUserError(err) + errorEvent = &ErrorEvent{ + UpdateID: update.UpdateID, + UpdateType: update.Type, + Plugin: plugin.name, + HandlerKind: HandlerCommandKind, + HandlerName: cmd, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + Err: err, + UserFacing: handlerEndEvent.UserFacing, + } + } + bot.safeEmitEvent(ctx.Context(), handlerEndEvent) + if errorEvent != nil { + bot.safeEmitEvent(ctx.Context(), *errorEvent) + } + return true } } + return false } -func (bot *Bot[T]) handleCallback(update *tgapi.Update, ctx *MsgContext) { +func (bot *Bot[T]) handleCallback(update *tgapi.Update, ctx *MsgContext) bool { data, err := bot.decodePayload(update.CallbackQuery.Data) if err != nil { bot.logger.Errorln(err) - return + bot.safeEmitEvent(ctx.Context(), ErrorEvent{ + UpdateID: update.UpdateID, + UpdateType: update.Type, + Plugin: "bot", + HandlerKind: HandlerPayloadKind, + HandlerName: "decodePayload", + FromID: ctx.FromID, + ChatID: ctx.ChatID, + Err: err, + UserFacing: false, + }) + return false } ctx.Args = data.Args @@ -75,12 +132,57 @@ func (bot *Bot[T]) handleCallback(update *tgapi.Update, ctx *MsgContext) { if ctx.Logger == nil { ctx.Logger = bot.logger } + if !plugin.executeMiddlewares(ctx, bot.appData) { - return + return false } - plugin.executePayload(data.Command, ctx, bot.appData) - return + + startTime := time.Now() + bot.safeEmitEvent(ctx.Context(), HandlerStartedEvent{ + UpdateID: update.UpdateID, + UpdateType: update.Type, + Plugin: plugin.name, + HandlerKind: HandlerPayloadKind, + HandlerName: data.Command, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + }) + err := plugin.executePayload(data.Command, ctx, bot.appData) + + endEvent := HandlerFinishedEvent{ + UpdateID: update.UpdateID, + UpdateType: update.Type, + Plugin: plugin.name, + HandlerKind: HandlerPayloadKind, + HandlerName: data.Command, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + Duration: time.Since(startTime), + } + var errorEvent *ErrorEvent = nil + if err != nil { + ctx.error(err) + errorEvent = &ErrorEvent{ + UpdateID: update.UpdateID, + UpdateType: update.Type, + Plugin: plugin.name, + HandlerKind: HandlerPayloadKind, + HandlerName: data.Command, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + Err: err, + UserFacing: IsUserError(err), + } + endEvent.Err = err + endEvent.UserFacing = errorEvent.UserFacing + } + bot.safeEmitEvent(ctx.Context(), endEvent) + if errorEvent != nil { + bot.safeEmitEvent(ctx.Context(), *errorEvent) + } + return true } + return false } func (bot *Bot[T]) checkPrefixes(text string) (string, bool) { diff --git a/observer.go b/observer.go new file mode 100644 index 0000000..65fd4e3 --- /dev/null +++ b/observer.go @@ -0,0 +1,183 @@ +package laniakea + +import ( + "context" + "fmt" + "time" + + "git.scuroneko.dev/scuroneko/laniakea/tgapi" +) + +// HandlerEventKind identifies the kind of handler observed by runtime events. +type HandlerEventKind string + +const ( + // HandlerCommandKind identifies a command handler. + HandlerCommandKind HandlerEventKind = "command" + // HandlerPayloadKind identifies a callback payload handler. + HandlerPayloadKind HandlerEventKind = "payload" + // HandlerUpdateKind identifies a generic update handler. + HandlerUpdateKind HandlerEventKind = "update" + // HandlerRunnerKind identifies a background runner execution. + HandlerRunnerKind HandlerEventKind = "runner" + // HandlerPollingKind identifies polling and getUpdates runtime work. + HandlerPollingKind HandlerEventKind = "polling" + // HandlerSceneKind identifies a scene runtime handler wrapper. + HandlerSceneKind HandlerEventKind = "scene" + // HandlerSceneStepKind identifies a scene step handler. + HandlerSceneStepKind HandlerEventKind = "scene_step" + // HandlerSceneCommandKind identifies a scene-local command handler. + HandlerSceneCommandKind HandlerEventKind = "scene_command" + // HandlerSceneMessageKind identifies a scene message fallback handler. + HandlerSceneMessageKind HandlerEventKind = "scene_message" +) + +type Event interface { + isEvent() +} + +// UpdateReceivedEvent describes an update entering the bot runtime. +type UpdateReceivedEvent struct { + UpdateID int + UpdateType tgapi.UpdateType + FromID int64 + ChatID int64 +} + +// UpdateHandledEvent describes a completed update execution path. +type UpdateHandledEvent struct { + UpdateID int + UpdateType tgapi.UpdateType + FromID int64 + ChatID int64 + Duration time.Duration + Handled bool +} + +// HandlerStartedEvent describes a handler about to execute. +type HandlerStartedEvent struct { + UpdateID int + UpdateType tgapi.UpdateType + Plugin string + HandlerKind HandlerEventKind + HandlerName string + FromID int64 + ChatID int64 +} + +// HandlerFinishedEvent describes a handler that has completed. +type HandlerFinishedEvent struct { + UpdateID int + UpdateType tgapi.UpdateType + Plugin string + HandlerKind HandlerEventKind + HandlerName string + FromID int64 + ChatID int64 + Duration time.Duration + Err error + UserFacing bool +} + +// SceneTransitionEvent describes a scene state transition. +type SceneTransitionEvent struct { + Plugin string + Scene string + From string + To string + Action SceneAction + FromID int64 + ChatID int64 +} + +// PolicyCheckedEvent describes the result of a policy evaluation. +type PolicyCheckedEvent struct { + Name string + Plugin string + FromID int64 + ChatID int64 + Passed bool + Err error + Internal bool +} + +// RunnerFinishedEvent describes a completed background runner execution. +type RunnerFinishedEvent struct { + Name string + Duration time.Duration + Err error +} + +// PollingRetryEvent describes a polling retry after a failed getUpdates call. +type PollingRetryEvent struct { + Attempt int + Delay time.Duration + Err error +} + +// ErrorEvent describes an error routed through framework error handling. +type ErrorEvent struct { + UpdateID int + UpdateType tgapi.UpdateType + Plugin string + HandlerKind HandlerEventKind + HandlerName string + FromID int64 + ChatID int64 + Err error + UserFacing bool +} + +func (UpdateReceivedEvent) isEvent() {} +func (UpdateHandledEvent) isEvent() {} +func (HandlerStartedEvent) isEvent() {} +func (HandlerFinishedEvent) isEvent() {} +func (SceneTransitionEvent) isEvent() {} +func (PolicyCheckedEvent) isEvent() {} +func (RunnerFinishedEvent) isEvent() {} +func (PollingRetryEvent) isEvent() {} +func (ErrorEvent) isEvent() {} + +// Observer receives best-effort runtime instrumentation events. +type Observer interface { + OnReceiveUpdate(ctx context.Context, event UpdateReceivedEvent) + OnHandledUpdate(ctx context.Context, event UpdateHandledEvent) + OnHandlerStarted(ctx context.Context, event HandlerStartedEvent) + OnHandlerFinished(ctx context.Context, event HandlerFinishedEvent) + OnSceneTransition(ctx context.Context, event SceneTransitionEvent) + OnPolicyChecked(ctx context.Context, event PolicyCheckedEvent) + OnRunnerFinished(ctx context.Context, event RunnerFinishedEvent) + OnPollingRetry(ctx context.Context, event PollingRetryEvent) + OnError(ctx context.Context, event ErrorEvent) +} + +func (bot *Bot[T]) safeEmitEvent(ctx context.Context, event Event) { + if bot.observer == nil { + return + } + defer func() { + if r := recover(); r != nil { + bot.logger.Errorln(fmt.Sprintf("panic in observer: %v", r)) + } + }() + switch e := event.(type) { + case UpdateReceivedEvent: + bot.observer.OnReceiveUpdate(ctx, e) + case UpdateHandledEvent: + bot.observer.OnHandledUpdate(ctx, e) + case HandlerStartedEvent: + bot.observer.OnHandlerStarted(ctx, e) + case HandlerFinishedEvent: + bot.observer.OnHandlerFinished(ctx, e) + case SceneTransitionEvent: + bot.observer.OnSceneTransition(ctx, e) + case PolicyCheckedEvent: + bot.observer.OnPolicyChecked(ctx, e) + case RunnerFinishedEvent: + bot.observer.OnRunnerFinished(ctx, e) + case PollingRetryEvent: + bot.observer.OnPollingRetry(ctx, e) + case ErrorEvent: + bot.observer.OnError(ctx, e) + } +} diff --git a/plugins.go b/plugins.go index 14ff52f..f988580 100644 --- a/plugins.go +++ b/plugins.go @@ -322,55 +322,47 @@ func (p *Plugin[T]) Close() error { } // Internal helper that validates and executes a command handler. -func (p *Plugin[T]) executeCmd(cmd string, ctx *MsgContext, db T) { +func (p *Plugin[T]) executeCmd(cmd string, ctx *MsgContext, db T) error { command, exists := p.commands[cmd] if !exists { - ctx.error(AsInternalError(errCommandNotFound)) - return + return AsInternalError(errCommandNotFound) } if err := command.validateArgs(ctx.Args); err != nil { - ctx.error(err) - return + return AsUserError(err) } // Run command-specific middlewares for _, m := range command.middlewares { if !m.Execute(ctx, db) { - return + return AsInternalError(errors.New("middleware blocked call")) } } // Execute command - if err := command.exec(ctx, db); err != nil { - ctx.error(err) - } + return command.exec(ctx, db) } // Internal helper that validates and executes a payload handler. -func (p *Plugin[T]) executePayload(payload string, ctx *MsgContext, db T) { +func (p *Plugin[T]) executePayload(payload string, ctx *MsgContext, db T) error { command, exists := p.payloads[payload] if !exists { - ctx.error(AsInternalError(errPayloadNotFound)) - return + return AsInternalError(errPayloadNotFound) } if err := command.validateArgs(ctx.Args); err != nil { - ctx.error(err) - return + return AsUserError(err) } // Run command-specific middlewares for _, m := range command.middlewares { if !m.Execute(ctx, db) { - return + return AsInternalError(errors.New("middleware blocked call")) } } // Execute payload - if err := command.exec(ctx, db); err != nil { - ctx.error(err) - } + return command.exec(ctx, db) } // Internal helper that runs plugin middlewares in order. diff --git a/policy.go b/policy.go index 30d0a59..d93fe9d 100644 --- a/policy.go +++ b/policy.go @@ -14,9 +14,23 @@ type Policy[T AppData] func(ctx *MsgContext, data T) error func RequirePolicy[T AppData](name string, p Policy[T]) Middleware[T] { return NewMiddleware(name, func(ctx *MsgContext, data T) bool { if err := p(ctx, data); err != nil { + ctx.emitPolicyChecked(PolicyCheckedEvent{ + Name: name, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + Passed: false, + Err: err, + Internal: IsInternalError(err), + }) ctx.error(err) return false } + ctx.emitPolicyChecked(PolicyCheckedEvent{ + Name: name, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + Passed: true, + }) return true }) } diff --git a/policy_test.go b/policy_test.go index 6e32ca3..a7d6fd8 100644 --- a/policy_test.go +++ b/policy_test.go @@ -1,6 +1,7 @@ package laniakea import ( + "context" "encoding/json" "errors" "io" @@ -227,3 +228,54 @@ func TestNotPolicyInvertsUserDenyButPreservesInternalErrors(t *testing.T) { t.Fatalf("expected internal error to be preserved, got %v", err) } } + +func TestRequirePolicyEmitsObserverEvents(t *testing.T) { + t.Run("allow", func(t *testing.T) { + observer := &recordingObserver{} + ctx := &MsgContext{ + Logger: slog.CreateLogger(), + ctx: context.Background(), + observer: observer, + FromID: 10, + ChatID: 20, + } + + mw := RequirePolicy[NoData]("allow", func(ctx *MsgContext, data NoData) error { + return nil + }) + + if !mw.Execute(ctx, NoData{}) { + t.Fatal("expected allowed policy middleware to continue execution") + } + if len(observer.policies) != 1 { + t.Fatalf("expected one policy event, got %d", len(observer.policies)) + } + if got := observer.policies[0]; got.Name != "allow" || !got.Passed || got.Err != nil || got.Internal { + t.Fatalf("unexpected policy event: %#v", got) + } + }) + + t.Run("deny", func(t *testing.T) { + observer := &recordingObserver{} + ctx := &MsgContext{ + Logger: slog.CreateLogger(), + ctx: context.Background(), + observer: observer, + errorTemplate: "%s", + } + + mw := RequirePolicy[NoData]("deny", func(ctx *MsgContext, data NoData) error { + return AsInternalError(errors.New("blocked")) + }) + + if mw.Execute(ctx, NoData{}) { + t.Fatal("expected denied policy middleware to stop execution") + } + if len(observer.policies) != 1 { + t.Fatalf("expected one policy event, got %d", len(observer.policies)) + } + if got := observer.policies[0]; got.Name != "deny" || got.Passed || got.Err == nil || !got.Internal { + t.Fatalf("unexpected policy event: %#v", got) + } + }) +} diff --git a/runners.go b/runners.go index 11c1d1d..7d50ce5 100644 --- a/runners.go +++ b/runners.go @@ -107,8 +107,21 @@ func (bot *Bot[T]) ExecRunners(ctx context.Context) { bot.runnerOnceWG.Add(1) go func(r Runner[T]) { defer bot.runnerOnceWG.Done() + startedAt := time.Now() err := r.fn(bot) + bot.safeEmitEvent(ctx, RunnerFinishedEvent{ + Name: r.name, + Duration: time.Since(startedAt), + Err: err, + }) if err != nil { + bot.safeEmitEvent(ctx, ErrorEvent{ + Plugin: "bot", + HandlerKind: HandlerRunnerKind, + HandlerName: r.name, + Err: err, + UserFacing: false, + }) bot.logger.Warnf("Runner %s failed: %s\n", r.name, err) } }(runner) @@ -116,10 +129,22 @@ func (bot *Bot[T]) ExecRunners(ctx context.Context) { // One-time sync: block until done t := time.Now() err := runner.fn(bot) + elapsed := time.Since(t) + bot.safeEmitEvent(ctx, RunnerFinishedEvent{ + Name: runner.name, + Duration: elapsed, + Err: err, + }) if err != nil { + bot.safeEmitEvent(ctx, ErrorEvent{ + Plugin: "bot", + HandlerKind: HandlerRunnerKind, + HandlerName: runner.name, + Err: err, + UserFacing: false, + }) bot.logger.Warnf("Runner %s failed: %s\n", runner.name, err) } - elapsed := time.Since(t) if elapsed > time.Second*2 { bot.logger.Warnf("Runner %s too slow. Elapsed time %v >= 2s\n", runner.name, elapsed) } @@ -135,8 +160,21 @@ func (bot *Bot[T]) ExecRunners(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: + startedAt := time.Now() err := r.fn(bot) + bot.safeEmitEvent(ctx, RunnerFinishedEvent{ + Name: r.name, + Duration: time.Since(startedAt), + Err: err, + }) if err != nil { + bot.safeEmitEvent(ctx, ErrorEvent{ + Plugin: "bot", + HandlerKind: HandlerRunnerKind, + HandlerName: r.name, + Err: err, + UserFacing: false, + }) bot.logger.Warnf("Runner %s failed: %s\n", r.name, err) } } diff --git a/runners_test.go b/runners_test.go index 5223203..cbeefd4 100644 --- a/runners_test.go +++ b/runners_test.go @@ -2,6 +2,7 @@ package laniakea import ( "context" + "errors" "sync/atomic" "testing" "time" @@ -9,6 +10,10 @@ import ( "git.scuroneko.dev/scuroneko/slog" ) +type runnerObserver struct { + recordingObserver +} + func TestExecRunnersRunsOnetimeSyncRunner(t *testing.T) { var calls atomic.Int32 bot := &Bot[NoData]{ @@ -60,3 +65,33 @@ func TestExecRunnersStopsBackgroundRunnerOnCancel(t *testing.T) { t.Fatal("expected background runner to be called at least once") } } + +func TestExecRunnersEmitObserverEvents(t *testing.T) { + observer := &runnerObserver{} + wantErr := errors.New("runner failed") + + bot := &Bot[NoData]{ + logger: slog.CreateLogger(), + observer: observer, + runners: []Runner[NoData]{ + NewRunner("sync-once", func(*Bot[NoData]) error { + return wantErr + }).Onetime(true).Async(false), + }, + } + + bot.ExecRunners(context.Background()) + + if len(observer.runners) != 1 { + t.Fatalf("expected one runner-finished event, got %d", len(observer.runners)) + } + if got := observer.runners[0]; got.Name != "sync-once" || !errors.Is(got.Err, wantErr) { + t.Fatalf("unexpected runner-finished event: %#v", got) + } + if len(observer.errors) != 1 { + t.Fatalf("expected one error event, got %d", len(observer.errors)) + } + if got := observer.errors[0]; got.HandlerKind != HandlerRunnerKind || got.HandlerName != "sync-once" || !errors.Is(got.Err, wantErr) { + t.Fatalf("unexpected runner error event: %#v", got) + } +} diff --git a/scene_handler.go b/scene_handler.go index 9ddcc81..12c6f00 100644 --- a/scene_handler.go +++ b/scene_handler.go @@ -4,6 +4,7 @@ import ( "errors" "fmt" "strings" + "time" ) func (bot *Bot[T]) tryHandleScene(ctx *MsgContext) (bool, error) { @@ -34,12 +35,13 @@ func (bot *Bot[T]) tryHandleScene(ctx *MsgContext) (bool, error) { sess: session, key: key, } - return bot.executeScene(scene, sceneCtx) + + return bot.executeScene(sceneCtx, scene) } return false, ErrSceneNotFound } -func (bot *Bot[T]) executeScene(scene *Scene[T], ctx *SceneContext) (bool, error) { +func (bot *Bot[T]) executeScene(ctx *SceneContext, scene *Scene[T]) (bool, error) { if ctx.MsgContext == nil || ctx.sess.Scene == "" { return false, nil } @@ -59,37 +61,143 @@ func (bot *Bot[T]) executeScene(scene *Scene[T], ctx *SceneContext) (bool, error ctx.Text = args ctx.Args = strings.Fields(args) - res, matched, err := scene.executeCommand(cmd, ctx, bot.appData) - if err != nil { - return false, err - } - if matched { - return bot.applySceneResult(scene, ctx, res) + if _, ok := scene.commands[cmd]; ok { + startTime := time.Now() + bot.emitSceneStarted(ctx, scene, HandlerSceneCommandKind, cmd) + res, _, err := scene.executeCommand(cmd, ctx, bot.appData) + if err != nil { + bot.emitSceneFinished(ctx, scene, HandlerSceneCommandKind, cmd, startTime, err) + bot.emitSceneError(ctx, scene, HandlerSceneCommandKind, cmd, err) + return false, err + } + from := ctx.sess.Step + ok, err := bot.applySceneResult(scene, ctx, res) + bot.emitSceneFinished(ctx, scene, HandlerSceneCommandKind, cmd, startTime, err) + if err != nil { + bot.emitSceneError(ctx, scene, HandlerSceneCommandKind, cmd, err) + } + if ok { + bot.emitSceneTransition(ctx, scene, from, res) + } + return ok, err } } ctx.Text = text ctx.Args = nil ctx.Prefix = "" if ctx.sess.Step != "" { - res, matched, err := scene.executeStep(ctx.sess.Step, ctx, bot.appData) - if err != nil { - return false, err - } - if matched { - return bot.applySceneResult(scene, ctx, res) + step := ctx.sess.Step + if _, ok := scene.steps[step]; ok { + startTime := time.Now() + bot.emitSceneStarted(ctx, scene, HandlerSceneStepKind, step) + res, _, err := scene.executeStep(step, ctx, bot.appData) + if err != nil { + bot.emitSceneFinished(ctx, scene, HandlerSceneStepKind, step, startTime, err) + bot.emitSceneError(ctx, scene, HandlerSceneStepKind, step, err) + return false, err + } + from := step + ok, err := bot.applySceneResult(scene, ctx, res) + bot.emitSceneFinished(ctx, scene, HandlerSceneStepKind, step, startTime, err) + if err != nil { + bot.emitSceneError(ctx, scene, HandlerSceneStepKind, from, err) + } + if ok { + bot.emitSceneTransition(ctx, scene, from, res) + } + return ok, err } } - res, matched, err := scene.executeMessage(ctx, bot.appData) - if err != nil { - return false, err - } - if matched { - return bot.applySceneResult(scene, ctx, res) + if scene.message != nil { + startTime := time.Now() + bot.emitSceneStarted(ctx, scene, HandlerSceneMessageKind, "message_fallback") + res, _, err := scene.executeMessage(ctx, bot.appData) + if err != nil { + bot.emitSceneFinished(ctx, scene, HandlerSceneMessageKind, "message_fallback", startTime, err) + bot.emitSceneError(ctx, scene, HandlerSceneMessageKind, "message_fallback", err) + return false, err + } + from := ctx.sess.Step + ok, err := bot.applySceneResult(scene, ctx, res) + bot.emitSceneFinished(ctx, scene, HandlerSceneMessageKind, "message_fallback", startTime, err) + if err != nil { + bot.emitSceneError(ctx, scene, HandlerSceneMessageKind, "message_fallback", err) + } + if ok { + bot.emitSceneTransition(ctx, scene, from, res) + } + return ok, err } return false, nil } + +func (bot *Bot[T]) emitSceneStarted(ctx *SceneContext, scene *Scene[T], kind HandlerEventKind, name string) { + bot.safeEmitEvent(ctx.Context(), HandlerStartedEvent{ + UpdateID: ctx.Update.UpdateID, + UpdateType: ctx.Update.Type, + Plugin: scene.PluginName, + HandlerKind: kind, + HandlerName: name, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + }) +} + +func (bot *Bot[T]) emitSceneFinished(ctx *SceneContext, scene *Scene[T], kind HandlerEventKind, name string, startedAt time.Time, err error) { + bot.safeEmitEvent(ctx.Context(), HandlerFinishedEvent{ + UpdateID: ctx.Update.UpdateID, + UpdateType: ctx.Update.Type, + Plugin: scene.PluginName, + HandlerKind: kind, + HandlerName: name, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + Duration: time.Since(startedAt), + Err: err, + UserFacing: IsUserError(err), + }) +} + +func (bot *Bot[T]) emitSceneError(ctx *SceneContext, scene *Scene[T], kind HandlerEventKind, name string, err error) { + bot.safeEmitEvent(ctx.Context(), ErrorEvent{ + UpdateID: ctx.Update.UpdateID, + UpdateType: ctx.Update.Type, + Plugin: scene.PluginName, + HandlerKind: kind, + HandlerName: name, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + Err: err, + UserFacing: IsUserError(err), + }) +} + +func (bot *Bot[T]) emitSceneTransition(ctx *SceneContext, scene *Scene[T], from string, result SceneResult) { + if result.Action == SceneActionPass { + return + } + + to := from + switch result.Action { + case SceneActionNext: + to = result.Next + case SceneActionExit: + to = "" + } + + bot.safeEmitEvent(ctx.Context(), SceneTransitionEvent{ + Plugin: scene.PluginName, + Scene: scene.Name, + From: from, + To: to, + Action: result.Action, + FromID: ctx.FromID, + ChatID: ctx.ChatID, + }) +} + func (bot *Bot[T]) applySceneResult(scene *Scene[T], ctx *SceneContext, result SceneResult) (bool, error) { switch result.Action { case SceneActionStay: diff --git a/scene_test.go b/scene_test.go index 0137ba8..ea9903b 100644 --- a/scene_test.go +++ b/scene_test.go @@ -266,6 +266,170 @@ func TestSceneCommandHandlerRunsBeforeStep(t *testing.T) { } } +func TestSceneCommandObserverEmitsLifecycleEvents(t *testing.T) { + observer := &recordingObserver{} + plugin := NewPlugin[NoData]("wizard") + plugin.NewScene("signup"). + SetEntry("start"). + OnStep("start", func(ctx *SceneContext, db NoData) (SceneResult, error) { + return ctx.Stay(), nil + }). + OnCommand("cancel", func(ctx *SceneContext, db NoData) (SceneResult, error) { + return ctx.Exit(), nil + }) + + bot := &Bot[NoData]{ + logger: slog.CreateLogger(), + prefixes: []string{"/"}, + sessionStore: NewMemorySessionStore(), + sceneScopePriority: []SceneScope{SceneScopeUserChat, SceneScopeChat, SceneScopeUser}, + observer: observer, + } + bot.AddPlugins(plugin) + + enterCtx := &MsgContext{ + Msg: &tgapi.Message{Chat: &tgapi.Chat{ID: 100, Type: tgapi.ChatTypePrivate}}, + FromID: 42, + sceneRuntime: bot, + } + if err := enterCtx.EnterScene("signup"); err != nil { + t.Fatalf("EnterScene returned error: %v", err) + } + + bot.handle(context.Background(), &tgapi.Update{ + UpdateID: 22, + Type: tgapi.UpdateTypeMessage, + Message: &tgapi.Message{ + MessageID: 9, + Text: "/cancel", + Chat: &tgapi.Chat{ID: 100, Type: tgapi.ChatTypePrivate}, + From: &tgapi.User{ID: 42}, + }, + }) + + if len(observer.started) != 1 { + t.Fatalf("expected one scene started event, got %d", len(observer.started)) + } + if got := observer.started[0]; got.HandlerKind != HandlerSceneCommandKind || got.HandlerName != "cancel" || got.Plugin != "wizard" { + t.Fatalf("unexpected scene started event: %#v", got) + } + if len(observer.finished) != 1 { + t.Fatalf("expected one scene finished event, got %d", len(observer.finished)) + } + if got := observer.finished[0]; got.HandlerKind != HandlerSceneCommandKind || got.HandlerName != "cancel" || got.Plugin != "wizard" || got.Err != nil { + t.Fatalf("unexpected scene finished event: %#v", got) + } +} + +func TestSceneStepObserverEmitsLifecycleEvents(t *testing.T) { + observer := &recordingObserver{} + plugin := NewPlugin[NoData]("wizard") + plugin.NewScene("signup"). + SetEntry("start"). + OnStep("start", func(ctx *SceneContext, db NoData) (SceneResult, error) { + return ctx.Stay(), nil + }) + + bot := &Bot[NoData]{ + logger: slog.CreateLogger(), + prefixes: []string{"/"}, + sessionStore: NewMemorySessionStore(), + sceneScopePriority: []SceneScope{SceneScopeUserChat, SceneScopeChat, SceneScopeUser}, + observer: observer, + } + bot.AddPlugins(plugin) + + enterCtx := &MsgContext{ + Msg: &tgapi.Message{Chat: &tgapi.Chat{ID: 100, Type: tgapi.ChatTypePrivate}}, + FromID: 42, + sceneRuntime: bot, + } + if err := enterCtx.EnterScene("signup"); err != nil { + t.Fatalf("EnterScene returned error: %v", err) + } + + bot.handle(context.Background(), &tgapi.Update{ + UpdateID: 23, + Type: tgapi.UpdateTypeMessage, + Message: &tgapi.Message{ + MessageID: 10, + Text: "hello there", + Chat: &tgapi.Chat{ID: 100, Type: tgapi.ChatTypePrivate}, + From: &tgapi.User{ID: 42}, + }, + }) + + if len(observer.started) != 1 { + t.Fatalf("expected one scene started event, got %d", len(observer.started)) + } + if got := observer.started[0]; got.HandlerKind != HandlerSceneStepKind || got.HandlerName != "start" || got.Plugin != "wizard" { + t.Fatalf("unexpected scene step started event: %#v", got) + } + if len(observer.finished) != 1 { + t.Fatalf("expected one scene finished event, got %d", len(observer.finished)) + } + if got := observer.finished[0]; got.HandlerKind != HandlerSceneStepKind || got.HandlerName != "start" || got.Plugin != "wizard" || got.Err != nil { + t.Fatalf("unexpected scene step finished event: %#v", got) + } +} + +func TestSceneMessageObserverEmitsLifecycleEvents(t *testing.T) { + observer := &recordingObserver{} + plugin := NewPlugin[NoData]("wizard") + scene := plugin.NewScene("signup"). + SetEntry("start"). + OnStep("start", func(ctx *SceneContext, db NoData) (SceneResult, error) { + return ctx.Stay(), nil + }). + OnMessage(func(ctx *SceneContext, db NoData) (SceneResult, error) { + return ctx.Exit(), nil + }) + + bot := &Bot[NoData]{ + logger: slog.CreateLogger(), + prefixes: []string{"/"}, + sessionStore: NewMemorySessionStore(), + sceneScopePriority: []SceneScope{SceneScopeUserChat, SceneScopeChat, SceneScopeUser}, + observer: observer, + } + bot.AddPlugins(plugin) + + key, ok := buildSceneKey(SceneScopeUserChat, &MsgContext{ + Msg: &tgapi.Message{Chat: &tgapi.Chat{ID: 100, Type: tgapi.ChatTypePrivate}}, + FromID: 42, + }) + if !ok { + t.Fatal("expected scene key to be built") + } + if err := bot.sessionStore.Set(key, SceneSession{Scene: scene.Name}); err != nil { + t.Fatalf("failed to seed scene session: %v", err) + } + + bot.handle(context.Background(), &tgapi.Update{ + UpdateID: 24, + Type: tgapi.UpdateTypeMessage, + Message: &tgapi.Message{ + MessageID: 11, + Text: "hello there", + Chat: &tgapi.Chat{ID: 100, Type: tgapi.ChatTypePrivate}, + From: &tgapi.User{ID: 42}, + }, + }) + + if len(observer.started) != 1 { + t.Fatalf("expected one scene started event, got %d", len(observer.started)) + } + if got := observer.started[0]; got.HandlerKind != HandlerSceneMessageKind || got.HandlerName != "message_fallback" || got.Plugin != "wizard" { + t.Fatalf("unexpected scene message started event: %#v", got) + } + if len(observer.finished) != 1 { + t.Fatalf("expected one scene finished event, got %d", len(observer.finished)) + } + if got := observer.finished[0]; got.HandlerKind != HandlerSceneMessageKind || got.HandlerName != "message_fallback" || got.Plugin != "wizard" || got.Err != nil { + t.Fatalf("unexpected scene message finished event: %#v", got) + } +} + func TestScenePassDoesNotPersistSessionData(t *testing.T) { commandCalled := false diff --git a/update_context.go b/update_context.go index 99eee95..150e7ac 100644 --- a/update_context.go +++ b/update_context.go @@ -1,8 +1,13 @@ package laniakea -import "git.scuroneko.dev/scuroneko/laniakea/tgapi" +import ( + "time" -func (bot *Bot[T]) handleUpdate(u *tgapi.Update, ctx *MsgContext) { + "git.scuroneko.dev/scuroneko/laniakea/tgapi" +) + +func (bot *Bot[T]) handleUpdate(u *tgapi.Update, ctx *MsgContext) bool { + handled := false for _, plugin := range bot.plugins { handler, ok := plugin.handlers[u.Type] if !ok { @@ -16,10 +21,49 @@ func (bot *Bot[T]) handleUpdate(u *tgapi.Update, ctx *MsgContext) { if !plugin.executeMiddlewares(pluginCtx, bot.appData) { continue } - if err := handler(pluginCtx, bot.appData); err != nil { + startTime := time.Now() + bot.safeEmitEvent(pluginCtx.Context(), HandlerStartedEvent{ + UpdateID: u.UpdateID, + UpdateType: u.Type, + Plugin: plugin.name, + HandlerKind: HandlerUpdateKind, + HandlerName: string(u.Type), + FromID: pluginCtx.FromID, + ChatID: pluginCtx.ChatID, + }) + err := handler(pluginCtx, bot.appData) + endEvent := HandlerFinishedEvent{ + UpdateID: u.UpdateID, + UpdateType: u.Type, + Plugin: plugin.name, + HandlerKind: HandlerUpdateKind, + HandlerName: string(u.Type), + FromID: pluginCtx.FromID, + ChatID: pluginCtx.ChatID, + Duration: time.Since(startTime), + } + if err != nil { + endEvent.Err = err + endEvent.UserFacing = IsUserError(err) + } + bot.safeEmitEvent(pluginCtx.Context(), endEvent) + if err != nil { + bot.safeEmitEvent(pluginCtx.Context(), ErrorEvent{ + UpdateID: u.UpdateID, + UpdateType: u.Type, + Plugin: plugin.name, + HandlerKind: HandlerUpdateKind, + HandlerName: string(u.Type), + FromID: pluginCtx.FromID, + ChatID: pluginCtx.ChatID, + Err: err, + UserFacing: IsUserError(err), + }) pluginCtx.error(err) } + handled = true } + return handled } func (bot *Bot[T]) prepareUpdateCtx(u *tgapi.Update, ctx *MsgContext) {