v1.0.0 beta 8; ratelimt war

bot.go

@@ -1,41 +1,63 @@
 package laniakea
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"sort"
+	"strconv"
 	"strings"
-	"time"
+	"sync"
 
 	"git.nix13.pw/scuroneko/extypes"
 	"git.nix13.pw/scuroneko/laniakea/tgapi"
+	"git.nix13.pw/scuroneko/laniakea/utils"
 	"git.nix13.pw/scuroneko/slog"
+	"github.com/alitto/pond/v2"
 )
 
 type BotOpts struct {
-	Token            string
-	Debug            bool
-	ErrorTemplate    string
-	Prefixes         []string
-	UpdateTypes      []string
+	Token       string
+	UpdateTypes []string
+
+	Debug         bool
+	ErrorTemplate string
+	Prefixes      []string
+
 	LoggerBasePath   string
 	UseRequestLogger bool
 	WriteToFile      bool
-	UseTestServer    bool
-	APIUrl           string
+
+	UseTestServer bool
+	APIUrl        string
+
+	RateLimit      int
+	DropRLOverflow bool
 }
 
+func NewOpts() *BotOpts { return new(BotOpts) }
 func LoadOptsFromEnv() *BotOpts {
+	rateLimit := 30
+	if rl := os.Getenv("RATE_LIMIT"); rl != "" {
+		rateLimit, _ = strconv.Atoi(rl)
+	}
+
 	return &BotOpts{
-		Token:            os.Getenv("TG_TOKEN"),
-		Debug:            os.Getenv("DEBUG") == "true",
-		ErrorTemplate:    os.Getenv("ERROR_TEMPLATE"),
-		Prefixes:         LoadPrefixesFromEnv(),
-		UpdateTypes:      strings.Split(os.Getenv("UPDATE_TYPES"), ";"),
+		Token:       os.Getenv("TG_TOKEN"),
+		UpdateTypes: strings.Split(os.Getenv("UPDATE_TYPES"), ";"),
+
+		Debug:         os.Getenv("DEBUG") == "true",
+		ErrorTemplate: os.Getenv("ERROR_TEMPLATE"),
+		Prefixes:      LoadPrefixesFromEnv(),
+
 		UseRequestLogger: os.Getenv("USE_REQ_LOG") == "true",
 		WriteToFile:      os.Getenv("WRITE_TO_FILE") == "true",
-		UseTestServer:    os.Getenv("USE_TEST_SERVER") == "true",
-		APIUrl:           os.Getenv("API_URL"),
+
+		UseTestServer: os.Getenv("USE_TEST_SERVER") == "true",
+		APIUrl:        os.Getenv("API_URL"),
+
+		RateLimit:      rateLimit,
+		DropRLOverflow: os.Getenv("DROP_RL_OVERFLOW") == "true",
 	}
 }
 func LoadPrefixesFromEnv() []string {
@@ -47,6 +69,7 @@ func LoadPrefixesFromEnv() []string {
 }
 
 type DbContext interface{}
+type NoDB struct{ DbContext }
 type Bot[T DbContext] struct {
 	token         string
 	debug         bool
@@ -54,48 +77,55 @@ type Bot[T DbContext] struct {
 
 	logger        *slog.Logger
 	RequestLogger *slog.Logger
+	extraLoggers  extypes.Slice[*slog.Logger]
 
 	plugins     []Plugin[T]
 	middlewares []Middleware[T]
 	prefixes    []string
 	runners     []Runner[T]
 
-	api       *tgapi.API
-	uploader  *tgapi.Uploader
-	dbContext *T
-	l10n      *L10n
+	api           *tgapi.API
+	uploader      *tgapi.Uploader
+	dbContext     *T
+	l10n          *L10n
+	draftProvider *DraftProvider
 
-	dbWriterRequested extypes.Slice[*slog.Logger]
-
-	updateOffset int
-	updateTypes  []tgapi.UpdateType
-	updateQueue  *extypes.Queue[*tgapi.Update]
+	updateOffsetMu sync.Mutex
+	updateOffset   int
+	updateTypes    []tgapi.UpdateType
+	updateQueue    chan *tgapi.Update
 }
 
 func NewBot[T any](opts *BotOpts) *Bot[T] {
-	updateQueue := extypes.CreateQueue[*tgapi.Update](512)
+	updateQueue := make(chan *tgapi.Update, 512)
 
-	apiOpts := tgapi.NewAPIOpts(opts.Token).SetAPIUrl(opts.APIUrl).UseTestServer(opts.UseTestServer)
+	var limiter *utils.RateLimiter
+	if opts.RateLimit > 0 {
+		limiter = utils.NewRateLimiter()
+	}
+
+	apiOpts := tgapi.NewAPIOpts(opts.Token).SetAPIUrl(opts.APIUrl).UseTestServer(opts.UseTestServer).SetLimiter(limiter)
 	api := tgapi.NewAPI(apiOpts)
 
 	uploader := tgapi.NewUploader(api)
 
 	bot := &Bot[T]{
-		updateOffset:      0,
-		errorTemplate:     "%s",
-		updateQueue:       updateQueue,
-		api:               api,
-		uploader:          uploader,
-		debug:             opts.Debug,
-		prefixes:          opts.Prefixes,
-		token:             opts.Token,
-		plugins:           make([]Plugin[T], 0),
-		updateTypes:       make([]tgapi.UpdateType, 0),
-		runners:           make([]Runner[T], 0),
-		dbWriterRequested: make([]*slog.Logger, 0),
-		l10n:              &L10n{},
+		updateOffset:  0,
+		errorTemplate: "%s",
+		updateQueue:   updateQueue,
+		api:           api,
+		uploader:      uploader,
+		debug:         opts.Debug,
+		prefixes:      opts.Prefixes,
+		token:         opts.Token,
+		plugins:       make([]Plugin[T], 0),
+		updateTypes:   make([]tgapi.UpdateType, 0),
+		runners:       make([]Runner[T], 0),
+		extraLoggers:  make([]*slog.Logger, 0),
+		l10n:          &L10n{},
+		draftProvider: NewRandomDraftProvider(api),
 	}
-	bot.dbWriterRequested = bot.dbWriterRequested.Push(api.GetLogger()).Push(uploader.GetLogger())
+	bot.extraLoggers = bot.extraLoggers.Push(api.GetLogger()).Push(uploader.GetLogger())
 
 	if len(opts.ErrorTemplate) > 0 {
 		bot.errorTemplate = opts.ErrorTemplate
@@ -107,8 +137,7 @@ func NewBot[T any](opts *BotOpts) *Bot[T] {
 
 	u, err := api.GetMe()
 	if err != nil {
-		_ = api.CloseApi()
-		_ = uploader.Close()
+		_ = bot.Close()
 		bot.logger.Fatal(err)
 	}
 	bot.logger.Infof("Authorized as %s\n", u.FirstName)
@@ -161,21 +190,34 @@ func (bot *Bot[T]) initLoggers(opts *BotOpts) {
 	}
 }
 
-func (bot *Bot[T]) GetUpdateOffset() int                    { return bot.updateOffset }
-func (bot *Bot[T]) SetUpdateOffset(offset int)              { bot.updateOffset = offset }
-func (bot *Bot[T]) GetUpdateTypes() []tgapi.UpdateType      { return bot.updateTypes }
-func (bot *Bot[T]) GetQueue() *extypes.Queue[*tgapi.Update] { return bot.updateQueue }
-func (bot *Bot[T]) GetLogger() *slog.Logger                 { return bot.logger }
-func (bot *Bot[T]) GetDBContext() *T                        { return bot.dbContext }
-func (bot *Bot[T]) L10n(lang, key string) string            { return bot.l10n.Translate(lang, key) }
+func (bot *Bot[T]) GetUpdateOffset() int {
+	bot.updateOffsetMu.Lock()
+	defer bot.updateOffsetMu.Unlock()
+	return bot.updateOffset
+}
+func (bot *Bot[T]) SetUpdateOffset(offset int) {
+	bot.updateOffsetMu.Lock()
+	defer bot.updateOffsetMu.Unlock()
+	bot.updateOffset = offset
+}
+func (bot *Bot[T]) GetUpdateTypes() []tgapi.UpdateType { return bot.updateTypes }
+func (bot *Bot[T]) GetLogger() *slog.Logger            { return bot.logger }
+func (bot *Bot[T]) GetDBContext() *T                   { return bot.dbContext }
+func (bot *Bot[T]) L10n(lang, key string) string       { return bot.l10n.Translate(lang, key) }
+func (bot *Bot[T]) SetDraftProvider(p *DraftProvider) *Bot[T] {
+	bot.draftProvider = p
+	return bot
+}
 
-func (bot *Bot[T]) AddDatabaseLogger(writer func(db *T) slog.LoggerWriter) *Bot[T] {
+type DbLogger[T DbContext] func(db *T) slog.LoggerWriter
+
+func (bot *Bot[T]) AddDatabaseLoggerWriter(writer DbLogger[T]) *Bot[T] {
 	w := writer(bot.dbContext)
 	bot.logger.AddWriter(w)
 	if bot.RequestLogger != nil {
 		bot.RequestLogger.AddWriter(w)
 	}
-	for _, l := range bot.dbWriterRequested {
+	for _, l := range bot.extraLoggers {
 		l.AddWriter(w)
 	}
 	return bot
@@ -209,7 +251,7 @@ func (bot *Bot[T]) Debug(debug bool) *Bot[T] {
 func (bot *Bot[T]) AddPlugins(plugin ...*Plugin[T]) *Bot[T] {
 	for _, p := range plugin {
 		bot.plugins = append(bot.plugins, *p)
-		bot.logger.Debugln(fmt.Sprintf("plugins with name \"%s\" registered", p.Name))
+		bot.logger.Debugln(fmt.Sprintf("plugins with name \"%s\" registered", p.name))
 	}
 	return bot
 }
@@ -240,7 +282,15 @@ func (bot *Bot[T]) AddL10n(l *L10n) *Bot[T] {
 	return bot
 }
 
-func (bot *Bot[T]) Run() {
+func (bot *Bot[T]) enqueueUpdate(u *tgapi.Update) error {
+	select {
+	case bot.updateQueue <- u:
+		return nil
+	default:
+		return extypes.QueueFullErr
+	}
+}
+func (bot *Bot[T]) RunWithContext(ctx context.Context) {
 	if len(bot.prefixes) == 0 {
 		bot.logger.Fatalln("no prefixes defined")
 		return
@@ -256,26 +306,35 @@ func (bot *Bot[T]) Run() {
 	bot.logger.Infoln("Bot running. Press CTRL+C to exit.")
 	go func() {
 		for {
-			_, err := bot.Updates()
-			if err != nil {
-				bot.logger.Errorln(err)
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				updates, err := bot.Updates()
+				if err != nil {
+					bot.logger.Errorln(err)
+					continue
+				}
+
+				for _, u := range updates {
+					select {
+					case bot.updateQueue <- new(u):
+					case <-ctx.Done():
+						return
+					}
+				}
 			}
 		}
 	}()
 
-	for {
-		queue := bot.updateQueue
-		if queue.IsEmpty() {
-			time.Sleep(time.Millisecond * 25)
-			continue
-		}
-
-		u := queue.Dequeue()
-		if u == nil {
-			bot.logger.Errorln("update is nil")
-			continue
-		}
-
-		bot.handle(u)
+	pool := pond.NewPool(16)
+	for update := range bot.updateQueue {
+		update := update
+		pool.Submit(func() {
+			bot.handle(update)
+		})
 	}
 }
+func (bot *Bot[T]) Run() {
+	bot.RunWithContext(context.Background())
+}

cmd_generator.go

@@ -27,7 +27,7 @@ func generateBotCommand[T any](cmd Command[T]) tgapi.BotCommand {
 
 func generateBotCommandForPlugin[T any](pl Plugin[T]) []tgapi.BotCommand {
 	commands := make([]tgapi.BotCommand, 0)
-	for _, cmd := range pl.Commands {
+	for _, cmd := range pl.commands {
 		if cmd.skipAutoCmd {
 			continue
 		}
@@ -46,6 +46,10 @@ func (bot *Bot[T]) AutoGenerateCommands() error {
 
 	commands := make([]tgapi.BotCommand, 0)
 	for _, pl := range bot.plugins {
+		if pl.skipAutoCmd {
+			continue
+		}
+
 		commands = append(commands, generateBotCommandForPlugin(pl)...)
 	}
 	if len(commands) > 100 {

drafts.go

@@ -0,0 +1,106 @@
+package laniakea
+
+import (
+	"math"
+	"math/rand/v2"
+	"sync/atomic"
+
+	"git.nix13.pw/scuroneko/laniakea/tgapi"
+)
+
+type draftIdGenerator interface {
+	Next() uint64
+}
+
+type RandomDraftIdGenerator struct {
+	draftIdGenerator
+}
+
+func (g *RandomDraftIdGenerator) Next() uint64 {
+	return rand.Uint64N(math.MaxUint64)
+}
+
+type LinearDraftIdGenerator struct {
+	draftIdGenerator
+	lastId uint64
+}
+
+func (g *LinearDraftIdGenerator) Next() uint64 {
+	return atomic.AddUint64(&g.lastId, 1)
+}
+
+type DraftProvider struct {
+	api *tgapi.API
+
+	chatID          int64
+	messageThreadID int
+	parseMode       tgapi.ParseMode
+	entities        []tgapi.MessageEntity
+
+	drafts    map[uint64]*Draft
+	generator draftIdGenerator
+}
+type Draft struct {
+	api *tgapi.API
+
+	chatID          int64
+	messageThreadID int
+	parseMode       tgapi.ParseMode
+	entities        []tgapi.MessageEntity
+
+	ID      uint64
+	Message string
+}
+
+func NewRandomDraftProvider(api *tgapi.API) *DraftProvider {
+	return &DraftProvider{
+		api: api, generator: &RandomDraftIdGenerator{},
+		drafts: make(map[uint64]*Draft),
+	}
+}
+func NewLinearDraftProvider(api *tgapi.API, startValue uint64) *DraftProvider {
+	return &DraftProvider{
+		api:       api,
+		generator: &LinearDraftIdGenerator{lastId: startValue},
+		drafts:    make(map[uint64]*Draft),
+	}
+}
+func (d *DraftProvider) NewDraft() *Draft {
+	id := d.generator.Next()
+	draft := &Draft{d.api, d.chatID, d.messageThreadID, d.parseMode, d.entities, id, ""}
+	d.drafts[id] = draft
+	return draft
+}
+
+func (d *Draft) Push(newText string) error {
+	d.Message += newText
+	params := tgapi.SendMessageDraftP{
+		ChatID:    d.chatID,
+		DraftID:   d.ID,
+		Text:      d.Message,
+		ParseMode: d.parseMode,
+		Entities:  d.entities,
+	}
+	if d.messageThreadID > 0 {
+		params.MessageThreadID = d.messageThreadID
+	}
+	_, err := d.api.SendMessageDraft(params)
+	return err
+}
+func (d *Draft) Flush() error {
+	if d.Message == "" {
+		return nil
+	}
+
+	params := tgapi.SendMessageP{
+		ChatID:    d.chatID,
+		ParseMode: d.parseMode,
+		Entities:  d.entities,
+		Text:      d.Message,
+	}
+	if d.messageThreadID > 0 {
+		params.MessageThreadID = d.messageThreadID
+	}
+	_, err := d.api.SendMessage(params)
+	return err
+}

go.mod

@@ -3,8 +3,10 @@ module git.nix13.pw/scuroneko/laniakea
 go 1.26
 
 require (
-	git.nix13.pw/scuroneko/extypes v1.2.0
+	git.nix13.pw/scuroneko/extypes v1.2.1
 	git.nix13.pw/scuroneko/slog v1.0.2
+	github.com/alitto/pond/v2 v2.6.2
+	golang.org/x/time v0.14.0
 )
 
 require (

go.sum

@@ -1,7 +1,9 @@
-git.nix13.pw/scuroneko/extypes v1.2.0 h1:2n2hD6KsMAted+6MGhAyeWyli2Qzc9G2y+pQNB7C1dM=
-git.nix13.pw/scuroneko/extypes v1.2.0/go.mod h1:uZVs8Yo3RrYAG9dMad6qR6lsYY67t+459D9c65QAYAw=
+git.nix13.pw/scuroneko/extypes v1.2.1 h1:IYrOjnWKL2EAuJYtYNa+luB1vBe6paE8VY/YD+5/RpQ=
+git.nix13.pw/scuroneko/extypes v1.2.1/go.mod h1:uZVs8Yo3RrYAG9dMad6qR6lsYY67t+459D9c65QAYAw=
 git.nix13.pw/scuroneko/slog v1.0.2 h1:vZyUROygxC2d5FJHUQM/30xFEHY1JT/aweDZXA4rm2g=
 git.nix13.pw/scuroneko/slog v1.0.2/go.mod h1:3Qm2wzkR5KjwOponMfG7TcGSDjmYaFqRAmLvSPTuWJI=
+github.com/alitto/pond/v2 v2.6.2 h1:Sphe40g0ILeM1pA2c2K+Th0DGU+pt0A/Kprr+WB24Pw=
+github.com/alitto/pond/v2 v2.6.2/go.mod h1:xkjYEgQ05RSpWdfSd1nM3OVv7TBhLdy7rMp3+2Nq+yE=
 github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=
 github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU=
 github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE=
@@ -11,3 +13,5 @@ github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
 golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
+golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=

handler.go

@@ -9,7 +9,13 @@ import (
 )
 
 func (bot *Bot[T]) handle(u *tgapi.Update) {
-	ctx := &MsgContext{Update: *u, Api: bot.api, botLogger: bot.logger, errorTemplate: bot.errorTemplate, l10n: bot.l10n}
+	ctx := &MsgContext{
+		Update: *u, Api: bot.api,
+		botLogger:     bot.logger,
+		errorTemplate: bot.errorTemplate,
+		l10n:          bot.l10n,
+		draftProvider: bot.draftProvider,
+	}
 	for _, middleware := range bot.middlewares {
 		middleware.Execute(ctx, bot.dbContext)
 	}
@@ -46,7 +52,7 @@ func (bot *Bot[T]) handleMessage(update *tgapi.Update, ctx *MsgContext) {
 	text = strings.TrimSpace(text[len(prefix):])
 
 	for _, plugin := range bot.plugins {
-		for cmd := range plugin.Commands {
+		for cmd := range plugin.commands {
 			if !strings.HasPrefix(text, cmd) {
 				continue
 			}
@@ -96,7 +102,7 @@ func (bot *Bot[T]) handleCallback(update *tgapi.Update, ctx *MsgContext) {
 	ctx.Args = data.Args
 
 	for _, plugin := range bot.plugins {
-		_, ok := plugin.Payloads[data.Command]
+		_, ok := plugin.payloads[data.Command]
 		if !ok {
 			continue
 		}

methods.go

@@ -19,14 +19,8 @@ func (bot *Bot[T]) Updates() ([]tgapi.Update, error) {
 		return nil, err
 	}
 
-	for _, u := range updates {
-		bot.SetUpdateOffset(u.UpdateID + 1)
-		err = bot.GetQueue().Enqueue(&u)
-		if err != nil {
-			return nil, err
-		}
-
-		if bot.RequestLogger != nil {
+	if bot.RequestLogger != nil {
+		for _, u := range updates {
 			j, err := json.Marshal(u)
 			if err != nil {
 				bot.GetLogger().Error(err)
@@ -34,5 +28,8 @@ func (bot *Bot[T]) Updates() ([]tgapi.Update, error) {
 			bot.RequestLogger.Debugf("UPDATE %s\n", j)
 		}
 	}
+	if len(updates) > 0 {
+		bot.SetUpdateOffset(updates[len(updates)-1].UpdateID + 1)
+	}
 	return updates, err
 }

msg_context.go

@@ -1,6 +1,7 @@
 package laniakea
 
 import (
+	"context"
 	"fmt"
 
 	"git.nix13.pw/scuroneko/laniakea/tgapi"
@@ -24,6 +25,7 @@ type MsgContext struct {
 	errorTemplate string
 	botLogger     *slog.Logger
 	l10n          *L10n
+	draftProvider *DraftProvider
 }
 
 type AnswerMessage struct {
@@ -77,6 +79,7 @@ func (ctx *MsgContext) editPhotoText(messageId int, text string, kb *InlineKeybo
 	if kb != nil {
 		params.ReplyMarkup = kb.Get()
 	}
+
 	msg, _, err := ctx.Api.EditMessageCaption(params)
 	if err != nil {
 		ctx.botLogger.Errorln(err)
@@ -105,7 +108,18 @@ func (ctx *MsgContext) answer(text string, keyboard *InlineKeyboard) *AnswerMess
 	if keyboard != nil {
 		params.ReplyMarkup = keyboard.Get()
 	}
+	if ctx.Msg.MessageThreadID > 0 {
+		params.MessageThreadID = ctx.Msg.MessageThreadID
+	}
+	if ctx.Msg.DirectMessageTopic != nil {
+		params.DirectMessagesTopicID = ctx.Msg.DirectMessageTopic.TopicID
+	}
 
+	cont := context.Background()
+	if err := ctx.Api.Limiter.Wait(cont, ctx.Msg.Chat.ID); err != nil {
+		ctx.botLogger.Errorln(err)
+		return nil
+	}
 	msg, err := ctx.Api.SendMessage(params)
 	if err != nil {
 		ctx.botLogger.Errorln(err)
@@ -135,6 +149,10 @@ func (ctx *MsgContext) answerPhoto(photoId, text string, kb *InlineKeyboard) *An
 	if kb != nil {
 		params.ReplyMarkup = kb.Get()
 	}
+	if ctx.Msg.MessageThreadID > 0 {
+		params.MessageThreadID = ctx.Msg.MessageThreadID
+	}
+
 	msg, err := ctx.Api.SendPhoto(params)
 	if err != nil {
 		ctx.botLogger.Errorln(err)
@@ -162,12 +180,8 @@ func (ctx *MsgContext) delete(messageId int) {
 		ctx.botLogger.Errorln(err)
 	}
 }
-func (m *AnswerMessage) Delete() {
-	m.ctx.delete(m.MessageID)
-}
-func (ctx *MsgContext) CallbackDelete() {
-	ctx.delete(ctx.CallbackMsgId)
-}
+func (m *AnswerMessage) Delete()        { m.ctx.delete(m.MessageID) }
+func (ctx *MsgContext) CallbackDelete() { ctx.delete(ctx.CallbackMsgId) }
 
 func (ctx *MsgContext) answerCallbackQuery(url, text string, showAlert bool) {
 	if len(ctx.CallbackQueryId) == 0 {
@@ -181,23 +195,19 @@ func (ctx *MsgContext) answerCallbackQuery(url, text string, showAlert bool) {
 		ctx.botLogger.Errorln(err)
 	}
 }
-func (ctx *MsgContext) AnswerCbQuery() {
-	ctx.answerCallbackQuery("", "", false)
-}
-func (ctx *MsgContext) AnswerCbQueryText(text string) {
-	ctx.answerCallbackQuery("", text, false)
-}
-func (ctx *MsgContext) AnswerCbQueryAlert(text string) {
-	ctx.answerCallbackQuery("", text, true)
-}
-func (ctx *MsgContext) AnswerCbQueryUrl(u string) {
-	ctx.answerCallbackQuery(u, "", false)
-}
+func (ctx *MsgContext) AnswerCbQuery()                 { ctx.answerCallbackQuery("", "", false) }
+func (ctx *MsgContext) AnswerCbQueryText(text string)  { ctx.answerCallbackQuery("", text, false) }
+func (ctx *MsgContext) AnswerCbQueryAlert(text string) { ctx.answerCallbackQuery("", text, true) }
+func (ctx *MsgContext) AnswerCbQueryUrl(u string)      { ctx.answerCallbackQuery(u, "", false) }
 
 func (ctx *MsgContext) SendAction(action tgapi.ChatActionType) {
-	_, err := ctx.Api.SendChatAction(tgapi.SendChatActionP{
+	params := tgapi.SendChatActionP{
 		ChatID: ctx.Msg.Chat.ID, Action: action,
-	})
+	}
+	if ctx.Msg.MessageThreadID > 0 {
+		params.MessageThreadID = ctx.Msg.MessageThreadID
+	}
+	_, err := ctx.Api.SendChatAction(params)
 	if err != nil {
 		ctx.botLogger.Errorln(err)
 	}
@@ -213,10 +223,20 @@ func (ctx *MsgContext) error(err error) {
 	}
 	ctx.botLogger.Errorln(err)
 }
-func (ctx *MsgContext) Error(err error) {
-	ctx.error(err)
-}
+func (ctx *MsgContext) Error(err error) { ctx.error(err) }
 
+func (ctx *MsgContext) NewDraft() *Draft {
+	c := context.Background()
+	if err := ctx.Api.Limiter.Wait(c, ctx.Msg.Chat.ID); err != nil {
+		ctx.botLogger.Errorln(err)
+		return nil
+	}
+
+	draft := ctx.draftProvider.NewDraft()
+	draft.chatID = ctx.Msg.Chat.ID
+	draft.messageThreadID = ctx.Msg.MessageThreadID
+	return draft
+}
 func (ctx *MsgContext) Translate(key string) string {
 	if ctx.From == nil {
 		return key

plugins.go

@@ -93,37 +93,42 @@ func (c *Command[T]) validateArgs(args []string) error {
 }
 
 type Plugin[T DbContext] struct {
-	Name        string
-	Commands    map[string]Command[T]
-	Payloads    map[string]Command[T]
-	Middlewares extypes.Slice[Middleware[T]]
+	name        string
+	commands    map[string]Command[T]
+	payloads    map[string]Command[T]
+	middlewares extypes.Slice[Middleware[T]]
+	skipAutoCmd bool
 }
 
 func NewPlugin[T DbContext](name string) *Plugin[T] {
 	return &Plugin[T]{
 		name, map[string]Command[T]{},
-		map[string]Command[T]{}, extypes.Slice[Middleware[T]]{},
+		map[string]Command[T]{}, extypes.Slice[Middleware[T]]{}, false,
 	}
 }
 
 func (p *Plugin[T]) AddCommand(command *Command[T]) *Plugin[T] {
-	p.Commands[command.command] = *command
+	p.commands[command.command] = *command
 	return p
 }
 func (p *Plugin[T]) NewCommand(exec CommandExecutor[T], command string, args ...CommandArg) *Command[T] {
 	return NewCommand(exec, command, args...)
 }
 func (p *Plugin[T]) AddPayload(command *Command[T]) *Plugin[T] {
-	p.Payloads[command.command] = *command
+	p.payloads[command.command] = *command
 	return p
 }
 func (p *Plugin[T]) AddMiddleware(middleware Middleware[T]) *Plugin[T] {
-	p.Middlewares = p.Middlewares.Push(middleware)
+	p.middlewares = p.middlewares.Push(middleware)
+	return p
+}
+func (p *Plugin[T]) SkipCommandAutoGen() *Plugin[T] {
+	p.skipAutoCmd = true
 	return p
 }
 
 func (p *Plugin[T]) executeCmd(cmd string, ctx *MsgContext, dbContext *T) {
-	command := p.Commands[cmd]
+	command := p.commands[cmd]
 	if err := command.validateArgs(ctx.Args); err != nil {
 		ctx.error(err)
 		return
@@ -131,7 +136,7 @@ func (p *Plugin[T]) executeCmd(cmd string, ctx *MsgContext, dbContext *T) {
 	command.exec(ctx, dbContext)
 }
 func (p *Plugin[T]) executePayload(payload string, ctx *MsgContext, dbContext *T) {
-	pl := p.Payloads[payload]
+	pl := p.payloads[payload]
 	if err := pl.validateArgs(ctx.Args); err != nil {
 		ctx.error(err)
 		return
@@ -139,7 +144,7 @@ func (p *Plugin[T]) executePayload(payload string, ctx *MsgContext, dbContext *T
 	pl.exec(ctx, dbContext)
 }
 func (p *Plugin[T]) executeMiddlewares(ctx *MsgContext, db *T) bool {
-	for _, m := range p.Middlewares {
+	for _, m := range p.middlewares {
 		if !m.Execute(ctx, db) {
 			return false
 		}

tgapi/api.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -18,8 +19,13 @@ type APIOpts struct {
 	client        *http.Client
 	useTestServer bool
 	apiUrl        string
+
+	limiter           *utils.RateLimiter
+	dropOverflowLimit bool
 }
 
+var ErrPoolUnexpected = errors.New("unexpected response from pool")
+
 func NewAPIOpts(token string) *APIOpts {
 	return &APIOpts{token: token, client: nil, useTestServer: false, apiUrl: "https://api.telegram.org"}
 }
@@ -39,6 +45,14 @@ func (opts *APIOpts) SetAPIUrl(apiUrl string) *APIOpts {
 	}
 	return opts
 }
+func (opts *APIOpts) SetLimiter(limiter *utils.RateLimiter) *APIOpts {
+	opts.limiter = limiter
+	return opts
+}
+func (opts *APIOpts) SetLimiterDrop(b bool) *APIOpts {
+	opts.dropOverflowLimit = b
+	return opts
+}
 
 type API struct {
 	token         string
@@ -46,6 +60,10 @@ type API struct {
 	logger        *slog.Logger
 	useTestServer bool
 	apiUrl        string
+
+	pool              *WorkerPool
+	Limiter           *utils.RateLimiter
+	dropOverflowLimit bool
 }
 
 func NewAPI(opts *APIOpts) *API {
@@ -55,27 +73,58 @@ func NewAPI(opts *APIOpts) *API {
 	if client == nil {
 		client = &http.Client{Timeout: time.Second * 45}
 	}
-	return &API{opts.token, client, l, opts.useTestServer, opts.apiUrl}
+	pool := NewWorkerPool(16, 256)
+	pool.Start(context.Background())
+	return &API{
+		opts.token, client, l,
+		opts.useTestServer, opts.apiUrl,
+		pool, opts.limiter, opts.dropOverflowLimit,
+	}
+}
+func (api *API) CloseApi() error {
+	api.pool.Stop()
+	return api.logger.Close()
 }
-func (api *API) CloseApi() error         { return api.logger.Close() }
 func (api *API) GetLogger() *slog.Logger { return api.logger }
 
+type ResponseParameters struct {
+	MigrateToChatID *int64 `json:"migrate_to_chat_id,omitempty"`
+	RetryAfter      *int   `json:"retry_after,omitempty"`
+}
 type ApiResponse[R any] struct {
 	Ok          bool   `json:"ok"`
 	Description string `json:"description,omitempty"`
 	Result      R      `json:"result,omitempty"`
 	ErrorCode   int    `json:"error_code,omitempty"`
+
+	Parameters *ResponseParameters `json:"parameters,omitempty"`
 }
 type TelegramRequest[R, P any] struct {
 	method string
 	params P
+	chatId int64
 }
 
 func NewRequest[R, P any](method string, params P) TelegramRequest[R, P] {
-	return TelegramRequest[R, P]{method: method, params: params}
+	return TelegramRequest[R, P]{method, params, 0}
 }
-func (r TelegramRequest[R, P]) DoWithContext(ctx context.Context, api *API) (R, error) {
+func NewRequestWithChatID[R, P any](method string, params P, chatId int64) TelegramRequest[R, P] {
+	return TelegramRequest[R, P]{method, params, chatId}
+}
+func (r TelegramRequest[R, P]) doRequest(ctx context.Context, api *API) (R, error) {
 	var zero R
+	if api.Limiter != nil {
+		if api.dropOverflowLimit {
+			if !api.Limiter.GlobalAllow() {
+				return zero, errors.New("rate limited")
+			}
+		} else {
+			if err := api.Limiter.GlobalWait(ctx); err != nil {
+				return zero, err
+			}
+		}
+	}
+
 	data, err := json.Marshal(r.params)
 	if err != nil {
 		return zero, err
@@ -109,11 +158,50 @@ func (r TelegramRequest[R, P]) DoWithContext(ctx context.Context, api *API) (R,
 		return zero, err
 	}
 	api.logger.Debugln("RES", r.method, string(data))
-	if res.StatusCode != http.StatusOK {
+	if res.StatusCode != http.StatusOK && res.StatusCode != http.StatusTooManyRequests {
 		return zero, fmt.Errorf("unexpected status code: %d, %s", res.StatusCode, string(data))
 	}
-	return parseBody[R](data)
 
+	responseData, err := parseBody[R](data)
+	if errors.Is(err, ErrRateLimit) {
+		if responseData.Parameters != nil {
+			after := 0
+			if responseData.Parameters.RetryAfter != nil {
+				after = *responseData.Parameters.RetryAfter
+			}
+			if r.chatId > 0 {
+				api.Limiter.SetChatLock(r.chatId, after)
+			} else {
+				api.Limiter.SetGlobalLock(after)
+			}
+			time.Sleep(time.Duration(after) * time.Second)
+			return r.doRequest(ctx, api)
+		}
+		return zero, ErrRateLimit
+	}
+	return responseData.Result, err
+}
+func (r TelegramRequest[R, P]) DoWithContext(ctx context.Context, api *API) (R, error) {
+	var zero R
+	result, err := api.pool.Submit(ctx, func(ctx context.Context) (any, error) {
+		return r.doRequest(ctx, api)
+	})
+	if err != nil {
+		return zero, err
+	}
+
+	select {
+	case <-ctx.Done():
+		return zero, ctx.Err()
+	case res := <-result:
+		if res.Err != nil {
+			return zero, res.Err
+		}
+		if val, ok := res.Value.(R); ok {
+			return val, nil
+		}
+		return zero, ErrPoolUnexpected
+	}
 }
 func (r TelegramRequest[R, P]) Do(api *API) (R, error) {
 	return r.DoWithContext(context.Background(), api)
@@ -123,15 +211,17 @@ func readBody(body io.ReadCloser) ([]byte, error) {
 	reader := io.LimitReader(body, 10<<20)
 	return io.ReadAll(reader)
 }
-func parseBody[R any](data []byte) (R, error) {
-	var zero R
+func parseBody[R any](data []byte) (ApiResponse[R], error) {
 	var resp ApiResponse[R]
 	err := json.Unmarshal(data, &resp)
 	if err != nil {
-		return zero, err
+		return resp, err
 	}
 	if !resp.Ok {
-		return zero, fmt.Errorf("[%d] %s", resp.ErrorCode, resp.Description)
+		if resp.ErrorCode == 429 {
+			return resp, ErrRateLimit
+		}
+		return resp, fmt.Errorf("[%d] %s", resp.ErrorCode, resp.Description)
 	}
-	return resp.Result, nil
+	return resp, nil
 }

tgapi/attachments_methods.go

@@ -2,7 +2,7 @@ package tgapi
 
 type SendPhotoP struct {
 	BusinessConnectionID  string `json:"business_connection_id,omitempty"`
-	ChatID                int    `json:"chat_id"`
+	ChatID                int64  `json:"chat_id"`
 	MessageThreadID       int    `json:"message_thread_id,omitempty"`
 	DirectMessagesTopicID int    `json:"direct_messages_topic_id,omitempty"`
 

tgapi/chat_methods.go

@@ -56,6 +56,7 @@ type PromoteChatMember struct {
 	CanPinMessages          bool `json:"can_pin_messages,omitempty"`
 	CanManageTopics         bool `json:"can_manage_topics,omitempty"`
 	CanManageDirectMessages bool `json:"can_manage_direct_messages,omitempty"`
+	CanManageTags           bool `json:"can_manage_tags,omitempty"`
 }
 
 func (api *API) PromoteChatMember(params PromoteChatMember) (bool, error) {
@@ -74,6 +75,17 @@ func (api *API) SetChatAdministratorCustomTitle(params SetChatAdministratorCusto
 	return req.Do(api)
 }
 
+type SetChatMemberTagP struct {
+	ChatID int    `json:"chat_id"`
+	UserID int    `json:"user_id"`
+	Tag    string `json:"tag,omitempty"`
+}
+
+func (api *API) SetChatMemberTag(params SetChatMemberTagP) (bool, error) {
+	req := NewRequest[bool]("setChatMemberTag", params)
+	return req.Do(api)
+}
+
 type BanChatSenderChatP struct {
 	ChatID       int `json:"chat_id"`
 	SenderChatID int `json:"sender_chat_id"`

tgapi/chat_types.go

@@ -1,7 +1,7 @@
 package tgapi
 
 type Chat struct {
-	ID               int     `json:"id"`
+	ID               int64   `json:"id"`
 	Type             string  `json:"type"`
 	Title            *string `json:"title,omitempty"`
 	Username         *string `json:"username,omitempty"`
@@ -99,6 +99,7 @@ type ChatPermissions struct {
 	CanSendPolls         bool `json:"can_send_polls"`
 	CanSendOtherMessages bool `json:"can_send_other_messages"`
 	CanAddWebPagePreview bool `json:"can_add_web_page_preview"`
+	CatEditTag           bool `json:"cat_edit_tag"`
 	CanChangeInfo        bool `json:"can_change_info"`
 	CanInviteUsers       bool `json:"can_invite_users"`
 	CanPinMessages       bool `json:"can_pin_messages"`
@@ -137,6 +138,7 @@ const (
 type ChatMember struct {
 	Status ChatMemberStatusType `json:"status"`
 	User   User                 `json:"user"`
+	Tag    string               `json:"tag,omitempty"`
 
 	// Owner
 	IsAnonymous *bool   `json:"is_anonymous"`
@@ -160,6 +162,7 @@ type ChatMember struct {
 	CanPinMessages          *bool `json:"can_pin_messages,omitempty"`
 	CanManageTopics         *bool `json:"can_manage_topics,omitempty"`
 	CanManageDirectMessages *bool `json:"can_manage_direct_messages,omitempty"`
+	CanManageTags           *bool `json:"can_manage_tags,omitempty"`
 
 	// Member
 	UntilDate *int `json:"until_date,omitempty"`
@@ -175,6 +178,7 @@ type ChatMember struct {
 	CanSendPolls         *bool `json:"can_send_polls,omitempty"`
 	CanSendOtherMessages *bool `json:"can_send_other_messages,omitempty"`
 	CanAddWebPagePreview *bool `json:"can_add_web_page_preview,omitempty"`
+	CanEditTag           *bool `json:"can_edit_tag,omitempty"`
 }
 
 type ChatBoostSource struct {
@@ -215,6 +219,7 @@ type ChatAdministratorRights struct {
 	CanPinMessages          *bool `json:"can_pin_messages,omitempty"`
 	CanManageTopics         *bool `json:"can_manage_topics,omitempty"`
 	CanManageDirectMessages *bool `json:"can_manage_direct_messages,omitempty"`
+	CanManageTags           *bool `json:"can_manage_tags,omitempty"`
 }
 
 type ChatBoostUpdated struct {

tgapi/errors.go

@@ -0,0 +1,5 @@
+package tgapi
+
+import "errors"
+
+var ErrRateLimit = errors.New("rate limit exceeded")

tgapi/messages_methods.go

@@ -2,9 +2,9 @@ package tgapi
 
 type SendMessageP struct {
 	BusinessConnectionID  string `json:"business_connection_id,omitempty"`
-	ChatID                int    `json:"chat_id"`
+	ChatID                int64  `json:"chat_id"`
 	MessageThreadID       int    `json:"message_thread_id,omitempty"`
-	DirectMessagesTopicID int    `json:"direct_messages_topic_id,omitempty"`
+	DirectMessagesTopicID int64  `json:"direct_messages_topic_id,omitempty"`
 
 	Text                 string              `json:"text"`
 	ParseMode            ParseMode           `json:"parse_mode,omitempty"`
@@ -21,7 +21,7 @@ type SendMessageP struct {
 }
 
 func (api *API) SendMessage(params SendMessageP) (Message, error) {
-	req := NewRequest[Message, SendMessageP]("sendMessage", params)
+	req := NewRequestWithChatID[Message, SendMessageP]("sendMessage", params, params.ChatID)
 	return req.Do(api)
 }
 
@@ -266,22 +266,22 @@ func (api *API) SendDice(params SendDiceP) (Message, error) {
 }
 
 type SendMessageDraftP struct {
-	ChatID          int             `json:"chat_id"`
+	ChatID          int64           `json:"chat_id"`
 	MessageThreadID int             `json:"message_thread_id,omitempty"`
-	DraftID         int             `json:"draft_id"`
+	DraftID         uint64          `json:"draft_id"`
 	Text            string          `json:"text"`
 	ParseMode       ParseMode       `json:"parse_mode,omitempty"`
 	Entities        []MessageEntity `json:"entities,omitempty"`
 }
 
 func (api *API) SendMessageDraft(params SendMessageDraftP) (bool, error) {
-	req := NewRequest[bool]("sendMessageDraft", params)
+	req := NewRequestWithChatID[bool]("sendMessageDraft", params, params.ChatID)
 	return req.Do(api)
 }
 
 type SendChatActionP struct {
 	BusinessConnectionID string         `json:"business_connection_id,omitempty"`
-	ChatID               int            `json:"chat_id"`
+	ChatID               int64          `json:"chat_id"`
 	MessageThreadID      int            `json:"message_thread_id,omitempty"`
 	Action               ChatActionType `json:"action"`
 }
@@ -307,7 +307,7 @@ func (api *API) SetMessageReaction(params SetMessageReactionP) (bool, error) {
 
 type EditMessageTextP struct {
 	BusinessConnectionID string       `json:"business_connection_id,omitempty"`
-	ChatID               int          `json:"chat_id,omitempty"`
+	ChatID               int64        `json:"chat_id,omitempty"`
 	MessageID            int          `json:"message_id,omitempty"`
 	InlineMessageID      string       `json:"inline_message_id,omitempty"`
 	Text                 string       `json:"text"`
@@ -331,7 +331,7 @@ func (api *API) EditMessageText(params EditMessageTextP) (Message, bool, error)
 
 type EditMessageCaptionP struct {
 	BusinessConnectionID string       `json:"business_connection_id,omitempty"`
-	ChatID               int          `json:"chat_id,omitempty"`
+	ChatID               int64        `json:"chat_id,omitempty"`
 	MessageID            int          `json:"message_id,omitempty"`
 	InlineMessageID      string       `json:"inline_message_id,omitempty"`
 	Caption              string       `json:"caption"`
@@ -495,8 +495,8 @@ func (api *API) DeclineSuggestedPost(params DeclineSuggestedPostP) (bool, error)
 }
 
 type DeleteMessageP struct {
-	ChatID    int `json:"chat_id"`
-	MessageID int `json:"message_id"`
+	ChatID    int64 `json:"chat_id"`
+	MessageID int   `json:"message_id"`
 }
 
 func (api *API) DeleteMessage(params DeleteMessageP) (bool, error) {

tgapi/messages_types.go

@@ -6,16 +6,23 @@ type MessageReplyMarkup struct {
 	InlineKeyboard [][]InlineKeyboardButton `json:"inline_keyboard"`
 }
 
-type Message struct {
-	MessageID            int    `json:"message_id"`
-	MessageThreadID      int    `json:"message_thread_id,omitempty"`
-	BusinessConnectionId string `json:"business_connection_id,omitempty"`
-	From                 *User  `json:"from,omitempty"`
+type DirectMessageTopic struct {
+	TopicID int64 `json:"topic_id"`
+	User    *User `json:"user,omitempty"`
+}
 
-	SenderChat        *Chat `json:"sender_chat,omitempty"`
-	SenderBoostCount  int   `json:"sender_boost_count,omitempty"`
-	SenderBusinessBot *User `json:"sender_business_bot,omitempty"`
-	Chat              *Chat `json:"chat,omitempty"`
+type Message struct {
+	MessageID            int                 `json:"message_id"`
+	MessageThreadID      int                 `json:"message_thread_id,omitempty"`
+	DirectMessageTopic   *DirectMessageTopic `json:"direct_message_topic,omitempty"`
+	BusinessConnectionId string              `json:"business_connection_id,omitempty"`
+	From                 *User               `json:"from,omitempty"`
+
+	SenderChat        *Chat  `json:"sender_chat,omitempty"`
+	SenderBoostCount  int    `json:"sender_boost_count,omitempty"`
+	SenderBusinessBot *User  `json:"sender_business_bot,omitempty"`
+	SenderTag         string `json:"sender_tag,omitempty"`
+	Chat              *Chat  `json:"chat,omitempty"`
 
 	IsTopicMessage     bool     `json:"is_topic_message,omitempty"`
 	IsAutomaticForward bool     `json:"is_automatic_forward,omitempty"`
@@ -74,6 +81,7 @@ const (
 	MessageEntityTextLink             MessageEntityType = "text_link"
 	MessageEntityTextMention          MessageEntityType = "text_mention"
 	MessageEntityCustomEmoji          MessageEntityType = "custom_emoji"
+	MessageEntityDateTime             MessageEntityType = "date_time"
 )
 
 type MessageEntity struct {
@@ -85,6 +93,9 @@ type MessageEntity struct {
 	User          *User  `json:"user,omitempty"`
 	Language      string `json:"language,omitempty"`
 	CustomEmojiID string `json:"custom_emoji_id,omitempty"`
+
+	UnixTime       int    `json:"unix_time,omitempty"`
+	DateTimeFormat string `json:"date_time_format,omitempty"`
 }
 
 type ReplyParameters struct {

tgapi/pool.go

@@ -0,0 +1,92 @@
+package tgapi
+
+import (
+	"context"
+	"errors"
+	"sync"
+)
+
+var ErrPoolQueueFull = errors.New("worker pool queue full")
+
+type RequestEnvelope struct {
+	DoFunc   func(context.Context) (any, error) // функция, которая выполнит запрос и вернет any
+	ResultCh chan RequestResult                 // канал для результата
+}
+type RequestResult struct {
+	Value any
+	Err   error
+}
+
+// WorkerPool управляет воркерами и очередью
+type WorkerPool struct {
+	taskCh    chan RequestEnvelope
+	queueSize int
+	workers   int
+	wg        sync.WaitGroup
+	quit      chan struct{}
+	started   bool
+	startedMu sync.Mutex
+}
+
+func NewWorkerPool(workers int, queueSize int) *WorkerPool {
+	return &WorkerPool{
+		taskCh:    make(chan RequestEnvelope, queueSize),
+		queueSize: queueSize,
+		workers:   workers,
+		quit:      make(chan struct{}),
+	}
+}
+
+// Start запускает воркеров
+func (p *WorkerPool) Start(ctx context.Context) {
+	p.startedMu.Lock()
+	defer p.startedMu.Unlock()
+	if p.started {
+		return
+	}
+	p.started = true
+	for i := 0; i < p.workers; i++ {
+		p.wg.Add(1)
+		go p.worker(ctx)
+	}
+}
+
+// Stop останавливает пул (ждет завершения текущих задач)
+func (p *WorkerPool) Stop() {
+	close(p.quit)
+	p.wg.Wait()
+}
+
+// Submit отправляет задачу в очередь и возвращает канал для результата
+func (p *WorkerPool) Submit(ctx context.Context, do func(context.Context) (any, error)) (<-chan RequestResult, error) {
+	if len(p.taskCh) >= p.queueSize {
+		return nil, ErrPoolQueueFull
+	}
+
+	resultCh := make(chan RequestResult, 1) // буфер 1, чтобы не блокировать воркера
+	envelope := RequestEnvelope{do, resultCh}
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case p.taskCh <- envelope:
+		return resultCh, nil
+	default:
+		return nil, ErrPoolQueueFull
+	}
+}
+
+// worker выполняет задачи
+func (p *WorkerPool) worker(ctx context.Context) {
+	defer p.wg.Done()
+	for {
+		select {
+		case <-p.quit:
+			return
+		case envelope := <-p.taskCh:
+			// Выполняем задачу с переданным контекстом (или можно использовать свой)
+			val, err := envelope.DoFunc(ctx)
+			envelope.ResultCh <- RequestResult{Value: val, Err: err}
+			close(envelope.ResultCh)
+		}
+	}
+}

tgapi/uploader_api.go

@@ -3,6 +3,7 @@ package tgapi
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"mime/multipart"
 	"net/http"
@@ -63,10 +64,21 @@ type UploaderRequest[R, P any] struct {
 func NewUploaderRequest[R, P any](method string, params P, files ...UploaderFile) UploaderRequest[R, P] {
 	return UploaderRequest[R, P]{method, files, params}
 }
-func (u UploaderRequest[R, P]) DoWithContext(ctx context.Context, up *Uploader) (R, error) {
+func (r UploaderRequest[R, P]) doRequest(ctx context.Context, up *Uploader) (R, error) {
 	var zero R
+	if up.api.Limiter != nil {
+		if up.api.dropOverflowLimit {
+			if !up.api.Limiter.GlobalAllow() {
+				return zero, errors.New("rate limited")
+			}
+		} else {
+			if err := up.api.Limiter.GlobalWait(ctx); err != nil {
+				return zero, err
+			}
+		}
+	}
 
-	buf, contentType, err := prepareMultipart(u.files, u.params)
+	buf, contentType, err := prepareMultipart(r.files, r.params)
 	if err != nil {
 		return zero, err
 	}
@@ -75,7 +87,7 @@ func (u UploaderRequest[R, P]) DoWithContext(ctx context.Context, up *Uploader)
 	if up.api.useTestServer {
 		methodPrefix = "/test"
 	}
-	url := fmt.Sprintf("%s/bot%s%s/%s", up.api.apiUrl, up.api.token, methodPrefix, u.method)
+	url := fmt.Sprintf("%s/bot%s%s/%s", up.api.apiUrl, up.api.token, methodPrefix, r.method)
 	req, err := http.NewRequestWithContext(ctx, "POST", url, buf)
 	if err != nil {
 		return zero, err
@@ -84,7 +96,7 @@ func (u UploaderRequest[R, P]) DoWithContext(ctx context.Context, up *Uploader)
 	req.Header.Set("Accept", "application/json")
 	req.Header.Set("User-Agent", fmt.Sprintf("Laniakea/%s", utils.VersionString))
 
-	up.logger.Debugln("UPLOADER REQ", u.method)
+	up.logger.Debugln("UPLOADER REQ", r.method)
 	res, err := up.api.client.Do(req)
 	if err != nil {
 		return zero, err
@@ -92,15 +104,42 @@ func (u UploaderRequest[R, P]) DoWithContext(ctx context.Context, up *Uploader)
 	defer res.Body.Close()
 
 	body, err := readBody(res.Body)
-	up.logger.Debugln("UPLOADER RES", u.method, string(body))
+	up.logger.Debugln("UPLOADER RES", r.method, string(body))
 	if res.StatusCode != http.StatusOK {
 		return zero, fmt.Errorf("unexpected status code: %d, %s", res.StatusCode, string(body))
 	}
 
-	return parseBody[R](body)
+	respBody, err := parseBody[R](body)
+	if err != nil {
+		return zero, err
+	}
+	return respBody.Result, nil
 }
-func (u UploaderRequest[R, P]) Do(up *Uploader) (R, error) {
-	return u.DoWithContext(context.Background(), up)
+func (r UploaderRequest[R, P]) DoWithContext(ctx context.Context, up *Uploader) (R, error) {
+	var zero R
+
+	result, err := up.api.pool.Submit(ctx, func(ctx context.Context) (any, error) {
+		return r.doRequest(ctx, up)
+	})
+	if err != nil {
+		return zero, err
+	}
+
+	select {
+	case <-ctx.Done():
+		return zero, ctx.Err()
+	case res := <-result:
+		if res.Err != nil {
+			return zero, res.Err
+		}
+		if val, ok := res.Value.(R); ok {
+			return val, nil
+		}
+		return zero, ErrPoolUnexpected
+	}
+}
+func (r UploaderRequest[R, P]) Do(up *Uploader) (R, error) {
+	return r.DoWithContext(context.Background(), up)
 }
 
 func prepareMultipart[P any](files []UploaderFile, params P) (*bytes.Buffer, string, error) {

tgapi/uploader_methods.go

@@ -2,7 +2,7 @@ package tgapi
 
 type UploadPhotoP struct {
 	BusinessConnectionID  string `json:"business_connection_id,omitempty"`
-	ChatID                int    `json:"chat_id"`
+	ChatID                int64  `json:"chat_id"`
 	MessageThreadID       int    `json:"message_thread_id,omitempty"`
 	DirectMessagesTopicID int    `json:"direct_messages_topic_id,omitempty"`
 

utils/limiter.go

@@ -0,0 +1,155 @@
+package utils
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"golang.org/x/time/rate"
+)
+
+type RateLimiter struct {
+	globalLockUntil time.Time
+	globalLimiter   *rate.Limiter
+	globalMu        sync.RWMutex
+
+	chatLocks    map[int64]time.Time
+	chatLimiters map[int64]*rate.Limiter
+	chatMu       sync.Mutex
+}
+
+func NewRateLimiter() *RateLimiter {
+	return &RateLimiter{
+		globalLimiter: rate.NewLimiter(30, 30),
+		chatLimiters:  make(map[int64]*rate.Limiter),
+		chatLocks:     make(map[int64]time.Time), // инициализация!
+	}
+}
+
+func (rl *RateLimiter) SetGlobalLock(retryAfter int) {
+	if retryAfter <= 0 {
+		return
+	}
+	rl.globalMu.Lock()
+	defer rl.globalMu.Unlock()
+	rl.globalLockUntil = time.Now().Add(time.Duration(retryAfter) * time.Second)
+}
+
+func (rl *RateLimiter) SetChatLock(chatID int64, retryAfter int) {
+	if retryAfter <= 0 {
+		return
+	}
+	rl.chatMu.Lock()
+	defer rl.chatMu.Unlock()
+	rl.chatLocks[chatID] = time.Now().Add(time.Duration(retryAfter) * time.Second)
+}
+
+// GlobalWait блокирует до возможности сделать глобальный запрос.
+func (rl *RateLimiter) GlobalWait(ctx context.Context) error {
+	// Ждём окончания глобальной блокировки, если она есть
+	if err := rl.waitForGlobalUnlock(ctx); err != nil {
+		return err
+	}
+	// Ждём разрешения rate limiter'а
+	return rl.globalLimiter.Wait(ctx)
+}
+
+// Wait блокирует до возможности сделать запрос для конкретного чата.
+func (rl *RateLimiter) Wait(ctx context.Context, chatID int64) error {
+	// Ждём окончания блокировки чата
+	if err := rl.waitForChatUnlock(ctx, chatID); err != nil {
+		return err
+	}
+	// Затем глобальной блокировки
+	if err := rl.waitForGlobalUnlock(ctx); err != nil {
+		return err
+	}
+	// Получаем или создаём лимитер для чата
+	limiter := rl.getChatLimiter(chatID)
+	return limiter.Wait(ctx)
+}
+
+// GlobalAllow неблокирующая проверка глобального запроса.
+func (rl *RateLimiter) GlobalAllow() bool {
+	rl.globalMu.RLock()
+	until := rl.globalLockUntil
+	rl.globalMu.RUnlock()
+
+	if !until.IsZero() && time.Now().Before(until) {
+		return false
+	}
+	return rl.globalLimiter.Allow()
+}
+
+// Allow неблокирующая проверка запроса для чата.
+func (rl *RateLimiter) Allow(chatID int64) bool {
+	// Проверяем глобальную блокировку
+	rl.globalMu.RLock()
+	globalUntil := rl.globalLockUntil
+	rl.globalMu.RUnlock()
+	if !globalUntil.IsZero() && time.Now().Before(globalUntil) {
+		return false
+	}
+
+	// Проверяем блокировку чата
+	rl.chatMu.Lock()
+	chatUntil, ok := rl.chatLocks[chatID]
+	rl.chatMu.Unlock()
+	if ok && !chatUntil.IsZero() && time.Now().Before(chatUntil) {
+		return false
+	}
+
+	// Проверяем глобальный лимитер
+	if !rl.globalLimiter.Allow() {
+		return false
+	}
+
+	// Проверяем лимитер чата
+	limiter := rl.getChatLimiter(chatID)
+	return limiter.Allow()
+}
+
+// Внутренние вспомогательные методы
+
+func (rl *RateLimiter) waitForGlobalUnlock(ctx context.Context) error {
+	rl.globalMu.RLock()
+	until := rl.globalLockUntil
+	rl.globalMu.RUnlock()
+
+	if until.IsZero() || time.Now().After(until) {
+		return nil
+	}
+	select {
+	case <-time.After(time.Until(until)):
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (rl *RateLimiter) waitForChatUnlock(ctx context.Context, chatID int64) error {
+	rl.chatMu.Lock()
+	until, ok := rl.chatLocks[chatID]
+	rl.chatMu.Unlock()
+
+	if !ok || until.IsZero() || time.Now().After(until) {
+		return nil
+	}
+	select {
+	case <-time.After(time.Until(until)):
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (rl *RateLimiter) getChatLimiter(chatID int64) *rate.Limiter {
+	rl.chatMu.Lock()
+	defer rl.chatMu.Unlock()
+	if lim, ok := rl.chatLimiters[chatID]; ok {
+		return lim
+	}
+	lim := rate.NewLimiter(1, 1) // 1 запрос/сек
+	rl.chatLimiters[chatID] = lim
+	return lim
+}

utils/version.go

@@ -1,9 +1,9 @@
 package utils
 
 const (
-	VersionString = "0.8.0-beta.2"
-	VersionMajor  = 0
-	VersionMinor  = 8
+	VersionString = "1.0.0-beta.8"
+	VersionMajor  = 1
+	VersionMinor  = 0
 	VersionPatch  = 0
-	Beta          = 2
+	Beta          = 8
 )