119 lines
2.5 KiB
Go
119 lines
2.5 KiB
Go
package core
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
mcontext "github.com/abelith/gomax/context"
|
|
"github.com/abelith/gomax/state"
|
|
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"
|
|
"time"
|
|
)
|
|
|
|
type FSMStorage interface {
|
|
Get(ctx context.Context, userID int) (state.Snapshot, error)
|
|
Put(ctx context.Context, state state.Snapshot) error
|
|
Delete(ctx context.Context, userID int) error
|
|
}
|
|
|
|
type UpdateHandler interface {
|
|
HandleUpdate(c *mcontext.Context) error
|
|
Use(mw ...Middleware)
|
|
}
|
|
|
|
type Dispatcher struct {
|
|
PollInterval time.Duration
|
|
Api *maxbot.Api
|
|
Handler UpdateHandler
|
|
FSMStorage FSMStorage
|
|
}
|
|
|
|
func (d *Dispatcher) getUpdates(ctx context.Context) <-chan model.Update {
|
|
var updates []model.Update
|
|
var marker int64
|
|
var err error
|
|
|
|
ch := make(chan model.Update)
|
|
|
|
go func() {
|
|
defer close(ch)
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.Tick(d.PollInterval):
|
|
start := time.Now()
|
|
updates, marker, err = d.Api.Subscriptions.GetUpdates(ctx, marker)
|
|
latency := time.Since(start)
|
|
if err != nil && !errors.Is(err, context.Canceled) {
|
|
log.Error().Err(err).Send()
|
|
}
|
|
log.Ctx(ctx).Debug().
|
|
Int("received", len(updates)).
|
|
Dur("latency", latency).
|
|
Msg("received updates")
|
|
|
|
for _, update := range updates {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case ch <- update:
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
|
|
return ch
|
|
}
|
|
|
|
func (d *Dispatcher) handleUpdates(ch <-chan model.Update) {
|
|
for update := range ch {
|
|
go d.handleUpdate(update)
|
|
}
|
|
}
|
|
|
|
func (d *Dispatcher) StartPolling(ctx context.Context) {
|
|
d.handleUpdates(d.getUpdates(ctx))
|
|
<-ctx.Done()
|
|
}
|
|
|
|
func (d *Dispatcher) handleUpdate(upd model.Update) {
|
|
ctx := context.Background()
|
|
var st *state.Context
|
|
|
|
snapshot, err := d.FSMStorage.Get(ctx, int(upd.UserID))
|
|
if err != nil && !errors.Is(err, state.ErrStateNotFound) {
|
|
log.Error().Err(err).Send()
|
|
return
|
|
}
|
|
if errors.Is(err, state.ErrStateNotFound) {
|
|
st = state.New(int(upd.UserID))
|
|
} else {
|
|
st = state.Restore(snapshot)
|
|
}
|
|
|
|
c := mcontext.New(ctx, upd, st, d.Api)
|
|
|
|
if err := d.Handler.HandleUpdate(c); err != nil {
|
|
log.Error().Err(err).Send()
|
|
return
|
|
}
|
|
|
|
st = c.State()
|
|
if st.State() == state.ClearState {
|
|
if err := d.FSMStorage.Delete(ctx, int(upd.UserID)); err != nil {
|
|
log.Error().Err(err).Send()
|
|
return
|
|
}
|
|
}
|
|
if st.State() != "" {
|
|
if err := d.FSMStorage.Put(ctx, st.Snapshot()); err != nil {
|
|
log.Error().Err(err).Send()
|
|
return
|
|
}
|
|
}
|
|
}
|