astra/owpengram-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: sync bot keyboards and callbacks

Browse source 
Sync telesrv b96f2dd (feat(bot): complete keyboards callbacks and durable delivery).

Skipped private docs and preserved public README files per sync rules; normalized the appearance seed log label for public naming.
This commit is contained in:
A 2026-07-19 20:38:48 +08:00
parent 0c99ae0a9d
commit bf965f610c
80 changed files with 7212 additions and 349 deletions
Download patch file Download diff file Expand all files Collapse all files

36
internal/store/bot_callback.go Normal file
View file

@ -0,0 +1,36 @@
package store
import (
"context"
"time"
"telesrv/internal/domain"
)
// BotCallbackPending is the short-lived, protocol-neutral ownership record for
// one messages.getBotCallbackAnswer request. It is deliberately ephemeral: the
// durable Bot API update remains in BotAPIUpdateStore, while this record only
// coordinates the synchronous client answer across server instances.
type BotCallbackPending struct {
QueryID int64
BotUserID int64
UserID int64
CreatedAt time.Time
}
type BotCallbackAnswerPush struct {
QueryID int64
BotUserID int64
Answer domain.BotCallbackAnswer
}
// BotCallbackRegistryStore coordinates callback waiters across processes.
// Implementations must make Put and Resolve atomic: query ids cannot be
// overwritten and at most one answer may win for the owning bot.
type BotCallbackRegistryStore interface {
PutBotCallbackPending(ctx context.Context, pending BotCallbackPending, ttl time.Duration) (bool, error)
ResolveBotCallback(ctx context.Context, botUserID, queryID int64, answer domain.BotCallbackAnswer) (bool, error)
GetBotCallbackAnswer(ctx context.Context, botUserID, queryID int64) (domain.BotCallbackAnswer, bool, error)
DeleteBotCallbackPending(ctx context.Context, botUserID, queryID int64) error
SubscribeBotCallbackAnswers(ctx context.Context, handle func(context.Context, BotCallbackAnswerPush)) error
}

26
internal/store/botapi_update.go
View file

@ -2,14 +2,40 @@ package store
import (
"context"
"time"
"telesrv/internal/domain"
)
// BotAPIPollLeaseStore serializes getUpdates across all HTTP gateway
// instances. owner is an opaque per-request token; Release must be compare-and-
// delete so a stale request cannot release a successor's lease.
type BotAPIPollLeaseStore interface {
AcquireBotAPIPollLease(ctx context.Context, botUserID int64, owner string, ttl time.Duration) (bool, error)
ReleaseBotAPIPollLease(ctx context.Context, botUserID int64, owner string) error
}
// BotAPIWebhookStore coordinates durable webhook configuration and delivery
// leases. Delivery itself stays at the HTTP edge; this store only owns state.
type BotAPIWebhookStore interface {
SetBotAPIWebhook(ctx context.Context, config domain.BotAPIWebhook, dropPending bool) error
DeleteBotAPIWebhook(ctx context.Context, botUserID int64, dropPending bool) error
BotAPIWebhook(ctx context.Context, botUserID int64) (domain.BotAPIWebhook, bool, error)
ListDueBotAPIWebhooks(ctx context.Context, limit int) ([]domain.BotAPIWebhook, error)
AcquireBotAPIWebhookLease(ctx context.Context, botUserID int64, owner string, ttl time.Duration) (bool, error)
ReleaseBotAPIWebhookLease(ctx context.Context, botUserID int64, owner string) error
RecordBotAPIWebhookFailure(ctx context.Context, botUserID int64, owner string, nextAttempt time.Time, message string) error
RecordBotAPIWebhookSuccess(ctx context.Context, botUserID int64, owner string, nextAttempt time.Time) error
}
// BotAPIUpdateStore persists update_id based Bot API delivery queues.
type BotAPIUpdateStore interface {
EnqueueBotAPIUpdate(ctx context.Context, req domain.EnqueueBotAPIUpdateRequest) (domain.BotAPIUpdate, bool, error)
ListBotAPIUpdates(ctx context.Context, botUserID, fromUpdateID int64, limit int) ([]domain.BotAPIUpdate, error)
ListTailBotAPIUpdates(ctx context.Context, botUserID int64, tail, limit int) ([]domain.BotAPIUpdate, error)
ConfirmBotAPIUpdates(ctx context.Context, botUserID, confirmedUpdateID int64) error
ConfirmedBotAPIUpdateID(ctx context.Context, botUserID int64) (int64, bool, error)
SetBotAPIAllowedUpdates(ctx context.Context, botUserID int64, allowed []domain.BotAPIUpdateKind) error
DropPendingBotAPIUpdates(ctx context.Context, botUserID int64) error
PendingBotAPIUpdateCount(ctx context.Context, botUserID int64) (int, error)
}

372
internal/store/memory/botapi_update.go
View file

@ -3,29 +3,228 @@ package memory
import (
"context"
"fmt"
"slices"
"sync"
"time"
"telesrv/internal/domain"
)
// BotAPIUpdateStore is an in-memory implementation of store.BotAPIUpdateStore.
type BotAPIUpdateStore struct {
mu sync.RWMutex
nextID int64
rows []domain.BotAPIUpdate
state map[int64]int64
byKey map[string]int64
mu sync.RWMutex
nextID int64
rows []domain.BotAPIUpdate
state map[int64]int64
cursorInitialized map[int64]bool
allowed map[int64]map[domain.BotAPIUpdateKind]struct{}
byKey map[string]int64
pollLeases map[int64]botAPIPollLease
webhooks map[int64]domain.BotAPIWebhook
webhookLeases map[int64]botAPIPollLease
}
type botAPIPollLease struct {
owner string
expiresAt time.Time
}
// NewBotAPIUpdateStore creates an in-memory Bot API update queue.
func NewBotAPIUpdateStore() *BotAPIUpdateStore {
return &BotAPIUpdateStore{
nextID: 1,
state: make(map[int64]int64),
byKey: make(map[string]int64),
nextID: 1,
state: make(map[int64]int64),
cursorInitialized: make(map[int64]bool),
allowed: make(map[int64]map[domain.BotAPIUpdateKind]struct{}),
byKey: make(map[string]int64),
pollLeases: make(map[int64]botAPIPollLease),
webhooks: make(map[int64]domain.BotAPIWebhook),
webhookLeases: make(map[int64]botAPIPollLease),
}
}
func (s *BotAPIUpdateStore) SetBotAPIWebhook(_ 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")
}
s.mu.Lock()
defer s.mu.Unlock()
if !config.AllowedUpdatesSet {
config.AllowedUpdates = allowedUpdateKinds(s.allowed[config.BotUserID])
}
config.AllowedUpdates = append([]domain.BotAPIUpdateKind(nil), config.AllowedUpdates...)
config.FailureCount, config.LastErrorDate, config.LastErrorMessage = 0, 0, ""
config.NextAttemptAt = time.Now()
s.webhooks[config.BotUserID] = config
if config.AllowedUpdates == nil {
delete(s.allowed, config.BotUserID)
} else {
allowed := make(map[domain.BotAPIUpdateKind]struct{}, len(config.AllowedUpdates))
for _, kind := range config.AllowedUpdates {
allowed[kind] = struct{}{}
}
s.allowed[config.BotUserID] = allowed
}
if dropPending {
s.dropPendingLocked(config.BotUserID)
}
return nil
}
func allowedUpdateKinds(items map[domain.BotAPIUpdateKind]struct{}) []domain.BotAPIUpdateKind {
if len(items) == 0 {
return nil
}
out := make([]domain.BotAPIUpdateKind, 0, len(items))
for kind := range items {
out = append(out, kind)
}
slices.Sort(out)
return out
}
func (s *BotAPIUpdateStore) DeleteBotAPIWebhook(_ context.Context, botUserID int64, dropPending bool) error {
s.mu.Lock()
delete(s.webhooks, botUserID)
delete(s.webhookLeases, botUserID)
if dropPending {
s.dropPendingLocked(botUserID)
}
s.mu.Unlock()
return nil
}
func (s *BotAPIUpdateStore) BotAPIWebhook(_ context.Context, botUserID int64) (domain.BotAPIWebhook, bool, error) {
s.mu.RLock()
config, found := s.webhooks[botUserID]
s.mu.RUnlock()
config.AllowedUpdates = append([]domain.BotAPIUpdateKind(nil), config.AllowedUpdates...)
return config, found, nil
}
func (s *BotAPIUpdateStore) ListDueBotAPIWebhooks(_ context.Context, limit int) ([]domain.BotAPIWebhook, error) {
if limit <= 0 || limit > 1000 {
limit = 100
}
now := time.Now()
s.mu.RLock()
out := make([]domain.BotAPIWebhook, 0, min(limit, len(s.webhooks)))
for botID, config := range s.webhooks {
lease := s.webhookLeases[botID]
if config.NextAttemptAt.After(now) || (lease.owner != "" && lease.expiresAt.After(now)) {
continue
}
config.AllowedUpdates = append([]domain.BotAPIUpdateKind(nil), config.AllowedUpdates...)
out = append(out, config)
if len(out) == limit {
break
}
}
s.mu.RUnlock()
return out, nil
}
func (s *BotAPIUpdateStore) AcquireBotAPIWebhookLease(_ 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")
}
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
if _, found := s.webhooks[botUserID]; !found {
return false, nil
}
current := s.webhookLeases[botUserID]
if current.owner != "" && current.owner != owner && current.expiresAt.After(now) {
return false, nil
}
s.webhookLeases[botUserID] = botAPIPollLease{owner: owner, expiresAt: now.Add(ttl)}
return true, nil
}
func (s *BotAPIUpdateStore) ReleaseBotAPIWebhookLease(_ context.Context, botUserID int64, owner string) error {
s.mu.Lock()
if current := s.webhookLeases[botUserID]; current.owner == owner {
delete(s.webhookLeases, botUserID)
}
s.mu.Unlock()
return nil
}
func (s *BotAPIUpdateStore) RecordBotAPIWebhookFailure(_ context.Context, botUserID int64, owner string, nextAttempt time.Time, message string) error {
s.mu.Lock()
defer s.mu.Unlock()
if current := s.webhookLeases[botUserID]; current.owner != owner {
return nil
}
config, found := s.webhooks[botUserID]
if !found {
delete(s.webhookLeases, botUserID)
return nil
}
config.FailureCount++
config.LastErrorDate = int(time.Now().Unix())
if len(message) > 512 {
message = message[:512]
}
config.LastErrorMessage = message
config.NextAttemptAt = nextAttempt
s.webhooks[botUserID] = config
delete(s.webhookLeases, botUserID)
return nil
}
func (s *BotAPIUpdateStore) RecordBotAPIWebhookSuccess(_ context.Context, botUserID int64, owner string, nextAttempt time.Time) error {
s.mu.Lock()
defer s.mu.Unlock()
if current := s.webhookLeases[botUserID]; current.owner != owner {
return nil
}
if config, found := s.webhooks[botUserID]; found {
config.FailureCount, config.LastErrorDate, config.LastErrorMessage = 0, 0, ""
config.NextAttemptAt = nextAttempt
s.webhooks[botUserID] = config
}
delete(s.webhookLeases, botUserID)
return nil
}
func (s *BotAPIUpdateStore) dropPendingLocked(botUserID int64) {
for _, row := range s.rows {
if row.BotUserID == botUserID && row.ID > s.state[botUserID] {
s.state[botUserID] = row.ID
}
}
s.cursorInitialized[botUserID] = true
}
func (s *BotAPIUpdateStore) AcquireBotAPIPollLease(_ 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")
}
now := time.Now()
s.mu.Lock()
defer s.mu.Unlock()
current, found := s.pollLeases[botUserID]
if found && current.owner != owner && current.expiresAt.After(now) {
return false, nil
}
s.pollLeases[botUserID] = botAPIPollLease{owner: owner, expiresAt: now.Add(ttl)}
return true, nil
}
func (s *BotAPIUpdateStore) ReleaseBotAPIPollLease(_ context.Context, botUserID int64, owner string) error {
if botUserID <= 0 || owner == "" {
return nil
}
s.mu.Lock()
if current, found := s.pollLeases[botUserID]; found && current.owner == owner {
delete(s.pollLeases, botUserID)
}
s.mu.Unlock()
return nil
}
func (s *BotAPIUpdateStore) EnqueueBotAPIUpdate(_ context.Context, req domain.EnqueueBotAPIUpdateRequest) (domain.BotAPIUpdate, bool, error) {
if err := validateBotAPIUpdateRequest(req); err != nil {
return domain.BotAPIUpdate{}, false, err
@ -33,6 +232,11 @@ func (s *BotAPIUpdateStore) EnqueueBotAPIUpdate(_ context.Context, req domain.En
key := botAPIUpdateKey(req)
s.mu.Lock()
defer s.mu.Unlock()
if allowed, configured := s.allowed[req.BotUserID]; configured {
if _, ok := allowed[req.Kind]; !ok {
return domain.BotAPIUpdate{}, false, nil
}
}
if existingID, ok := s.byKey[key]; ok {
for _, row := range s.rows {
if row.ID == existingID {
@ -48,13 +252,56 @@ func (s *BotAPIUpdateStore) EnqueueBotAPIUpdate(_ context.Context, req domain.En
MessageID: req.MessageID,
SourcePts: req.SourcePts,
Date: req.Date,
Callback: cloneBotAPICallback(req.Callback),
}
s.nextID++
s.rows = append(s.rows, row)
s.byKey[key] = row.ID
if config, found := s.webhooks[req.BotUserID]; found {
config.NextAttemptAt = time.Now()
s.webhooks[req.BotUserID] = config
}
return cloneBotAPIUpdate(row), true, nil
}
func (s *BotAPIUpdateStore) ListTailBotAPIUpdates(_ 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
}
s.mu.RLock()
defer s.mu.RUnlock()
confirmed := s.state[botUserID]
matching := make([]domain.BotAPIUpdate, 0, min(tail, limit))
start := 0
count := 0
for _, row := range s.rows {
if row.BotUserID == botUserID && row.ID > confirmed {
count++
}
}
if count > tail {
start = count - tail
}
seen := 0
for _, row := range s.rows {
if row.BotUserID != botUserID || row.ID <= confirmed {
continue
}
if seen < start {
seen++
continue
}
matching = append(matching, cloneBotAPIUpdate(row))
if len(matching) >= limit {
break
}
}
return matching, nil
}
func (s *BotAPIUpdateStore) ListBotAPIUpdates(_ context.Context, botUserID, fromUpdateID int64, limit int) ([]domain.BotAPIUpdate, error) {
if botUserID == 0 {
return nil, nil
@ -85,13 +332,78 @@ func (s *BotAPIUpdateStore) ConfirmBotAPIUpdates(_ context.Context, botUserID, c
return nil
}
s.mu.Lock()
maxExisting := int64(0)
for _, row := range s.rows {
if row.BotUserID == botUserID && row.ID > maxExisting {
maxExisting = row.ID
}
}
if confirmedUpdateID > maxExisting && s.cursorInitialized[botUserID] {
s.mu.Unlock()
return nil
}
if confirmedUpdateID > maxExisting {
confirmedUpdateID = maxExisting
}
if confirmedUpdateID > s.state[botUserID] {
s.state[botUserID] = confirmedUpdateID
}
s.cursorInitialized[botUserID] = true
s.mu.Unlock()
return nil
}
func (s *BotAPIUpdateStore) SetBotAPIAllowedUpdates(_ context.Context, botUserID int64, allowed []domain.BotAPIUpdateKind) error {
if botUserID == 0 {
return nil
}
s.mu.Lock()
if len(allowed) == 0 {
delete(s.allowed, botUserID)
} else {
set := make(map[domain.BotAPIUpdateKind]struct{}, len(allowed))
for _, kind := range allowed {
if kind != "" {
set[kind] = struct{}{}
}
}
s.allowed[botUserID] = set
}
s.mu.Unlock()
return nil
}
func (s *BotAPIUpdateStore) DropPendingBotAPIUpdates(ctx context.Context, botUserID int64) error {
if botUserID == 0 {
return nil
}
s.mu.Lock()
for _, row := range s.rows {
if row.BotUserID == botUserID && row.ID > s.state[botUserID] {
s.state[botUserID] = row.ID
}
}
s.cursorInitialized[botUserID] = true
s.mu.Unlock()
return nil
}
func (s *BotAPIUpdateStore) PendingBotAPIUpdateCount(_ context.Context, botUserID int64) (int, error) {
if botUserID == 0 {
return 0, nil
}
s.mu.RLock()
defer s.mu.RUnlock()
confirmed := s.state[botUserID]
count := 0
for _, row := range s.rows {
if row.BotUserID == botUserID && row.ID > confirmed {
count++
}
}
return count, nil
}
func (s *BotAPIUpdateStore) ConfirmedBotAPIUpdateID(_ context.Context, botUserID int64) (int64, bool, error) {
if botUserID == 0 {
return 0, false, nil
@ -103,27 +415,65 @@ func (s *BotAPIUpdateStore) ConfirmedBotAPIUpdateID(_ context.Context, botUserID
}
func validateBotAPIUpdateRequest(req domain.EnqueueBotAPIUpdateRequest) error {
if req.BotUserID == 0 || req.MessageID <= 0 {
if req.BotUserID == 0 {
return fmt.Errorf("invalid bot api update")
}
if req.Kind != domain.BotAPIUpdateMessage && req.Kind != domain.BotAPIUpdateEditedMessage {
if req.Kind != domain.BotAPIUpdateMessage && req.Kind != domain.BotAPIUpdateEditedMessage && req.Kind != domain.BotAPIUpdateCallbackQuery {
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 {
if req.Peer.ID <= 0 || req.MessageID <= 0 {
return fmt.Errorf("invalid bot api update peer")
}
case "":
if req.Kind != domain.BotAPIUpdateCallbackQuery || 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 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 botAPIUpdateKey(req domain.EnqueueBotAPIUpdateRequest) string {
if req.Kind == domain.BotAPIUpdateCallbackQuery && req.Callback != nil {
return fmt.Sprintf("%d:%s:%d", req.BotUserID, req.Kind, req.Callback.ID)
}
return fmt.Sprintf("%d:%s:%s:%d:%d:%d", req.BotUserID, req.Kind, req.Peer.Type, req.Peer.ID, req.MessageID, req.SourcePts)
}
func cloneBotAPIUpdate(row domain.BotAPIUpdate) domain.BotAPIUpdate {
row.Callback = cloneBotAPICallback(row.Callback)
return row
}
func cloneBotAPICallback(in *domain.BotCallbackQuery) *domain.BotCallbackQuery {
if in == nil {
return nil
}
out := *in
out.Data = append([]byte(nil), in.Data...)
if in.InlineMessage != nil {
inline := *in.InlineMessage
out.InlineMessage = &inline
}
return &out
}

191
internal/store/memory/botapi_update_test.go Normal file
View file

@ -0,0 +1,191 @@
package memory
import (
"context"
"testing"
"time"
"telesrv/internal/domain"
)
func botAPIMessageRequest(botID int64, kind domain.BotAPIUpdateKind, messageID int) domain.EnqueueBotAPIUpdateRequest {
return domain.EnqueueBotAPIUpdateRequest{
BotUserID: botID,
Kind: kind,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: 2001},
MessageID: messageID,
SourcePts: messageID,
Date: 1700000000 + messageID,
}
}
func TestBotAPIPollLeaseCompareOwnerAndExpiry(t *testing.T) {
ctx := context.Background()
store := NewBotAPIUpdateStore()
if acquired, err := store.AcquireBotAPIPollLease(ctx, 1001, "one", 20*time.Millisecond); err != nil || !acquired {
t.Fatalf("first acquire=%v err=%v", acquired, err)
}
if acquired, err := store.AcquireBotAPIPollLease(ctx, 1001, "two", time.Second); err != nil || acquired {
t.Fatalf("competing acquire=%v err=%v", acquired, err)
}
if err := store.ReleaseBotAPIPollLease(ctx, 1001, "stale"); err != nil {
t.Fatal(err)
}
if acquired, _ := store.AcquireBotAPIPollLease(ctx, 1001, "two", time.Second); acquired {
t.Fatal("stale release removed active owner")
}
time.Sleep(25 * time.Millisecond)
if acquired, err := store.AcquireBotAPIPollLease(ctx, 1001, "two", time.Second); err != nil || !acquired {
t.Fatalf("expired acquire=%v err=%v", acquired, err)
}
}
func TestBotAPIWebhookLeaseWakeAndAtomicDrop(t *testing.T) {
ctx := context.Background()
store := NewBotAPIUpdateStore()
if _, created, err := store.EnqueueBotAPIUpdate(ctx, botAPIMessageRequest(1001, domain.BotAPIUpdateMessage, 1)); err != nil || !created {
t.Fatalf("enqueue initial created=%v err=%v", created, err)
}
config := domain.BotAPIWebhook{BotUserID: 1001, URL: "https://example.test/hook", MaxConnections: 8}
if err := store.SetBotAPIWebhook(ctx, config, true); err != nil {
t.Fatal(err)
}
if count, _ := store.PendingBotAPIUpdateCount(ctx, 1001); count != 0 {
t.Fatalf("pending after atomic drop=%d", count)
}
if acquired, err := store.AcquireBotAPIWebhookLease(ctx, 1001, "worker-1", time.Second); err != nil || !acquired {
t.Fatalf("lease acquire=%v err=%v", acquired, err)
}
if acquired, _ := store.AcquireBotAPIWebhookLease(ctx, 1001, "worker-2", time.Second); acquired {
t.Fatal("second webhook worker acquired active lease")
}
if err := store.RecordBotAPIWebhookSuccess(ctx, 1001, "worker-1", time.Now().Add(time.Hour)); err != nil {
t.Fatal(err)
}
if due, err := store.ListDueBotAPIWebhooks(ctx, 10); err != nil || len(due) != 0 {
t.Fatalf("idle due=%#v err=%v", due, err)
}
if _, created, err := store.EnqueueBotAPIUpdate(ctx, botAPIMessageRequest(1001, domain.BotAPIUpdateMessage, 2)); err != nil || !created {
t.Fatalf("enqueue wake created=%v err=%v", created, err)
}
if due, err := store.ListDueBotAPIWebhooks(ctx, 10); err != nil || len(due) != 1 || due[0].BotUserID != 1001 {
t.Fatalf("woken due=%#v err=%v", due, err)
}
}
func TestBotAPIWebhookAllowedUpdatesOmissionPreservesPolicy(t *testing.T) {
ctx := context.Background()
store := NewBotAPIUpdateStore()
if err := store.SetBotAPIAllowedUpdates(ctx, 1001, []domain.BotAPIUpdateKind{domain.BotAPIUpdateCallbackQuery}); err != nil {
t.Fatal(err)
}
config := domain.BotAPIWebhook{BotUserID: 1001, URL: "https://example.test/one", MaxConnections: 8}
if err := store.SetBotAPIWebhook(ctx, config, false); err != nil {
t.Fatal(err)
}
stored, found, err := store.BotAPIWebhook(ctx, 1001)
if err != nil || !found || len(stored.AllowedUpdates) != 1 || stored.AllowedUpdates[0] != domain.BotAPIUpdateCallbackQuery {
t.Fatalf("preserved webhook=%#v found=%v err=%v", stored, found, err)
}
if row, created, err := store.EnqueueBotAPIUpdate(ctx, botAPIMessageRequest(1001, domain.BotAPIUpdateMessage, 1)); err != nil || created || row.ID != 0 {
t.Fatalf("message bypassed preserved policy: row=%#v created=%v err=%v", row, created, err)
}
config.URL = "https://example.test/two"
config.AllowedUpdatesSet = true // Explicit empty resets to the default/all policy.
if err := store.SetBotAPIWebhook(ctx, config, false); err != nil {
t.Fatal(err)
}
stored, found, err = store.BotAPIWebhook(ctx, 1001)
if err != nil || !found || stored.AllowedUpdates != nil {
t.Fatalf("explicit empty webhook=%#v found=%v err=%v", stored, found, err)
}
if _, created, err := store.EnqueueBotAPIUpdate(ctx, botAPIMessageRequest(1001, domain.BotAPIUpdateMessage, 2)); err != nil || !created {
t.Fatalf("message after explicit reset created=%v err=%v", created, err)
}
}
func TestBotAPIUpdateCursorClampDropAndTail(t *testing.T) {
ctx := context.Background()
store := NewBotAPIUpdateStore()
for id := 1; id <= 5; id++ {
if _, created, err := store.EnqueueBotAPIUpdate(ctx, botAPIMessageRequest(1001, domain.BotAPIUpdateMessage, id)); err != nil || !created {
t.Fatalf("enqueue %d: created=%v err=%v", id, created, err)
}
}
tail, err := store.ListTailBotAPIUpdates(ctx, 1001, 2, 100)
if err != nil || len(tail) != 2 || tail[0].MessageID != 4 || tail[1].MessageID != 5 {
t.Fatalf("tail = %#v err=%v", tail, err)
}
if err := store.ConfirmBotAPIUpdates(ctx, 1001, 1<<60); err != nil {
t.Fatalf("confirm huge offset: %v", err)
}
confirmed, found, err := store.ConfirmedBotAPIUpdateID(ctx, 1001)
if err != nil || !found || confirmed != 5 {
t.Fatalf("confirmed = %d found=%v err=%v, want 5", confirmed, found, err)
}
row, created, err := store.EnqueueBotAPIUpdate(ctx, botAPIMessageRequest(1001, domain.BotAPIUpdateMessage, 6))
if err != nil || !created {
t.Fatalf("enqueue after huge offset: row=%#v created=%v err=%v", row, created, err)
}
if err := store.ConfirmBotAPIUpdates(ctx, 1001, 1<<60); err != nil {
t.Fatalf("repeat foreign offset: %v", err)
}
if confirmed, _, _ := store.ConfirmedBotAPIUpdateID(ctx, 1001); confirmed != 5 {
t.Fatalf("repeat foreign offset advanced cursor to %d, want 5", confirmed)
}
pending, err := store.ListBotAPIUpdates(ctx, 1001, confirmed+1, 100)
if err != nil || len(pending) != 1 || pending[0].MessageID != 6 {
t.Fatalf("pending after huge offset = %#v err=%v", pending, err)
}
if err := store.DropPendingBotAPIUpdates(ctx, 1001); err != nil {
t.Fatalf("drop pending: %v", err)
}
count, err := store.PendingBotAPIUpdateCount(ctx, 1001)
if err != nil || count != 0 {
t.Fatalf("pending count = %d err=%v", count, err)
}
}
func TestBotAPIAllowedUpdatesOnlyAffectsFutureEnqueue(t *testing.T) {
ctx := context.Background()
store := NewBotAPIUpdateStore()
first, created, err := store.EnqueueBotAPIUpdate(ctx, botAPIMessageRequest(1001, domain.BotAPIUpdateMessage, 1))
if err != nil || !created {
t.Fatalf("enqueue pre-policy: %#v created=%v err=%v", first, created, err)
}
if err := store.SetBotAPIAllowedUpdates(ctx, 1001, []domain.BotAPIUpdateKind{domain.BotAPIUpdateEditedMessage}); err != nil {
t.Fatalf("set policy: %v", err)
}
if row, created, err := store.EnqueueBotAPIUpdate(ctx, botAPIMessageRequest(1001, domain.BotAPIUpdateMessage, 2)); err != nil || created || row.ID != 0 {
t.Fatalf("filtered message = %#v created=%v err=%v", row, created, err)
}
if _, created, err := store.EnqueueBotAPIUpdate(ctx, botAPIMessageRequest(1001, domain.BotAPIUpdateEditedMessage, 3)); err != nil || !created {
t.Fatalf("allowed edit created=%v err=%v", created, err)
}
rows, err := store.ListBotAPIUpdates(ctx, 1001, 1, 100)
if err != nil || len(rows) != 2 || rows[0].ID != first.ID || rows[1].Kind != domain.BotAPIUpdateEditedMessage {
t.Fatalf("rows = %#v err=%v", rows, err)
}
}
func TestBotAPIInlineCallbackRoundTrip(t *testing.T) {
ctx := context.Background()
store := NewBotAPIUpdateStore()
callback := &domain.BotCallbackQuery{
ID: 77, BotUserID: 1001, UserID: 2001, ChatInstance: 99, Data: []byte("tap"),
InlineMessage: &domain.BotInlineMessageID{DCID: 2, OwnerID: 2001, ID: 15, AccessHash: 1234},
}
row, created, err := store.EnqueueBotAPIUpdate(ctx, domain.EnqueueBotAPIUpdateRequest{
BotUserID: 1001, Kind: domain.BotAPIUpdateCallbackQuery, Date: int(time.Now().Unix()), Callback: callback,
})
if err != nil || !created || row.MessageID != 0 || row.Peer != (domain.Peer{}) || row.Callback == nil ||
row.Callback.InlineMessage == nil || *row.Callback.InlineMessage != *callback.InlineMessage {
t.Fatalf("inline callback row=%#v created=%v err=%v", row, created, err)
}
callback.Data[0] = 'X'
callback.InlineMessage.ID = 99
rows, err := store.ListBotAPIUpdates(ctx, 1001, 1, 100)
if err != nil || len(rows) != 1 || string(rows[0].Callback.Data) != "tap" || rows[0].Callback.InlineMessage.ID != 15 {
t.Fatalf("inline callback rows=%#v err=%v", rows, err)
}
}

34
internal/store/memory/message_helpers.go
View file

@ -91,6 +91,7 @@ func normalizeMemoryMessageIDs(ids []int) []int {
func cloneMessage(msg domain.Message) domain.Message {
msg.Entities = append([]domain.MessageEntity(nil), msg.Entities...)
msg.Media = cloneRequestedPeerMedia(msg.Media)
msg.ReplyTo = cloneMessageReply(msg.ReplyTo)
msg.Forward = cloneMessageForward(msg.Forward)
msg.Reactions = cloneChannelMessageReactionsPtr(msg.Reactions)
@ -99,13 +100,36 @@ func cloneMessage(msg domain.Message) domain.Message {
return msg
}
// cloneReplyMarkup 深拷 inline keyboard 快照：与 postgres 每盒独立 decode 对齐
// cloneRequestedPeerMedia isolates the immutable disclosure snapshot carried by
// messageActionRequestedPeer. Other media payloads retain their established
// copy behavior; this helper only deep-copies the newly mutable peer/photo slices.
func cloneRequestedPeerMedia(media *domain.MessageMedia) *domain.MessageMedia {
if media == nil {
return nil
}
clone := *media
if media.ServiceAction == nil || media.ServiceAction.RequestedPeer == nil {
return &clone
}
action := *media.ServiceAction
requested := *media.ServiceAction.RequestedPeer
requested.Peers = append([]domain.Peer(nil), requested.Peers...)
requested.Details = append([]domain.MessageRequestedPeerDetails(nil), requested.Details...)
for i := range requested.Details {
requested.Details[i].Photo = domain.ClonePhotoPtr(requested.Details[i].Photo)
}
action.RequestedPeer = &requested
clone.ServiceAction = &action
return &clone
}
// cloneReplyMarkup 深拷 reply markup 快照：与 postgres 每盒独立 decode 对齐
// （双 store 行为一致），避免发送方/接收方两行共享底层 rows/Data 切片。
func cloneReplyMarkup(m *domain.MessageReplyMarkup) *domain.MessageReplyMarkup {
if m == nil {
return nil
}
clone := domain.MessageReplyMarkup{}
clone := *m
if m.Inline != nil {
clone.Inline = make([][]domain.MarkupButton, len(m.Inline))
for i, row := range m.Inline {
@ -117,6 +141,12 @@ func cloneReplyMarkup(m *domain.MessageReplyMarkup) *domain.MessageReplyMarkup {
clone.Inline[i] = cloneRow
}
}
if m.Keyboard != nil {
clone.Keyboard = make([][]domain.MarkupButton, len(m.Keyboard))
for i, row := range m.Keyboard {
clone.Keyboard[i] = append([]domain.MarkupButton(nil), row...)
}
}
return &clone
}

27
internal/store/memory/message_history.go
View file

@ -35,6 +35,33 @@ func (s *MessageStore) GetByIDs(_ context.Context, userID int64, ids []int) (dom
return out, nil
}
// GetByUID resolves one owner's box row by the shared private message id. Callback delivery
// uses it to translate the clicker's box id to the bot's box id without scanning history.
func (s *MessageStore) GetByUID(_ context.Context, userID, uid int64) (domain.Message, bool, error) {
if userID == 0 || uid == 0 {
return domain.Message{}, false, nil
}
s.mu.RLock()
var found domain.Message
for _, msg := range s.m[userID] {
if msg.UID == uid {
found = cloneMessage(msg)
reactions := s.privateMessageReactionsForMessageLocked(found)
if len(reactions.Results) > 0 || len(reactions.Recent) > 0 {
found.Reactions = cloneChannelMessageReactionsPtr(&reactions)
}
break
}
}
s.mu.RUnlock()
if found.ID == 0 {
return domain.Message{}, false, nil
}
items := []domain.Message{found}
s.enrichPrivateMessagePolls(items, int(time.Now().Unix()))
return items[0], true, nil
}
func (s *MessageStore) ListByUser(_ context.Context, userID int64, filter domain.MessageFilter) (domain.MessageList, error) {
s.mu.RLock()
messages := cloneMessages(s.m[userID])

43
internal/store/memory/message_markup_test.go Normal file
View file

@ -0,0 +1,43 @@
package memory
import (
"context"
"testing"
"telesrv/internal/domain"
)
func TestPrivateReplyKeyboardSurvivesBothBoxesAndHistory(t *testing.T) {
store := NewMessageStore(NewDialogStore())
markup := &domain.MessageReplyMarkup{
Type: domain.MessageReplyMarkupKeyboard,
Keyboard: [][]domain.MarkupButton{{{Type: domain.MarkupButtonText, Text: "Help"}}},
Resize: true,
SingleUse: true,
Persistent: true,
Placeholder: "Choose",
}
res, err := store.SendPrivateText(context.Background(), domain.SendPrivateTextRequest{
SenderUserID: 10, RecipientUserID: 20, RandomID: 30,
Message: "pick", Date: 40, ReplyMarkup: markup,
})
if err != nil {
t.Fatalf("SendPrivateText: %v", err)
}
assertReplyKeyboard := func(name string, got *domain.MessageReplyMarkup) {
t.Helper()
if got == nil || got.Kind() != domain.MessageReplyMarkupKeyboard || len(got.Keyboard) != 1 ||
len(got.Keyboard[0]) != 1 || got.Keyboard[0][0].Text != "Help" || !got.Resize ||
!got.SingleUse || !got.Persistent || got.Placeholder != "Choose" {
t.Fatalf("%s = %#v", name, got)
}
}
assertReplyKeyboard("sender", res.SenderMessage.ReplyMarkup)
assertReplyKeyboard("recipient", res.RecipientMessage.ReplyMarkup)
markup.Keyboard[0][0].Text = "mutated"
list, err := store.GetByIDs(context.Background(), 20, []int{res.RecipientMessage.ID})
if err != nil || len(list.Messages) != 1 {
t.Fatalf("GetByIDs = %+v, %v", list, err)
}
assertReplyKeyboard("recipient history", list.Messages[0].ReplyMarkup)
}

3
internal/store/memory/message_send.go
View file

@ -84,7 +84,7 @@ func (s *MessageStore) SendPrivateText(_ context.Context, req domain.SendPrivate
NoForwards: req.NoForwards,
Body: req.Message,
Entities: append([]domain.MessageEntity(nil), req.Entities...),
Media: req.Media,
Media: cloneRequestedPeerMedia(req.Media),
ViaBotID: req.ViaBotID,
GroupedID: req.GroupedID,
Effect: req.Effect,
@ -110,6 +110,7 @@ func (s *MessageStore) SendPrivateText(_ context.Context, req domain.SendPrivate
recipient.Peer = domain.Peer{Type: domain.PeerTypeUser, ID: req.SenderUserID}
recipient.Out = false
recipient.ReplyTo = cloneMessageReply(recipientReply)
recipient.Media = cloneRequestedPeerMedia(sender.Media)
// recipient = sender 是值拷贝，共享 sender.ReplyMarkup 指针/Data 切片——深拷
// 让双盒各持独立快照（与 postgres 每盒独立 decode 对齐，I3/I2）。
recipient.ReplyMarkup = cloneReplyMarkup(sender.ReplyMarkup)

59
internal/store/memory/message_test.go
View file

@ -222,6 +222,65 @@ func TestMessageStoreWebViewDataServiceActionRoundTrip(t *testing.T) {
assertWebViewData("recipient history", recipientHistory.Messages[0])
}
func TestMessageStoreRequestedPeerDisclosureSnapshotRoundTrip(t *testing.T) {
ctx := context.Background()
messages := NewMessageStore()
photo := domain.Photo{ID: 8101, Sizes: []domain.PhotoSize{{
Kind: domain.PhotoSizeKindDefault, Type: "m", W: 320, H: 320, Size: 4096,
}}}
requestedPeer := domain.Peer{Type: domain.PeerTypeUser, ID: 1000000003}
req := domain.SendPrivateTextRequest{
SenderUserID: 1000000001, RecipientUserID: 1000000002, RandomID: 200, Date: 1700000121,
Media: &domain.MessageMedia{Kind: domain.MessageMediaKindService, ServiceAction: &domain.MessageServiceAction{
Kind: domain.MessageServiceActionRequestedPeer,
RequestedPeer: &domain.MessageRequestedPeerAction{
ButtonID: 77, Peers: []domain.Peer{requestedPeer},
Details: []domain.MessageRequestedPeerDetails{{
Peer: requestedPeer, FirstName: "Shared", Username: "shared_user", Photo: &photo,
}},
NameRequested: true, UsernameRequested: true, PhotoRequested: true,
},
}},
}
got, err := messages.SendPrivateText(ctx, req)
if err != nil {
t.Fatalf("SendPrivateText: %v", err)
}
assertSnapshot := func(name string, msg domain.Message) {
t.Helper()
if msg.Media == nil || msg.Media.ServiceAction == nil || msg.Media.ServiceAction.RequestedPeer == nil {
t.Fatalf("%s media=%+v, want requested-peer action", name, msg.Media)
}
action := msg.Media.ServiceAction.RequestedPeer
if action.ButtonID != 77 || len(action.Peers) != 1 || action.Peers[0] != requestedPeer ||
len(action.Details) != 1 || action.Details[0].FirstName != "Shared" ||
action.Details[0].Username != "shared_user" || action.Details[0].Photo == nil ||
len(action.Details[0].Photo.Sizes) != 1 || action.Details[0].Photo.Sizes[0].W != 320 ||
!action.NameRequested || !action.UsernameRequested || !action.PhotoRequested {
t.Fatalf("%s requested-peer=%+v", name, action)
}
}
assertSnapshot("sender", got.SenderMessage)
assertSnapshot("recipient", got.RecipientMessage)
// Mutating either the request or one returned box must not alter the other
// box or the immutable store snapshot.
req.Media.ServiceAction.RequestedPeer.Details[0].FirstName = "mutated-request"
req.Media.ServiceAction.RequestedPeer.Details[0].Photo.Sizes[0].W = 1
got.SenderMessage.Media.ServiceAction.RequestedPeer.Details[0].FirstName = "mutated-result"
got.SenderMessage.Media.ServiceAction.RequestedPeer.Details[0].Photo.Sizes[0].W = 2
assertSnapshot("isolated recipient result", got.RecipientMessage)
for _, owner := range []int64{req.SenderUserID, req.RecipientUserID} {
history, err := messages.ListByUser(ctx, owner, domain.MessageFilter{Limit: 10})
if err != nil || len(history.Messages) != 1 {
t.Fatalf("owner %d history=%+v err=%v", owner, history, err)
}
assertSnapshot("stored history", history.Messages[0])
}
}
func TestMessageStorePrivateMessageReactionsAreSharedAcrossOwnerBoxes(t *testing.T) {
ctx := context.Background()
messages := NewMessageStore()

16
internal/store/postgres/bot_integration_test.go
View file

@ -183,6 +183,22 @@ func TestBotStoreRoundTripPostgres(t *testing.T) {
if flagBot, _, _ := bots.GetBot(ctx, bot1.ID); !flagBot.Nochats || !flagBot.ChatHistory {
t.Fatalf("flags = nochats=%v chat_history=%v, want both true", flagBot.Nochats, flagBot.ChatHistory)
}
requestedButton := domain.BotRequestedWebViewButton{
WebAppReqID: fmt.Sprintf("pg-requested-%d", suffix), BotUserID: bot1.ID, UserID: owner.ID,
ButtonID: 45, Text: "Share", PeerType: "user", MaxQuantity: 2,
NameRequested: true, UsernameRequested: true, PhotoRequested: true,
CreatedAt: time.Now(), ExpiresAt: time.Now().Add(time.Hour),
}
t.Cleanup(func() {
_ = bots.DeleteRequestedWebViewButton(ctx, bot1.ID, owner.ID, requestedButton.WebAppReqID)
})
if err := bots.SaveRequestedWebViewButton(ctx, requestedButton); err != nil {
t.Fatalf("save requested button: %v", err)
}
storedButton, found, err := bots.GetRequestedWebViewButton(ctx, bot1.ID, owner.ID, requestedButton.WebAppReqID)
if err != nil || !found || !storedButton.NameRequested || !storedButton.UsernameRequested || !storedButton.PhotoRequested {
t.Fatalf("requested button=%#v found=%v err=%v", storedButton, found, err)
}
if can, err := bots.CanBotSendMessage(ctx, bot1.ID, owner.ID); err != nil || can {
t.Fatalf("CanBotSendMessage before allow = %v,%v, want false,nil", can, err)
}

35
internal/store/postgres/bot_miniapp.go
View file

@ -416,16 +416,29 @@ func (s *BotStore) SaveRequestedWebViewButton(ctx context.Context, button domain
if button.BotUserID == 0 || button.UserID == 0 || button.WebAppReqID == "" || button.ExpiresAt.IsZero() {
return domain.ErrBotRequestedButtonInvalid
}
_, err := s.db.Exec(ctx, `
INSERT INTO webview_requested_buttons (webapp_req_id, bot_user_id, user_id, button_id, text, peer_type, max_quantity, created_at, expires_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
peerFilter, err := json.Marshal(button.PeerFilter)
if err != nil {
return domain.ErrBotRequestedButtonInvalid
}
_, err = s.db.Exec(ctx, `
INSERT INTO webview_requested_buttons (
webapp_req_id, bot_user_id, user_id, button_id, text, peer_type, max_quantity,
peer_filter, name_requested, username_requested, photo_requested, created_at, expires_at
)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
ON CONFLICT (webapp_req_id) DO UPDATE SET
button_id=EXCLUDED.button_id,
text=EXCLUDED.text,
peer_type=EXCLUDED.peer_type,
max_quantity=EXCLUDED.max_quantity,
peer_filter=EXCLUDED.peer_filter,
name_requested=EXCLUDED.name_requested,
username_requested=EXCLUDED.username_requested,
photo_requested=EXCLUDED.photo_requested,
expires_at=EXCLUDED.expires_at`,
button.WebAppReqID, button.BotUserID, button.UserID, button.ButtonID, button.Text, button.PeerType, button.MaxQuantity, button.CreatedAt, button.ExpiresAt)
button.WebAppReqID, button.BotUserID, button.UserID, button.ButtonID, button.Text,
button.PeerType, button.MaxQuantity, peerFilter, button.NameRequested,
button.UsernameRequested, button.PhotoRequested, button.CreatedAt, button.ExpiresAt)
if err != nil {
return fmt.Errorf("save requested webview button: %w", err)
}
@ -435,18 +448,28 @@ ON CONFLICT (webapp_req_id) DO UPDATE SET
func (s *BotStore) GetRequestedWebViewButton(ctx context.Context, botUserID, userID int64, webAppReqID string) (domain.BotRequestedWebViewButton, bool, error) {
_, _ = s.db.Exec(ctx, `DELETE FROM webview_requested_buttons WHERE expires_at <= now()`)
var button domain.BotRequestedWebViewButton
var peerFilter []byte
err := s.db.QueryRow(ctx, `
SELECT webapp_req_id, bot_user_id, user_id, button_id, text, peer_type, max_quantity, created_at, expires_at
SELECT webapp_req_id, bot_user_id, user_id, button_id, text, peer_type, max_quantity,
peer_filter, name_requested, username_requested, photo_requested, created_at, expires_at
FROM webview_requested_buttons
WHERE bot_user_id=$1 AND user_id=$2 AND webapp_req_id=$3 AND expires_at > now()`,
botUserID, userID, webAppReqID).
Scan(&button.WebAppReqID, &button.BotUserID, &button.UserID, &button.ButtonID, &button.Text, &button.PeerType, &button.MaxQuantity, &button.CreatedAt, &button.ExpiresAt)
Scan(&button.WebAppReqID, &button.BotUserID, &button.UserID, &button.ButtonID,
&button.Text, &button.PeerType, &button.MaxQuantity, &peerFilter,
&button.NameRequested, &button.UsernameRequested, &button.PhotoRequested,
&button.CreatedAt, &button.ExpiresAt)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return domain.BotRequestedWebViewButton{}, false, nil
}
return domain.BotRequestedWebViewButton{}, false, fmt.Errorf("get requested webview button: %w", err)
}
if string(peerFilter) != "{}" && string(peerFilter) != "null" {
if err := json.Unmarshal(peerFilter, &button.PeerFilter); err != nil {
return domain.BotRequestedWebViewButton{}, false, fmt.Errorf("decode requested webview button filter: %w", err)
}
}
return button, true, nil
}

521
internal/store/postgres/botapi_update.go
View file

@ -20,17 +20,320 @@ 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
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
}
}
row, err := s.scanBotAPIUpdate(s.db.QueryRow(ctx, `
INSERT INTO bot_api_updates (
bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date
) VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts) DO NOTHING
RETURNING id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date
`, req.BotUserID, string(req.Kind), string(req.Peer.Type), req.Peer.ID, req.MessageID, req.SourcePts, req.Date))
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
) SELECT $1, $2::varchar(32), $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15
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
), 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
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))
if err == nil {
return row, true, nil
}
@ -38,21 +341,69 @@ RETURNING id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_p
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
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
FROM bot_api_updates
WHERE bot_user_id = $1
AND update_kind = $2
AND peer_type = $3
AND peer_id = $4
AND message_id = $5
AND source_pts = $6
`, req.BotUserID, string(req.Kind), string(req.Peer.Type), req.Peer.ID, req.MessageID, req.SourcePts))
AND (
(update_kind = 'callback_query' AND callback_query_id = $7)
OR
(update_kind <> 'callback_query' AND peer_type = $3 AND peer_id = $4 AND message_id = $5 AND source_pts = $6)
)
`, req.BotUserID, string(req.Kind), string(req.Peer.Type), req.Peer.ID, req.MessageID, req.SourcePts, callbackQueryID))
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
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
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
@ -64,7 +415,9 @@ func (s *BotAPIUpdateStore) ListBotAPIUpdates(ctx context.Context, botUserID, fr
limit = 100
}
rows, err := s.db.Query(ctx, `
SELECT id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date
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
FROM bot_api_updates
WHERE bot_user_id = $1 AND id >= $2
ORDER BY id
@ -93,23 +446,98 @@ func (s *BotAPIUpdateStore) ConfirmBotAPIUpdates(ctx context.Context, botUserID,
return nil
}
if _, err := s.db.Exec(ctx, `
INSERT INTO bot_api_update_states (bot_user_id, confirmed_update_id)
VALUES ($1, $2)
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, EXCLUDED.confirmed_update_id),
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()
WHERE bot_api_update_states.confirmed_update_id < EXCLUDED.confirmed_update_id
`, 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 回收 Bot API 投递队列的死行（性能审计 H1）：
// 1. 已确认（id <= bot_api_update_states.confirmed_update_id）且入队超过 confirmedGrace 的行——
// 官方 Bot API 语义下确认即弃，getUpdates 的 fromID 恒 > confirmed，删除不影响任何读路径；
// 宽限仅防御 offset 回拨调试场景。
// 2. 按消息 date 超过 maxAge 的行（无论确认与否）——对齐官方「updates 服务器最多保留 24 小时」
// 2. 按队列 created_at 超过 maxAge 的行（无论确认与否）——对齐官方「updates 服务器最多保留 24 小时」
// 语义，同时封顶 MTProto-only bot（从不调 getUpdates、无 state 行）成员身份带来的无界增长。
//
// 与 user_update_events 的「永久保留」约束无关：那是 TDesktop 账号级 differenceTooLong 缺陷所迫，
@ -139,15 +567,15 @@ WHERE id IN (
total += int(tag.RowsAffected())
}
if maxAge > 0 {
cutoff := time.Now().Add(-maxAge).Unix()
// 走 bot_api_updates_retention_idx(date, id)。
cutoff := time.Now().Add(-maxAge)
// 走 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 date < $1
ORDER BY date, id
WHERE created_at < $1
ORDER BY created_at, id
LIMIT $2
)`, cutoff, limit)
if err != nil {
@ -187,28 +615,69 @@ type botAPIUpdateScanner interface {
func scanBotAPIUpdateRows(row botAPIUpdateScanner) (domain.BotAPIUpdate, error) {
var item domain.BotAPIUpdate
var kind, peerType string
if err := row.Scan(&item.ID, &item.BotUserID, &kind, &peerType, &item.Peer.ID, &item.MessageID, &item.SourcePts, &item.Date); err != nil {
var callbackQueryID, callbackUserID, callbackChatInstance int64
var callbackInlineDCID, callbackInlineMessageID int
var callbackInlineOwnerID, callbackInlineAccessHash int64
var callbackData []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); 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}
}
}
return item, nil
}
func validateBotAPIUpdateRequest(req domain.EnqueueBotAPIUpdateRequest) error {
if req.BotUserID == 0 || req.MessageID <= 0 {
if req.BotUserID == 0 {
return fmt.Errorf("invalid bot api update")
}
if req.Kind != domain.BotAPIUpdateMessage && req.Kind != domain.BotAPIUpdateEditedMessage {
if req.Kind != domain.BotAPIUpdateMessage && req.Kind != domain.BotAPIUpdateEditedMessage && req.Kind != domain.BotAPIUpdateCallbackQuery {
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 {
if req.Peer.ID <= 0 || req.MessageID <= 0 {
return fmt.Errorf("invalid bot api update peer")
}
case "":
if req.Kind != domain.BotAPIUpdateCallbackQuery || 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 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
}

296
internal/store/postgres/botapi_update_integration_test.go
View file

@ -1,6 +1,7 @@
package postgres
import (
"bytes"
"context"
"testing"
"time"
@ -8,10 +9,292 @@ import (
"telesrv/internal/domain"
)
func TestBotAPICallbackQueryQueueRoundTrip(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
suffix := randomSuffix(t)
users := NewUserStore(pool)
bot, err := users.Create(ctx, domain.User{
AccessHash: 921, Phone: "+1921" + suffix + "01", FirstName: "CallbackQueueBot",
})
if err != nil {
t.Fatalf("create bot user: %v", err)
}
clicker, err := users.Create(ctx, domain.User{
AccessHash: 922, Phone: "+1922" + suffix + "02", FirstName: "CallbackClicker",
})
if err != nil {
t.Fatalf("create callback user: %v", err)
}
if _, err := pool.Exec(ctx, `
INSERT INTO bots (bot_user_id, owner_user_id, token_secret)
VALUES ($1, $1, 'callback-queue-secret')`, bot.ID); err != nil {
t.Fatalf("seed bot: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM bot_api_updates WHERE bot_user_id = $1", bot.ID)
_, _ = pool.Exec(ctx, "DELETE FROM bot_api_update_states WHERE bot_user_id = $1", bot.ID)
_, _ = pool.Exec(ctx, "DELETE FROM bots WHERE bot_user_id = $1", bot.ID)
})
callback := &domain.BotCallbackQuery{
ID: 880011, BotUserID: bot.ID, UserID: clicker.ID,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: clicker.ID}, MessageID: 17,
ChatInstance: 990022, Data: []byte{0, 1, 0xff, 'x'},
}
req := domain.EnqueueBotAPIUpdateRequest{
BotUserID: bot.ID, Kind: domain.BotAPIUpdateCallbackQuery,
Peer: callback.Peer, MessageID: callback.MessageID, Date: int(time.Now().Unix()), Callback: callback,
}
store := NewBotAPIUpdateStore(pool)
first, created, err := store.EnqueueBotAPIUpdate(ctx, req)
if err != nil || !created {
t.Fatalf("enqueue callback: row=%+v created=%v err=%v", first, created, err)
}
again, created, err := store.EnqueueBotAPIUpdate(ctx, req)
if err != nil || created || again.ID != first.ID {
t.Fatalf("dedupe callback: row=%+v created=%v err=%v", again, created, err)
}
items, err := store.ListBotAPIUpdates(ctx, bot.ID, first.ID, 100)
if err != nil || len(items) != 1 {
t.Fatalf("list callback = %+v, %v", items, err)
}
got := items[0].Callback
if got == nil || got.ID != callback.ID || got.BotUserID != bot.ID || got.UserID != clicker.ID ||
got.Peer != callback.Peer || got.MessageID != callback.MessageID || got.ChatInstance != callback.ChatInstance ||
!bytes.Equal(got.Data, callback.Data) {
t.Fatalf("callback round trip = %+v, want %+v", got, callback)
}
}
func TestBotAPIInlineCallbackAndWebhookStateRoundTrip(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
suffix := randomSuffix(t)
users := NewUserStore(pool)
bot, err := users.Create(ctx, domain.User{AccessHash: 931, Phone: "+1931" + suffix + "01", FirstName: "WebhookBot"})
if err != nil {
t.Fatal(err)
}
clicker, err := users.Create(ctx, domain.User{AccessHash: 932, Phone: "+1932" + suffix + "02", FirstName: "InlineClicker"})
if err != nil {
t.Fatal(err)
}
if _, err := pool.Exec(ctx, `INSERT INTO bots (bot_user_id, owner_user_id, token_secret) VALUES ($1, $1, 'webhook-secret')`, bot.ID); err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM bot_api_webhooks WHERE bot_user_id = $1", bot.ID)
_, _ = pool.Exec(ctx, "DELETE FROM bot_api_updates WHERE bot_user_id = $1", bot.ID)
_, _ = pool.Exec(ctx, "DELETE FROM bot_api_update_states WHERE bot_user_id = $1", bot.ID)
_, _ = pool.Exec(ctx, "DELETE FROM bots WHERE bot_user_id = $1", bot.ID)
})
s := NewBotAPIUpdateStore(pool)
inline := &domain.BotInlineMessageID{DCID: 2, OwnerID: clicker.ID, ID: 17, AccessHash: 445566}
callback := &domain.BotCallbackQuery{
ID: 9911, BotUserID: bot.ID, UserID: clicker.ID, ChatInstance: 8811,
Data: []byte{0, 1, 0xff}, InlineMessage: inline,
}
row, created, err := s.EnqueueBotAPIUpdate(ctx, domain.EnqueueBotAPIUpdateRequest{
BotUserID: bot.ID, Kind: domain.BotAPIUpdateCallbackQuery, Date: int(time.Now().Unix()), Callback: callback,
})
if err != nil || !created {
t.Fatalf("enqueue inline callback row=%#v created=%v err=%v", row, created, err)
}
items, err := s.ListBotAPIUpdates(ctx, bot.ID, row.ID, 100)
if err != nil || len(items) != 1 || items[0].Peer != (domain.Peer{}) || items[0].MessageID != 0 ||
items[0].Callback == nil || items[0].Callback.InlineMessage == nil || *items[0].Callback.InlineMessage != *inline ||
!bytes.Equal(items[0].Callback.Data, callback.Data) {
t.Fatalf("inline callback items=%#v err=%v", items, err)
}
config := domain.BotAPIWebhook{
BotUserID: bot.ID, URL: "https://example.test/hook", SecretToken: "safe_secret",
MaxConnections: 8, AllowedUpdates: []domain.BotAPIUpdateKind{domain.BotAPIUpdateCallbackQuery}, AllowedUpdatesSet: true,
}
if err := s.SetBotAPIWebhook(ctx, config, false); err != nil {
t.Fatal(err)
}
stored, found, err := s.BotAPIWebhook(ctx, bot.ID)
if err != nil || !found || stored.URL != config.URL || stored.SecretToken != config.SecretToken ||
stored.MaxConnections != 8 || len(stored.AllowedUpdates) != 1 {
t.Fatalf("webhook=%#v found=%v err=%v", stored, found, err)
}
config.URL = "https://example.test/reconfigured"
config.AllowedUpdates = nil
config.AllowedUpdatesSet = false
if err := s.SetBotAPIWebhook(ctx, config, false); err != nil {
t.Fatal(err)
}
stored, found, err = s.BotAPIWebhook(ctx, bot.ID)
if err != nil || !found || stored.URL != config.URL || len(stored.AllowedUpdates) != 1 || stored.AllowedUpdates[0] != domain.BotAPIUpdateCallbackQuery {
t.Fatalf("preserved webhook=%#v found=%v err=%v", stored, found, err)
}
if acquired, err := s.AcquireBotAPIWebhookLease(ctx, bot.ID, "one", time.Minute); err != nil || !acquired {
t.Fatalf("first lease=%v err=%v", acquired, err)
}
if acquired, err := s.AcquireBotAPIWebhookLease(ctx, bot.ID, "two", time.Minute); err != nil || acquired {
t.Fatalf("second lease=%v err=%v", acquired, err)
}
if err := s.ReleaseBotAPIWebhookLease(ctx, bot.ID, "stale"); err != nil {
t.Fatal(err)
}
if acquired, _ := s.AcquireBotAPIWebhookLease(ctx, bot.ID, "two", time.Minute); acquired {
t.Fatal("stale webhook release removed active lease")
}
next := time.Now().Add(time.Hour)
if err := s.RecordBotAPIWebhookSuccess(ctx, bot.ID, "one", next); err != nil {
t.Fatal(err)
}
if due, err := s.ListDueBotAPIWebhooks(ctx, 10); err != nil || len(due) != 0 {
t.Fatalf("idle due=%#v err=%v", due, err)
}
// A newly inserted allowed callback wakes the idle webhook in the same SQL statement.
callback2 := *callback
callback2.ID++
callback2.InlineMessage = &domain.BotInlineMessageID{DCID: 2, OwnerID: clicker.ID, ID: 18, AccessHash: 556677}
if _, created, err := s.EnqueueBotAPIUpdate(ctx, domain.EnqueueBotAPIUpdateRequest{
BotUserID: bot.ID, Kind: domain.BotAPIUpdateCallbackQuery, Date: int(time.Now().Unix()), Callback: &callback2,
}); err != nil || !created {
t.Fatalf("enqueue wake created=%v err=%v", created, err)
}
if due, err := s.ListDueBotAPIWebhooks(ctx, 10); err != nil || len(due) != 1 || due[0].BotUserID != bot.ID {
t.Fatalf("woken due=%#v err=%v", due, err)
}
if err := s.DeleteBotAPIWebhook(ctx, bot.ID, true); err != nil {
t.Fatal(err)
}
if _, found, err := s.BotAPIWebhook(ctx, bot.ID); err != nil || found {
t.Fatalf("webhook after delete found=%v err=%v", found, err)
}
if pending, err := s.PendingBotAPIUpdateCount(ctx, bot.ID); err != nil || pending != 0 {
t.Fatalf("pending after delete/drop=%d err=%v", pending, err)
}
}
func TestBotAPIPollLeaseCrossStoreInstance(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
suffix := randomSuffix(t)
users := NewUserStore(pool)
bot, err := users.Create(ctx, domain.User{AccessHash: 933, Phone: "+1933" + suffix + "01", FirstName: "PollLeaseBot"})
if err != nil {
t.Fatal(err)
}
if _, err := pool.Exec(ctx, `INSERT INTO bots (bot_user_id, owner_user_id, token_secret) VALUES ($1, $1, 'poll-lease-secret')`, bot.ID); err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM bot_api_update_states WHERE bot_user_id = $1", bot.ID)
_, _ = pool.Exec(ctx, "DELETE FROM bots WHERE bot_user_id = $1", bot.ID)
})
a, b := NewBotAPIUpdateStore(pool), NewBotAPIUpdateStore(pool)
if acquired, err := a.AcquireBotAPIPollLease(ctx, bot.ID, "one", time.Minute); err != nil || !acquired {
t.Fatalf("first acquire=%v err=%v", acquired, err)
}
if acquired, err := b.AcquireBotAPIPollLease(ctx, bot.ID, "two", time.Minute); err != nil || acquired {
t.Fatalf("cross-instance acquire=%v err=%v", acquired, err)
}
if err := b.ReleaseBotAPIPollLease(ctx, bot.ID, "stale"); err != nil {
t.Fatal(err)
}
if acquired, _ := b.AcquireBotAPIPollLease(ctx, bot.ID, "two", time.Minute); acquired {
t.Fatal("stale release removed active poll lease")
}
if err := a.ReleaseBotAPIPollLease(ctx, bot.ID, "one"); err != nil {
t.Fatal(err)
}
if acquired, err := b.AcquireBotAPIPollLease(ctx, bot.ID, "two", time.Minute); err != nil || !acquired {
t.Fatalf("successor acquire=%v err=%v", acquired, err)
}
}
func TestBotAPIPollingStateClampFilterTailAndDrop(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
users := NewUserStore(pool)
suffix := randomSuffix(t)
bot, err := users.Create(ctx, domain.User{
AccessHash: 923, Phone: "+1923" + suffix + "01", FirstName: "PollingStateBot",
})
if err != nil {
t.Fatalf("create bot user: %v", err)
}
if _, err := pool.Exec(ctx, `INSERT INTO bots (bot_user_id, owner_user_id, token_secret) VALUES ($1, $1, 'poll-state-secret')`, bot.ID); err != nil {
t.Fatalf("seed bot: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM bot_api_updates WHERE bot_user_id = $1", bot.ID)
_, _ = pool.Exec(ctx, "DELETE FROM bot_api_update_states WHERE bot_user_id = $1", bot.ID)
_, _ = pool.Exec(ctx, "DELETE FROM bots WHERE bot_user_id = $1", bot.ID)
})
s := NewBotAPIUpdateStore(pool)
enqueue := func(kind domain.BotAPIUpdateKind, messageID int) (domain.BotAPIUpdate, bool) {
t.Helper()
row, created, err := s.EnqueueBotAPIUpdate(ctx, domain.EnqueueBotAPIUpdateRequest{
BotUserID: bot.ID, Kind: kind,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: bot.ID + 1},
MessageID: messageID, SourcePts: messageID, Date: int(time.Now().Unix()),
})
if err != nil {
t.Fatalf("enqueue %s/%d: %v", kind, messageID, err)
}
return row, created
}
for id := 1; id <= 3; id++ {
if _, created := enqueue(domain.BotAPIUpdateMessage, id); !created {
t.Fatalf("initial message %d was not created", id)
}
}
if err := s.SetBotAPIAllowedUpdates(ctx, bot.ID, []domain.BotAPIUpdateKind{domain.BotAPIUpdateEditedMessage}); err != nil {
t.Fatalf("set allowed updates: %v", err)
}
if row, created := enqueue(domain.BotAPIUpdateMessage, 4); created || row.ID != 0 {
t.Fatalf("filtered row=%+v created=%v", row, created)
}
lastBeforeBaseline, created := enqueue(domain.BotAPIUpdateEditedMessage, 5)
if !created {
t.Fatal("allowed edit was filtered")
}
if err := s.ConfirmBotAPIUpdates(ctx, bot.ID, 1<<60); err != nil {
t.Fatalf("initialize external cursor: %v", err)
}
confirmed, found, err := s.ConfirmedBotAPIUpdateID(ctx, bot.ID)
if err != nil || !found || confirmed != lastBeforeBaseline.ID {
t.Fatalf("baseline confirmed=%d found=%v err=%v want=%d", confirmed, found, err, lastBeforeBaseline.ID)
}
pendingRow, created := enqueue(domain.BotAPIUpdateEditedMessage, 6)
if !created {
t.Fatal("post-baseline edit was filtered")
}
if err := s.ConfirmBotAPIUpdates(ctx, bot.ID, 1<<60); err != nil {
t.Fatalf("repeat external cursor: %v", err)
}
confirmed, _, _ = s.ConfirmedBotAPIUpdateID(ctx, bot.ID)
if confirmed != lastBeforeBaseline.ID {
t.Fatalf("repeat external cursor advanced to %d, want %d", confirmed, lastBeforeBaseline.ID)
}
tail, err := s.ListTailBotAPIUpdates(ctx, bot.ID, 1, 100)
if err != nil || len(tail) != 1 || tail[0].ID != pendingRow.ID {
t.Fatalf("tail=%+v err=%v want=%d", tail, err, pendingRow.ID)
}
if count, err := s.PendingBotAPIUpdateCount(ctx, bot.ID); err != nil || count != 1 {
t.Fatalf("pending count=%d err=%v", count, err)
}
if err := s.DropPendingBotAPIUpdates(ctx, bot.ID); err != nil {
t.Fatalf("drop pending: %v", err)
}
if count, err := s.PendingBotAPIUpdateCount(ctx, bot.ID); err != nil || count != 0 {
t.Fatalf("pending after drop=%d err=%v", count, err)
}
}
// TestBotAPIUpdateRetention 锁定 H1 场景矩阵：
// - 已确认 + 超宽限 → 删；已确认 + 宽限内 → 留；
// - 未确认 + date 超保留期 → 删（含无 state 行的 MTProto-only bot）；
// - 未确认 + date 在保留期内 → 留；
// - 未确认 + created_at 超保留期 → 删（含无 state 行的 MTProto-only bot）；
// - 未确认 + created_at 在保留期内 → 留；
// - 删除后 getUpdates 读路径（fromID > confirmed）不受影响。
func TestBotAPIUpdateRetention(t *testing.T) {
pool := testPool(t)
@ -47,7 +330,6 @@ ON CONFLICT (bot_user_id) DO NOTHING`, u.ID); err != nil {
s := NewBotAPIUpdateStore(pool)
now := time.Now().Unix()
stale := now - int64((48 * time.Hour).Seconds())
enqueue := func(botID int64, messageID int, date int64) domain.BotAPIUpdate {
t.Helper()
row, created, err := s.EnqueueBotAPIUpdate(ctx, domain.EnqueueBotAPIUpdateRequest{
@ -67,7 +349,7 @@ ON CONFLICT (bot_user_id) DO NOTHING`, u.ID); err != nil {
confirmedOld := enqueue(confirmedBot, 1, now) // 已确认 + created_at 回拨超宽限 → 删
confirmedFresh := enqueue(confirmedBot, 2, now) // 已确认 + 宽限内 → 留
unconfirmedFresh := enqueue(confirmedBot, 3, now)
expiredNoState := enqueue(mtprotoOnlyBot, 4, stale) // 无 state 行 + date 超保留期 → 删
expiredNoState := enqueue(mtprotoOnlyBot, 4, now) // 无 state 行 + created_at 超保留期 → 删
freshNoState := enqueue(mtprotoOnlyBot, 5, now)
if err := s.ConfirmBotAPIUpdates(ctx, confirmedBot, confirmedFresh.ID); err != nil {
@ -77,6 +359,10 @@ ON CONFLICT (bot_user_id) DO NOTHING`, u.ID); err != nil {
"UPDATE bot_api_updates SET created_at = now() - interval '1 hour' WHERE id = $1", confirmedOld.ID); err != nil {
t.Fatalf("backdate confirmed row: %v", err)
}
if _, err := pool.Exec(ctx,
"UPDATE bot_api_updates SET created_at = now() - interval '48 hours' WHERE id = $1", expiredNoState.ID); err != nil {
t.Fatalf("backdate expired row: %v", err)
}
deleted, err := s.DeleteDeliveredOrExpired(ctx, 15*time.Minute, 24*time.Hour, 1000)
if err != nil {
@ -85,7 +371,7 @@ ON CONFLICT (bot_user_id) DO NOTHING`, u.ID); err != nil {
// 共享测试库可能有其它历史行同被回收，只要求至少删掉本测试的 2 行；
// 精确归属由下方 remaining 断言保证。
if deleted < 2 {
t.Fatalf("deleted = %d, want >= 2 (confirmed+grace expired, date expired)", deleted)
t.Fatalf("deleted = %d, want >= 2 (confirmed+grace expired, created_at expired)", deleted)
}
remaining := map[int64]bool{}

50
internal/store/postgres/media.go
View file

@ -550,6 +550,56 @@ func (s *MediaStore) GetPhoto(ctx context.Context, id int64) (domain.Photo, bool
return photo, true, nil
}
// GetPhotos resolves a bounded set of immutable photo metadata with one indexed
// ANY query. Missing ids are omitted and the result follows first-seen caller order.
func (s *MediaStore) GetPhotos(ctx context.Context, ids []int64) ([]domain.Photo, error) {
if len(ids) == 0 {
return nil, nil
}
unique := make([]int64, 0, len(ids))
seen := make(map[int64]struct{}, len(ids))
for _, id := range ids {
if id == 0 {
continue
}
if _, ok := seen[id]; ok {
continue
}
seen[id] = struct{}{}
unique = append(unique, id)
}
if len(unique) == 0 {
return nil, nil
}
rows, err := s.db.Query(ctx, `
SELECT id, access_hash, file_reference, date, dc_id, has_stickers, sizes::text
FROM photos
WHERE id = ANY($1::bigint[])
`, unique)
if err != nil {
return nil, err
}
defer rows.Close()
byID := make(map[int64]domain.Photo, len(unique))
for rows.Next() {
photo, err := scanPhotoRow(rows)
if err != nil {
return nil, err
}
byID[photo.ID] = photo
}
if err := rows.Err(); err != nil {
return nil, err
}
out := make([]domain.Photo, 0, len(byID))
for _, id := range unique {
if photo, ok := byID[id]; ok {
out = append(out, photo)
}
}
return out, nil
}
type photoScanner interface {
Scan(dest ...any) error
}

9
internal/store/postgres/media_codec.go
View file

@ -2,6 +2,7 @@ package postgres
import (
"encoding/json"
"fmt"
"reflect"
"telesrv/internal/domain"
@ -44,9 +45,12 @@ func decodeMessageMedia(s string) (*domain.MessageMedia, error) {
return &m, nil
}
// encodeReplyMarkup 把 inline keyboard 快照序列化为 JSONB；空 markup 序列化为 "{}"。
// encodeReplyMarkup 把 reply/inline keyboard 快照序列化为 JSONB；空 markup 序列化为 "{}"。
// callback data 是 []byte，json.Marshal 自动 base64（保证经 JSONB 字节级 round-trip）。
func encodeReplyMarkup(m *domain.MessageReplyMarkup) ([]byte, error) {
if err := domain.ValidateReplyMarkup(m); err != nil {
return nil, fmt.Errorf("encode reply markup: %w", err)
}
if m.IsZero() {
return []byte("{}"), nil
}
@ -63,6 +67,9 @@ func decodeReplyMarkup(s string) (*domain.MessageReplyMarkup, error) {
if err := json.Unmarshal([]byte(s), &m); err != nil {
return nil, err
}
if err := domain.ValidateReplyMarkup(&m); err != nil {
return nil, fmt.Errorf("decode reply markup: %w", err)
}
if m.IsZero() {
return nil, nil
}

4
internal/store/postgres/media_integration_test.go
View file

@ -91,6 +91,10 @@ func TestMediaStoreRoundTrip(t *testing.T) {
if err != nil || !ok || len(gotPhoto.Sizes) != 1 || gotPhoto.Sizes[0].Type != "x" {
t.Fatalf("get photo mismatch: ok=%v err=%v photo=%+v", ok, err, gotPhoto)
}
photos, err := s.GetPhotos(ctx, []int64{photoID, 0, photoID, photoID + 99})
if err != nil || len(photos) != 1 || photos[0].ID != photoID || len(photos[0].Sizes) != 1 {
t.Fatalf("get photos mismatch: photos=%+v err=%v", photos, err)
}
// ---- sticker set ----
set := domain.StickerSet{

21
internal/store/postgres/message_history.go
View file

@ -51,6 +51,27 @@ func (s *MessageStore) GetByIDs(ctx context.Context, userID int64, ids []int) (d
return out, nil
}
// GetByUID resolves one owner's box row by the indexed shared private_message_id.
func (s *MessageStore) GetByUID(ctx context.Context, userID, uid int64) (domain.Message, bool, error) {
if userID == 0 || uid == 0 {
return domain.Message{}, false, nil
}
row, err := s.q.GetMessageBoxByPrivateMessage(ctx, sqlcgen.GetMessageBoxByPrivateMessageParams{
OwnerUserID: userID,
PrivateMessageID: uid,
})
if errors.Is(err, pgx.ErrNoRows) {
return domain.Message{}, false, nil
}
if err != nil {
return domain.Message{}, false, fmt.Errorf("get message by uid: %w", err)
}
if _, err := decodeReplyMarkup(row.ReplyMarkupJson); err != nil {
return domain.Message{}, false, fmt.Errorf("get message by uid reply markup: %w", err)
}
return messageFromGetBoxRow(row), true, nil
}
func (s *MessageStore) ListByUser(ctx context.Context, userID int64, filter domain.MessageFilter) (domain.MessageList, error) {
limit := filter.Limit
if limit <= 0 {

42
internal/store/postgres/message_markup_codec_test.go Normal file
View file

@ -0,0 +1,42 @@
package postgres
import (
"testing"
"telesrv/internal/domain"
)
func TestReplyMarkupCodecValidatesTaggedUnion(t *testing.T) {
keyboard := &domain.MessageReplyMarkup{
Type: domain.MessageReplyMarkupKeyboard,
Keyboard: [][]domain.MarkupButton{{{Type: domain.MarkupButtonText, Text: "Help"}}},
Resize: true,
Placeholder: "Choose",
}
raw, err := encodeReplyMarkup(keyboard)
if err != nil {
t.Fatalf("encode reply keyboard: %v", err)
}
got, err := decodeReplyMarkup(string(raw))
if err != nil || got == nil || got.Kind() != domain.MessageReplyMarkupKeyboard ||
len(got.Keyboard) != 1 || got.Keyboard[0][0].Text != "Help" || !got.Resize || got.Placeholder != "Choose" {
t.Fatalf("decoded reply keyboard = %#v, err=%v", got, err)
}
// Pre-union inline snapshots intentionally remain readable.
legacy, err := decodeReplyMarkup(`{"inline":[[{"type":"callback","text":"OK","data":"b2s="}]]}`)
if err != nil || legacy == nil || legacy.Kind() != domain.MessageReplyMarkupInline {
t.Fatalf("legacy inline markup = %#v, err=%v", legacy, err)
}
malformed := &domain.MessageReplyMarkup{
Type: domain.MessageReplyMarkupInline,
Keyboard: [][]domain.MarkupButton{{{Type: domain.MarkupButtonText, Text: "wrong"}}},
}
if _, err := encodeReplyMarkup(malformed); err == nil {
t.Fatal("malformed union must fail at the write boundary")
}
if _, err := decodeReplyMarkup(`{"type":"inline","keyboard":[[{"type":"text","text":"wrong"}]]}`); err == nil {
t.Fatal("malformed stored union must fail at the read boundary")
}
}

2
internal/store/postgres/message_send.go
View file

@ -117,7 +117,7 @@ func (s *MessageStore) sendPrivateTextOnce(ctx context.Context, req domain.SendP
if err != nil {
return domain.SendPrivateTextResult{}, err
}
// reply_markup（bot inline keyboard）随消息一并入双盒；普通用户发送恒 nil → "{}"。
// reply_markup（bot reply/inline keyboard）随消息一并入双盒；普通用户发送恒 nil → "{}"。
replyMarkupJSON, err := encodeReplyMarkup(req.ReplyMarkup)
if err != nil {
return domain.SendPrivateTextResult{}, err

64
internal/store/postgres/message_send_integration_test.go
View file

@ -205,6 +205,70 @@ func TestMessageStoreWebViewDataServiceActionRoundTrip(t *testing.T) {
assertWebViewData("recipient event", events[0].Message)
}
func TestMessageStoreRequestedPeerDisclosureSnapshotRoundTrip(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
suffix := randomSuffix(t)
users := NewUserStore(pool)
sender := createTestUser(t, ctx, users, "+1666"+suffix+"33", "RequestedSender", "")
recipient := createTestUser(t, ctx, users, "+1666"+suffix+"34", "RequestedRecipient", "")
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM users WHERE id = ANY($1::bigint[])", []int64{sender.ID, recipient.ID})
})
requestedPeer := domain.Peer{Type: domain.PeerTypeChannel, ID: 5501}
photo := domain.Photo{ID: 8201, Sizes: []domain.PhotoSize{{
Kind: domain.PhotoSizeKindDefault, Type: "m", W: 320, H: 320, Size: 4096,
}}}
messages := NewMessageStore(pool)
got, err := messages.SendPrivateText(ctx, domain.SendPrivateTextRequest{
SenderUserID: sender.ID, RecipientUserID: recipient.ID, RandomID: 9002, Date: 1700000212,
Media: &domain.MessageMedia{Kind: domain.MessageMediaKindService, ServiceAction: &domain.MessageServiceAction{
Kind: domain.MessageServiceActionRequestedPeer,
RequestedPeer: &domain.MessageRequestedPeerAction{
ButtonID: 88, Peers: []domain.Peer{requestedPeer},
Details: []domain.MessageRequestedPeerDetails{{
Peer: requestedPeer, Title: "Shared Chat", Username: "shared_chat", Photo: &photo,
}},
NameRequested: true, UsernameRequested: true, PhotoRequested: true,
},
}},
})
if err != nil {
t.Fatalf("SendPrivateText: %v", err)
}
assertSnapshot := func(name string, msg domain.Message) {
t.Helper()
if msg.Media == nil || msg.Media.ServiceAction == nil || msg.Media.ServiceAction.RequestedPeer == nil {
t.Fatalf("%s media=%+v, want requested-peer action", name, msg.Media)
}
action := msg.Media.ServiceAction.RequestedPeer
if action.ButtonID != 88 || len(action.Peers) != 1 || action.Peers[0] != requestedPeer ||
len(action.Details) != 1 || action.Details[0].Title != "Shared Chat" ||
action.Details[0].Username != "shared_chat" || action.Details[0].Photo == nil ||
len(action.Details[0].Photo.Sizes) != 1 || action.Details[0].Photo.Sizes[0].W != 320 ||
!action.NameRequested || !action.UsernameRequested || !action.PhotoRequested {
t.Fatalf("%s requested-peer=%+v", name, action)
}
}
assertSnapshot("sender", got.SenderMessage)
assertSnapshot("recipient", got.RecipientMessage)
history, err := messages.ListByUser(ctx, recipient.ID, domain.MessageFilter{
HasPeer: true, Peer: domain.Peer{Type: domain.PeerTypeUser, ID: sender.ID}, Limit: 10,
})
if err != nil || len(history.Messages) != 1 {
t.Fatalf("recipient history=%+v err=%v", history, err)
}
assertSnapshot("recipient history", history.Messages[0])
events, err := NewUpdateEventStore(pool).ListAfter(ctx, recipient.ID, 0, 10)
if err != nil || len(events) != 1 {
t.Fatalf("recipient events=%+v err=%v", events, err)
}
assertSnapshot("recipient event", events[0].Message)
}
func TestMessageStorePhoneCallServiceFirstMessageFeedsDialogsAndUpdates(t *testing.T) {
pool := testPool(t)
ctx := context.Background()

30
internal/store/postgres/sqlcgen/models.go
View file

@ -294,21 +294,33 @@ type Bot struct {
}
type BotApiUpdate struct {
ID int64
BotUserID int64
UpdateKind string
PeerType string
PeerID int64
MessageID int32
SourcePts int32
Date int32
CreatedAt pgtype.Timestamptz
ID int64
BotUserID int64
UpdateKind string
PeerType string
PeerID int64
MessageID int32
SourcePts int32
Date int32
CreatedAt pgtype.Timestamptz
CallbackQueryID int64
CallbackUserID int64
CallbackChatInstance int64
CallbackData []byte
CallbackInlineDcID int32
CallbackInlineOwnerID int64
CallbackInlineMessageID int32
CallbackInlineAccessHash int64
}
type BotApiUpdateState struct {
BotUserID int64
ConfirmedUpdateID int64
UpdatedAt pgtype.Timestamptz
AllowedUpdates []string
CursorInitialized bool
PollOwner string
PollExpiresAt pgtype.Timestamptz
}
type BotApp struct {

150
internal/store/redisstore/bot_callback.go Normal file
View file

@ -0,0 +1,150 @@
package redisstore
import (
"context"
"encoding/json"
"fmt"
"strconv"
"time"
"github.com/redis/go-redis/v9"
"telesrv/internal/domain"
"telesrv/internal/store"
)
const botCallbackAnswerChannel = "telesrv:bot_callback:answers"
type BotCallbackRegistryStore struct {
c redis.UniversalClient
}
func NewBotCallbackRegistryStore(c redis.UniversalClient) *BotCallbackRegistryStore {
return &BotCallbackRegistryStore{c: c}
}
func botCallbackKey(queryID int64) string {
return fmt.Sprintf("telesrv:bot_callback:%d", queryID)
}
var putBotCallbackScript = redis.NewScript(`
if redis.call('EXISTS', KEYS[1]) ~= 0 then
return 0
end
redis.call('HSET', KEYS[1],
'bot_user_id', ARGV[1],
'user_id', ARGV[2],
'created_at_unix_nano', ARGV[3])
redis.call('PEXPIRE', KEYS[1], ARGV[4])
return 1
`)
func (s *BotCallbackRegistryStore) PutBotCallbackPending(ctx context.Context, pending store.BotCallbackPending, ttl time.Duration) (bool, error) {
if s == nil || s.c == nil || pending.QueryID == 0 || pending.BotUserID <= 0 || pending.UserID <= 0 || ttl <= 0 {
return false, fmt.Errorf("invalid bot callback pending")
}
createdAt := pending.CreatedAt
if createdAt.IsZero() {
createdAt = time.Now()
}
result, err := putBotCallbackScript.Run(ctx, s.c, []string{botCallbackKey(pending.QueryID)},
pending.BotUserID, pending.UserID, createdAt.UnixNano(), ttl.Milliseconds()).Int64()
if err != nil {
return false, fmt.Errorf("put bot callback pending: %w", err)
}
return result == 1, nil
}
var resolveBotCallbackScript = redis.NewScript(`
if redis.call('HGET', KEYS[1], 'bot_user_id') ~= ARGV[1] then
return 0
end
if redis.call('HEXISTS', KEYS[1], 'answer') ~= 0 then
return 0
end
redis.call('HSET', KEYS[1], 'answer', ARGV[2])
redis.call('PUBLISH', ARGV[3], ARGV[4])
return 1
`)
func (s *BotCallbackRegistryStore) ResolveBotCallback(ctx context.Context, botUserID, queryID int64, answer domain.BotCallbackAnswer) (bool, error) {
if s == nil || s.c == nil || botUserID <= 0 || queryID == 0 {
return false, nil
}
answerJSON, err := json.Marshal(answer)
if err != nil {
return false, fmt.Errorf("marshal bot callback answer: %w", err)
}
pushJSON, err := json.Marshal(store.BotCallbackAnswerPush{QueryID: queryID, BotUserID: botUserID, Answer: answer})
if err != nil {
return false, fmt.Errorf("marshal bot callback answer push: %w", err)
}
result, err := resolveBotCallbackScript.Run(ctx, s.c, []string{botCallbackKey(queryID)},
strconv.FormatInt(botUserID, 10), answerJSON, botCallbackAnswerChannel, pushJSON).Int64()
if err != nil {
return false, fmt.Errorf("resolve bot callback: %w", err)
}
return result == 1, nil
}
func (s *BotCallbackRegistryStore) GetBotCallbackAnswer(ctx context.Context, botUserID, queryID int64) (domain.BotCallbackAnswer, bool, error) {
if s == nil || s.c == nil || botUserID <= 0 || queryID == 0 {
return domain.BotCallbackAnswer{}, false, nil
}
values, err := s.c.HMGet(ctx, botCallbackKey(queryID), "bot_user_id", "answer").Result()
if err != nil {
return domain.BotCallbackAnswer{}, false, fmt.Errorf("get bot callback answer: %w", err)
}
if len(values) != 2 || values[0] == nil || values[1] == nil || fmt.Sprint(values[0]) != strconv.FormatInt(botUserID, 10) {
return domain.BotCallbackAnswer{}, false, nil
}
var answer domain.BotCallbackAnswer
if err := json.Unmarshal([]byte(fmt.Sprint(values[1])), &answer); err != nil {
return domain.BotCallbackAnswer{}, false, fmt.Errorf("decode bot callback answer: %w", err)
}
return answer, true, nil
}
var deleteBotCallbackScript = redis.NewScript(`
if redis.call('HGET', KEYS[1], 'bot_user_id') ~= ARGV[1] then
return 0
end
return redis.call('DEL', KEYS[1])
`)
func (s *BotCallbackRegistryStore) DeleteBotCallbackPending(ctx context.Context, botUserID, queryID int64) error {
if s == nil || s.c == nil || botUserID <= 0 || queryID == 0 {
return nil
}
if _, err := deleteBotCallbackScript.Run(ctx, s.c, []string{botCallbackKey(queryID)}, strconv.FormatInt(botUserID, 10)).Result(); err != nil && err != redis.Nil {
return fmt.Errorf("delete bot callback pending: %w", err)
}
return nil
}
func (s *BotCallbackRegistryStore) SubscribeBotCallbackAnswers(ctx context.Context, handle func(context.Context, store.BotCallbackAnswerPush)) error {
if s == nil || s.c == nil || handle == nil {
return nil
}
pubsub := s.c.Subscribe(ctx, botCallbackAnswerChannel)
defer pubsub.Close()
if _, err := pubsub.Receive(ctx); err != nil {
return fmt.Errorf("subscribe bot callback answers: %w", err)
}
channel := pubsub.Channel(redis.WithChannelSize(256))
for {
select {
case <-ctx.Done():
return nil
case message, ok := <-channel:
if !ok {
return nil
}
var push store.BotCallbackAnswerPush
if err := json.Unmarshal([]byte(message.Payload), &push); err != nil || push.QueryID == 0 || push.BotUserID <= 0 {
continue
}
handle(ctx, push)
}
}
}

76
internal/store/redisstore/bot_callback_integration_test.go Normal file
View file

@ -0,0 +1,76 @@
package redisstore
import (
"context"
"os"
"testing"
"time"
"telesrv/internal/domain"
"telesrv/internal/store"
)
func TestRedisBotCallbackRegistryCrossInstanceCASAndPubSub(t *testing.T) {
addr := os.Getenv("TELESRV_TEST_REDIS_ADDR")
if addr == "" {
t.Skip("set TELESRV_TEST_REDIS_ADDR to run redis integration test")
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
clientA, err := Open(ctx, addr, "", 0)
if err != nil {
t.Fatal(err)
}
defer clientA.Close()
clientB, err := Open(ctx, addr, "", 0)
if err != nil {
t.Fatal(err)
}
defer clientB.Close()
a, b := NewBotCallbackRegistryStore(clientA), NewBotCallbackRegistryStore(clientB)
queryID := time.Now().UnixNano()
defer a.DeleteBotCallbackPending(context.Background(), 1001, queryID)
pushes := make(chan store.BotCallbackAnswerPush, 1)
subscribed := make(chan struct{})
go func() {
_ = b.SubscribeBotCallbackAnswers(ctx, func(_ context.Context, push store.BotCallbackAnswerPush) {
select {
case pushes <- push:
default:
}
})
}()
// Subscribe uses Redis' acknowledgement before consuming Channel. Give that
// acknowledgement one bounded scheduling turn before publishing.
time.AfterFunc(50*time.Millisecond, func() { close(subscribed) })
<-subscribed
created, err := a.PutBotCallbackPending(ctx, store.BotCallbackPending{QueryID: queryID, BotUserID: 1001, UserID: 2001}, time.Second)
if err != nil || !created {
t.Fatalf("put created=%v err=%v", created, err)
}
if duplicate, err := b.PutBotCallbackPending(ctx, store.BotCallbackPending{QueryID: queryID, BotUserID: 1001, UserID: 2002}, time.Second); err != nil || duplicate {
t.Fatalf("duplicate=%v err=%v", duplicate, err)
}
answer := domain.BotCallbackAnswer{Message: "done", CacheTime: 3}
if resolved, err := b.ResolveBotCallback(ctx, 9999, queryID, answer); err != nil || resolved {
t.Fatalf("foreign resolve=%v err=%v", resolved, err)
}
if resolved, err := b.ResolveBotCallback(ctx, 1001, queryID, answer); err != nil || !resolved {
t.Fatalf("owner resolve=%v err=%v", resolved, err)
}
if second, err := a.ResolveBotCallback(ctx, 1001, queryID, domain.BotCallbackAnswer{Message: "second"}); err != nil || second {
t.Fatalf("second resolve=%v err=%v", second, err)
}
stored, found, err := a.GetBotCallbackAnswer(ctx, 1001, queryID)
if err != nil || !found || stored.Message != "done" {
t.Fatalf("stored=%#v found=%v err=%v", stored, found, err)
}
select {
case push := <-pushes:
if push.QueryID != queryID || push.BotUserID != 1001 || push.Answer.Message != "done" {
t.Fatalf("push=%#v", push)
}
case <-ctx.Done():
t.Fatal("missing cross-instance callback pubsub")
}
}