feat: sync ephemeral transient messages
Sync telesrv 570ccf8 (feat(ephemeral): implement Layer 228 transient messages). Skipped telesrv docs changes per public sync rules; normalized the public appearance seed label.
This commit is contained in:
A
parent
3f78eaa2c6
commit
f49c817def
53 changed files with 5793 additions and 112 deletions
54
internal/store/ephemeral.go
Normal file
54
internal/store/ephemeral.go
Normal file
|
|
@ -0,0 +1,54 @@
|
|||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
||||
// EphemeralMessageStore is the short-lived authoritative state used for
|
||||
// idempotency, callback, edit, delete and report lookups. Implementations must
|
||||
// make Create atomic across the message ID and random-ID indexes.
|
||||
type EphemeralMessageStore interface {
|
||||
CreateEphemeralMessage(ctx context.Context, message domain.EphemeralMessage) (stored domain.EphemeralMessage, created bool, err error)
|
||||
GetEphemeralMessage(ctx context.Context, peer domain.Peer, id int, now time.Time) (domain.EphemeralMessage, bool, error)
|
||||
EditEphemeralMessage(ctx context.Context, peer domain.Peer, id int, expectedVersion uint64, content domain.EphemeralContent, editDate int, now time.Time) (domain.EphemeralMessage, error)
|
||||
DeleteEphemeralMessage(ctx context.Context, peer domain.Peer, id int, expectedVersion uint64, now time.Time) (domain.EphemeralMessage, bool, error)
|
||||
PruneExpiredEphemeralMessages(ctx context.Context, now time.Time, limit int) (int, error)
|
||||
PutEphemeralCallbackAction(ctx context.Context, action domain.EphemeralCallbackAction) (bool, error)
|
||||
GetEphemeralCallbackAction(ctx context.Context, botUserID, queryID int64, now time.Time) (domain.EphemeralCallbackAction, bool, error)
|
||||
}
|
||||
|
||||
// EphemeralReportStore is deliberately durable: transient messages disappear
|
||||
// after 48 hours, while a submitted abuse report must retain review evidence.
|
||||
type EphemeralReportStore interface {
|
||||
CreateEphemeralReport(ctx context.Context, report domain.EphemeralAbuseReport) (created bool, err error)
|
||||
}
|
||||
|
||||
type EphemeralPushKind string
|
||||
|
||||
const (
|
||||
EphemeralPushNew EphemeralPushKind = "new"
|
||||
EphemeralPushEdit EphemeralPushKind = "edit"
|
||||
EphemeralPushDelete EphemeralPushKind = "delete"
|
||||
EphemeralPushCallback EphemeralPushKind = "callback"
|
||||
)
|
||||
|
||||
// EphemeralPush is a process-to-process online accelerator. It is deliberately
|
||||
// not a durable event: Redis Pub/Sub and ready Layer 228 sessions are the only
|
||||
// consumers, while EphemeralMessageStore remains the short-lived lookup truth.
|
||||
type EphemeralPush struct {
|
||||
SourceID string
|
||||
Kind EphemeralPushKind
|
||||
TargetUserID int64
|
||||
TargetBusinessAuthKey [8]byte
|
||||
Message domain.EphemeralMessage
|
||||
Callback *domain.BotCallbackQuery
|
||||
Date int
|
||||
}
|
||||
|
||||
type EphemeralPushBroker interface {
|
||||
PublishEphemeralPush(ctx context.Context, event EphemeralPush) error
|
||||
SubscribeEphemeralPushes(ctx context.Context, handle func(context.Context, EphemeralPush)) error
|
||||
}
|
||||
|
|
@ -253,6 +253,7 @@ func (s *BotAPIUpdateStore) EnqueueBotAPIUpdate(_ context.Context, req domain.En
|
|||
SourcePts: req.SourcePts,
|
||||
Date: req.Date,
|
||||
Callback: cloneBotAPICallback(req.Callback),
|
||||
Ephemeral: cloneBotAPIEphemeral(req.Ephemeral),
|
||||
}
|
||||
s.nextID++
|
||||
s.rows = append(s.rows, row)
|
||||
|
|
@ -433,6 +434,17 @@ func validateBotAPIUpdateRequest(req domain.EnqueueBotAPIUpdateRequest) error {
|
|||
default:
|
||||
return fmt.Errorf("invalid bot api update peer type %q", req.Peer.Type)
|
||||
}
|
||||
if req.Ephemeral != nil {
|
||||
message := req.Ephemeral.Message
|
||||
if req.Ephemeral.Validate() != nil || message.ID != req.MessageID || message.Peer != req.Peer || message.Expired(time.Unix(int64(req.Date), 0)) ||
|
||||
req.Peer.Type != domain.PeerTypeChannel || req.SourcePts != 0 {
|
||||
return fmt.Errorf("invalid bot api ephemeral update")
|
||||
}
|
||||
if (req.Kind == domain.BotAPIUpdateCallbackQuery && message.SenderUserID != req.BotUserID) ||
|
||||
(req.Kind != domain.BotAPIUpdateCallbackQuery && message.ReceiverUserID != req.BotUserID) {
|
||||
return fmt.Errorf("invalid bot api ephemeral target")
|
||||
}
|
||||
}
|
||||
if req.Kind == domain.BotAPIUpdateCallbackQuery {
|
||||
cb := req.Callback
|
||||
if cb == nil || cb.ID == 0 || cb.BotUserID != req.BotUserID || cb.UserID <= 0 ||
|
||||
|
|
@ -457,14 +469,25 @@ 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)
|
||||
}
|
||||
if req.Ephemeral != nil {
|
||||
return fmt.Sprintf("%d:%s:ephemeral:%s:%d:%d:%d", req.BotUserID, req.Kind, req.Peer.Type, req.Peer.ID, req.MessageID, req.Ephemeral.Message.Version)
|
||||
}
|
||||
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)
|
||||
row.Ephemeral = cloneBotAPIEphemeral(row.Ephemeral)
|
||||
return row
|
||||
}
|
||||
|
||||
func cloneBotAPIEphemeral(in *domain.BotAPIEphemeralPayload) *domain.BotAPIEphemeralPayload {
|
||||
if in == nil {
|
||||
return nil
|
||||
}
|
||||
return domain.NewBotAPIEphemeralPayload(cloneEphemeralMessage(in.EphemeralMessage()))
|
||||
}
|
||||
|
||||
func cloneBotAPICallback(in *domain.BotCallbackQuery) *domain.BotCallbackQuery {
|
||||
if in == nil {
|
||||
return nil
|
||||
|
|
|
|||
|
|
@ -189,3 +189,57 @@ func TestBotAPIInlineCallbackRoundTrip(t *testing.T) {
|
|||
t.Fatalf("inline callback rows=%#v err=%v", rows, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBotAPIEphemeralMessageVersionsAndCallbackRoundTrip(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := NewBotAPIUpdateStore()
|
||||
now := time.Now()
|
||||
peer := domain.Peer{Type: domain.PeerTypeChannel, ID: 3001}
|
||||
incoming := domain.EphemeralMessage{
|
||||
ID: 71, Peer: peer, SenderUserID: 2001, ReceiverUserID: 1001,
|
||||
Date: int(now.Unix()), RandomID: 1, Content: domain.EphemeralContent{Message: "/private"},
|
||||
Version: 1, CreatedAt: now, ExpiresAt: now.Add(domain.EphemeralMessageRetention),
|
||||
}
|
||||
request := domain.EnqueueBotAPIUpdateRequest{
|
||||
BotUserID: 1001, Kind: domain.BotAPIUpdateMessage, Peer: peer,
|
||||
MessageID: incoming.ID, Date: incoming.Date,
|
||||
Ephemeral: domain.NewBotAPIEphemeralPayload(incoming),
|
||||
}
|
||||
first, created, err := store.EnqueueBotAPIUpdate(ctx, request)
|
||||
if err != nil || !created || first.SourcePts != 0 || first.Ephemeral == nil {
|
||||
t.Fatalf("first=%+v created=%v err=%v", first, created, err)
|
||||
}
|
||||
if replay, created, err := store.EnqueueBotAPIUpdate(ctx, request); err != nil || created || replay.ID != first.ID {
|
||||
t.Fatalf("replay=%+v created=%v err=%v", replay, created, err)
|
||||
}
|
||||
incoming.Version = 2
|
||||
incoming.EditDate = incoming.Date + 1
|
||||
incoming.Content.Message = "edited"
|
||||
request.Kind = domain.BotAPIUpdateEditedMessage
|
||||
request.Ephemeral = domain.NewBotAPIEphemeralPayload(incoming)
|
||||
edited, created, err := store.EnqueueBotAPIUpdate(ctx, request)
|
||||
if err != nil || !created || edited.ID <= first.ID {
|
||||
t.Fatalf("edited=%+v created=%v err=%v", edited, created, err)
|
||||
}
|
||||
|
||||
outgoing := incoming
|
||||
outgoing.ID, outgoing.SenderUserID, outgoing.ReceiverUserID = 72, 1001, 2001
|
||||
outgoing.Version, outgoing.Content.Message = 1, "button"
|
||||
callback := &domain.BotCallbackQuery{
|
||||
ID: 9001, BotUserID: 1001, UserID: 2001, Peer: peer,
|
||||
MessageID: outgoing.ID, ChatInstance: 901, Data: []byte("tap"),
|
||||
}
|
||||
callbackRow, created, err := store.EnqueueBotAPIUpdate(ctx, domain.EnqueueBotAPIUpdateRequest{
|
||||
BotUserID: 1001, Kind: domain.BotAPIUpdateCallbackQuery, Peer: peer,
|
||||
MessageID: outgoing.ID, Date: outgoing.Date, Callback: callback,
|
||||
Ephemeral: domain.NewBotAPIEphemeralPayload(outgoing),
|
||||
})
|
||||
if err != nil || !created || callbackRow.Callback == nil || callbackRow.Ephemeral == nil {
|
||||
t.Fatalf("callback=%+v created=%v err=%v", callbackRow, created, err)
|
||||
}
|
||||
rows, err := store.ListBotAPIUpdates(ctx, 1001, first.ID, 100)
|
||||
if err != nil || len(rows) != 3 || rows[0].Ephemeral.Message.Content.Message != "/private" ||
|
||||
rows[1].Ephemeral.Message.Content.Message != "edited" || string(rows[2].Callback.Data) != "tap" {
|
||||
t.Fatalf("rows=%+v err=%v", rows, err)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
380
internal/store/memory/ephemeral.go
Normal file
380
internal/store/memory/ephemeral.go
Normal file
|
|
@ -0,0 +1,380 @@
|
|||
package memory
|
||||
|
||||
import (
|
||||
"container/heap"
|
||||
"context"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
||||
const ephemeralShardCount = 64
|
||||
|
||||
type ephemeralMessageKey struct {
|
||||
peerType domain.PeerType
|
||||
peerID int64
|
||||
id int
|
||||
}
|
||||
|
||||
type ephemeralRandomKey struct {
|
||||
peerType domain.PeerType
|
||||
peerID int64
|
||||
senderID int64
|
||||
receiverID int64
|
||||
randomID int64
|
||||
}
|
||||
|
||||
type ephemeralEntry struct {
|
||||
message domain.EphemeralMessage
|
||||
generation uint64
|
||||
}
|
||||
|
||||
type ephemeralExpiry struct {
|
||||
key ephemeralMessageKey
|
||||
expiresAt int64
|
||||
generation uint64
|
||||
}
|
||||
|
||||
type ephemeralExpiryHeap []ephemeralExpiry
|
||||
|
||||
func (h ephemeralExpiryHeap) Len() int { return len(h) }
|
||||
func (h ephemeralExpiryHeap) Less(i, j int) bool { return h[i].expiresAt < h[j].expiresAt }
|
||||
func (h ephemeralExpiryHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
|
||||
|
||||
func (h *ephemeralExpiryHeap) Push(value any) {
|
||||
*h = append(*h, value.(ephemeralExpiry))
|
||||
}
|
||||
|
||||
func (h *ephemeralExpiryHeap) Pop() any {
|
||||
old := *h
|
||||
n := len(old)
|
||||
value := old[n-1]
|
||||
old[n-1] = ephemeralExpiry{}
|
||||
*h = old[:n-1]
|
||||
return value
|
||||
}
|
||||
|
||||
type ephemeralShard struct {
|
||||
mu sync.RWMutex
|
||||
messages map[ephemeralMessageKey]ephemeralEntry
|
||||
random map[ephemeralRandomKey]ephemeralMessageKey
|
||||
expiry ephemeralExpiryHeap
|
||||
nextGeneration uint64
|
||||
}
|
||||
|
||||
type ephemeralCallbackActionShard struct {
|
||||
mu sync.RWMutex
|
||||
actions map[int64]ephemeralCallbackActionEntry
|
||||
expiry ephemeralCallbackExpiryHeap
|
||||
nextGeneration uint64
|
||||
}
|
||||
|
||||
type ephemeralCallbackActionEntry struct {
|
||||
action domain.EphemeralCallbackAction
|
||||
generation uint64
|
||||
}
|
||||
|
||||
type ephemeralCallbackExpiry struct {
|
||||
queryID int64
|
||||
expiresAt int64
|
||||
generation uint64
|
||||
}
|
||||
|
||||
type ephemeralCallbackExpiryHeap []ephemeralCallbackExpiry
|
||||
|
||||
func (h ephemeralCallbackExpiryHeap) Len() int { return len(h) }
|
||||
func (h ephemeralCallbackExpiryHeap) Less(i, j int) bool { return h[i].expiresAt < h[j].expiresAt }
|
||||
func (h ephemeralCallbackExpiryHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
|
||||
|
||||
func (h *ephemeralCallbackExpiryHeap) Push(value any) {
|
||||
*h = append(*h, value.(ephemeralCallbackExpiry))
|
||||
}
|
||||
|
||||
func (h *ephemeralCallbackExpiryHeap) Pop() any {
|
||||
old := *h
|
||||
n := len(old)
|
||||
value := old[n-1]
|
||||
old[n-1] = ephemeralCallbackExpiry{}
|
||||
*h = old[:n-1]
|
||||
return value
|
||||
}
|
||||
|
||||
// EphemeralMessageStore shards by peer. A create touches one shard, so the ID
|
||||
// and random-ID indexes can be updated atomically without a process-wide lock.
|
||||
type EphemeralMessageStore struct {
|
||||
shards [ephemeralShardCount]ephemeralShard
|
||||
callbackActions [ephemeralShardCount]ephemeralCallbackActionShard
|
||||
messageCursor atomic.Uint32
|
||||
callbackCursor atomic.Uint32
|
||||
}
|
||||
|
||||
func NewEphemeralMessageStore() *EphemeralMessageStore {
|
||||
s := &EphemeralMessageStore{}
|
||||
for i := range s.shards {
|
||||
s.shards[i].messages = make(map[ephemeralMessageKey]ephemeralEntry)
|
||||
s.shards[i].random = make(map[ephemeralRandomKey]ephemeralMessageKey)
|
||||
s.callbackActions[i].actions = make(map[int64]ephemeralCallbackActionEntry)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) PutEphemeralCallbackAction(_ context.Context, action domain.EphemeralCallbackAction) (bool, error) {
|
||||
if action.QueryID == 0 || action.BotUserID <= 0 || action.UserID <= 0 || action.Peer.Type != domain.PeerTypeChannel ||
|
||||
action.Peer.ID <= 0 || action.MessageID <= 0 || action.Device.UserID != action.UserID ||
|
||||
action.Device.BusinessAuthKeyID == ([8]byte{}) || action.CreatedAt.IsZero() || !action.ExpiresAt.After(action.CreatedAt) ||
|
||||
action.ExpiresAt.Sub(action.CreatedAt) > domain.EphemeralReplyWindow {
|
||||
return false, domain.ErrEphemeralInvalid
|
||||
}
|
||||
shard := &s.callbackActions[uint64(action.QueryID)&(ephemeralShardCount-1)]
|
||||
shard.mu.Lock()
|
||||
defer shard.mu.Unlock()
|
||||
if existing, ok := shard.actions[action.QueryID]; ok && action.CreatedAt.Before(existing.action.ExpiresAt) {
|
||||
return false, nil
|
||||
}
|
||||
shard.nextGeneration++
|
||||
entry := ephemeralCallbackActionEntry{action: action, generation: shard.nextGeneration}
|
||||
shard.actions[action.QueryID] = entry
|
||||
heap.Push(&shard.expiry, ephemeralCallbackExpiry{
|
||||
queryID: action.QueryID, expiresAt: action.ExpiresAt.UnixNano(), generation: entry.generation,
|
||||
})
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) GetEphemeralCallbackAction(_ context.Context, botUserID, queryID int64, now time.Time) (domain.EphemeralCallbackAction, bool, error) {
|
||||
if botUserID <= 0 || queryID == 0 {
|
||||
return domain.EphemeralCallbackAction{}, false, nil
|
||||
}
|
||||
shard := &s.callbackActions[uint64(queryID)&(ephemeralShardCount-1)]
|
||||
shard.mu.RLock()
|
||||
entry, ok := shard.actions[queryID]
|
||||
if ok && entry.action.BotUserID == botUserID && now.Before(entry.action.ExpiresAt) {
|
||||
shard.mu.RUnlock()
|
||||
return entry.action, true, nil
|
||||
}
|
||||
shard.mu.RUnlock()
|
||||
if !ok || entry.action.BotUserID != botUserID {
|
||||
return domain.EphemeralCallbackAction{}, false, nil
|
||||
}
|
||||
shard.mu.Lock()
|
||||
if current, exists := shard.actions[queryID]; exists && !now.Before(current.action.ExpiresAt) {
|
||||
delete(shard.actions, queryID)
|
||||
}
|
||||
shard.mu.Unlock()
|
||||
return domain.EphemeralCallbackAction{}, false, nil
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) CreateEphemeralMessage(_ context.Context, message domain.EphemeralMessage) (domain.EphemeralMessage, bool, error) {
|
||||
now := message.CreatedAt
|
||||
if err := message.ValidateForCreate(now); err != nil {
|
||||
return domain.EphemeralMessage{}, false, err
|
||||
}
|
||||
shard := s.shard(message.Peer)
|
||||
messageKey := ephemeralKey(message.Peer, message.ID)
|
||||
randomKey := ephemeralRandom(message)
|
||||
shard.mu.Lock()
|
||||
defer shard.mu.Unlock()
|
||||
|
||||
if existingKey, ok := shard.random[randomKey]; ok {
|
||||
if existing, found := shard.messages[existingKey]; found && !existing.message.Expired(now) {
|
||||
if existing.message.PayloadHash != message.PayloadHash {
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralRandomIDConflict
|
||||
}
|
||||
return cloneEphemeralMessage(existing.message), false, nil
|
||||
}
|
||||
delete(shard.random, randomKey)
|
||||
delete(shard.messages, existingKey)
|
||||
}
|
||||
if existing, ok := shard.messages[messageKey]; ok {
|
||||
if !existing.message.Expired(now) {
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralIDCollision
|
||||
}
|
||||
delete(shard.random, ephemeralRandom(existing.message))
|
||||
delete(shard.messages, messageKey)
|
||||
}
|
||||
stored := cloneEphemeralMessage(message)
|
||||
stored.BotAPIReply = nil
|
||||
shard.nextGeneration++
|
||||
entry := ephemeralEntry{message: stored, generation: shard.nextGeneration}
|
||||
shard.messages[messageKey] = entry
|
||||
shard.random[randomKey] = messageKey
|
||||
heap.Push(&shard.expiry, ephemeralExpiry{
|
||||
key: messageKey,
|
||||
expiresAt: stored.ExpiresAt.UnixNano(),
|
||||
generation: entry.generation,
|
||||
})
|
||||
return cloneEphemeralMessage(stored), true, nil
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) GetEphemeralMessage(_ context.Context, peer domain.Peer, id int, now time.Time) (domain.EphemeralMessage, bool, error) {
|
||||
key := ephemeralKey(peer, id)
|
||||
shard := s.shard(peer)
|
||||
shard.mu.RLock()
|
||||
entry, ok := shard.messages[key]
|
||||
if ok && !entry.message.Expired(now) {
|
||||
message := cloneEphemeralMessage(entry.message)
|
||||
shard.mu.RUnlock()
|
||||
return message, true, nil
|
||||
}
|
||||
shard.mu.RUnlock()
|
||||
if !ok {
|
||||
return domain.EphemeralMessage{}, false, nil
|
||||
}
|
||||
shard.mu.Lock()
|
||||
if entry, ok = shard.messages[key]; ok && entry.message.Expired(now) {
|
||||
delete(shard.messages, key)
|
||||
delete(shard.random, ephemeralRandom(entry.message))
|
||||
}
|
||||
shard.mu.Unlock()
|
||||
return domain.EphemeralMessage{}, false, nil
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) EditEphemeralMessage(_ context.Context, peer domain.Peer, id int, expectedVersion uint64, content domain.EphemeralContent, editDate int, now time.Time) (domain.EphemeralMessage, error) {
|
||||
key := ephemeralKey(peer, id)
|
||||
shard := s.shard(peer)
|
||||
shard.mu.Lock()
|
||||
defer shard.mu.Unlock()
|
||||
entry, ok := shard.messages[key]
|
||||
if !ok {
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralNotFound
|
||||
}
|
||||
if entry.message.Expired(now) {
|
||||
delete(shard.messages, key)
|
||||
delete(shard.random, ephemeralRandom(entry.message))
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralExpired
|
||||
}
|
||||
if entry.message.Deleted {
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralDeleted
|
||||
}
|
||||
if expectedVersion == 0 || entry.message.Version != expectedVersion {
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralVersionConflict
|
||||
}
|
||||
if domain.ValidateEphemeralContent(content) != nil {
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralInvalid
|
||||
}
|
||||
entry.message.Content = cloneEphemeralContent(content)
|
||||
entry.message.EditDate = editDate
|
||||
entry.message.Version++
|
||||
shard.messages[key] = entry
|
||||
return cloneEphemeralMessage(entry.message), nil
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) DeleteEphemeralMessage(_ context.Context, peer domain.Peer, id int, expectedVersion uint64, now time.Time) (domain.EphemeralMessage, bool, error) {
|
||||
key := ephemeralKey(peer, id)
|
||||
shard := s.shard(peer)
|
||||
shard.mu.Lock()
|
||||
defer shard.mu.Unlock()
|
||||
entry, ok := shard.messages[key]
|
||||
if !ok {
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralNotFound
|
||||
}
|
||||
if entry.message.Expired(now) {
|
||||
delete(shard.messages, key)
|
||||
delete(shard.random, ephemeralRandom(entry.message))
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralExpired
|
||||
}
|
||||
if entry.message.Deleted {
|
||||
return cloneEphemeralMessage(entry.message), false, nil
|
||||
}
|
||||
if expectedVersion == 0 || entry.message.Version != expectedVersion {
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralVersionConflict
|
||||
}
|
||||
entry.message.Deleted = true
|
||||
entry.message.Version++
|
||||
// Keep a small tombstone until the original TTL. It prevents a delayed
|
||||
// random-id retry from resurrecting a message after delete.
|
||||
entry.message.Content = domain.EphemeralContent{}
|
||||
shard.messages[key] = entry
|
||||
return cloneEphemeralMessage(entry.message), true, nil
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) PruneExpiredEphemeralMessages(_ context.Context, now time.Time, limit int) (int, error) {
|
||||
if limit <= 0 {
|
||||
return 0, nil
|
||||
}
|
||||
deleted := 0
|
||||
nowUnixNano := now.UnixNano()
|
||||
start := int(s.messageCursor.Add(1)-1) & (ephemeralShardCount - 1)
|
||||
for offset := range ephemeralShardCount {
|
||||
shard := &s.shards[(start+offset)&(ephemeralShardCount-1)]
|
||||
shard.mu.Lock()
|
||||
for deleted < limit && shard.expiry.Len() > 0 && shard.expiry[0].expiresAt <= nowUnixNano {
|
||||
expiry := heap.Pop(&shard.expiry).(ephemeralExpiry)
|
||||
entry, ok := shard.messages[expiry.key]
|
||||
if !ok || entry.generation != expiry.generation {
|
||||
continue
|
||||
}
|
||||
delete(shard.messages, expiry.key)
|
||||
delete(shard.random, ephemeralRandom(entry.message))
|
||||
deleted++
|
||||
}
|
||||
shard.mu.Unlock()
|
||||
if deleted >= limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
// Callback authorizations have an independent 15-second TTL. Give their
|
||||
// heap an independent bounded budget so a hot message shard cannot starve
|
||||
// callback cleanup and cause an in-memory deployment to grow forever.
|
||||
callbackDeleted := 0
|
||||
callbackStart := int(s.callbackCursor.Add(1)-1) & (ephemeralShardCount - 1)
|
||||
for offset := range ephemeralShardCount {
|
||||
shard := &s.callbackActions[(callbackStart+offset)&(ephemeralShardCount-1)]
|
||||
shard.mu.Lock()
|
||||
for callbackDeleted < limit && shard.expiry.Len() > 0 && shard.expiry[0].expiresAt <= nowUnixNano {
|
||||
expiry := heap.Pop(&shard.expiry).(ephemeralCallbackExpiry)
|
||||
entry, ok := shard.actions[expiry.queryID]
|
||||
if !ok || entry.generation != expiry.generation {
|
||||
continue
|
||||
}
|
||||
delete(shard.actions, expiry.queryID)
|
||||
callbackDeleted++
|
||||
}
|
||||
shard.mu.Unlock()
|
||||
if callbackDeleted >= limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) shard(peer domain.Peer) *ephemeralShard {
|
||||
// Peer IDs are already uniformly allocated monotonically; multiplicative
|
||||
// mixing avoids adjacent hot groups concentrating in neighboring low bits.
|
||||
index := (uint64(peer.ID) * 11400714819323198485) >> (64 - 6)
|
||||
return &s.shards[index]
|
||||
}
|
||||
|
||||
func ephemeralKey(peer domain.Peer, id int) ephemeralMessageKey {
|
||||
return ephemeralMessageKey{peerType: peer.Type, peerID: peer.ID, id: id}
|
||||
}
|
||||
|
||||
func ephemeralRandom(message domain.EphemeralMessage) ephemeralRandomKey {
|
||||
return ephemeralRandomKey{
|
||||
peerType: message.Peer.Type,
|
||||
peerID: message.Peer.ID,
|
||||
senderID: message.SenderUserID,
|
||||
receiverID: message.ReceiverUserID,
|
||||
randomID: message.RandomID,
|
||||
}
|
||||
}
|
||||
|
||||
func cloneEphemeralMessage(message domain.EphemeralMessage) domain.EphemeralMessage {
|
||||
message.Content = cloneEphemeralContent(message.Content)
|
||||
if message.BotAPIReply != nil {
|
||||
reply := *message.BotAPIReply
|
||||
reply.Content = cloneEphemeralContent(reply.Content)
|
||||
reply.BotAPIReply = nil
|
||||
message.BotAPIReply = &reply
|
||||
}
|
||||
return message
|
||||
}
|
||||
|
||||
func cloneEphemeralContent(content domain.EphemeralContent) domain.EphemeralContent {
|
||||
content.Entities = append([]domain.MessageEntity(nil), content.Entities...)
|
||||
content.Media = cloneRequestedPeerMedia(content.Media)
|
||||
content.ReplyMarkup = cloneReplyMarkup(content.ReplyMarkup)
|
||||
content.RichMessage = cloneRichMessage(content.RichMessage)
|
||||
return content
|
||||
}
|
||||
53
internal/store/memory/ephemeral_report.go
Normal file
53
internal/store/memory/ephemeral_report.go
Normal file
|
|
@ -0,0 +1,53 @@
|
|||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
||||
type ephemeralReportKey struct {
|
||||
reporterUserID int64
|
||||
channelID int64
|
||||
messageID int
|
||||
option string
|
||||
commentHash [32]byte
|
||||
}
|
||||
|
||||
// EphemeralReportStore is the deterministic in-memory test implementation.
|
||||
type EphemeralReportStore struct {
|
||||
mu sync.Mutex
|
||||
reports map[ephemeralReportKey]domain.EphemeralAbuseReport
|
||||
}
|
||||
|
||||
func NewEphemeralReportStore() *EphemeralReportStore {
|
||||
return &EphemeralReportStore{reports: make(map[ephemeralReportKey]domain.EphemeralAbuseReport)}
|
||||
}
|
||||
|
||||
func (s *EphemeralReportStore) CreateEphemeralReport(_ context.Context, report domain.EphemeralAbuseReport) (bool, error) {
|
||||
if err := report.Validate(); err != nil {
|
||||
return false, err
|
||||
}
|
||||
key := ephemeralReportKey{
|
||||
reporterUserID: report.ReporterUserID, channelID: report.Evidence.Peer.ID,
|
||||
messageID: report.Evidence.MessageID, option: report.Option, commentHash: report.CommentHash,
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if _, exists := s.reports[key]; exists {
|
||||
return false, nil
|
||||
}
|
||||
s.reports[key] = report
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (s *EphemeralReportStore) Reports() []domain.EphemeralAbuseReport {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
out := make([]domain.EphemeralAbuseReport, 0, len(s.reports))
|
||||
for _, report := range s.reports {
|
||||
out = append(out, report)
|
||||
}
|
||||
return out
|
||||
}
|
||||
184
internal/store/memory/ephemeral_test.go
Normal file
184
internal/store/memory/ephemeral_test.go
Normal file
|
|
@ -0,0 +1,184 @@
|
|||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
||||
func TestEphemeralMessageStoreCreateReplayEditDeleteAndExpiry(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := NewEphemeralMessageStore()
|
||||
now := time.Unix(1_800_000_000, 0)
|
||||
message := testEphemeralMessage(now)
|
||||
created, fresh, err := store.CreateEphemeralMessage(ctx, message)
|
||||
if err != nil || !fresh || created.ID != message.ID {
|
||||
t.Fatalf("create = %+v fresh=%v err=%v", created, fresh, err)
|
||||
}
|
||||
|
||||
replayed, fresh, err := store.CreateEphemeralMessage(ctx, message)
|
||||
if err != nil || fresh || replayed.Version != 1 {
|
||||
t.Fatalf("replay = %+v fresh=%v err=%v", replayed, fresh, err)
|
||||
}
|
||||
conflict := message
|
||||
conflict.ID++
|
||||
conflict.PayloadHash = sha256.Sum256([]byte("different"))
|
||||
if _, _, err := store.CreateEphemeralMessage(ctx, conflict); !errors.Is(err, domain.ErrEphemeralRandomIDConflict) {
|
||||
t.Fatalf("random-id conflict err=%v", err)
|
||||
}
|
||||
|
||||
edited, err := store.EditEphemeralMessage(ctx, message.Peer, message.ID, 1, domain.EphemeralContent{Message: "edited"}, int(now.Unix())+1, now)
|
||||
if err != nil || edited.Version != 2 || edited.Content.Message != "edited" {
|
||||
t.Fatalf("edit = %+v err=%v", edited, err)
|
||||
}
|
||||
if _, err := store.EditEphemeralMessage(ctx, message.Peer, message.ID, 1, domain.EphemeralContent{Message: "stale"}, int(now.Unix())+2, now); !errors.Is(err, domain.ErrEphemeralVersionConflict) {
|
||||
t.Fatalf("stale edit err=%v", err)
|
||||
}
|
||||
|
||||
deleted, changed, err := store.DeleteEphemeralMessage(ctx, message.Peer, message.ID, 2, now)
|
||||
if err != nil || !changed || !deleted.Deleted || deleted.Version != 3 || deleted.Content.Message != "" {
|
||||
t.Fatalf("delete = %+v changed=%v err=%v", deleted, changed, err)
|
||||
}
|
||||
deleted, changed, err = store.DeleteEphemeralMessage(ctx, message.Peer, message.ID, 3, now)
|
||||
if err != nil || changed || !deleted.Deleted {
|
||||
t.Fatalf("repeat delete = %+v changed=%v err=%v", deleted, changed, err)
|
||||
}
|
||||
if _, err := store.EditEphemeralMessage(ctx, message.Peer, message.ID, 3, domain.EphemeralContent{Message: "resurrect"}, int(now.Unix())+3, now); !errors.Is(err, domain.ErrEphemeralDeleted) {
|
||||
t.Fatalf("edit deleted err=%v", err)
|
||||
}
|
||||
|
||||
if _, found, err := store.GetEphemeralMessage(ctx, message.Peer, message.ID, message.ExpiresAt); err != nil || found {
|
||||
t.Fatalf("expired found=%v err=%v", found, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEphemeralMessageStoreIDCollisionAndBoundedPrune(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := NewEphemeralMessageStore()
|
||||
now := time.Unix(1_800_000_100, 0)
|
||||
first := testEphemeralMessage(now)
|
||||
if _, _, err := store.CreateEphemeralMessage(ctx, first); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
second := first
|
||||
second.RandomID++
|
||||
second.PayloadHash = sha256.Sum256([]byte("second"))
|
||||
if _, _, err := store.CreateEphemeralMessage(ctx, second); !errors.Is(err, domain.ErrEphemeralIDCollision) {
|
||||
t.Fatalf("id collision err=%v", err)
|
||||
}
|
||||
if got, err := store.PruneExpiredEphemeralMessages(ctx, first.ExpiresAt, 1); err != nil || got != 1 {
|
||||
t.Fatalf("prune=%d err=%v", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEphemeralCallbackActionExactBotAndExpiry(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := NewEphemeralMessageStore()
|
||||
now := time.Unix(1_800_000_000, 0)
|
||||
action := domain.EphemeralCallbackAction{
|
||||
QueryID: 81, BotUserID: 2001, UserID: 3001,
|
||||
Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: 1001}, MessageID: 17, TopMessageID: 42,
|
||||
Device: domain.EphemeralDevice{UserID: 3001, BusinessAuthKeyID: [8]byte{1}, SessionID: 9},
|
||||
CreatedAt: now, ExpiresAt: now.Add(domain.EphemeralReplyWindow),
|
||||
}
|
||||
if created, err := store.PutEphemeralCallbackAction(ctx, action); err != nil || !created {
|
||||
t.Fatalf("put created=%v err=%v", created, err)
|
||||
}
|
||||
if created, err := store.PutEphemeralCallbackAction(ctx, action); err != nil || created {
|
||||
t.Fatalf("duplicate created=%v err=%v", created, err)
|
||||
}
|
||||
if _, found, err := store.GetEphemeralCallbackAction(ctx, action.BotUserID+1, action.QueryID, now); err != nil || found {
|
||||
t.Fatalf("wrong bot found=%v err=%v", found, err)
|
||||
}
|
||||
got, found, err := store.GetEphemeralCallbackAction(ctx, action.BotUserID, action.QueryID, now)
|
||||
if err != nil || !found || got.TopMessageID != 42 {
|
||||
t.Fatalf("get=%+v found=%v err=%v", got, found, err)
|
||||
}
|
||||
if _, found, err := store.GetEphemeralCallbackAction(ctx, action.BotUserID, action.QueryID, action.ExpiresAt); err != nil || found {
|
||||
t.Fatalf("expired found=%v err=%v", found, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEphemeralCallbackActionBoundedHeapPrune(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := NewEphemeralMessageStore()
|
||||
now := time.Unix(1_800_000_000, 0)
|
||||
action := domain.EphemeralCallbackAction{
|
||||
QueryID: 82, BotUserID: 2001, UserID: 3001,
|
||||
Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: 1001}, MessageID: 17,
|
||||
Device: domain.EphemeralDevice{UserID: 3001, BusinessAuthKeyID: [8]byte{1}, SessionID: 9},
|
||||
CreatedAt: now, ExpiresAt: now.Add(domain.EphemeralReplyWindow),
|
||||
}
|
||||
if created, err := store.PutEphemeralCallbackAction(ctx, action); err != nil || !created {
|
||||
t.Fatalf("put created=%v err=%v", created, err)
|
||||
}
|
||||
if _, err := store.PruneExpiredEphemeralMessages(ctx, action.ExpiresAt, 1); err != nil {
|
||||
t.Fatalf("prune err=%v", err)
|
||||
}
|
||||
shard := &store.callbackActions[uint64(action.QueryID)&(ephemeralShardCount-1)]
|
||||
shard.mu.RLock()
|
||||
_, found := shard.actions[action.QueryID]
|
||||
shard.mu.RUnlock()
|
||||
if found {
|
||||
t.Fatal("expired callback action survived bounded heap prune")
|
||||
}
|
||||
}
|
||||
|
||||
func TestEphemeralReportStoreIdempotency(t *testing.T) {
|
||||
store := NewEphemeralReportStore()
|
||||
now := time.Unix(1_800_000_000, 0)
|
||||
message := testEphemeralMessage(now)
|
||||
message.ReceiverUserID = 3001
|
||||
report := domain.NewEphemeralAbuseReport(message.ReceiverUserID, "spam", "evidence", message, now)
|
||||
if created, err := store.CreateEphemeralReport(context.Background(), report); err != nil || !created {
|
||||
t.Fatalf("create=%v err=%v", created, err)
|
||||
}
|
||||
if created, err := store.CreateEphemeralReport(context.Background(), report); err != nil || created {
|
||||
t.Fatalf("retry create=%v err=%v", created, err)
|
||||
}
|
||||
reports := store.Reports()
|
||||
if len(reports) != 1 || reports[0].Evidence.Content.Message != message.Content.Message {
|
||||
t.Fatalf("reports=%+v", reports)
|
||||
}
|
||||
}
|
||||
|
||||
func testEphemeralMessage(now time.Time) domain.EphemeralMessage {
|
||||
return domain.EphemeralMessage{
|
||||
ID: 17,
|
||||
Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: 1001},
|
||||
SenderUserID: 2001,
|
||||
ReceiverUserID: 3001,
|
||||
Date: int(now.Unix()),
|
||||
RandomID: 99,
|
||||
Content: domain.EphemeralContent{Message: "/private"},
|
||||
PayloadHash: sha256.Sum256([]byte("payload")),
|
||||
Version: 1,
|
||||
CreatedAt: now,
|
||||
ExpiresAt: now.Add(domain.EphemeralMessageRetention),
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkEphemeralMessageStoreParallelCreate(b *testing.B) {
|
||||
store := NewEphemeralMessageStore()
|
||||
base := time.Unix(1_800_000_000, 0)
|
||||
ctx := context.Background()
|
||||
var sequence atomic.Int64
|
||||
b.RunParallel(func(pb *testing.PB) {
|
||||
for pb.Next() {
|
||||
n := sequence.Add(1)
|
||||
message := testEphemeralMessage(base)
|
||||
message.ID = int(n%1_000_000) + 1
|
||||
message.Peer.ID += n
|
||||
message.RandomID += n
|
||||
message.PayloadHash = sha256.Sum256([]byte{byte(n), byte(n >> 8), byte(n >> 16), byte(n >> 24)})
|
||||
if _, _, err := store.CreateEphemeralMessage(ctx, message); err != nil {
|
||||
b.Errorf("create: %v", err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
@ -108,6 +108,12 @@ func cloneRequestedPeerMedia(media *domain.MessageMedia) *domain.MessageMedia {
|
|||
return nil
|
||||
}
|
||||
clone := *media
|
||||
if media.LivePhotoVideo != nil {
|
||||
video := *media.LivePhotoVideo
|
||||
video.FileReference = append([]byte(nil), media.LivePhotoVideo.FileReference...)
|
||||
video.Attributes = append([]domain.DocumentAttribute(nil), media.LivePhotoVideo.Attributes...)
|
||||
clone.LivePhotoVideo = &video
|
||||
}
|
||||
if media.ServiceAction == nil || media.ServiceAction.RequestedPeer == nil {
|
||||
return &clone
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,8 +1,11 @@
|
|||
package postgres
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
|
|
@ -291,6 +294,7 @@ func (s *BotAPIUpdateStore) EnqueueBotAPIUpdate(ctx context.Context, req domain.
|
|||
var callbackInlineDCID, callbackInlineMessageID int
|
||||
var callbackInlineOwnerID, callbackInlineAccessHash int64
|
||||
var callbackData []byte
|
||||
var ephemeralPayload []byte
|
||||
if req.Callback != nil {
|
||||
callbackQueryID = req.Callback.ID
|
||||
callbackUserID = req.Callback.UserID
|
||||
|
|
@ -303,13 +307,21 @@ func (s *BotAPIUpdateStore) EnqueueBotAPIUpdate(ctx context.Context, req domain.
|
|||
callbackInlineAccessHash = req.Callback.InlineMessage.AccessHash
|
||||
}
|
||||
}
|
||||
if req.Ephemeral != nil {
|
||||
var err error
|
||||
ephemeralPayload, err = json.Marshal(req.Ephemeral)
|
||||
if err != nil {
|
||||
return domain.BotAPIUpdate{}, false, fmt.Errorf("marshal bot api ephemeral payload: %w", err)
|
||||
}
|
||||
}
|
||||
row, err := s.scanBotAPIUpdate(s.db.QueryRow(ctx, `
|
||||
WITH inserted AS (
|
||||
INSERT INTO bot_api_updates (
|
||||
bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date,
|
||||
callback_query_id, callback_user_id, callback_chat_instance, callback_data,
|
||||
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash
|
||||
) SELECT $1, $2::varchar(32), $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15
|
||||
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
|
||||
ephemeral_payload
|
||||
) SELECT $1, $2::varchar(32), $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16::jsonb
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM bot_api_update_states
|
||||
|
|
@ -320,7 +332,8 @@ WHERE NOT EXISTS (
|
|||
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
|
||||
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
|
||||
ephemeral_payload
|
||||
), wake_webhook AS (
|
||||
UPDATE bot_api_webhooks
|
||||
SET next_attempt_at = now(), updated_at = now()
|
||||
|
|
@ -329,11 +342,12 @@ WHERE NOT EXISTS (
|
|||
)
|
||||
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
|
||||
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
|
||||
ephemeral_payload
|
||||
FROM inserted
|
||||
`, req.BotUserID, string(req.Kind), string(req.Peer.Type), req.Peer.ID, req.MessageID, req.SourcePts, req.Date,
|
||||
callbackQueryID, callbackUserID, callbackChatInstance, callbackData,
|
||||
callbackInlineDCID, callbackInlineOwnerID, callbackInlineMessageID, callbackInlineAccessHash))
|
||||
callbackInlineDCID, callbackInlineOwnerID, callbackInlineMessageID, callbackInlineAccessHash, ephemeralPayload))
|
||||
if err == nil {
|
||||
return row, true, nil
|
||||
}
|
||||
|
|
@ -343,16 +357,20 @@ FROM inserted
|
|||
row, err = s.scanBotAPIUpdate(s.db.QueryRow(ctx, `
|
||||
SELECT id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date,
|
||||
callback_query_id, callback_user_id, callback_chat_instance, callback_data,
|
||||
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash
|
||||
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
|
||||
ephemeral_payload
|
||||
FROM bot_api_updates
|
||||
WHERE bot_user_id = $1
|
||||
AND update_kind = $2
|
||||
AND (
|
||||
(update_kind = 'callback_query' AND callback_query_id = $7)
|
||||
OR
|
||||
(update_kind <> 'callback_query' AND peer_type = $3 AND peer_id = $4 AND message_id = $5 AND source_pts = $6)
|
||||
(update_kind <> 'callback_query' AND peer_type = $3 AND peer_id = $4 AND message_id = $5 AND (
|
||||
(ephemeral_payload IS NULL AND $8::jsonb IS NULL AND source_pts = $6)
|
||||
OR (ephemeral_payload = $8::jsonb)
|
||||
))
|
||||
)
|
||||
`, req.BotUserID, string(req.Kind), string(req.Peer.Type), req.Peer.ID, req.MessageID, req.SourcePts, callbackQueryID))
|
||||
`, req.BotUserID, string(req.Kind), string(req.Peer.Type), req.Peer.ID, req.MessageID, req.SourcePts, callbackQueryID, ephemeralPayload))
|
||||
if err != nil {
|
||||
if err == pgx.ErrNoRows {
|
||||
return domain.BotAPIUpdate{}, false, nil
|
||||
|
|
@ -372,11 +390,13 @@ func (s *BotAPIUpdateStore) ListTailBotAPIUpdates(ctx context.Context, botUserID
|
|||
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
|
||||
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
|
||||
ephemeral_payload
|
||||
FROM (
|
||||
SELECT id, bot_user_id, update_kind, peer_type, peer_id, message_id, source_pts, date,
|
||||
callback_query_id, callback_user_id, callback_chat_instance, callback_data,
|
||||
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash
|
||||
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
|
||||
ephemeral_payload
|
||||
FROM bot_api_updates
|
||||
WHERE bot_user_id = $1
|
||||
AND id > COALESCE((SELECT confirmed_update_id FROM bot_api_update_states WHERE bot_user_id = $1), 0)
|
||||
|
|
@ -417,7 +437,8 @@ func (s *BotAPIUpdateStore) ListBotAPIUpdates(ctx context.Context, botUserID, fr
|
|||
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
|
||||
callback_inline_dc_id, callback_inline_owner_id, callback_inline_message_id, callback_inline_access_hash,
|
||||
ephemeral_payload
|
||||
FROM bot_api_updates
|
||||
WHERE bot_user_id = $1 AND id >= $2
|
||||
ORDER BY id
|
||||
|
|
@ -619,9 +640,11 @@ func scanBotAPIUpdateRows(row botAPIUpdateScanner) (domain.BotAPIUpdate, error)
|
|||
var callbackInlineDCID, callbackInlineMessageID int
|
||||
var callbackInlineOwnerID, callbackInlineAccessHash int64
|
||||
var callbackData []byte
|
||||
var ephemeralPayload []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 {
|
||||
&callbackInlineDCID, &callbackInlineOwnerID, &callbackInlineMessageID, &callbackInlineAccessHash,
|
||||
&ephemeralPayload); err != nil {
|
||||
return domain.BotAPIUpdate{}, err
|
||||
}
|
||||
item.Kind = domain.BotAPIUpdateKind(kind)
|
||||
|
|
@ -640,6 +663,21 @@ func scanBotAPIUpdateRows(row botAPIUpdateScanner) (domain.BotAPIUpdate, error)
|
|||
item.Callback.InlineMessage = &domain.BotInlineMessageID{DCID: callbackInlineDCID, OwnerID: callbackInlineOwnerID, ID: callbackInlineMessageID, AccessHash: callbackInlineAccessHash}
|
||||
}
|
||||
}
|
||||
if len(ephemeralPayload) != 0 {
|
||||
var payload domain.BotAPIEphemeralPayload
|
||||
decoder := json.NewDecoder(bytes.NewReader(ephemeralPayload))
|
||||
decoder.DisallowUnknownFields()
|
||||
if err := decoder.Decode(&payload); err != nil {
|
||||
return domain.BotAPIUpdate{}, fmt.Errorf("decode bot api ephemeral payload: %w", err)
|
||||
}
|
||||
if err := decoder.Decode(&struct{}{}); err != io.EOF {
|
||||
return domain.BotAPIUpdate{}, fmt.Errorf("decode bot api ephemeral payload: trailing JSON")
|
||||
}
|
||||
if err := validateBotAPIEphemeralPayload(item.BotUserID, item.Kind, item.Peer, item.MessageID, item.SourcePts, item.Date, &payload); err != nil {
|
||||
return domain.BotAPIUpdate{}, err
|
||||
}
|
||||
item.Ephemeral = &payload
|
||||
}
|
||||
return item, nil
|
||||
}
|
||||
|
||||
|
|
@ -662,6 +700,9 @@ func validateBotAPIUpdateRequest(req domain.EnqueueBotAPIUpdateRequest) error {
|
|||
default:
|
||||
return fmt.Errorf("invalid bot api update peer type %q", req.Peer.Type)
|
||||
}
|
||||
if err := validateBotAPIEphemeralPayload(req.BotUserID, req.Kind, req.Peer, req.MessageID, req.SourcePts, req.Date, req.Ephemeral); err != nil {
|
||||
return err
|
||||
}
|
||||
if req.Kind == domain.BotAPIUpdateCallbackQuery {
|
||||
cb := req.Callback
|
||||
if cb == nil || cb.ID == 0 || cb.BotUserID != req.BotUserID || cb.UserID <= 0 ||
|
||||
|
|
@ -681,3 +722,27 @@ func validateBotAPIUpdateRequest(req domain.EnqueueBotAPIUpdateRequest) error {
|
|||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateBotAPIEphemeralPayload(botUserID int64, kind domain.BotAPIUpdateKind, peer domain.Peer, messageID, sourcePts, date int, payload *domain.BotAPIEphemeralPayload) error {
|
||||
if payload == nil {
|
||||
return nil
|
||||
}
|
||||
message := payload.Message
|
||||
if payload.Validate() != nil || peer.Type != domain.PeerTypeChannel || message.ID != messageID || message.Peer != peer ||
|
||||
message.Expired(time.Unix(int64(date), 0)) || sourcePts != 0 {
|
||||
return fmt.Errorf("invalid bot api ephemeral update")
|
||||
}
|
||||
if kind == domain.BotAPIUpdateCallbackQuery {
|
||||
if message.SenderUserID != botUserID {
|
||||
return fmt.Errorf("invalid bot api ephemeral callback target")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if kind != domain.BotAPIUpdateMessage && kind != domain.BotAPIUpdateEditedMessage {
|
||||
return fmt.Errorf("invalid bot api ephemeral update kind")
|
||||
}
|
||||
if message.ReceiverUserID != botUserID {
|
||||
return fmt.Errorf("invalid bot api ephemeral receiver")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -406,3 +406,70 @@ ON CONFLICT (bot_user_id) DO NOTHING`, u.ID); err != nil {
|
|||
t.Fatalf("post-retention list = %+v, want only unconfirmed fresh row %d", items, unconfirmedFresh.ID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBotAPIEphemeralEnvelopeRoundTrip(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
ctx := context.Background()
|
||||
suffix := randomSuffix(t)
|
||||
users := NewUserStore(pool)
|
||||
bot, err := users.Create(ctx, domain.User{AccessHash: 941, Phone: "+1941" + suffix + "01", FirstName: "EphemeralQueueBot"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
human, err := users.Create(ctx, domain.User{AccessHash: 942, Phone: "+1942" + suffix + "02", FirstName: "EphemeralHuman"})
|
||||
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, 'ephemeral-secret')`, bot.ID); err != nil {
|
||||
t.Fatal(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)
|
||||
})
|
||||
|
||||
now := time.Now()
|
||||
peer := domain.Peer{Type: domain.PeerTypeChannel, ID: 3901}
|
||||
message := domain.EphemeralMessage{
|
||||
ID: 81, Peer: peer, SenderUserID: human.ID, ReceiverUserID: bot.ID,
|
||||
Date: int(now.Unix()), RandomID: 11, Content: domain.EphemeralContent{Message: "/private"},
|
||||
Version: 1, CreatedAt: now, ExpiresAt: now.Add(domain.EphemeralMessageRetention),
|
||||
}
|
||||
store := NewBotAPIUpdateStore(pool)
|
||||
request := domain.EnqueueBotAPIUpdateRequest{
|
||||
BotUserID: bot.ID, Kind: domain.BotAPIUpdateMessage, Peer: peer,
|
||||
MessageID: message.ID, Date: message.Date,
|
||||
Ephemeral: domain.NewBotAPIEphemeralPayload(message),
|
||||
}
|
||||
first, created, err := store.EnqueueBotAPIUpdate(ctx, request)
|
||||
if err != nil || !created || first.Ephemeral == nil {
|
||||
t.Fatalf("first=%+v created=%v err=%v", first, created, err)
|
||||
}
|
||||
var leakedPrivateRoutingState bool
|
||||
if err := pool.QueryRow(ctx, `
|
||||
SELECT (ephemeral_payload -> 'Message') ?| ARRAY[
|
||||
'RandomID', 'OriginDevice', 'PayloadHash', 'CreatedAt', 'Deleted'
|
||||
]
|
||||
FROM bot_api_updates WHERE id = $1`, first.ID).Scan(&leakedPrivateRoutingState); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if leakedPrivateRoutingState {
|
||||
t.Fatal("durable Bot API envelope contains private ephemeral routing fields")
|
||||
}
|
||||
if replay, created, err := store.EnqueueBotAPIUpdate(ctx, request); err != nil || created || replay.ID != first.ID {
|
||||
t.Fatalf("replay=%+v created=%v err=%v", replay, created, err)
|
||||
}
|
||||
message.Version, message.EditDate, message.Content.Message = 2, message.Date+1, "edited"
|
||||
request.Kind = domain.BotAPIUpdateEditedMessage
|
||||
request.Ephemeral = domain.NewBotAPIEphemeralPayload(message)
|
||||
second, created, err := store.EnqueueBotAPIUpdate(ctx, request)
|
||||
if err != nil || !created || second.ID <= first.ID {
|
||||
t.Fatalf("second=%+v created=%v err=%v", second, created, err)
|
||||
}
|
||||
rows, err := store.ListBotAPIUpdates(ctx, bot.ID, first.ID, 100)
|
||||
if err != nil || len(rows) != 2 || rows[0].SourcePts != 0 || rows[0].Ephemeral == nil ||
|
||||
rows[0].Ephemeral.Message.Content.Message != "/private" || rows[1].Ephemeral.Message.Content.Message != "edited" {
|
||||
t.Fatalf("rows=%+v err=%v", rows, err)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
49
internal/store/postgres/ephemeral_report.go
Normal file
49
internal/store/postgres/ephemeral_report.go
Normal file
|
|
@ -0,0 +1,49 @@
|
|||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
"telesrv/internal/store/postgres/sqlcgen"
|
||||
)
|
||||
|
||||
// EphemeralReportStore persists the low-volume abuse-review evidence path.
|
||||
// The hot ephemeral send/edit/delete path remains entirely in Redis.
|
||||
type EphemeralReportStore struct {
|
||||
db sqlcgen.DBTX
|
||||
}
|
||||
|
||||
func NewEphemeralReportStore(db sqlcgen.DBTX) *EphemeralReportStore {
|
||||
return &EphemeralReportStore{db: db}
|
||||
}
|
||||
|
||||
func (s *EphemeralReportStore) CreateEphemeralReport(ctx context.Context, report domain.EphemeralAbuseReport) (bool, error) {
|
||||
if s == nil || s.db == nil {
|
||||
return false, fmt.Errorf("ephemeral report store is not configured")
|
||||
}
|
||||
if err := report.Validate(); err != nil {
|
||||
return false, err
|
||||
}
|
||||
evidence, err := json.Marshal(report.Evidence)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("marshal ephemeral report evidence: %w", err)
|
||||
}
|
||||
tag, err := s.db.Exec(ctx, `
|
||||
INSERT INTO ephemeral_abuse_reports (
|
||||
reporter_user_id, channel_id, ephemeral_message_id, sender_user_id,
|
||||
receiver_user_id, report_option, report_comment, comment_hash,
|
||||
payload_hash, evidence, created_at
|
||||
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11)
|
||||
ON CONFLICT (
|
||||
reporter_user_id, channel_id, ephemeral_message_id, report_option, comment_hash
|
||||
) DO NOTHING
|
||||
`, report.ReporterUserID, report.Evidence.Peer.ID, report.Evidence.MessageID,
|
||||
report.Evidence.SenderUserID, report.Evidence.ReceiverUserID,
|
||||
report.Option, report.Comment, report.CommentHash[:], report.Evidence.PayloadHash[:], evidence, report.CreatedAt)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("insert ephemeral abuse report: %w", err)
|
||||
}
|
||||
return tag.RowsAffected() == 1, nil
|
||||
}
|
||||
60
internal/store/postgres/ephemeral_report_integration_test.go
Normal file
60
internal/store/postgres/ephemeral_report_integration_test.go
Normal file
|
|
@ -0,0 +1,60 @@
|
|||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
||||
func TestEphemeralReportStoreDurableEvidenceAndIdempotency(t *testing.T) {
|
||||
pool := testPool(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now()
|
||||
reporter := now.UnixNano()&0x3fffffff + 1000
|
||||
sender := reporter + 1
|
||||
messageID := int(now.UnixNano()&0x3fffffff) + 1
|
||||
message := domain.EphemeralMessage{
|
||||
ID: messageID, Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: reporter + 2},
|
||||
SenderUserID: sender, ReceiverUserID: reporter, Date: int(now.Unix()), RandomID: 99,
|
||||
Content: domain.EphemeralContent{Message: "abuse evidence"},
|
||||
OriginDevice: domain.EphemeralDevice{UserID: reporter, BusinessAuthKeyID: [8]byte{1, 2, 3}, SessionID: 44},
|
||||
Version: 1, CreatedAt: now, ExpiresAt: now.Add(domain.EphemeralMessageRetention),
|
||||
}
|
||||
report := domain.NewEphemeralAbuseReport(reporter, "spam", "review this", message, now)
|
||||
store := NewEphemeralReportStore(pool)
|
||||
created, err := store.CreateEphemeralReport(ctx, report)
|
||||
if err != nil || !created {
|
||||
t.Fatalf("create=%v err=%v", created, err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_, _ = pool.Exec(ctx, "DELETE FROM ephemeral_abuse_reports WHERE reporter_user_id = $1", reporter)
|
||||
})
|
||||
if created, err := store.CreateEphemeralReport(ctx, report); err != nil || created {
|
||||
t.Fatalf("retry create=%v err=%v", created, err)
|
||||
}
|
||||
var evidenceRaw []byte
|
||||
var count int
|
||||
if err := pool.QueryRow(ctx, `
|
||||
SELECT evidence, count(*) OVER ()
|
||||
FROM ephemeral_abuse_reports
|
||||
WHERE reporter_user_id = $1 AND channel_id = $2 AND ephemeral_message_id = $3
|
||||
`, reporter, message.Peer.ID, message.ID).Scan(&evidenceRaw, &count); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count != 1 {
|
||||
t.Fatalf("rows=%d", count)
|
||||
}
|
||||
var evidence map[string]any
|
||||
if err := json.Unmarshal(evidenceRaw, &evidence); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if evidence["MessageID"] != float64(message.ID) || evidence["Content"] == nil {
|
||||
t.Fatalf("evidence=%s", evidenceRaw)
|
||||
}
|
||||
if _, leaked := evidence["OriginDevice"]; leaked {
|
||||
t.Fatalf("device identity leaked into report evidence: %s", evidenceRaw)
|
||||
}
|
||||
}
|
||||
|
|
@ -14,7 +14,7 @@ func TestStarGiftLifecycleMigrationsApply(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatalf("migrate star gift lifecycle schema: %v", err)
|
||||
}
|
||||
if status.Dirty || status.Empty || status.Version != 119 {
|
||||
t.Fatalf("migration status = %+v, want clean version 119", status)
|
||||
if status.Dirty || status.Empty || status.Version != 121 {
|
||||
t.Fatalf("migration status = %+v, want clean version 121", status)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
458
internal/store/redisstore/ephemeral.go
Normal file
458
internal/store/redisstore/ephemeral.go
Normal file
|
|
@ -0,0 +1,458 @@
|
|||
package redisstore
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
"telesrv/internal/store"
|
||||
)
|
||||
|
||||
const (
|
||||
maxEncodedEphemeralMessageBytes = 2 << 20
|
||||
ephemeralPushChannel = "telesrv:ephemeral:push:v1"
|
||||
)
|
||||
|
||||
type EphemeralMessageStore struct {
|
||||
c redis.UniversalClient
|
||||
}
|
||||
|
||||
func NewEphemeralMessageStore(c redis.UniversalClient) *EphemeralMessageStore {
|
||||
return &EphemeralMessageStore{c: c}
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) PublishEphemeralPush(ctx context.Context, event store.EphemeralPush) error {
|
||||
if s == nil || s.c == nil {
|
||||
return errors.New("redis ephemeral push broker is not configured")
|
||||
}
|
||||
if !validEphemeralPush(event) {
|
||||
return errors.New("invalid ephemeral push")
|
||||
}
|
||||
raw, err := json.Marshal(event)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal ephemeral push: %w", err)
|
||||
}
|
||||
if len(raw) > maxEncodedEphemeralMessageBytes {
|
||||
return errors.New("ephemeral push exceeds encoded size limit")
|
||||
}
|
||||
if err := s.c.Publish(ctx, ephemeralPushChannel, raw).Err(); err != nil {
|
||||
return fmt.Errorf("redis publish ephemeral push: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) SubscribeEphemeralPushes(ctx context.Context, handle func(context.Context, store.EphemeralPush)) error {
|
||||
if s == nil || s.c == nil {
|
||||
return errors.New("redis ephemeral push broker is not configured")
|
||||
}
|
||||
if handle == nil {
|
||||
return errors.New("ephemeral push handler is nil")
|
||||
}
|
||||
pubsub := s.c.Subscribe(ctx, ephemeralPushChannel)
|
||||
defer func() { _ = pubsub.Close() }()
|
||||
if _, err := pubsub.Receive(ctx); err != nil {
|
||||
return fmt.Errorf("redis subscribe ephemeral push: %w", err)
|
||||
}
|
||||
messages := pubsub.Channel()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case item, ok := <-messages:
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
if len(item.Payload) > maxEncodedEphemeralMessageBytes {
|
||||
continue
|
||||
}
|
||||
var event store.EphemeralPush
|
||||
if strictUnmarshalEphemeral([]byte(item.Payload), &event) != nil || !validEphemeralPush(event) {
|
||||
continue
|
||||
}
|
||||
handle(ctx, event)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func validEphemeralPush(event store.EphemeralPush) bool {
|
||||
if event.SourceID == "" || event.TargetUserID <= 0 || event.Date <= 0 || event.Message.ID <= 0 ||
|
||||
event.Message.Peer.Type != domain.PeerTypeChannel || event.Message.Peer.ID <= 0 || event.Message.ValidateStored() != nil {
|
||||
return false
|
||||
}
|
||||
if event.TargetBusinessAuthKey != ([8]byte{}) &&
|
||||
(event.Message.OriginDevice.UserID != event.TargetUserID || event.Message.OriginDevice.BusinessAuthKeyID != event.TargetBusinessAuthKey) {
|
||||
return false
|
||||
}
|
||||
switch event.Kind {
|
||||
case store.EphemeralPushNew, store.EphemeralPushEdit:
|
||||
return !event.Message.Deleted && event.Callback == nil && event.TargetUserID == event.Message.ReceiverUserID
|
||||
case store.EphemeralPushDelete:
|
||||
return event.Message.Deleted && event.Callback == nil &&
|
||||
(event.TargetUserID == event.Message.SenderUserID || event.TargetUserID == event.Message.ReceiverUserID)
|
||||
case store.EphemeralPushCallback:
|
||||
return event.Callback != nil && event.Callback.BotUserID == event.TargetUserID &&
|
||||
event.Callback.ID != 0 && event.Callback.UserID == event.Message.ReceiverUserID &&
|
||||
event.Callback.ChatInstance != 0 && len(event.Callback.Data) <= domain.MaxEphemeralCallbackDataBytes && event.Callback.InlineMessage == nil &&
|
||||
event.Callback.MessageID == event.Message.ID && event.Callback.Peer == event.Message.Peer &&
|
||||
event.TargetUserID == event.Message.SenderUserID
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func ephemeralPeerTag(peer domain.Peer) string {
|
||||
// A shared Redis Cluster hash tag keeps the message and random-id index in
|
||||
// the same slot, so the two-key Lua transaction remains cluster-safe.
|
||||
return fmt.Sprintf("{ephemeral:%s:%d}", peer.Type, peer.ID)
|
||||
}
|
||||
|
||||
func ephemeralMessageKey(peer domain.Peer, id int) string {
|
||||
return fmt.Sprintf("telesrv:%s:message:%d", ephemeralPeerTag(peer), id)
|
||||
}
|
||||
|
||||
func ephemeralRandomKey(message domain.EphemeralMessage) string {
|
||||
return fmt.Sprintf("telesrv:%s:random:%d:%d:%d", ephemeralPeerTag(message.Peer),
|
||||
message.SenderUserID, message.ReceiverUserID, message.RandomID)
|
||||
}
|
||||
|
||||
func ephemeralCallbackActionKey(queryID int64) string {
|
||||
return fmt.Sprintf("telesrv:ephemeral:callback_action:%d", queryID)
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) PutEphemeralCallbackAction(ctx context.Context, action domain.EphemeralCallbackAction) (bool, error) {
|
||||
if s == nil || s.c == nil || action.QueryID == 0 || action.BotUserID <= 0 || action.UserID <= 0 ||
|
||||
action.Peer.Type != domain.PeerTypeChannel || action.Peer.ID <= 0 || action.MessageID <= 0 ||
|
||||
action.Device.UserID != action.UserID || action.Device.BusinessAuthKeyID == ([8]byte{}) || action.CreatedAt.IsZero() ||
|
||||
!action.ExpiresAt.After(action.CreatedAt) || action.ExpiresAt.Sub(action.CreatedAt) > domain.EphemeralReplyWindow {
|
||||
return false, domain.ErrEphemeralInvalid
|
||||
}
|
||||
ttl := time.Until(action.ExpiresAt)
|
||||
if ttl <= 0 || ttl > domain.EphemeralReplyWindow {
|
||||
return false, domain.ErrEphemeralReplyExpired
|
||||
}
|
||||
raw, err := json.Marshal(action)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("marshal ephemeral callback action: %w", err)
|
||||
}
|
||||
created, err := s.c.SetNX(ctx, ephemeralCallbackActionKey(action.QueryID), raw, ttl).Result()
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("redis put ephemeral callback action: %w", err)
|
||||
}
|
||||
return created, nil
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) GetEphemeralCallbackAction(ctx context.Context, botUserID, queryID int64, now time.Time) (domain.EphemeralCallbackAction, bool, error) {
|
||||
if s == nil || s.c == nil || botUserID <= 0 || queryID == 0 {
|
||||
return domain.EphemeralCallbackAction{}, false, nil
|
||||
}
|
||||
key := ephemeralCallbackActionKey(queryID)
|
||||
raw, err := s.c.Get(ctx, key).Bytes()
|
||||
if errors.Is(err, redis.Nil) {
|
||||
return domain.EphemeralCallbackAction{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return domain.EphemeralCallbackAction{}, false, fmt.Errorf("redis get ephemeral callback action: %w", err)
|
||||
}
|
||||
var action domain.EphemeralCallbackAction
|
||||
if strictUnmarshalEphemeral(raw, &action) != nil || action.QueryID != queryID || action.BotUserID != botUserID ||
|
||||
action.UserID <= 0 || action.Peer.Type != domain.PeerTypeChannel || action.Peer.ID <= 0 || action.MessageID <= 0 ||
|
||||
action.Device.UserID != action.UserID || !now.Before(action.ExpiresAt) {
|
||||
_ = s.c.Del(ctx, key).Err()
|
||||
return domain.EphemeralCallbackAction{}, false, nil
|
||||
}
|
||||
return action, true, nil
|
||||
}
|
||||
|
||||
var createEphemeralMessageScript = redis.NewScript(`
|
||||
local index = redis.call('GET', KEYS[2])
|
||||
if index then
|
||||
local separator = string.find(index, '\n', 1, true)
|
||||
if not separator then
|
||||
return {4, ''}
|
||||
end
|
||||
local target = string.sub(index, 1, separator - 1)
|
||||
local payload_hash = string.sub(index, separator + 1)
|
||||
local existing = redis.call('GET', target)
|
||||
if existing then
|
||||
if payload_hash ~= ARGV[2] then
|
||||
return {2, ''}
|
||||
end
|
||||
return {1, existing}
|
||||
end
|
||||
redis.call('DEL', KEYS[2])
|
||||
end
|
||||
if redis.call('EXISTS', KEYS[1]) ~= 0 then
|
||||
return {3, ''}
|
||||
end
|
||||
redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[3])
|
||||
redis.call('SET', KEYS[2], KEYS[1] .. '\n' .. ARGV[2], 'PX', ARGV[3])
|
||||
return {0, ARGV[1]}
|
||||
`)
|
||||
|
||||
func (s *EphemeralMessageStore) CreateEphemeralMessage(ctx context.Context, message domain.EphemeralMessage) (domain.EphemeralMessage, bool, error) {
|
||||
if s == nil || s.c == nil {
|
||||
return domain.EphemeralMessage{}, false, errors.New("redis ephemeral store is not configured")
|
||||
}
|
||||
now := message.CreatedAt
|
||||
if err := message.ValidateForCreate(now); err != nil {
|
||||
return domain.EphemeralMessage{}, false, err
|
||||
}
|
||||
ttl := message.ExpiresAt.Sub(now)
|
||||
if ttl <= 0 || ttl > domain.EphemeralMessageRetention {
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralInvalid
|
||||
}
|
||||
raw, err := marshalEphemeralMessage(message)
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, false, err
|
||||
}
|
||||
value, err := createEphemeralMessageScript.Run(ctx, s.c, []string{
|
||||
ephemeralMessageKey(message.Peer, message.ID), ephemeralRandomKey(message),
|
||||
}, raw, hex.EncodeToString(message.PayloadHash[:]), ttl.Milliseconds()).Result()
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, false, fmt.Errorf("redis create ephemeral message: %w", err)
|
||||
}
|
||||
status, encoded, err := decodeEphemeralScriptResult(value)
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, false, err
|
||||
}
|
||||
switch status {
|
||||
case 0, 1:
|
||||
stored, err := unmarshalEphemeralMessage(encoded)
|
||||
return stored, status == 0, err
|
||||
case 2:
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralRandomIDConflict
|
||||
case 3:
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralIDCollision
|
||||
default:
|
||||
return domain.EphemeralMessage{}, false, fmt.Errorf("redis ephemeral create index is corrupt")
|
||||
}
|
||||
}
|
||||
|
||||
func (s *EphemeralMessageStore) GetEphemeralMessage(ctx context.Context, peer domain.Peer, id int, now time.Time) (domain.EphemeralMessage, bool, error) {
|
||||
if s == nil || s.c == nil || peer.ID <= 0 || id <= 0 {
|
||||
return domain.EphemeralMessage{}, false, nil
|
||||
}
|
||||
raw, err := s.c.Get(ctx, ephemeralMessageKey(peer, id)).Bytes()
|
||||
if errors.Is(err, redis.Nil) {
|
||||
return domain.EphemeralMessage{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, false, fmt.Errorf("redis get ephemeral message: %w", err)
|
||||
}
|
||||
message, err := unmarshalEphemeralMessage(raw)
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, false, err
|
||||
}
|
||||
if message.Peer != peer || message.ID != id {
|
||||
return domain.EphemeralMessage{}, false, fmt.Errorf("redis ephemeral message identity mismatch")
|
||||
}
|
||||
if message.Expired(now) {
|
||||
return domain.EphemeralMessage{}, false, nil
|
||||
}
|
||||
return message, true, nil
|
||||
}
|
||||
|
||||
var editEphemeralMessageScript = redis.NewScript(`
|
||||
local raw = redis.call('GET', KEYS[1])
|
||||
if not raw then
|
||||
return {0, ''}
|
||||
end
|
||||
local decoded, record = pcall(cjson.decode, raw)
|
||||
if not decoded or type(record) ~= 'table' or tonumber(record.Version or 0) <= 0 then
|
||||
return {4, ''}
|
||||
end
|
||||
if record.Deleted == true then
|
||||
return {2, raw}
|
||||
end
|
||||
if tonumber(record.Version) ~= tonumber(ARGV[1]) then
|
||||
return {3, raw}
|
||||
end
|
||||
if redis.call('PTTL', KEYS[1]) <= 0 then
|
||||
return {4, ''}
|
||||
end
|
||||
redis.call('SET', KEYS[1], ARGV[2], 'KEEPTTL')
|
||||
return {1, ARGV[2]}
|
||||
`)
|
||||
|
||||
func (s *EphemeralMessageStore) EditEphemeralMessage(ctx context.Context, peer domain.Peer, id int, expectedVersion uint64, content domain.EphemeralContent, editDate int, now time.Time) (domain.EphemeralMessage, error) {
|
||||
current, found, err := s.GetEphemeralMessage(ctx, peer, id, now)
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, err
|
||||
}
|
||||
if !found {
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralNotFound
|
||||
}
|
||||
if current.Deleted {
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralDeleted
|
||||
}
|
||||
if expectedVersion == 0 || current.Version != expectedVersion {
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralVersionConflict
|
||||
}
|
||||
if domain.ValidateEphemeralContent(content) != nil {
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralInvalid
|
||||
}
|
||||
current.Content = content
|
||||
current.EditDate = editDate
|
||||
current.Version++
|
||||
replacement, err := marshalEphemeralMessage(current)
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, err
|
||||
}
|
||||
value, err := editEphemeralMessageScript.Run(ctx, s.c, []string{ephemeralMessageKey(peer, id)}, expectedVersion, replacement).Result()
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, fmt.Errorf("redis edit ephemeral message: %w", err)
|
||||
}
|
||||
status, encoded, err := decodeEphemeralScriptResult(value)
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, err
|
||||
}
|
||||
switch status {
|
||||
case 1:
|
||||
return unmarshalEphemeralMessage(encoded)
|
||||
case 0:
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralNotFound
|
||||
case 2:
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralDeleted
|
||||
case 3:
|
||||
return domain.EphemeralMessage{}, domain.ErrEphemeralVersionConflict
|
||||
default:
|
||||
return domain.EphemeralMessage{}, fmt.Errorf("redis ephemeral edit record is corrupt")
|
||||
}
|
||||
}
|
||||
|
||||
var deleteEphemeralMessageScript = redis.NewScript(`
|
||||
local raw = redis.call('GET', KEYS[1])
|
||||
if not raw then
|
||||
return {0, ''}
|
||||
end
|
||||
local decoded, record = pcall(cjson.decode, raw)
|
||||
if not decoded or type(record) ~= 'table' or tonumber(record.Version or 0) <= 0 then
|
||||
return {4, ''}
|
||||
end
|
||||
if record.Deleted == true then
|
||||
return {2, raw}
|
||||
end
|
||||
if tonumber(record.Version) ~= tonumber(ARGV[1]) then
|
||||
return {3, raw}
|
||||
end
|
||||
if redis.call('PTTL', KEYS[1]) <= 0 then
|
||||
return {4, ''}
|
||||
end
|
||||
redis.call('SET', KEYS[1], ARGV[2], 'KEEPTTL')
|
||||
return {1, ARGV[2]}
|
||||
`)
|
||||
|
||||
func (s *EphemeralMessageStore) DeleteEphemeralMessage(ctx context.Context, peer domain.Peer, id int, expectedVersion uint64, now time.Time) (domain.EphemeralMessage, bool, error) {
|
||||
current, found, err := s.GetEphemeralMessage(ctx, peer, id, now)
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, false, err
|
||||
}
|
||||
if !found {
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralNotFound
|
||||
}
|
||||
if current.Deleted {
|
||||
return current, false, nil
|
||||
}
|
||||
if expectedVersion == 0 || current.Version != expectedVersion {
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralVersionConflict
|
||||
}
|
||||
current.Deleted = true
|
||||
current.Version++
|
||||
current.Content = domain.EphemeralContent{}
|
||||
replacement, err := marshalEphemeralMessage(current)
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, false, err
|
||||
}
|
||||
value, err := deleteEphemeralMessageScript.Run(ctx, s.c, []string{ephemeralMessageKey(peer, id)}, expectedVersion, replacement).Result()
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, false, fmt.Errorf("redis delete ephemeral message: %w", err)
|
||||
}
|
||||
status, encoded, err := decodeEphemeralScriptResult(value)
|
||||
if err != nil {
|
||||
return domain.EphemeralMessage{}, false, err
|
||||
}
|
||||
switch status {
|
||||
case 1, 2:
|
||||
message, err := unmarshalEphemeralMessage(encoded)
|
||||
return message, status == 1, err
|
||||
case 0:
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralNotFound
|
||||
case 3:
|
||||
return domain.EphemeralMessage{}, false, domain.ErrEphemeralVersionConflict
|
||||
default:
|
||||
return domain.EphemeralMessage{}, false, fmt.Errorf("redis ephemeral delete record is corrupt")
|
||||
}
|
||||
}
|
||||
|
||||
func (*EphemeralMessageStore) PruneExpiredEphemeralMessages(context.Context, time.Time, int) (int, error) {
|
||||
// Redis key expiry is the authoritative O(1) cleanup path; no key scan is
|
||||
// permitted here because SCAN cost would grow with total ephemeral volume.
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
func marshalEphemeralMessage(message domain.EphemeralMessage) ([]byte, error) {
|
||||
raw, err := json.Marshal(message)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("marshal ephemeral message: %w", err)
|
||||
}
|
||||
if len(raw) == 0 || len(raw) > maxEncodedEphemeralMessageBytes {
|
||||
return nil, domain.ErrEphemeralInvalid
|
||||
}
|
||||
return raw, nil
|
||||
}
|
||||
|
||||
func unmarshalEphemeralMessage(raw []byte) (domain.EphemeralMessage, error) {
|
||||
if len(raw) == 0 || len(raw) > maxEncodedEphemeralMessageBytes {
|
||||
return domain.EphemeralMessage{}, fmt.Errorf("redis ephemeral message has invalid encoded size")
|
||||
}
|
||||
var message domain.EphemeralMessage
|
||||
if err := strictUnmarshalEphemeral(raw, &message); err != nil {
|
||||
return domain.EphemeralMessage{}, fmt.Errorf("decode redis ephemeral message: %w", err)
|
||||
}
|
||||
if message.ValidateStored() != nil {
|
||||
return domain.EphemeralMessage{}, fmt.Errorf("redis ephemeral message violates stored invariants")
|
||||
}
|
||||
return message, nil
|
||||
}
|
||||
|
||||
func strictUnmarshalEphemeral(raw []byte, value any) error {
|
||||
decoder := json.NewDecoder(bytes.NewReader(raw))
|
||||
decoder.DisallowUnknownFields()
|
||||
if err := decoder.Decode(value); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := decoder.Decode(&struct{}{}); err != io.EOF {
|
||||
return fmt.Errorf("trailing ephemeral JSON")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func decodeEphemeralScriptResult(value any) (int64, []byte, error) {
|
||||
items, ok := value.([]interface{})
|
||||
if !ok || len(items) != 2 {
|
||||
return 0, nil, fmt.Errorf("redis ephemeral script returned %T", value)
|
||||
}
|
||||
status, ok := items[0].(int64)
|
||||
if !ok {
|
||||
return 0, nil, fmt.Errorf("redis ephemeral script returned invalid status %T", items[0])
|
||||
}
|
||||
var raw []byte
|
||||
switch value := items[1].(type) {
|
||||
case string:
|
||||
raw = []byte(value)
|
||||
case []byte:
|
||||
raw = append([]byte(nil), value...)
|
||||
case nil:
|
||||
default:
|
||||
return 0, nil, fmt.Errorf("redis ephemeral script returned invalid payload %T", items[1])
|
||||
}
|
||||
return status, raw, nil
|
||||
}
|
||||
99
internal/store/redisstore/ephemeral_integration_test.go
Normal file
99
internal/store/redisstore/ephemeral_integration_test.go
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
package redisstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
"telesrv/internal/store"
|
||||
)
|
||||
|
||||
func TestRedisEphemeralAtomicLifecycleCallbackAndBroker(t *testing.T) {
|
||||
addr := os.Getenv("TELESRV_TEST_REDIS_ADDR")
|
||||
if addr == "" {
|
||||
t.Skip("set TELESRV_TEST_REDIS_ADDR to run redis integration test")
|
||||
}
|
||||
client := redis.NewClient(&redis.Options{Addr: addr})
|
||||
defer client.Close()
|
||||
ctx := context.Background()
|
||||
storeImpl := NewEphemeralMessageStore(client)
|
||||
now := time.Now()
|
||||
seed := now.UnixNano() & 0x3fffffff
|
||||
message := domain.EphemeralMessage{
|
||||
ID: int(seed) + 1, Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: seed + 2},
|
||||
SenderUserID: seed + 3, ReceiverUserID: seed + 4, Date: int(now.Unix()), RandomID: seed + 5,
|
||||
Content: domain.EphemeralContent{Message: "/private"}, PayloadHash: sha256.Sum256([]byte("payload")),
|
||||
Version: 1, CreatedAt: now, ExpiresAt: now.Add(domain.EphemeralMessageRetention),
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_ = client.Del(context.Background(), ephemeralMessageKey(message.Peer, message.ID), ephemeralRandomKey(message), ephemeralCallbackActionKey(seed+6)).Err()
|
||||
})
|
||||
created, fresh, err := storeImpl.CreateEphemeralMessage(ctx, message)
|
||||
if err != nil || !fresh || created.ID != message.ID {
|
||||
t.Fatalf("create=%+v fresh=%v err=%v", created, fresh, err)
|
||||
}
|
||||
replay, fresh, err := storeImpl.CreateEphemeralMessage(ctx, message)
|
||||
if err != nil || fresh || replay.ID != message.ID {
|
||||
t.Fatalf("replay=%+v fresh=%v err=%v", replay, fresh, err)
|
||||
}
|
||||
edited, err := storeImpl.EditEphemeralMessage(ctx, message.Peer, message.ID, 1, domain.EphemeralContent{Message: "edited"}, message.Date+1, now)
|
||||
if err != nil || edited.Version != 2 || edited.Content.Message != "edited" {
|
||||
t.Fatalf("edit=%+v err=%v", edited, err)
|
||||
}
|
||||
deleted, changed, err := storeImpl.DeleteEphemeralMessage(ctx, message.Peer, message.ID, 2, now)
|
||||
if err != nil || !changed || !deleted.Deleted || deleted.Version != 3 {
|
||||
t.Fatalf("delete=%+v changed=%v err=%v", deleted, changed, err)
|
||||
}
|
||||
|
||||
action := domain.EphemeralCallbackAction{
|
||||
QueryID: seed + 6, BotUserID: seed + 3, UserID: seed + 4, Peer: message.Peer,
|
||||
MessageID: message.ID, TopMessageID: 42,
|
||||
Device: domain.EphemeralDevice{UserID: seed + 4, BusinessAuthKeyID: [8]byte{7}, SessionID: 8},
|
||||
CreatedAt: now, ExpiresAt: now.Add(domain.EphemeralReplyWindow),
|
||||
}
|
||||
if created, err := storeImpl.PutEphemeralCallbackAction(ctx, action); err != nil || !created {
|
||||
t.Fatalf("put callback created=%v err=%v", created, err)
|
||||
}
|
||||
got, found, err := storeImpl.GetEphemeralCallbackAction(ctx, action.BotUserID, action.QueryID, now)
|
||||
if err != nil || !found || got.TopMessageID != 42 || got.Device.BusinessAuthKeyID != action.Device.BusinessAuthKeyID {
|
||||
t.Fatalf("callback=%+v found=%v err=%v", got, found, err)
|
||||
}
|
||||
|
||||
brokerCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
received := make(chan store.EphemeralPush, 1)
|
||||
go func() {
|
||||
_ = storeImpl.SubscribeEphemeralPushes(brokerCtx, func(_ context.Context, event store.EphemeralPush) {
|
||||
select {
|
||||
case received <- event:
|
||||
default:
|
||||
}
|
||||
})
|
||||
}()
|
||||
event := store.EphemeralPush{
|
||||
SourceID: "redis-test", Kind: store.EphemeralPushDelete,
|
||||
TargetUserID: message.ReceiverUserID, Message: deleted, Date: int(now.Unix()),
|
||||
}
|
||||
ticker := time.NewTicker(20 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
if err := storeImpl.PublishEphemeralPush(ctx, event); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case got := <-received:
|
||||
if got.SourceID != event.SourceID || got.Message.ID != event.Message.ID || got.Kind != event.Kind {
|
||||
t.Fatalf("broker event=%+v", got)
|
||||
}
|
||||
return
|
||||
case <-brokerCtx.Done():
|
||||
t.Fatal("redis ephemeral broker did not deliver")
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue