internal/store/postgres/botapi_update.go
Rama main · réplica de lectura del repositorio original.
package postgres
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"strconv"
"time"
"github.com/jackc/pgx/v5"
"telesrv/internal/domain"
"telesrv/internal/store/postgres/sqlcgen"
)
// BotAPIUpdateStore persists Bot API getUpdates queues in PostgreSQL.
type BotAPIUpdateStore struct {
db sqlcgen.DBTX
}
func NewBotAPIUpdateStore(db sqlcgen.DBTX) *BotAPIUpdateStore {
return &BotAPIUpdateStore{db: db}
}
func (s *BotAPIUpdateStore) SetBotAPIWebhook(ctx context.Context, config domain.BotAPIWebhook, dropPending bool) error {
if config.BotUserID <= 0 || config.URL == "" || config.MaxConnections < 1 || config.MaxConnections > 100 {
return fmt.Errorf("invalid bot api webhook")
}
var allowed []string
if len(config.AllowedUpdates) > 0 {
allowed = make([]string, 0, len(config.AllowedUpdates))
for _, kind := range config.AllowedUpdates {
if kind != "" {
allowed = append(allowed, string(kind))
}
}
}
if _, err := s.db.Exec(ctx, `
WITH policy AS (
SELECT CASE WHEN $6::boolean THEN $5::text[]
ELSE (SELECT allowed_updates FROM bot_api_update_states WHERE bot_user_id = $1)
END AS allowed_updates
), configured AS (
INSERT INTO bot_api_webhooks (
bot_user_id, url, secret_token, max_connections, allowed_updates,
failure_count, last_error_date, last_error_message, next_attempt_at,
delivery_owner, delivery_expires_at, updated_at
)
SELECT $1, $2, $3, $4, allowed_updates, 0, 0, '', now(), '', NULL, now()
FROM policy
ON CONFLICT (bot_user_id) DO UPDATE
SET url = EXCLUDED.url,
secret_token = EXCLUDED.secret_token,
max_connections = EXCLUDED.max_connections,
allowed_updates = EXCLUDED.allowed_updates,
failure_count = 0,
last_error_date = 0,
last_error_message = '',
next_attempt_at = now(),
delivery_owner = '',
delivery_expires_at = NULL,
updated_at = now()
RETURNING bot_user_id
), boundary AS (
SELECT CASE WHEN $7::boolean THEN COALESCE(MAX(id), 0) ELSE 0 END AS confirmed_update_id
FROM bot_api_updates
WHERE bot_user_id = $1
)
INSERT INTO bot_api_update_states (
bot_user_id, confirmed_update_id, allowed_updates, cursor_initialized
)
SELECT $1, confirmed_update_id, policy.allowed_updates, $7::boolean
FROM boundary, configured, policy
ON CONFLICT (bot_user_id) DO UPDATE
SET confirmed_update_id = CASE WHEN $7::boolean
THEN GREATEST(bot_api_update_states.confirmed_update_id, EXCLUDED.confirmed_update_id)
ELSE bot_api_update_states.confirmed_update_id
END,
allowed_updates = EXCLUDED.allowed_updates,
cursor_initialized = bot_api_update_states.cursor_initialized OR EXCLUDED.cursor_initialized,
updated_at = now()
`, config.BotUserID, config.URL, config.SecretToken, config.MaxConnections, allowed,
config.AllowedUpdatesSet, dropPending); err != nil {
return fmt.Errorf("set bot api webhook: %w", err)
}
return nil
}
func (s *BotAPIUpdateStore) DeleteBotAPIWebhook(ctx context.Context, botUserID int64, dropPending bool) error {
if botUserID <= 0 {
return nil
}
if _, err := s.db.Exec(ctx, `
WITH deleted AS (
DELETE FROM bot_api_webhooks WHERE bot_user_id = $1 RETURNING bot_user_id
), boundary AS (
SELECT CASE WHEN $2::boolean THEN COALESCE(MAX(id), 0) ELSE 0 END AS confirmed_update_id
FROM bot_api_updates
WHERE bot_user_id = $1
)
INSERT INTO bot_api_update_states (bot_user_id, confirmed_update_id, cursor_initialized)
SELECT $1, confirmed_update_id, $2::boolean
FROM boundary
ON CONFLICT (bot_user_id) DO UPDATE
SET confirmed_update_id = CASE WHEN $2::boolean
THEN GREATEST(bot_api_update_states.confirmed_update_id, EXCLUDED.confirmed_update_id)
ELSE bot_api_update_states.confirmed_update_id
END,
cursor_initialized = bot_api_update_states.cursor_initialized OR EXCLUDED.cursor_initialized,
updated_at = now()
`, botUserID, dropPending); err != nil {
return fmt.Errorf("delete bot api webhook: %w", err)
}
return nil
}
func (s *BotAPIUpdateStore) BotAPIWebhook(ctx context.Context, botUserID int64) (domain.BotAPIWebhook, bool, error) {
config, err := scanBotAPIWebhook(s.db.QueryRow(ctx, `
SELECT bot_user_id, url, secret_token, max_connections, allowed_updates,
failure_count, last_error_date, last_error_message, next_attempt_at
FROM bot_api_webhooks
WHERE bot_user_id = $1
`, botUserID))
if err == pgx.ErrNoRows {
return domain.BotAPIWebhook{}, false, nil
}
if err != nil {
return domain.BotAPIWebhook{}, false, fmt.Errorf("get bot api webhook: %w", err)
}
return config, true, nil
}
func (s *BotAPIUpdateStore) ListDueBotAPIWebhooks(ctx context.Context, limit int) ([]domain.BotAPIWebhook, error) {
if limit <= 0 || limit > 1000 {
limit = 100
}
rows, err := s.db.Query(ctx, `
SELECT bot_user_id, url, secret_token, max_connections, allowed_updates,
failure_count, last_error_date, last_error_message, next_attempt_at
FROM bot_api_webhooks
WHERE next_attempt_at <= now()
AND (delivery_owner = '' OR delivery_expires_at <= now())
ORDER BY next_attempt_at, bot_user_id
LIMIT $1
`, limit)
if err != nil {
return nil, fmt.Errorf("list due bot api webhooks: %w", err)
}
defer rows.Close()
out := make([]domain.BotAPIWebhook, 0, limit)
for rows.Next() {
config, err := scanBotAPIWebhook(rows)
if err != nil {
return nil, err
}
out = append(out, config)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("list due bot api webhook rows: %w", err)
}
return out, nil
}
func (s *BotAPIUpdateStore) AcquireBotAPIWebhookLease(ctx context.Context, botUserID int64, owner string, ttl time.Duration) (bool, error) {
if botUserID <= 0 || owner == "" || ttl <= 0 {
return false, fmt.Errorf("invalid bot api webhook lease")
}
var acquiredOwner string
err := s.db.QueryRow(ctx, `
UPDATE bot_api_webhooks
SET delivery_owner = $2,
delivery_expires_at = now() + make_interval(secs => $3),
updated_at = now()
WHERE bot_user_id = $1
AND (delivery_owner = $2 OR delivery_owner = '' OR delivery_expires_at <= now())
RETURNING delivery_owner
`, botUserID, owner, int64((ttl+time.Second-1)/time.Second)).Scan(&acquiredOwner)
if err == pgx.ErrNoRows {
return false, nil
}
if err != nil {
return false, fmt.Errorf("acquire bot api webhook lease: %w", err)
}
return acquiredOwner == owner, nil
}
func (s *BotAPIUpdateStore) ReleaseBotAPIWebhookLease(ctx context.Context, botUserID int64, owner string) error {
if botUserID <= 0 || owner == "" {
return nil
}
if _, err := s.db.Exec(ctx, `
UPDATE bot_api_webhooks
SET delivery_owner = '', delivery_expires_at = NULL, updated_at = now()
WHERE bot_user_id = $1 AND delivery_owner = $2
`, botUserID, owner); err != nil {
return fmt.Errorf("release bot api webhook lease: %w", err)
}
return nil
}
func (s *BotAPIUpdateStore) RecordBotAPIWebhookFailure(ctx context.Context, botUserID int64, owner string, nextAttempt time.Time, message string) error {
if len(message) > 512 {
message = message[:512]
}
if _, err := s.db.Exec(ctx, `
UPDATE bot_api_webhooks
SET failure_count = failure_count + 1,
last_error_date = EXTRACT(EPOCH FROM now())::integer,
last_error_message = $3,
next_attempt_at = $4,
delivery_owner = '', delivery_expires_at = NULL, updated_at = now()
WHERE bot_user_id = $1 AND delivery_owner = $2
`, botUserID, owner, message, nextAttempt); err != nil {
return fmt.Errorf("record bot api webhook failure: %w", err)
}
return nil
}
func (s *BotAPIUpdateStore) RecordBotAPIWebhookSuccess(ctx context.Context, botUserID int64, owner string, nextAttempt time.Time) error {
if _, err := s.db.Exec(ctx, `
UPDATE bot_api_webhooks
SET failure_count = 0, last_error_date = 0, last_error_message = '',
next_attempt_at = $3, delivery_owner = '', delivery_expires_at = NULL, updated_at = now()
WHERE bot_user_id = $1 AND delivery_owner = $2
`, botUserID, owner, nextAttempt); err != nil {
return fmt.Errorf("record bot api webhook success: %w", err)
}
return nil
}
func scanBotAPIWebhook(row botAPIUpdateScanner) (domain.BotAPIWebhook, error) {
var config domain.BotAPIWebhook
var allowed []string
if err := row.Scan(&config.BotUserID, &config.URL, &config.SecretToken, &config.MaxConnections, &allowed,
&config.FailureCount, &config.LastErrorDate, &config.LastErrorMessage, &config.NextAttemptAt); err != nil {
return domain.BotAPIWebhook{}, err
}
if allowed != nil {
config.AllowedUpdates = make([]domain.BotAPIUpdateKind, 0, len(allowed))
for _, kind := range allowed {
config.AllowedUpdates = append(config.AllowedUpdates, domain.BotAPIUpdateKind(kind))
}
}
return config, nil
}
func (s *BotAPIUpdateStore) AcquireBotAPIPollLease(ctx context.Context, botUserID int64, owner string, ttl time.Duration) (bool, error) {
if botUserID <= 0 || owner == "" || ttl <= 0 {
return false, fmt.Errorf("invalid bot api poll lease")
}
var acquiredOwner string
err := s.db.QueryRow(ctx, `
INSERT INTO bot_api_update_states (
bot_user_id, confirmed_update_id, poll_owner, poll_expires_at
) VALUES ($1, 0, $2, now() + make_interval(secs => $3))
ON CONFLICT (bot_user_id) DO UPDATE
SET poll_owner = EXCLUDED.poll_owner,
poll_expires_at = EXCLUDED.poll_expires_at,
updated_at = now()
WHERE bot_api_update_states.poll_owner = EXCLUDED.poll_owner
OR bot_api_update_states.poll_expires_at IS NULL
OR bot_api_update_states.poll_expires_at <= now()
RETURNING poll_owner
`, botUserID, owner, int64((ttl+time.Second-1)/time.Second)).Scan(&acquiredOwner)
if err == pgx.ErrNoRows {
return false, nil
}
if err != nil {
return false, fmt.Errorf("acquire bot api poll lease: %w", err)
}
return acquiredOwner == owner, nil
}
func (s *BotAPIUpdateStore) ReleaseBotAPIPollLease(ctx context.Context, botUserID int64, owner string) error {
if botUserID <= 0 || owner == "" {
return nil
}
if _, err := s.db.Exec(ctx, `
UPDATE bot_api_update_states
SET poll_owner = '', poll_expires_at = NULL, updated_at = now()
WHERE bot_user_id = $1 AND poll_owner = $2
`, botUserID, owner); err != nil {
return fmt.Errorf("release bot api poll lease: %w", err)
}
return nil
}
func (s *BotAPIUpdateStore) EnqueueBotAPIUpdate(ctx context.Context, req domain.EnqueueBotAPIUpdateRequest) (domain.BotAPIUpdate, bool, error) {
if err := validateBotAPIUpdateRequest(req); err != nil {
return domain.BotAPIUpdate{}, false, err
}
var callbackQueryID, callbackUserID, callbackChatInstance int64
var callbackInlineDCID, callbackInlineMessageID int
var callbackInlineOwnerID, callbackInlineAccessHash int64
var callbackData []byte
var ephemeralPayload []byte
var preCheckoutPayload []byte
var preCheckoutID string
if req.PreCheckout != nil {
var err error
preCheckoutPayload, err = json.Marshal(req.PreCheckout)
if err != nil {
return domain.BotAPIUpdate{}, false, fmt.Errorf("marshal bot api pre-checkout payload: %w", err)
}
preCheckoutID = strconv.FormatInt(req.PreCheckout.ID, 10)
}
if req.Callback != nil {
callbackQueryID = req.Callback.ID
callbackUserID = req.Callback.UserID
callbackChatInstance = req.Callback.ChatInstance
callbackData = req.Callback.Data
if req.Callback.InlineMessage != nil {
callbackInlineDCID = req.Callback.InlineMessage.DCID
callbackInlineOwnerID = req.Callback.InlineMessage.OwnerID
callbackInlineMessageID = req.Callback.InlineMessage.ID
callbackInlineAccessHash = req.Callback.InlineMessage.AccessHash
}
}
if req.Ephemeral != nil {
var err error
ephemeralPayload, err = json.Marshal(req.Ephemeral)
if err != nil {
return domain.BotAPIUpdate{}, false, fmt.Errorf("marshal bot api ephemeral payload: %w", err)
}
}
row, err := s.scanBotAPIUpdate(s.db.QueryRow(ctx, `
WITH inserted AS (
INSERT INTO bot_api_updates (
bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date,
callback_query_id, callback_user_id, callback_chat_instance, callback_data,
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
ephemeral_payload, pre_checkout_payload
) SELECT $1, $2::varchar(32), $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16::jsonb, $17::jsonb
WHERE NOT EXISTS (
SELECT 1
FROM bot_api_update_states
WHERE bot_user_id = $1
AND allowed_updates IS NOT NULL
AND NOT ($2::text = ANY(allowed_updates))
)
ON CONFLICT DO NOTHING
RETURNING id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date,
callback_query_id, callback_user_id, callback_chat_instance, callback_data,
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
ephemeral_payload, pre_checkout_payload
), wake_webhook AS (
UPDATE bot_api_webhooks
SET next_attempt_at = now(), updated_at = now()
WHERE bot_user_id = $1 AND EXISTS (SELECT 1 FROM inserted)
RETURNING bot_user_id
)
SELECT id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date,
callback_query_id, callback_user_id, callback_chat_instance, callback_data,
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
ephemeral_payload, pre_checkout_payload
FROM inserted
`, req.BotUserID, string(req.Kind), string(req.Peer.Type), req.Peer.ID, req.MessageID, req.SourcePts, req.Date,
callbackQueryID, callbackUserID, callbackChatInstance, callbackData,
callbackInlineDCID, callbackInlineOwnerID, callbackInlineMessageID, callbackInlineAccessHash, ephemeralPayload,
preCheckoutPayload))
if err == nil {
return row, true, nil
}
if err != pgx.ErrNoRows {
return domain.BotAPIUpdate{}, false, fmt.Errorf("insert bot api update: %w", err)
}
row, err = s.scanBotAPIUpdate(s.db.QueryRow(ctx, `
SELECT id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date,
callback_query_id, callback_user_id, callback_chat_instance, callback_data,
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
ephemeral_payload, pre_checkout_payload
FROM bot_api_updates
WHERE bot_user_id = $1
AND update_kind = $2
AND (
(update_kind = 'callback_query' AND callback_query_id = $7)
OR
(update_kind = 'pre_checkout_query' AND pre_checkout_payload->>'id' = $9::text)
OR
(update_kind <> 'callback_query' AND update_kind <> 'pre_checkout_query'
AND peer_type = $3 AND peer_id = $4 AND message_id = $5 AND (
(ephemeral_payload IS NULL AND $8::jsonb IS NULL AND source_pts = $6)
OR (ephemeral_payload = $8::jsonb)
))
)
`, req.BotUserID, string(req.Kind), string(req.Peer.Type), req.Peer.ID, req.MessageID, req.SourcePts,
callbackQueryID, ephemeralPayload, preCheckoutID))
if err != nil {
if err == pgx.ErrNoRows {
return domain.BotAPIUpdate{}, false, nil
}
return domain.BotAPIUpdate{}, false, fmt.Errorf("select existing bot api update: %w", err)
}
return row, false, nil
}
func (s *BotAPIUpdateStore) ListTailBotAPIUpdates(ctx context.Context, botUserID int64, tail, limit int) ([]domain.BotAPIUpdate, error) {
if botUserID == 0 || tail <= 0 {
return nil, nil
}
if limit <= 0 || limit > 100 {
limit = 100
}
rows, err := s.db.Query(ctx, `
SELECT id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date,
callback_query_id, callback_user_id, callback_chat_instance, callback_data,
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
ephemeral_payload, pre_checkout_payload
FROM (
SELECT id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date,
callback_query_id, callback_user_id, callback_chat_instance, callback_data,
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
ephemeral_payload, pre_checkout_payload
FROM bot_api_updates
WHERE bot_user_id = $1
AND id > COALESCE((SELECT confirmed_update_id FROM bot_api_update_states WHERE bot_user_id = $1), 0)
ORDER BY id DESC
LIMIT $2
) AS tail_updates
ORDER BY id
LIMIT $3
`, botUserID, tail, limit)
if err != nil {
return nil, fmt.Errorf("list bot api tail updates: %w", err)
}
defer rows.Close()
out := make([]domain.BotAPIUpdate, 0, limit)
for rows.Next() {
item, err := scanBotAPIUpdateRows(rows)
if err != nil {
return nil, err
}
out = append(out, item)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("list bot api tail update rows: %w", err)
}
return out, nil
}
func (s *BotAPIUpdateStore) ListBotAPIUpdates(ctx context.Context, botUserID, fromUpdateID int64, limit int) ([]domain.BotAPIUpdate, error) {
if botUserID == 0 {
return nil, nil
}
if fromUpdateID <= 0 {
fromUpdateID = 1
}
if limit <= 0 || limit > 100 {
limit = 100
}
rows, err := s.db.Query(ctx, `
SELECT id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date,
callback_query_id, callback_user_id, callback_chat_instance, callback_data,
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
ephemeral_payload, pre_checkout_payload
FROM bot_api_updates
WHERE bot_user_id = $1 AND id >= $2
ORDER BY id
LIMIT $3
`, botUserID, fromUpdateID, limit)
if err != nil {
return nil, fmt.Errorf("list bot api updates: %w", err)
}
defer rows.Close()
out := make([]domain.BotAPIUpdate, 0, limit)
for rows.Next() {
item, err := scanBotAPIUpdateRows(rows)
if err != nil {
return nil, err
}
out = append(out, item)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("list bot api updates rows: %w", err)
}
return out, nil
}
func (s *BotAPIUpdateStore) ConfirmBotAPIUpdates(ctx context.Context, botUserID, confirmedUpdateID int64) error {
if botUserID == 0 || confirmedUpdateID <= 0 {
return nil
}
if _, err := s.db.Exec(ctx, `
WITH bounded AS (
SELECT COALESCE(MAX(id), 0) AS max_update_id,
LEAST($2::bigint, COALESCE(MAX(id), 0)) AS confirmed_update_id
FROM bot_api_updates
WHERE bot_user_id = $1
)
INSERT INTO bot_api_update_states (bot_user_id, confirmed_update_id, cursor_initialized)
SELECT $1, confirmed_update_id, true
FROM bounded
ON CONFLICT (bot_user_id) DO UPDATE
SET confirmed_update_id = GREATEST(
bot_api_update_states.confirmed_update_id,
CASE
WHEN $2::bigint > (SELECT max_update_id FROM bounded)
AND bot_api_update_states.cursor_initialized
THEN bot_api_update_states.confirmed_update_id
ELSE EXCLUDED.confirmed_update_id
END
),
cursor_initialized = true,
updated_at = now()
`, botUserID, confirmedUpdateID); err != nil {
return fmt.Errorf("confirm bot api updates: %w", err)
}
return nil
}
func (s *BotAPIUpdateStore) SetBotAPIAllowedUpdates(ctx context.Context, botUserID int64, allowed []domain.BotAPIUpdateKind) error {
if botUserID == 0 {
return nil
}
var values []string
if len(allowed) > 0 {
values = make([]string, 0, len(allowed))
for _, kind := range allowed {
if kind != "" {
values = append(values, string(kind))
}
}
}
if _, err := s.db.Exec(ctx, `
INSERT INTO bot_api_update_states (bot_user_id, confirmed_update_id, allowed_updates)
VALUES ($1, 0, $2::text[])
ON CONFLICT (bot_user_id) DO UPDATE
SET allowed_updates = EXCLUDED.allowed_updates,
updated_at = now()
`, botUserID, values); err != nil {
return fmt.Errorf("set bot api allowed updates: %w", err)
}
return nil
}
func (s *BotAPIUpdateStore) DropPendingBotAPIUpdates(ctx context.Context, botUserID int64) error {
if botUserID == 0 {
return nil
}
if _, err := s.db.Exec(ctx, `
INSERT INTO bot_api_update_states (bot_user_id, confirmed_update_id, cursor_initialized)
SELECT $1, COALESCE(MAX(id), 0), true
FROM bot_api_updates
WHERE bot_user_id = $1
ON CONFLICT (bot_user_id) DO UPDATE
SET confirmed_update_id = GREATEST(bot_api_update_states.confirmed_update_id, EXCLUDED.confirmed_update_id),
cursor_initialized = true,
updated_at = now()
`, botUserID); err != nil {
return fmt.Errorf("drop pending bot api updates: %w", err)
}
return nil
}
func (s *BotAPIUpdateStore) PendingBotAPIUpdateCount(ctx context.Context, botUserID int64) (int, error) {
if botUserID == 0 {
return 0, nil
}
var count int
if err := s.db.QueryRow(ctx, `
SELECT COUNT(*)
FROM bot_api_updates
WHERE bot_user_id = $1
AND id > COALESCE((SELECT confirmed_update_id FROM bot_api_update_states WHERE bot_user_id = $1), 0)
`, botUserID).Scan(&count); err != nil {
return 0, fmt.Errorf("count pending bot api updates: %w", err)
}
return count, nil
}
// DeleteDeliveredOrExpired reclaims dead rows from the Bot API delivery queue (performance audit H1):
// 1. rows already confirmed (id <= bot_api_update_states.confirmed_update_id) and enqueued longer than confirmedGrace—
// under official Bot API semantics confirmation means discard, getUpdates' fromID is always > confirmed, so deletion affects no read path;
// the grace only guards the offset-rewind debugging scenario.
// 2. rows whose queue created_at exceeds maxAge (confirmed or not)—matching the official "the updates server keeps at most 24 hours"
// semantics, while also capping the unbounded growth caused by MTProto-only bots (which never call getUpdates and have no state row) being members.
//
// Unrelated to the "permanent retention" constraint of user_update_events: that is forced by a TDesktop account-level differenceTooLong defect,
// and the Bot API queue has no such constraint. Returns the total rows deleted across both steps.
func (s *BotAPIUpdateStore) DeleteDeliveredOrExpired(ctx context.Context, confirmedGrace, maxAge time.Duration, limit int) (int, error) {
if limit <= 0 {
limit = 10000
}
if limit > 100000 {
limit = 100000
}
total := 0
if confirmedGrace > 0 {
// Start from the small states table; per bot use a range scan of bot_api_updates_bot_scan_idx(bot_user_id, id).
tag, err := s.db.Exec(ctx, `
DELETE FROM bot_api_updates
WHERE id IN (
SELECT u.id
FROM bot_api_update_states s
JOIN bot_api_updates u ON u.bot_user_id = s.bot_user_id AND u.id <= s.confirmed_update_id
WHERE u.created_at < now() - make_interval(secs => $1)
LIMIT $2
)`, int64(confirmedGrace/time.Second), limit)
if err != nil {
return total, fmt.Errorf("delete confirmed bot api updates: %w", err)
}
total += int(tag.RowsAffected())
}
if maxAge > 0 {
cutoff := time.Now().Add(-maxAge)
// Use bot_api_updates_created_retention_idx(created_at, id).
tag, err := s.db.Exec(ctx, `
DELETE FROM bot_api_updates
WHERE id IN (
SELECT id
FROM bot_api_updates
WHERE created_at < $1
ORDER BY created_at, id
LIMIT $2
)`, cutoff, limit)
if err != nil {
return total, fmt.Errorf("delete expired bot api updates: %w", err)
}
total += int(tag.RowsAffected())
}
return total, nil
}
func (s *BotAPIUpdateStore) ConfirmedBotAPIUpdateID(ctx context.Context, botUserID int64) (int64, bool, error) {
if botUserID == 0 {
return 0, false, nil
}
var id int64
if err := s.db.QueryRow(ctx, `
SELECT confirmed_update_id
FROM bot_api_update_states
WHERE bot_user_id = $1
`, botUserID).Scan(&id); err != nil {
if err == pgx.ErrNoRows {
return 0, false, nil
}
return 0, false, fmt.Errorf("get bot api update state: %w", err)
}
return id, true, nil
}
func (s *BotAPIUpdateStore) scanBotAPIUpdate(row pgx.Row) (domain.BotAPIUpdate, error) {
return scanBotAPIUpdateRows(row)
}
type botAPIUpdateScanner interface {
Scan(dest ...any) error
}
func scanBotAPIUpdateRows(row botAPIUpdateScanner) (domain.BotAPIUpdate, error) {
var item domain.BotAPIUpdate
var kind, peerType string
var callbackQueryID, callbackUserID, callbackChatInstance int64
var callbackInlineDCID, callbackInlineMessageID int
var callbackInlineOwnerID, callbackInlineAccessHash int64
var callbackData []byte
var ephemeralPayload, preCheckoutPayload []byte
if err := row.Scan(&item.ID, &item.BotUserID, &kind, &peerType, &item.Peer.ID, &item.MessageID, &item.SourcePts, &item.Date,
&callbackQueryID, &callbackUserID, &callbackChatInstance, &callbackData,
&callbackInlineDCID, &callbackInlineOwnerID, &callbackInlineMessageID, &callbackInlineAccessHash,
&ephemeralPayload, &preCheckoutPayload); err != nil {
return domain.BotAPIUpdate{}, err
}
item.Kind = domain.BotAPIUpdateKind(kind)
item.Peer.Type = domain.PeerType(peerType)
if item.Kind == domain.BotAPIUpdateCallbackQuery {
item.Callback = &domain.BotCallbackQuery{
ID: callbackQueryID,
BotUserID: item.BotUserID,
UserID: callbackUserID,
Peer: item.Peer,
MessageID: item.MessageID,
ChatInstance: callbackChatInstance,
Data: append([]byte(nil), callbackData...),
}
if callbackInlineMessageID > 0 {
item.Callback.InlineMessage = &domain.BotInlineMessageID{DCID: callbackInlineDCID, OwnerID: callbackInlineOwnerID, ID: callbackInlineMessageID, AccessHash: callbackInlineAccessHash}
}
}
if len(preCheckoutPayload) != 0 && item.Kind == domain.BotAPIUpdatePreCheckoutQuery {
var query domain.BotPreCheckoutQuery
if err := json.Unmarshal(preCheckoutPayload, &query); err == nil {
item.PreCheckout = &query
}
}
if len(ephemeralPayload) != 0 {
var payload domain.BotAPIEphemeralPayload
decoder := json.NewDecoder(bytes.NewReader(ephemeralPayload))
decoder.DisallowUnknownFields()
if err := decoder.Decode(&payload); err != nil {
return domain.BotAPIUpdate{}, fmt.Errorf("decode bot api ephemeral payload: %w", err)
}
if err := decoder.Decode(&struct{}{}); err != io.EOF {
return domain.BotAPIUpdate{}, fmt.Errorf("decode bot api ephemeral payload: trailing JSON")
}
if err := validateBotAPIEphemeralPayload(item.BotUserID, item.Kind, item.Peer, item.MessageID, item.SourcePts, item.Date, &payload); err != nil {
return domain.BotAPIUpdate{}, err
}
item.Ephemeral = &payload
}
return item, nil
}
func validateBotAPIUpdateRequest(req domain.EnqueueBotAPIUpdateRequest) error {
if req.BotUserID == 0 {
return fmt.Errorf("invalid bot api update")
}
// A pre-checkout query is message-less by design: the bot is asked before any
// payment exists, so it carries no peer and no message id.
if req.Kind != domain.BotAPIUpdateMessage && req.Kind != domain.BotAPIUpdateEditedMessage &&
req.Kind != domain.BotAPIUpdateCallbackQuery && req.Kind != domain.BotAPIUpdatePreCheckoutQuery {
return fmt.Errorf("invalid bot api update kind %q", req.Kind)
}
switch req.Peer.Type {
case domain.PeerTypeUser, domain.PeerTypeChannel:
if req.Peer.ID <= 0 || req.MessageID <= 0 {
return fmt.Errorf("invalid bot api update peer")
}
case "":
if req.Peer.ID != 0 || req.MessageID != 0 {
return fmt.Errorf("invalid bot api update peer")
}
default:
return fmt.Errorf("invalid bot api update peer type %q", req.Peer.Type)
}
if err := validateBotAPIEphemeralPayload(req.BotUserID, req.Kind, req.Peer, req.MessageID, req.SourcePts, req.Date, req.Ephemeral); err != nil {
return err
}
if req.Kind == domain.BotAPIUpdateCallbackQuery {
cb := req.Callback
if cb == nil || cb.ID == 0 || cb.BotUserID != req.BotUserID || cb.UserID <= 0 ||
cb.Peer != req.Peer || cb.MessageID != req.MessageID || cb.ChatInstance == 0 ||
len(cb.Data) > domain.MaxCallbackDataLen || req.SourcePts != 0 {
return fmt.Errorf("invalid bot api callback query")
}
inline := cb.InlineMessage
if req.MessageID == 0 && (inline == nil || inline.DCID <= 0 || inline.OwnerID <= 0 || inline.ID <= 0 || inline.AccessHash == 0) {
return fmt.Errorf("invalid bot api inline callback query")
}
if req.MessageID > 0 && inline != nil {
return fmt.Errorf("ambiguous bot api callback query")
}
} else if req.Callback != nil {
return fmt.Errorf("unexpected bot api callback query")
}
return nil
}
func validateBotAPIEphemeralPayload(botUserID int64, kind domain.BotAPIUpdateKind, peer domain.Peer, messageID, sourcePts, date int, payload *domain.BotAPIEphemeralPayload) error {
if payload == nil {
return nil
}
message := payload.Message
if payload.Validate() != nil || peer.Type != domain.PeerTypeChannel || message.ID != messageID || message.Peer != peer ||
message.Expired(time.Unix(int64(date), 0)) || sourcePts != 0 {
return fmt.Errorf("invalid bot api ephemeral update")
}
if kind == domain.BotAPIUpdateCallbackQuery {
if message.SenderUserID != botUserID {
return fmt.Errorf("invalid bot api ephemeral callback target")
}
return nil
}
if kind != domain.BotAPIUpdateMessage && kind != domain.BotAPIUpdateEditedMessage {
return fmt.Errorf("invalid bot api ephemeral update kind")
}
if message.ReceiverUserID != botUserID {
return fmt.Errorf("invalid bot api ephemeral receiver")
}
return nil
}