From 522ba1b63933beb6ae4ffd9359b4adba715d43b3 Mon Sep 17 00:00:00 2001 From: Kirillkriv Date: Mon, 5 Oct 2026 16:37:50 +0300 Subject: [PATCH] feat: implemented webhook poller --- core/pollers.go | 169 +++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 166 insertions(+), 3 deletions(-) diff --git a/core/pollers.go b/core/pollers.go index 6f83b30..644d45e 100644 --- a/core/pollers.go +++ b/core/pollers.go @@ -2,21 +2,34 @@ package core import ( "context" + "crypto/tls" "errors" "fmt" maxbot "github.com/max-messenger/max-bot-api-client-go/v2" "github.com/max-messenger/max-bot-api-client-go/v2/model" "github.com/rs/zerolog/log" + "net" + "net/http" "time" ) -const defaultInterval = 200 * time.Millisecond +const ( + defaultInterval = 200 * time.Millisecond + defaultWhStartTimeout = 10 * time.Second +) + +type WebhookOption interface { + apply(opts *webhookOptions) +} type LongPollerOption interface { apply(opts *longPollerOptions) } -type Webhook struct{} +type Webhook struct { + srv *http.Server + ch chan model.Update +} type LongPoller struct { api *maxbot.Api @@ -25,12 +38,94 @@ type LongPoller struct { type TestPoller struct{} +type webhookOptions struct { + addr string + tlsConfig *tls.Config + readTimeout time.Duration + readHeaderTimeout time.Duration + writeTimeout time.Duration + idleTimeout time.Duration + maxHeaderBytes int + baseContext func(l net.Listener) context.Context + connContext func(ctx context.Context, conn net.Conn) context.Context +} + type longPollerOptions struct { pollInterval time.Duration } +type webhookOptionWrapper struct { + fn func(opts *webhookOptions) +} + +func NewWebhookPoller(api *maxbot.Api, secret string, options ...WebhookOption) *Webhook { + ch := make(chan model.Update, 1) + + handler := api.GetHandler(func(ctx context.Context, update model.Update) { + select { + case ch <- update: + case <-ctx.Done(): + } + }, secret) + + srv := &http.Server{ + Handler: handler, + } + + wh := &Webhook{ + srv: srv, + ch: ch, + } + + opts := newWebhookOptions() + for _, opt := range options { + opt.apply(opts) + } + + wh.srv.Addr = opts.addr + wh.srv.TLSConfig = opts.tlsConfig + wh.srv.ReadTimeout = opts.readTimeout + wh.srv.ReadHeaderTimeout = opts.readHeaderTimeout + wh.srv.WriteTimeout = opts.writeTimeout + wh.srv.IdleTimeout = opts.idleTimeout + wh.srv.MaxHeaderBytes = opts.maxHeaderBytes + wh.srv.BaseContext = opts.baseContext + wh.srv.ConnContext = opts.connContext + + return wh +} + func (w *Webhook) Start(_ context.Context) (<-chan model.Update, error) { - return nil, fmt.Errorf("unimplemented") + errCh := make(chan error, 1) + defer close(errCh) + + go func() { + var err error + defer func() { + errCh <- err + if err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Fatal().Err(err).Msg("webhook server error") + } + }() + + if w.srv.TLSConfig == nil { + err = w.srv.ListenAndServe() + return + } + + err = w.srv.ListenAndServeTLS("", "") + }() + + select { + case err := <-errCh: + return nil, err + case <-time.After(defaultWhStartTimeout): + return w.ch, nil + } +} + +func (w *Webhook) Stop(ctx context.Context) error { + return w.srv.Shutdown(ctx) } func NewLongPoller(api *maxbot.Api, options ...LongPollerOption) *LongPoller { @@ -94,8 +189,76 @@ func (tp *TestPoller) Start(_ context.Context) (<-chan model.Update, error) { return nil, fmt.Errorf("unimplemented") } +func WithAddress(addr string) WebhookOption { + return newWhOption(func(opts *webhookOptions) { + opts.addr = addr + }) +} + +func WithTLSConfig(cfg *tls.Config) WebhookOption { + return newWhOption(func(opts *webhookOptions) { + opts.tlsConfig = cfg + }) +} + +func WithReadTimeout(timeout time.Duration) WebhookOption { + return newWhOption(func(opts *webhookOptions) { + opts.readTimeout = timeout + }) +} + +func WithReadHeaderTimeout(timeout time.Duration) WebhookOption { + return newWhOption(func(opts *webhookOptions) { + opts.readHeaderTimeout = timeout + }) +} + +func WithWriteTimeout(timeout time.Duration) WebhookOption { + return newWhOption(func(opts *webhookOptions) { + opts.writeTimeout = timeout + }) +} + +func WithIdleTimeout(timeout time.Duration) WebhookOption { + return newWhOption(func(opts *webhookOptions) { + opts.idleTimeout = timeout + }) +} + +func WithMaxHeaderBytes(bytes int) WebhookOption { + return newWhOption(func(opts *webhookOptions) { + opts.maxHeaderBytes = bytes + }) +} + +func WithBaseContext(fn func(l net.Listener) context.Context) WebhookOption { + return newWhOption(func(opts *webhookOptions) { + opts.baseContext = fn + }) +} + +func WithConnContext(fn func(ctx context.Context, c net.Conn) context.Context) WebhookOption { + return newWhOption(func(opts *webhookOptions) { + opts.connContext = fn + }) +} + func newLongPollerOptions() *longPollerOptions { return &longPollerOptions{ pollInterval: defaultInterval, } } + +func newWebhookOptions() *webhookOptions { + return &webhookOptions{ + addr: ":8080", + } +} + +func newWhOption(fn func(options *webhookOptions)) WebhookOption { + return webhookOptionWrapper{fn: fn} +} + +func (wow webhookOptionWrapper) apply(opts *webhookOptions) { + wow.fn(opts) +}