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

feat: sync collectible star gifts

Browse source
This commit is contained in:
A 2026-07-16 12:38:53 +08:00
parent 47fcf0ea41
commit 5ecf4e912d
64 changed files with 7559 additions and 403 deletions
Download patch file Download diff file Expand all files Collapse all files

8
internal/store/postgres/authkey_integration_test.go
View file

@ -5,6 +5,7 @@ import (
"crypto/rand"
"errors"
"os"
"strings"
"testing"
"github.com/jackc/pgx/v5/pgxpool"
@ -19,6 +20,13 @@ func testPool(t *testing.T) *pgxpool.Pool {
if dsn == "" {
t.Skip("set TELESRV_TEST_POSTGRES_DSN to run postgres integration test")
}
parsed, err := pgxpool.ParseConfig(dsn)
if err != nil {
t.Fatalf("parse TELESRV_TEST_POSTGRES_DSN: %v", err)
}
if !strings.Contains(strings.ToLower(parsed.ConnConfig.Database), "test") {
t.Fatalf("TELESRV_TEST_POSTGRES_DSN must name a dedicated test database, got %q", parsed.ConnConfig.Database)
}
if err := Migrate(dsn); err != nil {
t.Fatalf("migrate: %v", err)
}

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

@ -76,8 +76,17 @@ ON CONFLICT (id) DO NOTHING
}
func (s *MessageStore) SendPrivateText(ctx context.Context, req domain.SendPrivateTextRequest) (res domain.SendPrivateTextResult, err error) {
return s.sendPrivateTextWithHooks(ctx, req, privateSendTxHooks{})
}
type privateSendTxHooks struct {
before func(context.Context, pgx.Tx, *domain.SendPrivateTextRequest) error
after func(context.Context, pgx.Tx, domain.SendPrivateTextResult) error
}
func (s *MessageStore) sendPrivateTextWithHooks(ctx context.Context, req domain.SendPrivateTextRequest, hooks privateSendTxHooks) (res domain.SendPrivateTextResult, err error) {
for attempt := 0; attempt < 2; attempt++ {
res, err = s.sendPrivateTextOnce(ctx, req)
res, err = s.sendPrivateTextOnce(ctx, req, hooks)
if err == nil {
return res, nil
}
@ -91,7 +100,7 @@ func (s *MessageStore) SendPrivateText(ctx context.Context, req domain.SendPriva
return domain.SendPrivateTextResult{}, err
}
func (s *MessageStore) sendPrivateTextOnce(ctx context.Context, req domain.SendPrivateTextRequest) (res domain.SendPrivateTextResult, err error) {
func (s *MessageStore) sendPrivateTextOnce(ctx context.Context, req domain.SendPrivateTextRequest, hooks privateSendTxHooks) (res domain.SendPrivateTextResult, err error) {
if req.SenderUserID == 0 || req.RecipientUserID == 0 {
return domain.SendPrivateTextResult{}, fmt.Errorf("send private text: missing user id")
}
@ -108,10 +117,6 @@ func (s *MessageStore) sendPrivateTextOnce(ctx context.Context, req domain.SendP
if err != nil {
return domain.SendPrivateTextResult{}, err
}
mediaJSON, err := encodeMessageMedia(req.Media)
if err != nil {
return domain.SendPrivateTextResult{}, err
}
// reply_markup（bot inline keyboard）随消息一并入双盒；普通用户发送恒 nil → "{}"。
replyMarkupJSON, err := encodeReplyMarkup(req.ReplyMarkup)
if err != nil {
@ -181,6 +186,15 @@ func (s *MessageStore) sendPrivateTextOnce(ctx context.Context, req domain.SendP
if err := lockUsersForUpdate(ctx, tx, req.SenderUserID, req.RecipientUserID); err != nil {
return domain.SendPrivateTextResult{}, fmt.Errorf("lock send users: %w", err)
}
if hooks.before != nil {
if err := hooks.before(ctx, tx, &req); err != nil {
return domain.SendPrivateTextResult{}, err
}
}
mediaJSON, err := encodeMessageMedia(req.Media)
if err != nil {
return domain.SendPrivateTextResult{}, err
}
ttlPeriod := req.TTLPeriod
if ttlPeriod == 0 {
ttlPeriod, err = privateHistoryTTLPeriod(ctx, tx, req.SenderUserID, req.RecipientUserID)
@ -298,12 +312,21 @@ func (s *MessageStore) sendPrivateTextOnce(ctx context.Context, req domain.SendP
if err := appendNewMessageEvent(ctx, qtx, sender); err != nil {
return domain.SendPrivateTextResult{}, err
}
originUserID := req.OriginUserID
if originUserID == 0 {
originUserID = req.SenderUserID
}
senderExcludeAuthKeyID, senderExcludeSessionID := int64(0), int64(0)
if originUserID == req.SenderUserID {
senderExcludeAuthKeyID = authKeyIDToInt64(req.OriginAuthKeyID)
senderExcludeSessionID = req.OriginSessionID
}
if err := enqueueDispatch(ctx, qtx, sqlcgen.EnqueueDispatchParams{
TargetUserID: req.SenderUserID,
Pts: int32(senderPts),
EventType: string(domain.UpdateEventNewMessage),
ExcludeAuthKeyID: authKeyIDToInt64(req.OriginAuthKeyID),
ExcludeSessionID: req.OriginSessionID,
ExcludeAuthKeyID: senderExcludeAuthKeyID,
ExcludeSessionID: senderExcludeSessionID,
}); err != nil {
return domain.SendPrivateTextResult{}, fmt.Errorf("enqueue sender dispatch: %w", err)
}
@ -360,12 +383,17 @@ func (s *MessageStore) sendPrivateTextOnce(ctx context.Context, req domain.SendP
if err := appendNewMessageEvent(ctx, qtx, recipient); err != nil {
return domain.SendPrivateTextResult{}, err
}
recipientExcludeAuthKeyID, recipientExcludeSessionID := int64(0), int64(0)
if originUserID == req.RecipientUserID {
recipientExcludeAuthKeyID = authKeyIDToInt64(req.OriginAuthKeyID)
recipientExcludeSessionID = req.OriginSessionID
}
if err := enqueueDispatch(ctx, qtx, sqlcgen.EnqueueDispatchParams{
TargetUserID: req.RecipientUserID,
Pts: int32(recipientPts),
EventType: string(domain.UpdateEventNewMessage),
ExcludeAuthKeyID: 0,
ExcludeSessionID: 0,
ExcludeAuthKeyID: recipientExcludeAuthKeyID,
ExcludeSessionID: recipientExcludeSessionID,
}); err != nil {
return domain.SendPrivateTextResult{}, fmt.Errorf("enqueue recipient dispatch: %w", err)
}
@ -397,17 +425,23 @@ WHERE sender_user_id = $1
if tag.RowsAffected() != 1 {
return domain.SendPrivateTextResult{}, fmt.Errorf("save private send receipt: private message %d already has or lost its immutable receipt", pm.ID)
}
result := domain.SendPrivateTextResult{
SenderMessage: sender,
RecipientMessage: recipient,
SenderEvent: eventFromMessage(sender),
RecipientEvent: eventFromMessage(recipient),
}
if hooks.after != nil {
if err := hooks.after(ctx, tx, result); err != nil {
return domain.SendPrivateTextResult{}, err
}
}
if err := tx.Commit(ctx); err != nil {
return domain.SendPrivateTextResult{}, fmt.Errorf("commit send message tx: %w", err)
}
committed = true
return domain.SendPrivateTextResult{
SenderMessage: sender,
RecipientMessage: recipient,
SenderEvent: eventFromMessage(sender),
RecipientEvent: eventFromMessage(recipient),
}, nil
return result, nil
}
// LookupPrivateSendReplay reads an existing receipt without permission checks, source/media

17
internal/store/postgres/read_model_listener.go
View file

@ -33,6 +33,12 @@ type ReadModelCacheSet struct {
RPCProjections RPCProjectionReadModelCache
BaseUsers BaseUserCache
BotProfiles BotProfileReadModelCache
StarGifts StarGiftCatalogCache
}
type StarGiftCatalogCache interface {
InvalidateStarGiftCatalog()
FlushStarGiftCatalog()
}
// BaseUserCache 是跨进程共享的 user:base 缓存(Redis)。user_base read-model 事件必须删除
@ -214,7 +220,8 @@ func (l *ReadModelChangeListener) empty() bool {
l.caches.PrivateMediaCounts == nil &&
l.caches.RPCProjections == nil &&
l.caches.BaseUsers == nil &&
l.caches.BotProfiles == nil
l.caches.BotProfiles == nil &&
l.caches.StarGifts == nil
}
func (l *ReadModelChangeListener) flush(reasons ...string) {
@ -287,6 +294,10 @@ func (l *ReadModelChangeListener) flush(reasons ...string) {
l.caches.BotProfiles.FlushBotProfileReadModel()
flushed = append(flushed, "bot_profiles")
}
if l.caches.StarGifts != nil {
l.caches.StarGifts.FlushStarGiftCatalog()
flushed = append(flushed, "star_gifts")
}
// 注意：BaseUsers(Redis) 刻意不在重连时 flush——它是跨实例共享缓存，整库清空会误伤
// 其它实例；漏掉的通知由其 5min TTL 兜底。
l.log.Info("read model caches flushed",
@ -315,6 +326,10 @@ func (l *ReadModelChangeListener) handlePayload(payload string) {
}
}
switch evt.Model {
case "star_gift_catalog":
if l.caches.StarGifts != nil {
l.caches.StarGifts.InvalidateStarGiftCatalog()
}
case "user_base":
if evt.PeerType == "user" && evt.PeerID != 0 {
if l.caches.RPCProjections != nil {

462
internal/store/postgres/star_gift.go
View file

@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"strings"
"github.com/jackc/pgx/v5"
@ -21,6 +22,290 @@ func NewStarGiftStore(db sqlcgen.DBTX) *StarGiftStore {
return &StarGiftStore{db: db}
}
const starGiftCatalogSelect = `
SELECT c.gift_id, r.id, r.stars, r.convert_stars, r.title,
COALESCE(cr.upgrade_stars, 0), COALESCE(cr.supply_total, 0), COALESCE(cr.issued, 0),
d.id, d.access_hash, d.file_reference, d.date, d.mime_type, d.size, d.dc_id,
d.attributes::text, d.thumbs::text
FROM star_gift_catalog c
JOIN star_gift_catalog_revisions r ON r.id = c.active_revision_id
LEFT JOIN star_gift_collectible_revisions cr ON cr.id = c.collectible_revision_id AND cr.status = 'published'
JOIN documents d ON d.id = r.document_id`
func (s *StarGiftStore) Catalog(ctx context.Context) ([]domain.StarGift, error) {
rows, err := s.db.Query(ctx, starGiftCatalogSelect+`
WHERE c.enabled
ORDER BY c.sort_order, c.gift_id`)
if err != nil {
return nil, fmt.Errorf("list star gift catalog: %w", err)
}
defer rows.Close()
out := make([]domain.StarGift, 0)
for rows.Next() {
gift, err := scanCatalogGift(rows)
if err != nil {
return nil, err
}
out = append(out, gift)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate star gift catalog: %w", err)
}
return out, nil
}
func (s *StarGiftStore) CatalogGift(ctx context.Context, giftID int64) (domain.StarGift, bool, error) {
if giftID <= 0 {
return domain.StarGift{}, false, nil
}
gift, err := scanCatalogGift(s.db.QueryRow(ctx, starGiftCatalogSelect+`
WHERE c.enabled AND c.gift_id = $1`, giftID))
if errors.Is(err, pgx.ErrNoRows) {
return domain.StarGift{}, false, nil
}
if err != nil {
return domain.StarGift{}, false, err
}
return gift, true, nil
}
func (s *StarGiftStore) CatalogRevision(ctx context.Context, revisionID int64) (domain.StarGift, bool, error) {
if revisionID <= 0 {
return domain.StarGift{}, false, nil
}
gift, err := scanCatalogGift(s.db.QueryRow(ctx, `
SELECT r.gift_id, r.id, r.stars, r.convert_stars, r.title,
COALESCE(cr.upgrade_stars, 0), COALESCE(cr.supply_total, 0), COALESCE(cr.issued, 0),
d.id, d.access_hash, d.file_reference, d.date, d.mime_type, d.size, d.dc_id,
d.attributes::text, d.thumbs::text
FROM star_gift_catalog_revisions r
JOIN star_gift_catalog c ON c.gift_id = r.gift_id
LEFT JOIN star_gift_collectible_revisions cr ON cr.id = c.collectible_revision_id AND cr.status = 'published'
JOIN documents d ON d.id = r.document_id
WHERE r.id = $1`, revisionID))
if errors.Is(err, pgx.ErrNoRows) {
return domain.StarGift{}, false, nil
}
if err != nil {
return domain.StarGift{}, false, err
}
return gift, true, nil
}
func scanCatalogGift(row rowScanner) (domain.StarGift, error) {
var gift domain.StarGift
var attrsJSON, thumbsJSON string
if err := row.Scan(
&gift.ID, &gift.RevisionID, &gift.Stars, &gift.ConvertStars, &gift.Title,
&gift.UpgradeStars, &gift.UpgradeTotal, &gift.UpgradeIssued,
&gift.Sticker.ID, &gift.Sticker.AccessHash, &gift.Sticker.FileReference, &gift.Sticker.Date,
&gift.Sticker.MimeType, &gift.Sticker.Size, &gift.Sticker.DCID, &attrsJSON, &thumbsJSON,
); err != nil {
return domain.StarGift{}, err
}
attrs, err := decodeDocumentAttributes(attrsJSON)
if err != nil {
return domain.StarGift{}, fmt.Errorf("decode star gift document attributes: %w", err)
}
thumbs, err := decodePhotoSizes(thumbsJSON)
if err != nil {
return domain.StarGift{}, fmt.Errorf("decode star gift document thumbs: %w", err)
}
gift.Sticker.Attributes = attrs
gift.Sticker.Thumbs = thumbs
if !gift.Sticker.IsSticker() || gift.Sticker.MimeType != "application/x-tgsticker" {
return domain.StarGift{}, fmt.Errorf("invalid star gift revision %d document %d", gift.RevisionID, gift.Sticker.ID)
}
return gift, nil
}
func (s *StarGiftStore) CreateCatalogRevision(ctx context.Context, write domain.StarGiftCatalogWrite) (domain.StarGiftCatalogEntry, error) {
if write.Stars <= 0 || write.ConvertStars < 0 || write.ConvertStars > write.Stars ||
write.Document.ID <= 0 || !write.Document.IsSticker() || write.Document.MimeType != "application/x-tgsticker" ||
len(write.Animation.JSON) == 0 || len(write.Animation.SHA256) != 32 {
return domain.StarGiftCatalogEntry{}, domain.ErrStarGiftInvalid
}
var entry domain.StarGiftCatalogEntry
err := withTx(ctx, s.db, "create star gift catalog revision", func(tx pgx.Tx) error {
giftID := write.GiftID
var revisionID int64
if err := tx.QueryRow(ctx, `SELECT nextval('star_gift_catalog_revision_id_seq')`).Scan(&revisionID); err != nil {
return fmt.Errorf("allocate star gift revision id: %w", err)
}
revision := 1
if giftID == 0 {
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended('star_gift_catalog', 0))`); err != nil {
return fmt.Errorf("lock star gift catalog capacity: %w", err)
}
var catalogCount int
if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM star_gift_catalog`).Scan(&catalogCount); err != nil {
return fmt.Errorf("count star gift catalog: %w", err)
}
if catalogCount >= domain.MaxStarGiftCatalogSize {
return domain.ErrStarGiftCatalogFull
}
if err := tx.QueryRow(ctx, `SELECT nextval('star_gift_catalog_gift_id_seq')`).Scan(&giftID); err != nil {
return fmt.Errorf("allocate star gift id: %w", err)
}
if _, err := tx.Exec(ctx, `
INSERT INTO star_gift_catalog (gift_id, active_revision_id, enabled, sort_order)
VALUES ($1,$2,$3,$4)`, giftID, revisionID, write.Enabled, write.SortOrder); err != nil {
return fmt.Errorf("insert star gift catalog: %w", err)
}
} else {
var ignored int64
if err := tx.QueryRow(ctx, `
SELECT active_revision_id FROM star_gift_catalog WHERE gift_id=$1 FOR UPDATE`, giftID).Scan(&ignored); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return domain.ErrStarGiftNotFound
}
return fmt.Errorf("lock star gift catalog: %w", err)
}
if err := tx.QueryRow(ctx, `
SELECT COALESCE(MAX(revision), 0) + 1
FROM star_gift_catalog_revisions
WHERE gift_id = $1`, giftID).Scan(&revision); err != nil {
return fmt.Errorf("lock star gift catalog: %w", err)
}
}
media := NewMediaStore(tx)
if err := media.PutDocument(ctx, write.Document); err != nil {
return fmt.Errorf("put star gift document: %w", err)
}
if err := media.PutFileBlob(ctx, write.Blob); err != nil {
return fmt.Errorf("put star gift blob: %w", err)
}
if _, err := tx.Exec(ctx, `
INSERT INTO star_gift_catalog_revisions (
id, gift_id, revision, title, stars, convert_stars, document_id,
animation_json, animation_sha256, source_name, source_format,
width, height, frame_rate, in_point, out_point, created_by, command_id
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8::jsonb,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18)`,
revisionID, giftID, revision, write.Title, write.Stars, write.ConvertStars, write.Document.ID,
string(write.Animation.JSON), write.Animation.SHA256, write.Animation.SourceName, string(write.Animation.SourceFormat),
write.Animation.Width, write.Animation.Height, write.Animation.FrameRate, write.Animation.InPoint, write.Animation.OutPoint,
write.Actor, write.CommandID,
); err != nil {
return fmt.Errorf("insert star gift revision: %w", err)
}
if write.GiftID != 0 {
if _, err := tx.Exec(ctx, `
UPDATE star_gift_catalog
SET active_revision_id=$2, enabled=$3, sort_order=$4, updated_at=now()
WHERE gift_id=$1`, giftID, revisionID, write.Enabled, write.SortOrder); err != nil {
return fmt.Errorf("activate star gift revision: %w", err)
}
}
write.GiftID = giftID
var err error
entry, err = catalogEntryByID(ctx, tx, giftID)
return err
})
if err != nil {
return domain.StarGiftCatalogEntry{}, err
}
return entry, nil
}
func (s *StarGiftStore) SetCatalogEnabled(ctx context.Context, giftID int64, enabled bool) (bool, error) {
tag, err := s.db.Exec(ctx, `
UPDATE star_gift_catalog SET enabled=$2, updated_at=now()
WHERE gift_id=$1 AND enabled IS DISTINCT FROM $2`, giftID, enabled)
if err != nil {
return false, fmt.Errorf("set star gift enabled: %w", err)
}
if tag.RowsAffected() > 0 {
return true, nil
}
var exists bool
if err := s.db.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM star_gift_catalog WHERE gift_id=$1)`, giftID).Scan(&exists); err != nil {
return false, fmt.Errorf("check star gift enabled target: %w", err)
}
if !exists {
return false, domain.ErrStarGiftNotFound
}
return false, nil
}
func (s *StarGiftStore) SetCatalogSortOrder(ctx context.Context, giftID int64, sortOrder int) (bool, error) {
tag, err := s.db.Exec(ctx, `
UPDATE star_gift_catalog SET sort_order=$2, updated_at=now()
WHERE gift_id=$1 AND sort_order IS DISTINCT FROM $2`, giftID, sortOrder)
if err != nil {
return false, fmt.Errorf("set star gift sort order: %w", err)
}
if tag.RowsAffected() > 0 {
return true, nil
}
var exists bool
if err := s.db.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM star_gift_catalog WHERE gift_id=$1)`, giftID).Scan(&exists); err != nil {
return false, fmt.Errorf("check star gift sort target: %w", err)
}
if !exists {
return false, domain.ErrStarGiftNotFound
}
return false, nil
}
func (s *StarGiftStore) AnimationJSON(ctx context.Context, giftID int64) ([]byte, bool, error) {
var raw []byte
err := s.db.QueryRow(ctx, `
SELECT r.animation_json::text
FROM star_gift_catalog c
JOIN star_gift_catalog_revisions r ON r.id = c.active_revision_id
WHERE c.gift_id=$1`, giftID).Scan(&raw)
if errors.Is(err, pgx.ErrNoRows) {
return nil, false, nil
}
if err != nil {
return nil, false, fmt.Errorf("get star gift animation: %w", err)
}
return raw, true, nil
}
func catalogEntryByID(ctx context.Context, db sqlcgen.DBTX, giftID int64) (domain.StarGiftCatalogEntry, error) {
row := db.QueryRow(ctx, `
SELECT c.gift_id, r.id, r.stars, r.convert_stars, r.title,
COALESCE(cr.upgrade_stars, 0), COALESCE(cr.supply_total, 0), COALESCE(cr.issued, 0),
d.id, d.access_hash, d.file_reference, d.date, d.mime_type, d.size, d.dc_id,
d.attributes::text, d.thumbs::text,
c.enabled, c.sort_order, r.revision, r.source_name, r.source_format,
r.animation_sha256, r.width, r.height, r.frame_rate, r.created_by, c.updated_at,
(SELECT COUNT(*) FROM peer_star_gifts p WHERE p.gift_id=c.gift_id)
FROM star_gift_catalog c
JOIN star_gift_catalog_revisions r ON r.id=c.active_revision_id
LEFT JOIN star_gift_collectible_revisions cr ON cr.id=c.collectible_revision_id AND cr.status='published'
JOIN documents d ON d.id=r.document_id
WHERE c.gift_id=$1`, giftID)
var entry domain.StarGiftCatalogEntry
var attrsJSON, thumbsJSON, sourceFormat string
if err := row.Scan(
&entry.Gift.ID, &entry.Gift.RevisionID, &entry.Gift.Stars, &entry.Gift.ConvertStars, &entry.Gift.Title,
&entry.Gift.UpgradeStars, &entry.Gift.UpgradeTotal, &entry.Gift.UpgradeIssued,
&entry.Gift.Sticker.ID, &entry.Gift.Sticker.AccessHash, &entry.Gift.Sticker.FileReference, &entry.Gift.Sticker.Date,
&entry.Gift.Sticker.MimeType, &entry.Gift.Sticker.Size, &entry.Gift.Sticker.DCID, &attrsJSON, &thumbsJSON,
&entry.Enabled, &entry.SortOrder, &entry.Revision, &entry.SourceName, &sourceFormat,
&entry.AnimationSHA, &entry.Width, &entry.Height, &entry.FrameRate, &entry.CreatedBy, &entry.UpdatedAt,
&entry.ReceivedCount,
); err != nil {
return domain.StarGiftCatalogEntry{}, err
}
attrs, err := decodeDocumentAttributes(attrsJSON)
if err != nil {
return domain.StarGiftCatalogEntry{}, err
}
thumbs, err := decodePhotoSizes(thumbsJSON)
if err != nil {
return domain.StarGiftCatalogEntry{}, err
}
entry.Gift.Sticker.Attributes = attrs
entry.Gift.Sticker.Thumbs = thumbs
entry.SourceFormat = domain.StarGiftAnimationFormat(sourceFormat)
entry.AnimationSize = entry.Gift.Sticker.Size
return entry, nil
}
func (s *StarGiftStore) Create(ctx context.Context, gift domain.SavedStarGift) (int64, error) {
if !validSavedStarGift(gift) {
return 0, domain.ErrStarGiftInvalid
@ -30,14 +315,14 @@ func (s *StarGiftStore) Create(ctx context.Context, gift domain.SavedStarGift) (
WITH next_id AS (
SELECT nextval(pg_get_serial_sequence('public.peer_star_gifts', 'id'))::bigint AS id
)
INSERT INTO peer_star_gifts (id, owner_peer_type, owner_peer_id, from_user_id, gift_id, msg_id, saved_id, gift_date, name_hidden, unsaved, converted, convert_stars, message)
SELECT next_id.id, $1,$2,$3,$4,$5,
CASE WHEN $1 = 'channel' AND $6::bigint = 0 THEN next_id.id ELSE $6::bigint END,
$7,$8,$9,false,$10,$11
INSERT INTO peer_star_gifts (id, owner_peer_type, owner_peer_id, from_user_id, gift_id, catalog_revision_id, msg_id, saved_id, gift_date, name_hidden, unsaved, converted, convert_stars, prepaid_upgrade_stars, message)
SELECT next_id.id, $1,$2,$3,$4,$5,$6,
CASE WHEN $1 = 'channel' AND $7::bigint = 0 THEN next_id.id ELSE $7::bigint END,
$8,$9,$10,false,$11,$12,$13
FROM next_id
RETURNING id`,
string(gift.Owner.Type), gift.Owner.ID, gift.FromUserID, gift.GiftID, gift.MsgID, gift.SavedID, gift.Date,
gift.NameHidden, gift.Unsaved, gift.ConvertStars, gift.Message).Scan(&id)
string(gift.Owner.Type), gift.Owner.ID, gift.FromUserID, gift.GiftID, gift.RevisionID, gift.MsgID, gift.SavedID, gift.Date,
gift.NameHidden, gift.Unsaved, gift.ConvertStars, gift.PrepaidUpgradeStars, gift.Message).Scan(&id)
if err != nil {
return 0, fmt.Errorf("create star gift: %w", err)
}
@ -45,38 +330,80 @@ RETURNING id`,
}
func (s *StarGiftStore) ListByOwner(ctx context.Context, owner domain.Peer, excludeUnsaved bool, offset string, limit int) (domain.SavedStarGiftPage, error) {
return s.ListByOwnerFiltered(ctx, domain.SavedStarGiftFilter{
Owner: owner, ExcludeUnsaved: excludeUnsaved, Offset: offset, Limit: limit,
})
}
func (s *StarGiftStore) ListByOwnerFiltered(ctx context.Context, filter domain.SavedStarGiftFilter) (domain.SavedStarGiftPage, error) {
owner, offset, limit := filter.Owner, filter.Offset, filter.Limit
if !validStarGiftOwner(owner) {
return domain.SavedStarGiftPage{}, nil
}
if limit <= 0 || limit > domain.MaxSavedStarGiftsLimit {
limit = domain.MaxSavedStarGiftsLimit
}
// 总数（未转换 + 可选 excludeUnsaved 过滤）。
countQuery := `SELECT COUNT(*) FROM peer_star_gifts WHERE owner_peer_type = $1 AND owner_peer_id = $2 AND NOT converted`
if excludeUnsaved {
countQuery += ` AND NOT unsaved`
joins := `
JOIN star_gift_catalog c ON c.gift_id = p.gift_id
LEFT JOIN star_gift_collectible_revisions acr
ON acr.id = c.collectible_revision_id AND acr.status = 'published'`
conditions := []string{"p.owner_peer_type = $1", "p.owner_peer_id = $2", "NOT p.converted"}
args := []any{string(owner.Type), owner.ID}
if filter.ExcludeUnsaved {
conditions = append(conditions, "NOT p.unsaved")
}
if filter.ExcludeSaved {
conditions = append(conditions, "p.unsaved")
}
if filter.ExcludeUnique {
conditions = append(conditions, "p.unique_gift_id IS NULL")
}
// telesrv ordinary catalog gifts are currently unlimited. Unique gifts are
// collectibles and therefore survive exclude_unlimited.
if filter.ExcludeUnlimited {
conditions = append(conditions, "p.unique_gift_id IS NOT NULL")
}
upgradable := `(p.unique_gift_id IS NULL AND acr.id IS NOT NULL AND acr.upgrade_stars > 0 AND acr.issued < acr.supply_total)`
if filter.ExcludeUpgradable {
conditions = append(conditions, "NOT "+upgradable)
}
if filter.ExcludeUnupgradable {
conditions = append(conditions, upgradable)
}
if filter.CollectionID > 0 {
args = append(args, filter.CollectionID)
conditions = append(conditions, fmt.Sprintf(`EXISTS (
SELECT 1 FROM star_gift_collection_items ci
JOIN star_gift_collections cc ON cc.collection_id = ci.collection_id
WHERE ci.saved_gift_id = p.id AND ci.collection_id = $%d
AND cc.owner_peer_type = p.owner_peer_type AND cc.owner_peer_id = p.owner_peer_id)`, len(args)))
}
where := strings.Join(conditions, " AND ")
countQuery := `SELECT COUNT(*) FROM peer_star_gifts p ` + joins + ` WHERE ` + where
var total int
if err := s.db.QueryRow(ctx, countQuery, string(owner.Type), owner.ID).Scan(&total); err != nil {
if err := s.db.QueryRow(ctx, countQuery, args...).Scan(&total); err != nil {
return domain.SavedStarGiftPage{}, fmt.Errorf("count star gifts: %w", err)
}
page := domain.SavedStarGiftPage{Count: total}
where := "owner_peer_type = $1 AND owner_peer_id = $2 AND NOT converted"
if excludeUnsaved {
where += " AND NOT unsaved"
}
args := []any{string(owner.Type), owner.ID, limit + 1}
if cursor, ok := domain.DecodeStarGiftCursor(offset); ok {
where += " AND id < $4"
args = append(args, cursor)
where += fmt.Sprintf(" AND p.id < $%d", len(args))
}
args = append(args, limit+1)
limitPlaceholder := len(args)
rows, err := s.db.Query(ctx, `
SELECT id, owner_peer_type, owner_peer_id, from_user_id, gift_id, msg_id, saved_id, gift_date, name_hidden, unsaved, converted, convert_stars, message
FROM peer_star_gifts
SELECT p.id, p.owner_peer_type, p.owner_peer_id, p.from_user_id, p.gift_id, p.catalog_revision_id,
p.msg_id, p.saved_id, p.gift_date, p.name_hidden, p.unsaved, p.converted, p.convert_stars, p.prepaid_upgrade_stars,
p.message, COALESCE(p.unique_gift_id, 0), p.upgrade_msg_id, p.pinned_order,
COALESCE((SELECT array_agg(i.collection_id ORDER BY c.sort_order, i.collection_id)
FROM star_gift_collection_items i
JOIN star_gift_collections c ON c.collection_id=i.collection_id
WHERE i.saved_gift_id=p.id), ARRAY[]::integer[])
FROM peer_star_gifts p `+joins+`
WHERE `+where+`
ORDER BY id DESC
LIMIT $3`, args...)
ORDER BY p.id DESC
LIMIT $`+fmt.Sprint(limitPlaceholder), args...)
if err != nil {
return domain.SavedStarGiftPage{}, fmt.Errorf("list star gifts: %w", err)
}
@ -100,14 +427,73 @@ LIMIT $3`, args...)
return page, nil
}
func (s *StarGiftStore) ResolveSavedIDs(ctx context.Context, owner domain.Peer, refs []domain.SavedStarGiftRef) ([]int64, error) {
if !validStarGiftOwner(owner) || len(refs) > domain.MaxStarGiftCollectionItems {
return nil, domain.ErrStarGiftCollectibleInvalid
}
if len(refs) == 0 {
return []int64{}, nil
}
values := make([]int64, 0, len(refs))
seenValues := make(map[int64]struct{}, len(refs))
column := "msg_id"
for _, ref := range refs {
if ref.Owner != owner || !ref.Valid() {
return nil, domain.ErrStarGiftNotFound
}
value := int64(ref.MsgID)
if owner.Type == domain.PeerTypeChannel {
column = "saved_id"
value = ref.SavedID
}
if _, duplicate := seenValues[value]; duplicate {
return nil, domain.ErrStarGiftCollectibleInvalid
}
seenValues[value] = struct{}{}
values = append(values, value)
}
rows, err := s.db.Query(ctx, `SELECT `+column+`::bigint, id FROM peer_star_gifts
WHERE owner_peer_type=$1 AND owner_peer_id=$2 AND NOT converted AND `+column+`::bigint=ANY($3::bigint[])`, string(owner.Type), owner.ID, values)
if err != nil {
return nil, fmt.Errorf("resolve saved star gifts: %w", err)
}
defer rows.Close()
resolved := make(map[int64]int64, len(values))
for rows.Next() {
var value, id int64
if err := rows.Scan(&value, &id); err != nil {
return nil, fmt.Errorf("scan resolved saved star gift: %w", err)
}
resolved[value] = id
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate resolved saved star gifts: %w", err)
}
out := make([]int64, 0, len(values))
for _, value := range values {
id := resolved[value]
if id == 0 {
return nil, domain.ErrStarGiftNotFound
}
out = append(out, id)
}
return out, nil
}
func (s *StarGiftStore) GetByRef(ctx context.Context, ref domain.SavedStarGiftRef) (domain.SavedStarGift, bool, error) {
if !ref.Valid() {
return domain.SavedStarGift{}, false, nil
}
where, args := savedStarGiftRefWhere(ref)
row := s.db.QueryRow(ctx, `
SELECT id, owner_peer_type, owner_peer_id, from_user_id, gift_id, msg_id, saved_id, gift_date, name_hidden, unsaved, converted, convert_stars, message
FROM peer_star_gifts
SELECT p.id, p.owner_peer_type, p.owner_peer_id, p.from_user_id, p.gift_id, p.catalog_revision_id,
p.msg_id, p.saved_id, p.gift_date, p.name_hidden, p.unsaved, p.converted, p.convert_stars, p.prepaid_upgrade_stars,
p.message, COALESCE(p.unique_gift_id, 0), p.upgrade_msg_id, p.pinned_order,
COALESCE((SELECT array_agg(i.collection_id ORDER BY c.sort_order, i.collection_id)
FROM star_gift_collection_items i
JOIN star_gift_collections c ON c.collection_id=i.collection_id
WHERE i.saved_gift_id=p.id), ARRAY[]::integer[])
FROM peer_star_gifts p
WHERE `+where, args...)
g, err := scanSavedStarGift(row)
if err != nil {
@ -151,10 +537,19 @@ func (s *StarGiftStore) MarkConverted(ctx context.Context, ref domain.SavedStarG
}
out := domain.SavedStarGift{}
err := withTx(ctx, s.db, "convert star gift", func(tx pgx.Tx) error {
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, starGiftCollectionLockKey(ref.Owner)); err != nil {
return fmt.Errorf("lock star gift owner collections: %w", err)
}
where, args := savedStarGiftRefWhere(ref)
row := tx.QueryRow(ctx, `
SELECT id, owner_peer_type, owner_peer_id, from_user_id, gift_id, msg_id, saved_id, gift_date, name_hidden, unsaved, converted, convert_stars, message
FROM peer_star_gifts
SELECT p.id, p.owner_peer_type, p.owner_peer_id, p.from_user_id, p.gift_id, p.catalog_revision_id,
p.msg_id, p.saved_id, p.gift_date, p.name_hidden, p.unsaved, p.converted, p.convert_stars, p.prepaid_upgrade_stars,
p.message, COALESCE(p.unique_gift_id, 0), p.upgrade_msg_id, p.pinned_order,
COALESCE((SELECT array_agg(i.collection_id ORDER BY c.sort_order, i.collection_id)
FROM star_gift_collection_items i
JOIN star_gift_collections c ON c.collection_id=i.collection_id
WHERE i.saved_gift_id=p.id), ARRAY[]::integer[])
FROM peer_star_gifts p
WHERE `+where+` FOR UPDATE`, args...)
g, err := scanSavedStarGift(row)
if err != nil {
@ -166,11 +561,19 @@ WHERE `+where+` FOR UPDATE`, args...)
if g.Converted {
return domain.ErrStarGiftAlreadyConverted
}
if _, err := tx.Exec(ctx, `UPDATE peer_star_gifts SET converted = true, unsaved = true WHERE id = $1`, g.ID); err != nil {
if g.UniqueGiftID != 0 {
return domain.ErrStarGiftAlreadyUpgraded
}
if _, err := tx.Exec(ctx, `UPDATE peer_star_gifts SET converted = true, unsaved = true, pinned_order = 0 WHERE id = $1`, g.ID); err != nil {
return fmt.Errorf("mark star gift converted: %w", err)
}
if err := removeSavedGiftFromCollections(ctx, tx, g.Owner, g.ID); err != nil {
return err
}
g.Converted = true
g.Unsaved = true
g.PinnedOrder = 0
g.CollectionIDs = nil
out = g
return nil
})
@ -183,8 +586,9 @@ WHERE `+where+` FOR UPDATE`, args...)
func scanSavedStarGift(row rowScanner) (domain.SavedStarGift, error) {
var g domain.SavedStarGift
var ownerType string
if err := row.Scan(&g.ID, &ownerType, &g.Owner.ID, &g.FromUserID, &g.GiftID, &g.MsgID, &g.SavedID, &g.Date,
&g.NameHidden, &g.Unsaved, &g.Converted, &g.ConvertStars, &g.Message); err != nil {
if err := row.Scan(&g.ID, &ownerType, &g.Owner.ID, &g.FromUserID, &g.GiftID, &g.RevisionID, &g.MsgID, &g.SavedID, &g.Date,
&g.NameHidden, &g.Unsaved, &g.Converted, &g.ConvertStars, &g.PrepaidUpgradeStars, &g.Message, &g.UniqueGiftID,
&g.UpgradeMsgID, &g.PinnedOrder, &g.CollectionIDs); err != nil {
return domain.SavedStarGift{}, err
}
g.Owner.Type = domain.PeerType(ownerType)
@ -204,7 +608,7 @@ func savedStarGiftRefWhere(ref domain.SavedStarGiftRef) (string, []any) {
}
func validSavedStarGift(g domain.SavedStarGift) bool {
if g.GiftID == 0 || !validStarGiftOwner(g.Owner) {
if g.GiftID == 0 || g.RevisionID == 0 || !validStarGiftOwner(g.Owner) {
return false
}
switch g.Owner.Type {

781
internal/store/postgres/star_gift_collectibles.go Normal file
View file

@ -0,0 +1,781 @@
package postgres
import (
"context"
"errors"
"fmt"
"sort"
"strings"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
"telesrv/internal/domain"
"telesrv/internal/store/postgres/sqlcgen"
)
func (s *StarGiftStore) PublishCollectibleRevision(ctx context.Context, write domain.StarGiftCollectibleWrite) (domain.StarGiftCollectibleRevision, error) {
write.SlugPrefix = strings.ToLower(strings.TrimSpace(write.SlugPrefix))
write.Actor = strings.TrimSpace(write.Actor)
write.CommandID = strings.TrimSpace(write.CommandID)
if err := domain.ValidateStarGiftCollectibleWrite(write); err != nil {
return domain.StarGiftCollectibleRevision{}, err
}
var result domain.StarGiftCollectibleRevision
err := withTx(ctx, s.db, "publish collectible star gift revision", func(tx pgx.Tx) error {
var ignored int64
if err := tx.QueryRow(ctx, `SELECT gift_id FROM star_gift_catalog WHERE gift_id=$1 FOR UPDATE`, write.GiftID).Scan(&ignored); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return domain.ErrStarGiftNotFound
}
return fmt.Errorf("lock collectible catalog gift: %w", err)
}
var revision int
if err := tx.QueryRow(ctx, `
SELECT COALESCE(MAX(revision), 0) + 1 FROM star_gift_collectible_revisions WHERE gift_id=$1`, write.GiftID).Scan(&revision); err != nil {
return fmt.Errorf("allocate collectible revision: %w", err)
}
var revisionID int64
if err := tx.QueryRow(ctx, `
INSERT INTO star_gift_collectible_revisions
(gift_id, revision, upgrade_stars, supply_total, slug_prefix, status, created_by, command_id)
VALUES ($1,$2,$3,$4,$5,'draft',$6,$7)
RETURNING id`, write.GiftID, revision, write.UpgradeStars, write.SupplyTotal, write.SlugPrefix, write.Actor, write.CommandID).Scan(&revisionID); err != nil {
return fmt.Errorf("insert collectible revision: %w", err)
}
media := NewMediaStore(tx)
insertAnimated := func(table string, attributes []domain.StarGiftCollectibleAttribute) error {
for _, attribute := range attributes {
if err := media.PutDocument(ctx, *attribute.Document); err != nil {
return fmt.Errorf("put collectible %s document: %w", attribute.Kind, err)
}
if err := media.PutFileBlob(ctx, *attribute.Blob); err != nil {
return fmt.Errorf("put collectible %s blob: %w", attribute.Kind, err)
}
animation := attribute.Animation
query := fmt.Sprintf(`
INSERT INTO %s
(collectible_revision_id, name, document_id, animation_json, animation_sha256,
source_name, source_format, width, height, frame_rate, in_point, out_point,
rarity_permille, sort_order)
VALUES ($1,$2,$3,$4::jsonb,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)`, table)
if _, err := tx.Exec(ctx, query, revisionID, strings.TrimSpace(attribute.Name), attribute.Document.ID,
string(animation.JSON), animation.SHA256, animation.SourceName, string(animation.SourceFormat),
animation.Width, animation.Height, animation.FrameRate, animation.InPoint, animation.OutPoint,
attribute.RarityPermille, attribute.SortOrder); err != nil {
return fmt.Errorf("insert collectible %s attribute: %w", attribute.Kind, err)
}
}
return nil
}
if err := insertAnimated("star_gift_collectible_models", write.Models); err != nil {
return err
}
if err := insertAnimated("star_gift_collectible_patterns", write.Patterns); err != nil {
return err
}
for _, attribute := range write.Backdrops {
if _, err := tx.Exec(ctx, `
INSERT INTO star_gift_collectible_backdrops
(collectible_revision_id, name, backdrop_id, center_color, edge_color, pattern_color,
text_color, rarity_permille, sort_order)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)`, revisionID, strings.TrimSpace(attribute.Name), attribute.BackdropID,
attribute.CenterColor, attribute.EdgeColor, attribute.PatternColor, attribute.TextColor,
attribute.RarityPermille, attribute.SortOrder); err != nil {
return fmt.Errorf("insert collectible backdrop: %w", err)
}
}
if _, err := tx.Exec(ctx, `
UPDATE star_gift_collectible_revisions SET status='published', published_at=now() WHERE id=$1`, revisionID); err != nil {
return fmt.Errorf("publish collectible revision: %w", err)
}
if _, err := tx.Exec(ctx, `
UPDATE star_gift_catalog SET collectible_revision_id=$2, updated_at=now() WHERE gift_id=$1`, write.GiftID, revisionID); err != nil {
return fmt.Errorf("activate collectible revision: %w", err)
}
var err error
result, err = collectibleRevisionByID(ctx, tx, revisionID)
return err
})
return result, err
}
func (s *StarGiftStore) ActiveCollectibleRevision(ctx context.Context, giftID int64) (domain.StarGiftCollectibleRevision, bool, error) {
var revisionID int64
err := s.db.QueryRow(ctx, `
SELECT collectible_revision_id FROM star_gift_catalog
WHERE gift_id=$1 AND collectible_revision_id IS NOT NULL`, giftID).Scan(&revisionID)
if errors.Is(err, pgx.ErrNoRows) {
return domain.StarGiftCollectibleRevision{}, false, nil
}
if err != nil {
return domain.StarGiftCollectibleRevision{}, false, fmt.Errorf("get active collectible revision: %w", err)
}
revision, err := collectibleRevisionByID(ctx, s.db, revisionID)
if err != nil {
return domain.StarGiftCollectibleRevision{}, false, err
}
return revision, true, nil
}
func (s *StarGiftStore) CollectibleAvailability(ctx context.Context, giftIDs []int64) (map[int64]domain.StarGiftCollectibleAvailability, error) {
out := make(map[int64]domain.StarGiftCollectibleAvailability, len(giftIDs))
if len(giftIDs) == 0 {
return out, nil
}
rows, err := s.db.Query(ctx, `
SELECT c.gift_id, r.upgrade_stars, r.supply_total, r.issued
FROM star_gift_catalog c
JOIN star_gift_collectible_revisions r ON r.id=c.collectible_revision_id
WHERE c.gift_id=ANY($1) AND r.status='published'`, giftIDs)
if err != nil {
return nil, fmt.Errorf("list collectible availability: %w", err)
}
defer rows.Close()
for rows.Next() {
var giftID int64
var availability domain.StarGiftCollectibleAvailability
if err := rows.Scan(&giftID, &availability.UpgradeStars, &availability.SupplyTotal, &availability.Issued); err != nil {
return nil, fmt.Errorf("scan collectible availability: %w", err)
}
out[giftID] = availability
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("list collectible availability rows: %w", err)
}
return out, nil
}
func collectibleRevisionByID(ctx context.Context, db sqlcgen.DBTX, revisionID int64) (domain.StarGiftCollectibleRevision, error) {
var revision domain.StarGiftCollectibleRevision
var status string
var publishedAt pgtype.Timestamptz
if err := db.QueryRow(ctx, `
SELECT id, gift_id, revision, upgrade_stars, supply_total, issued, slug_prefix, status,
created_by, created_at, published_at
FROM star_gift_collectible_revisions WHERE id=$1`, revisionID).Scan(
&revision.ID, &revision.GiftID, &revision.Revision, &revision.UpgradeStars, &revision.SupplyTotal,
&revision.Issued, &revision.SlugPrefix, &status, &revision.CreatedBy, &revision.CreatedAt, &publishedAt,
); err != nil {
return domain.StarGiftCollectibleRevision{}, fmt.Errorf("get collectible revision: %w", err)
}
revision.Published = status == "published"
if publishedAt.Valid {
revision.PublishedAt = publishedAt.Time
}
var err error
if revision.Models, err = listAnimatedCollectibleAttributes(ctx, db, revisionID, domain.StarGiftCollectibleModel); err != nil {
return domain.StarGiftCollectibleRevision{}, err
}
if revision.Patterns, err = listAnimatedCollectibleAttributes(ctx, db, revisionID, domain.StarGiftCollectiblePattern); err != nil {
return domain.StarGiftCollectibleRevision{}, err
}
if revision.Backdrops, err = listCollectibleBackdrops(ctx, db, revisionID); err != nil {
return domain.StarGiftCollectibleRevision{}, err
}
return revision, nil
}
func listAnimatedCollectibleAttributes(ctx context.Context, db sqlcgen.DBTX, revisionID int64, kind domain.StarGiftCollectibleAttributeKind) ([]domain.StarGiftCollectibleAttribute, error) {
table := "star_gift_collectible_models"
if kind == domain.StarGiftCollectiblePattern {
table = "star_gift_collectible_patterns"
} else if kind != domain.StarGiftCollectibleModel {
return nil, domain.ErrStarGiftCollectibleInvalid
}
rows, err := db.Query(ctx, fmt.Sprintf(`
SELECT a.id, a.collectible_revision_id, a.name, a.rarity_permille, a.sort_order,
a.animation_json::text, a.animation_sha256, a.source_name, a.source_format,
a.width, a.height, a.frame_rate, a.in_point, a.out_point,
d.id, d.access_hash, d.file_reference, d.date, d.mime_type, d.size, d.dc_id,
d.attributes::text, d.thumbs::text
FROM %s a JOIN documents d ON d.id=a.document_id
WHERE a.collectible_revision_id=$1 ORDER BY a.sort_order, a.id`, table), revisionID)
if err != nil {
return nil, fmt.Errorf("list collectible %s attributes: %w", kind, err)
}
defer rows.Close()
out := make([]domain.StarGiftCollectibleAttribute, 0)
for rows.Next() {
attribute := domain.StarGiftCollectibleAttribute{Kind: kind, Document: &domain.Document{}, Animation: &domain.StarGiftAnimation{}}
var attrsJSON, thumbsJSON, sourceFormat string
if err := rows.Scan(&attribute.ID, &attribute.CollectibleRevisionID, &attribute.Name, &attribute.RarityPermille, &attribute.SortOrder,
&attribute.Animation.JSON, &attribute.Animation.SHA256, &attribute.Animation.SourceName, &sourceFormat,
&attribute.Animation.Width, &attribute.Animation.Height, &attribute.Animation.FrameRate, &attribute.Animation.InPoint, &attribute.Animation.OutPoint,
&attribute.Document.ID, &attribute.Document.AccessHash, &attribute.Document.FileReference, &attribute.Document.Date,
&attribute.Document.MimeType, &attribute.Document.Size, &attribute.Document.DCID, &attrsJSON, &thumbsJSON); err != nil {
return nil, err
}
attribute.Animation.SourceFormat = domain.StarGiftAnimationFormat(sourceFormat)
if attribute.Document.Attributes, err = decodeDocumentAttributes(attrsJSON); err != nil {
return nil, err
}
if attribute.Document.Thumbs, err = decodePhotoSizes(thumbsJSON); err != nil {
return nil, err
}
out = append(out, attribute)
}
return out, rows.Err()
}
func listCollectibleBackdrops(ctx context.Context, db sqlcgen.DBTX, revisionID int64) ([]domain.StarGiftCollectibleAttribute, error) {
rows, err := db.Query(ctx, `
SELECT id, collectible_revision_id, name, backdrop_id, center_color, edge_color, pattern_color,
text_color, rarity_permille, sort_order
FROM star_gift_collectible_backdrops WHERE collectible_revision_id=$1 ORDER BY sort_order, id`, revisionID)
if err != nil {
return nil, fmt.Errorf("list collectible backdrops: %w", err)
}
defer rows.Close()
out := make([]domain.StarGiftCollectibleAttribute, 0)
for rows.Next() {
attribute := domain.StarGiftCollectibleAttribute{Kind: domain.StarGiftCollectibleBackdrop}
if err := rows.Scan(&attribute.ID, &attribute.CollectibleRevisionID, &attribute.Name, &attribute.BackdropID,
&attribute.CenterColor, &attribute.EdgeColor, &attribute.PatternColor, &attribute.TextColor,
&attribute.RarityPermille, &attribute.SortOrder); err != nil {
return nil, err
}
out = append(out, attribute)
}
return out, rows.Err()
}
func (s *StarGiftStore) CollectibleAnimationJSON(ctx context.Context, giftID int64, kind domain.StarGiftCollectibleAttributeKind, attributeID int64) ([]byte, bool, error) {
table := "star_gift_collectible_models"
if kind == domain.StarGiftCollectiblePattern {
table = "star_gift_collectible_patterns"
} else if kind != domain.StarGiftCollectibleModel {
return nil, false, nil
}
var raw []byte
err := s.db.QueryRow(ctx, fmt.Sprintf(`
SELECT a.animation_json::text FROM %s a
JOIN star_gift_catalog c ON c.collectible_revision_id=a.collectible_revision_id
WHERE c.gift_id=$1 AND a.id=$2`, table), giftID, attributeID).Scan(&raw)
if errors.Is(err, pgx.ErrNoRows) {
return nil, false, nil
}
if err != nil {
return nil, false, fmt.Errorf("get collectible animation: %w", err)
}
return raw, true, nil
}
func (s *StarGiftStore) UniqueBySlug(ctx context.Context, slug string) (domain.UniqueStarGift, bool, error) {
return s.uniqueByPredicate(ctx, "u.slug=$1", strings.ToLower(strings.TrimSpace(slug)))
}
func (s *StarGiftStore) UniqueByID(ctx context.Context, uniqueGiftID int64) (domain.UniqueStarGift, bool, error) {
return s.uniqueByPredicate(ctx, "u.id=$1", uniqueGiftID)
}
func (s *StarGiftStore) UniqueByIDs(ctx context.Context, uniqueGiftIDs []int64) (map[int64]domain.UniqueStarGift, error) {
out := make(map[int64]domain.UniqueStarGift, len(uniqueGiftIDs))
if len(uniqueGiftIDs) == 0 {
return out, nil
}
rows, err := s.db.Query(ctx, uniqueStarGiftQuery("u.id=ANY($1::bigint[])"), uniqueGiftIDs)
if err != nil {
return nil, fmt.Errorf("list unique star gifts: %w", err)
}
defer rows.Close()
for rows.Next() {
unique, err := scanUniqueStarGift(rows)
if err != nil {
return nil, err
}
out[unique.ID] = unique
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate unique star gifts: %w", err)
}
return out, nil
}
func (s *StarGiftStore) uniqueByPredicate(ctx context.Context, predicate string, value any) (domain.UniqueStarGift, bool, error) {
row := s.db.QueryRow(ctx, uniqueStarGiftQuery(predicate), value)
unique, err := scanUniqueStarGift(row)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return domain.UniqueStarGift{}, false, nil
}
return domain.UniqueStarGift{}, false, err
}
return unique, true, nil
}
func uniqueStarGiftQuery(predicate string) string {
return fmt.Sprintf(`
SELECT u.id, u.gift_id, u.collectible_revision_id, u.source_saved_gift_id, u.title, u.slug, u.num,
u.owner_peer_type, u.owner_peer_id, u.keep_original_details, u.created_at,
r.issued, r.supply_total, sg.from_user_id, sg.owner_peer_type, sg.owner_peer_id,
sg.gift_date, sg.message, sg.name_hidden,
m.id, m.name, m.rarity_permille, md.id, md.access_hash, md.file_reference, md.date,
md.mime_type, md.size, md.dc_id, md.attributes::text, md.thumbs::text,
p.id, p.name, p.rarity_permille, pd.id, pd.access_hash, pd.file_reference, pd.date,
pd.mime_type, pd.size, pd.dc_id, pd.attributes::text, pd.thumbs::text,
b.id, b.name, b.backdrop_id, b.center_color, b.edge_color, b.pattern_color, b.text_color, b.rarity_permille
FROM unique_star_gifts u
JOIN star_gift_collectible_revisions r ON r.id=u.collectible_revision_id
JOIN star_gift_collectible_models m ON m.id=u.model_attribute_id
JOIN documents md ON md.id=m.document_id
JOIN star_gift_collectible_patterns p ON p.id=u.pattern_attribute_id
JOIN documents pd ON pd.id=p.document_id
JOIN star_gift_collectible_backdrops b ON b.id=u.backdrop_attribute_id
JOIN peer_star_gifts sg ON sg.id=u.source_saved_gift_id
WHERE %s`, predicate)
}
func scanUniqueStarGift(row rowScanner) (domain.UniqueStarGift, error) {
var unique domain.UniqueStarGift
var ownerType, originalOwnerType string
unique.Model.Kind = domain.StarGiftCollectibleModel
unique.Pattern.Kind = domain.StarGiftCollectiblePattern
unique.Backdrop.Kind = domain.StarGiftCollectibleBackdrop
unique.Model.Document = &domain.Document{}
unique.Pattern.Document = &domain.Document{}
var modelAttrs, modelThumbs, patternAttrs, patternThumbs string
if err := row.Scan(&unique.ID, &unique.GiftID, &unique.CollectibleRevisionID, &unique.SourceSavedGiftID,
&unique.Title, &unique.Slug, &unique.Num, &ownerType, &unique.Owner.ID, &unique.KeepOriginalDetails,
&unique.CreatedAt, &unique.AvailabilityIssued, &unique.AvailabilityTotal,
&unique.OriginalFromUserID, &originalOwnerType, &unique.OriginalOwner.ID, &unique.OriginalDate,
&unique.OriginalMessage, &unique.OriginalNameHidden,
&unique.Model.ID, &unique.Model.Name, &unique.Model.RarityPermille,
&unique.Model.Document.ID, &unique.Model.Document.AccessHash, &unique.Model.Document.FileReference,
&unique.Model.Document.Date, &unique.Model.Document.MimeType, &unique.Model.Document.Size,
&unique.Model.Document.DCID, &modelAttrs, &modelThumbs,
&unique.Pattern.ID, &unique.Pattern.Name, &unique.Pattern.RarityPermille,
&unique.Pattern.Document.ID, &unique.Pattern.Document.AccessHash, &unique.Pattern.Document.FileReference,
&unique.Pattern.Document.Date, &unique.Pattern.Document.MimeType, &unique.Pattern.Document.Size,
&unique.Pattern.Document.DCID, &patternAttrs, &patternThumbs,
&unique.Backdrop.ID, &unique.Backdrop.Name, &unique.Backdrop.BackdropID, &unique.Backdrop.CenterColor,
&unique.Backdrop.EdgeColor, &unique.Backdrop.PatternColor, &unique.Backdrop.TextColor, &unique.Backdrop.RarityPermille); err != nil {
return domain.UniqueStarGift{}, fmt.Errorf("get unique star gift: %w", err)
}
unique.Owner.Type = domain.PeerType(ownerType)
unique.OriginalOwner.Type = domain.PeerType(originalOwnerType)
unique.Model.CollectibleRevisionID = unique.CollectibleRevisionID
unique.Pattern.CollectibleRevisionID = unique.CollectibleRevisionID
unique.Backdrop.CollectibleRevisionID = unique.CollectibleRevisionID
var err error
if unique.Model.Document.Attributes, err = decodeDocumentAttributes(modelAttrs); err != nil {
return domain.UniqueStarGift{}, err
}
if unique.Model.Document.Thumbs, err = decodePhotoSizes(modelThumbs); err != nil {
return domain.UniqueStarGift{}, err
}
if unique.Pattern.Document.Attributes, err = decodeDocumentAttributes(patternAttrs); err != nil {
return domain.UniqueStarGift{}, err
}
if unique.Pattern.Document.Thumbs, err = decodePhotoSizes(patternThumbs); err != nil {
return domain.UniqueStarGift{}, err
}
return unique, nil
}
func (s *StarGiftStore) ListCollections(ctx context.Context, owner domain.Peer) ([]domain.StarGiftCollection, error) {
rows, err := s.db.Query(ctx, `
SELECT c.collection_id, c.title, c.hash, c.sort_order, c.created_at, c.updated_at, i.saved_gift_id
FROM star_gift_collections c
LEFT JOIN star_gift_collection_items i ON i.collection_id=c.collection_id
WHERE c.owner_peer_type=$1 AND c.owner_peer_id=$2
ORDER BY c.sort_order, c.collection_id, i.sort_order, i.saved_gift_id`, string(owner.Type), owner.ID)
if err != nil {
return nil, fmt.Errorf("list star gift collections: %w", err)
}
defer rows.Close()
out := make([]domain.StarGiftCollection, 0)
index := make(map[int]int)
for rows.Next() {
var collection domain.StarGiftCollection
var giftID pgtype.Int8
if err := rows.Scan(&collection.CollectionID, &collection.Title, &collection.Hash, &collection.SortOrder,
&collection.CreatedAt, &collection.UpdatedAt, &giftID); err != nil {
return nil, err
}
position, ok := index[collection.CollectionID]
if !ok {
collection.Owner = owner
position = len(out)
index[collection.CollectionID] = position
out = append(out, collection)
}
if giftID.Valid {
out[position].GiftIDs = append(out[position].GiftIDs, giftID.Int64)
}
}
return out, rows.Err()
}
func (s *StarGiftStore) CreateCollection(ctx context.Context, owner domain.Peer, title string, savedGiftIDs []int64) (domain.StarGiftCollection, error) {
title = strings.TrimSpace(title)
if !validPostgresStarGiftOwner(owner) || title == "" || len([]rune(title)) > domain.MaxStarGiftCollectionTitleRunes {
return domain.StarGiftCollection{}, domain.ErrStarGiftCollectibleInvalid
}
var result domain.StarGiftCollection
err := withTx(ctx, s.db, "create star gift collection", func(tx pgx.Tx) error {
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, starGiftCollectionLockKey(owner)); err != nil {
return err
}
var count int
if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM star_gift_collections WHERE owner_peer_type=$1 AND owner_peer_id=$2`, string(owner.Type), owner.ID).Scan(&count); err != nil {
return err
}
if count >= domain.MaxStarGiftCollectionsPerPeer {
return domain.ErrStarGiftCollectionsFull
}
ids, err := validatePostgresCollectionGiftIDs(ctx, tx, owner, savedGiftIDs)
if err != nil {
return err
}
result = domain.StarGiftCollection{Owner: owner, Title: title, GiftIDs: ids, SortOrder: count}
result.Hash = domain.StarGiftCollectionHash(title, ids)
if err := tx.QueryRow(ctx, `
INSERT INTO star_gift_collections(owner_peer_type, owner_peer_id, title, sort_order, hash)
VALUES ($1,$2,$3,$4,$5) RETURNING collection_id, created_at, updated_at`, string(owner.Type), owner.ID,
title, count, result.Hash).Scan(&result.CollectionID, &result.CreatedAt, &result.UpdatedAt); err != nil {
return err
}
return replaceCollectionItems(ctx, tx, result.CollectionID, ids)
})
return result, err
}
func (s *StarGiftStore) UpdateCollection(ctx context.Context, owner domain.Peer, collectionID int, patch domain.StarGiftCollectionPatch) (domain.StarGiftCollection, error) {
var result domain.StarGiftCollection
err := withTx(ctx, s.db, "update star gift collection", func(tx pgx.Tx) error {
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, starGiftCollectionLockKey(owner)); err != nil {
return err
}
if err := tx.QueryRow(ctx, `
SELECT title, hash, sort_order, created_at, updated_at FROM star_gift_collections
WHERE owner_peer_type=$1 AND owner_peer_id=$2 AND collection_id=$3 FOR UPDATE`, string(owner.Type), owner.ID, collectionID).Scan(
&result.Title, &result.Hash, &result.SortOrder, &result.CreatedAt, &result.UpdatedAt); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return domain.ErrStarGiftCollectionNotFound
}
return err
}
result.Owner = owner
result.CollectionID = collectionID
rows, err := tx.Query(ctx, `SELECT saved_gift_id FROM star_gift_collection_items WHERE collection_id=$1 ORDER BY sort_order, saved_gift_id`, collectionID)
if err != nil {
return err
}
for rows.Next() {
var id int64
if err := rows.Scan(&id); err != nil {
rows.Close()
return err
}
result.GiftIDs = append(result.GiftIDs, id)
}
rows.Close()
if patch.Title != nil {
title := strings.TrimSpace(*patch.Title)
if title == "" || len([]rune(title)) > domain.MaxStarGiftCollectionTitleRunes {
return domain.ErrStarGiftCollectibleInvalid
}
result.Title = title
}
deleted := make(map[int64]struct{}, len(patch.DeleteIDs))
for _, id := range patch.DeleteIDs {
deleted[id] = struct{}{}
}
next := make([]int64, 0, len(result.GiftIDs)+len(patch.AddIDs))
for _, id := range result.GiftIDs {
if _, ok := deleted[id]; !ok {
next = append(next, id)
}
}
add, err := validatePostgresCollectionGiftIDs(ctx, tx, owner, patch.AddIDs)
if err != nil {
return err
}
next = appendUniquePostgresIDs(next, add...)
if patch.Order != nil {
order, err := validatePostgresCollectionGiftIDs(ctx, tx, owner, patch.Order)
if err != nil || !samePostgresIDSet(order, next) {
return domain.ErrStarGiftCollectibleInvalid
}
next = order
}
if len(next) > domain.MaxStarGiftCollectionItems {
return domain.ErrStarGiftCollectibleInvalid
}
result.GiftIDs = next
result.Hash = domain.StarGiftCollectionHash(result.Title, result.GiftIDs)
if err := tx.QueryRow(ctx, `
UPDATE star_gift_collections SET title=$4, hash=$5, updated_at=now()
WHERE owner_peer_type=$1 AND owner_peer_id=$2 AND collection_id=$3 RETURNING updated_at`,
string(owner.Type), owner.ID, collectionID, result.Title, result.Hash).Scan(&result.UpdatedAt); err != nil {
return err
}
return replaceCollectionItems(ctx, tx, collectionID, result.GiftIDs)
})
return result, err
}
func (s *StarGiftStore) DeleteCollection(ctx context.Context, owner domain.Peer, collectionID int) (bool, error) {
var changed bool
err := withTx(ctx, s.db, "delete star gift collection", func(tx pgx.Tx) error {
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, starGiftCollectionLockKey(owner)); err != nil {
return err
}
tag, err := tx.Exec(ctx, `DELETE FROM star_gift_collections WHERE owner_peer_type=$1 AND owner_peer_id=$2 AND collection_id=$3`, string(owner.Type), owner.ID, collectionID)
if err != nil {
return err
}
changed = tag.RowsAffected() > 0
if changed {
_, err = tx.Exec(ctx, `
WITH ordered AS (
SELECT collection_id, row_number() OVER (ORDER BY sort_order, collection_id) - 1 AS next_order
FROM star_gift_collections WHERE owner_peer_type=$1 AND owner_peer_id=$2
)
UPDATE star_gift_collections c SET sort_order=o.next_order, updated_at=now()
FROM ordered o WHERE c.collection_id=o.collection_id`, string(owner.Type), owner.ID)
}
return err
})
return changed, err
}
func (s *StarGiftStore) ReorderCollections(ctx context.Context, owner domain.Peer, collectionIDs []int) error {
return withTx(ctx, s.db, "reorder star gift collections", func(tx pgx.Tx) error {
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, starGiftCollectionLockKey(owner)); err != nil {
return err
}
rows, err := tx.Query(ctx, `SELECT collection_id FROM star_gift_collections WHERE owner_peer_type=$1 AND owner_peer_id=$2 FOR UPDATE`, string(owner.Type), owner.ID)
if err != nil {
return err
}
existing := make([]int, 0)
for rows.Next() {
var id int
if err := rows.Scan(&id); err != nil {
rows.Close()
return err
}
existing = append(existing, id)
}
rows.Close()
if !samePostgresIntSet(existing, collectionIDs) {
return domain.ErrStarGiftCollectibleInvalid
}
for order, id := range collectionIDs {
if _, err := tx.Exec(ctx, `UPDATE star_gift_collections SET sort_order=$4, updated_at=now() WHERE owner_peer_type=$1 AND owner_peer_id=$2 AND collection_id=$3`, string(owner.Type), owner.ID, id, order); err != nil {
return err
}
}
return nil
})
}
func (s *StarGiftStore) SetPinned(ctx context.Context, owner domain.Peer, savedGiftIDs []int64) error {
return withTx(ctx, s.db, "set pinned star gifts", func(tx pgx.Tx) error {
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, starGiftCollectionLockKey(owner)); err != nil {
return err
}
ids, err := validatePostgresCollectionGiftIDs(ctx, tx, owner, savedGiftIDs)
if err != nil {
return err
}
if _, err := tx.Exec(ctx, `UPDATE peer_star_gifts SET pinned_order=0 WHERE owner_peer_type=$1 AND owner_peer_id=$2 AND pinned_order<>0`, string(owner.Type), owner.ID); err != nil {
return err
}
for order, id := range ids {
if _, err := tx.Exec(ctx, `UPDATE peer_star_gifts SET pinned_order=$2 WHERE id=$1`, id, order+1); err != nil {
return err
}
}
return nil
})
}
func validatePostgresCollectionGiftIDs(ctx context.Context, db sqlcgen.DBTX, owner domain.Peer, ids []int64) ([]int64, error) {
ids = dedupePostgresIDs(ids)
if len(ids) > domain.MaxStarGiftCollectionItems {
return nil, domain.ErrStarGiftCollectibleInvalid
}
if len(ids) == 0 {
return []int64{}, nil
}
rows, err := db.Query(ctx, `
SELECT id FROM peer_star_gifts
WHERE owner_peer_type=$1 AND owner_peer_id=$2 AND NOT converted AND id=ANY($3::bigint[])
FOR UPDATE`, string(owner.Type), owner.ID, ids)
if err != nil {
return nil, err
}
defer rows.Close()
found := make(map[int64]struct{}, len(ids))
for rows.Next() {
var id int64
if err := rows.Scan(&id); err != nil {
return nil, err
}
found[id] = struct{}{}
}
if len(found) != len(ids) {
return nil, domain.ErrStarGiftNotFound
}
return ids, rows.Err()
}
// removeSavedGiftFromCollections runs under the owner advisory lock. It removes
// terminal gifts and updates every affected collection hash in bounded batches,
// so getStarGiftCollections cannot return NotModified for changed membership.
func removeSavedGiftFromCollections(ctx context.Context, tx pgx.Tx, owner domain.Peer, savedGiftID int64) error {
rows, err := tx.Query(ctx, `
SELECT c.collection_id, c.title
FROM star_gift_collections c
JOIN star_gift_collection_items i ON i.collection_id=c.collection_id
WHERE c.owner_peer_type=$1 AND c.owner_peer_id=$2 AND i.saved_gift_id=$3
ORDER BY c.collection_id
FOR UPDATE OF c`, string(owner.Type), owner.ID, savedGiftID)
if err != nil {
return fmt.Errorf("lock converted gift collections: %w", err)
}
titles := make(map[int]string)
ids := make([]int, 0)
for rows.Next() {
var id int
var title string
if err := rows.Scan(&id, &title); err != nil {
rows.Close()
return err
}
ids = append(ids, id)
titles[id] = title
}
if err := rows.Err(); err != nil {
rows.Close()
return err
}
rows.Close()
if len(ids) == 0 {
return nil
}
if _, err := tx.Exec(ctx, `DELETE FROM star_gift_collection_items WHERE saved_gift_id=$1`, savedGiftID); err != nil {
return fmt.Errorf("remove converted gift collection memberships: %w", err)
}
memberships := make(map[int][]int64, len(ids))
itemRows, err := tx.Query(ctx, `
SELECT collection_id, saved_gift_id
FROM star_gift_collection_items
WHERE collection_id=ANY($1::integer[])
ORDER BY collection_id, sort_order, saved_gift_id`, ids)
if err != nil {
return fmt.Errorf("list remaining collection memberships: %w", err)
}
for itemRows.Next() {
var collectionID int
var giftID int64
if err := itemRows.Scan(&collectionID, &giftID); err != nil {
itemRows.Close()
return err
}
memberships[collectionID] = append(memberships[collectionID], giftID)
}
if err := itemRows.Err(); err != nil {
itemRows.Close()
return err
}
itemRows.Close()
hashes := make([]int64, len(ids))
for i, collectionID := range ids {
hashes[i] = domain.StarGiftCollectionHash(titles[collectionID], memberships[collectionID])
}
if _, err := tx.Exec(ctx, `
UPDATE star_gift_collections c SET hash=x.hash, updated_at=now()
FROM unnest($1::integer[], $2::bigint[]) AS x(collection_id, hash)
WHERE c.collection_id=x.collection_id`, ids, hashes); err != nil {
return fmt.Errorf("refresh converted gift collection hashes: %w", err)
}
return nil
}
func replaceCollectionItems(ctx context.Context, tx pgx.Tx, collectionID int, ids []int64) error {
if _, err := tx.Exec(ctx, `DELETE FROM star_gift_collection_items WHERE collection_id=$1`, collectionID); err != nil {
return err
}
for order, id := range ids {
if _, err := tx.Exec(ctx, `INSERT INTO star_gift_collection_items(collection_id, saved_gift_id, sort_order) VALUES ($1,$2,$3)`, collectionID, id, order); err != nil {
return err
}
}
return nil
}
func validPostgresStarGiftOwner(owner domain.Peer) bool {
return owner.ID > 0 && (owner.Type == domain.PeerTypeUser || owner.Type == domain.PeerTypeChannel)
}
func starGiftCollectionLockKey(owner domain.Peer) string {
return fmt.Sprintf("star_gift_collection:%s:%d", owner.Type, owner.ID)
}
func dedupePostgresIDs(ids []int64) []int64 {
out := 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 {
seen[id] = struct{}{}
out = append(out, id)
}
}
return out
}
func appendUniquePostgresIDs(dst []int64, values ...int64) []int64 {
seen := make(map[int64]struct{}, len(dst)+len(values))
for _, id := range dst {
seen[id] = struct{}{}
}
for _, id := range values {
if _, ok := seen[id]; !ok {
seen[id] = struct{}{}
dst = append(dst, id)
}
}
return dst
}
func samePostgresIDSet(a, b []int64) bool {
if len(a) != len(b) {
return false
}
a = append([]int64(nil), a...)
b = append([]int64(nil), b...)
sort.Slice(a, func(i, j int) bool { return a[i] < a[j] })
sort.Slice(b, func(i, j int) bool { return b[i] < b[j] })
for i := range a {
if a[i] != b[i] {
return false
}
}
return true
}
func samePostgresIntSet(a, b []int) bool {
if len(a) != len(b) {
return false
}
seen := make(map[int]struct{}, len(a))
for _, id := range a {
seen[id] = struct{}{}
}
for _, id := range b {
if _, ok := seen[id]; !ok {
return false
}
delete(seen, id)
}
return len(seen) == 0
}

429
internal/store/postgres/star_gift_collectibles_integration_test.go Normal file
View file

@ -0,0 +1,429 @@
package postgres
import (
"context"
"errors"
"fmt"
"testing"
"time"
"telesrv/internal/domain"
)
func TestStarGiftCollectibleUpgradeAggregatePostgres(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
suffix := randomSuffix(t)
users := NewUserStore(pool)
sender := createTestUser(t, ctx, users, "+1778"+suffix+"41", "CollectibleSender", "")
owner := createTestUser(t, ctx, users, "+1778"+suffix+"42", "CollectibleOwner", "")
ownerPeer := domain.Peer{Type: domain.PeerTypeUser, ID: owner.ID}
gifts := NewStarGiftStore(pool)
baseDocumentID := time.Now().UnixNano() & 0x7ffffffffffff000
entry, err := gifts.CreateCatalogRevision(ctx, domain.StarGiftCatalogWrite{
Title: "Comet", Stars: 50, ConvertStars: 25, Enabled: true,
Document: collectibleTestDocument(baseDocumentID, "gift.tgs"),
Blob: collectibleTestBlob(baseDocumentID, "gift"),
Animation: collectibleTestAnimation("gift.tgs"),
Actor: "integration", CommandID: "catalog-" + suffix,
})
if err != nil {
t.Fatalf("create collectible catalog gift: %v", err)
}
poolRevision, err := gifts.PublishCollectibleRevision(ctx, domain.StarGiftCollectibleWrite{
GiftID: entry.Gift.ID, UpgradeStars: 100, SupplyTotal: 10, SlugPrefix: "comet-" + suffix,
Models: []domain.StarGiftCollectibleAttribute{{
Kind: domain.StarGiftCollectibleModel, Name: "Aurora", RarityPermille: 1000,
Document: collectibleTestDocumentPtr(baseDocumentID+1, "model.tgs"),
Blob: collectibleTestBlobPtr(baseDocumentID+1, "model"), Animation: collectibleTestAnimationPtr("model.tgs"),
}},
Patterns: []domain.StarGiftCollectibleAttribute{{
Kind: domain.StarGiftCollectiblePattern, Name: "Orbit", RarityPermille: 1000,
Document: collectibleTestDocumentPtr(baseDocumentID+2, "pattern.tgs"),
Blob: collectibleTestBlobPtr(baseDocumentID+2, "pattern"), Animation: collectibleTestAnimationPtr("pattern.tgs"),
}},
Backdrops: []domain.StarGiftCollectibleAttribute{{
Kind: domain.StarGiftCollectibleBackdrop, Name: "Midnight", BackdropID: 1,
CenterColor: 0x112233, EdgeColor: 0x223344, PatternColor: 0x334455, TextColor: 0xffffff,
RarityPermille: 1000,
}},
Actor: "integration", CommandID: "collectibles-" + suffix,
})
if err != nil {
t.Fatalf("publish collectible pool: %v", err)
}
if !poolRevision.Published || poolRevision.Issued != 0 || len(poolRevision.Models) != 1 || len(poolRevision.Patterns) != 1 || len(poolRevision.Backdrops) != 1 {
t.Fatalf("published pool = %+v", poolRevision)
}
availability, err := gifts.CollectibleAvailability(ctx, []int64{entry.Gift.ID, entry.Gift.ID + 1})
if err != nil {
t.Fatalf("collectible availability: %v", err)
}
if got, ok := availability[entry.Gift.ID]; !ok || got.UpgradeStars != 100 || got.SupplyTotal != 10 || got.Issued != 0 {
t.Fatalf("collectible availability = %+v, want active published pool", availability)
}
if _, ok := availability[entry.Gift.ID+1]; ok {
t.Fatalf("unknown gift must not have collectible availability: %+v", availability)
}
if _, err := pool.Exec(ctx, `UPDATE star_gift_collectible_revisions SET issued=issued WHERE id=$1`, poolRevision.ID); err == nil {
t.Fatal("published collectible revision accepted a non-advancing issuance update")
}
var guardedIssued int
if err := pool.QueryRow(ctx, `SELECT issued FROM star_gift_collectible_revisions WHERE id=$1`, poolRevision.ID).Scan(&guardedIssued); err != nil || guardedIssued != 0 {
t.Fatalf("issued after rejected manual update = %d err %v, want 0", guardedIssued, err)
}
savedID, err := gifts.Create(ctx, domain.SavedStarGift{
Owner: ownerPeer, FromUserID: sender.ID, GiftID: entry.Gift.ID, RevisionID: entry.Gift.RevisionID,
MsgID: 700001, Date: 1700001000, ConvertStars: 25, Message: "original",
})
if err != nil {
t.Fatalf("create saved gift: %v", err)
}
stars := NewStarsStore(pool)
if _, _, err := stars.EnsureGrant(ctx, owner.ID, 1000, 1700001001); err != nil {
t.Fatalf("grant upgrade stars: %v", err)
}
messages := NewMessageStore(pool)
upgrades := NewStarGiftUpgradeStore(pool, messages)
req := domain.StarGiftUpgradeRequest{
UserID: owner.ID, Ref: domain.SavedStarGiftRef{Owner: ownerPeer, MsgID: 700001},
KeepOriginalDetails: true, ChargeStars: 100, FormID: 991,
CommandKey: "paid-" + suffix, Date: 1700001002,
}
upgraded, err := upgrades.UpgradeStarGift(ctx, req)
if err != nil {
t.Fatalf("upgrade star gift: %v", err)
}
if upgraded.Duplicate || upgraded.Unique.Num != 1 || upgraded.Unique.Slug != "comet-"+suffix+"-1" ||
upgraded.Unique.Model.Name != "Aurora" || upgraded.Unique.Pattern.Name != "Orbit" ||
upgraded.Unique.Backdrop.Name != "Midnight" || upgraded.Balance.Balance != 900 ||
upgraded.Saved.ID != savedID || upgraded.Saved.UniqueGiftID != upgraded.Unique.ID || upgraded.Saved.UpgradeMsgID <= 0 {
t.Fatalf("upgrade result = %+v", upgraded)
}
ownerMessage := upgraded.Send.RecipientMessage
if ownerMessage.OwnerUserID != owner.ID || ownerMessage.Pts <= 0 || ownerMessage.Media == nil ||
ownerMessage.Media.ServiceAction == nil || ownerMessage.Media.ServiceAction.Kind != domain.MessageServiceActionStarGiftUnique ||
ownerMessage.Media.ServiceAction.StarGiftUnique == nil || ownerMessage.Media.ServiceAction.StarGiftUnique.Gift.ID != upgraded.Unique.ID {
t.Fatalf("owner upgrade service message = %+v", ownerMessage)
}
var (
issued, uniqueCount, commandCount int
reason string
)
if err := pool.QueryRow(ctx, `SELECT issued FROM star_gift_collectible_revisions WHERE id=$1`, poolRevision.ID).Scan(&issued); err != nil {
t.Fatal(err)
}
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM unique_star_gifts WHERE source_saved_gift_id=$1`, savedID).Scan(&uniqueCount); err != nil {
t.Fatal(err)
}
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM star_gift_upgrade_commands WHERE source_saved_gift_id=$1`, savedID).Scan(&commandCount); err != nil {
t.Fatal(err)
}
if err := pool.QueryRow(ctx, `SELECT reason FROM stars_transactions WHERE user_id=$1 ORDER BY id DESC LIMIT 1`, owner.ID).Scan(&reason); err != nil {
t.Fatal(err)
}
if issued != 1 || uniqueCount != 1 || commandCount != 1 || reason != string(domain.StarsReasonGiftUpgrade) {
t.Fatalf("durable aggregate issued=%d unique=%d command=%d reason=%q", issued, uniqueCount, commandCount, reason)
}
replayed, err := upgrades.UpgradeStarGift(ctx, req)
if err != nil {
t.Fatalf("replay upgrade: %v", err)
}
if !replayed.Duplicate || replayed.Unique.ID != upgraded.Unique.ID || replayed.Balance.Balance != 900 {
t.Fatalf("replayed upgrade = %+v", replayed)
}
conflictingReplay := req
conflictingReplay.KeepOriginalDetails = false
if _, err := upgrades.UpgradeStarGift(ctx, conflictingReplay); err == nil {
t.Fatal("same command key with a changed semantic payload must not replay")
}
if _, err := upgrades.UpgradeStarGift(ctx, domain.StarGiftUpgradeRequest{
UserID: owner.ID, Ref: req.Ref, ChargeStars: 100, FormID: 992,
CommandKey: "different-" + suffix, Date: 1700001003,
}); !errors.Is(err, domain.ErrStarGiftAlreadyUpgraded) {
t.Fatalf("second logical upgrade err = %v", err)
}
bal, err := stars.GetBalance(ctx, owner.ID)
if err != nil || bal.Balance != 900 {
t.Fatalf("balance after retries = %+v err %v", bal, err)
}
prepaidSavedID, err := gifts.Create(ctx, domain.SavedStarGift{
Owner: ownerPeer, FromUserID: sender.ID, GiftID: entry.Gift.ID, RevisionID: entry.Gift.RevisionID,
// A later pool revision may raise the current price; the historical paid
// amount remains an entitlement instead of being compared to that price.
MsgID: 700002, Date: 1700001004, ConvertStars: 25, PrepaidUpgradeStars: 50,
})
if err != nil {
t.Fatalf("create prepaid saved gift: %v", err)
}
prepaid, err := upgrades.UpgradeStarGift(ctx, domain.StarGiftUpgradeRequest{
UserID: owner.ID, Ref: domain.SavedStarGiftRef{Owner: ownerPeer, MsgID: 700002},
RequirePrepaid: true, CommandKey: "prepaid-" + suffix, Date: 1700001005,
})
if err != nil {
t.Fatalf("free prepaid upgrade: %v", err)
}
if prepaid.Saved.ID != prepaidSavedID || prepaid.Unique.Num != 2 || prepaid.Balance.Balance != 900 ||
prepaid.Send.RecipientMessage.Media.ServiceAction.StarGiftUnique == nil ||
!prepaid.Send.RecipientMessage.Media.ServiceAction.StarGiftUnique.PrepaidUpgrade {
t.Fatalf("prepaid upgrade = %+v", prepaid)
}
insufficientSavedID, err := gifts.Create(ctx, domain.SavedStarGift{
Owner: ownerPeer, FromUserID: sender.ID, GiftID: entry.Gift.ID, RevisionID: entry.Gift.RevisionID,
MsgID: 700003, Date: 1700001006, ConvertStars: 25,
})
if err != nil {
t.Fatalf("create insufficient saved gift: %v", err)
}
if _, err := stars.Debit(ctx, owner.ID, 850, domain.StarsReasonReaction,
domain.Peer{Type: domain.PeerTypeChannel, ID: 777001}, 1700001007, "paid reaction", ""); err != nil {
t.Fatalf("seed isolated paid reaction debit: %v", err)
}
if _, err := upgrades.UpgradeStarGift(ctx, domain.StarGiftUpgradeRequest{
UserID: owner.ID, Ref: domain.SavedStarGiftRef{Owner: ownerPeer, MsgID: 700003},
ChargeStars: 100, CommandKey: "insufficient-" + suffix, Date: 1700001008,
}); !errors.Is(err, domain.ErrStarsInsufficient) {
t.Fatalf("insufficient upgrade err = %v", err)
}
insufficientSaved, found, err := gifts.GetByRef(ctx, domain.SavedStarGiftRef{Owner: ownerPeer, MsgID: 700003})
if err != nil || !found || insufficientSaved.ID != insufficientSavedID || insufficientSaved.UniqueGiftID != 0 {
t.Fatalf("saved gift after rejected upgrade = %+v found %v err %v", insufficientSaved, found, err)
}
if err := pool.QueryRow(ctx, `SELECT issued FROM star_gift_collectible_revisions WHERE id=$1`, poolRevision.ID).Scan(&issued); err != nil || issued != 2 {
t.Fatalf("issued after rejected upgrade = %d err %v, want 2", issued, err)
}
if err := pool.QueryRow(ctx, `SELECT reason FROM stars_transactions WHERE user_id=$1 ORDER BY id DESC LIMIT 1`, owner.ID).Scan(&reason); err != nil || reason != string(domain.StarsReasonReaction) {
t.Fatalf("paid reaction ledger reason after rejected upgrade = %q err %v", reason, err)
}
collection, err := gifts.CreateCollection(ctx, ownerPeer, "Favorites", []int64{savedID})
if err != nil {
t.Fatalf("create unique collection: %v", err)
}
filtered, err := gifts.ListByOwnerFiltered(ctx, domain.SavedStarGiftFilter{Owner: ownerPeer, CollectionID: collection.CollectionID, Limit: 10})
if err != nil || filtered.Count != 1 || len(filtered.Gifts) != 1 || filtered.Gifts[0].UniqueGiftID != upgraded.Unique.ID {
t.Fatalf("collection filter = %+v err %v", filtered, err)
}
if err := gifts.SetPinned(ctx, ownerPeer, []int64{savedID}); err != nil {
t.Fatalf("pin unique gift: %v", err)
}
pinned, found, err := gifts.GetByRef(ctx, req.Ref)
if err != nil || !found || pinned.PinnedOrder != 1 || len(pinned.CollectionIDs) != 1 || pinned.CollectionIDs[0] != collection.CollectionID {
t.Fatalf("pinned saved gift = %+v found %v err %v", pinned, found, err)
}
concurrentOwner := createTestUser(t, ctx, users, "+1778"+suffix+"43", "ConcurrentOwner", "")
concurrentPeer := domain.Peer{Type: domain.PeerTypeUser, ID: concurrentOwner.ID}
if _, err := gifts.Create(ctx, domain.SavedStarGift{
Owner: concurrentPeer, FromUserID: sender.ID, GiftID: entry.Gift.ID, RevisionID: entry.Gift.RevisionID,
MsgID: 700004, Date: 1700001010, ConvertStars: 25,
}); err != nil {
t.Fatalf("create concurrent upgrade target: %v", err)
}
if _, _, err := stars.EnsureGrant(ctx, concurrentOwner.ID, 150, 1700001011); err != nil {
t.Fatalf("grant concurrent balance: %v", err)
}
type concurrentDebitResult struct {
kind string
err error
}
start := make(chan struct{})
results := make(chan concurrentDebitResult, 2)
go func() {
<-start
_, err := upgrades.UpgradeStarGift(ctx, domain.StarGiftUpgradeRequest{
UserID: concurrentOwner.ID, Ref: domain.SavedStarGiftRef{Owner: concurrentPeer, MsgID: 700004},
ChargeStars: 100, FormID: 993, CommandKey: "concurrent-upgrade-" + suffix, Date: 1700001012,
})
results <- concurrentDebitResult{kind: "gift_upgrade", err: err}
}()
go func() {
<-start
_, err := stars.Debit(ctx, concurrentOwner.ID, 100, domain.StarsReasonReaction,
domain.Peer{Type: domain.PeerTypeChannel, ID: 777002}, 1700001012, "paid reaction", "")
results <- concurrentDebitResult{kind: "paid_reaction", err: err}
}()
close(start)
firstResult, secondResult := <-results, <-results
successes := 0
for _, result := range []concurrentDebitResult{firstResult, secondResult} {
if result.err == nil {
successes++
continue
}
if !errors.Is(result.err, domain.ErrStarsInsufficient) {
t.Fatalf("concurrent %s err = %v, want Stars insufficient for loser", result.kind, result.err)
}
}
if successes != 1 {
t.Fatalf("concurrent debit results = %+v / %+v, want exactly one success", firstResult, secondResult)
}
concurrentBalance, err := stars.GetBalance(ctx, concurrentOwner.ID)
if err != nil || concurrentBalance.Balance != 50 {
t.Fatalf("concurrent balance = %+v err %v, want 50", concurrentBalance, err)
}
reasonRows, err := pool.Query(ctx, `SELECT reason FROM stars_transactions WHERE user_id=$1 AND amount<0 ORDER BY id`, concurrentOwner.ID)
if err != nil {
t.Fatalf("list concurrent debit reasons: %v", err)
}
var debitReasons []string
for reasonRows.Next() {
var got string
if err := reasonRows.Scan(&got); err != nil {
reasonRows.Close()
t.Fatal(err)
}
debitReasons = append(debitReasons, got)
}
if err := reasonRows.Err(); err != nil {
reasonRows.Close()
t.Fatal(err)
}
reasonRows.Close()
if len(debitReasons) != 1 || (debitReasons[0] != string(domain.StarsReasonGiftUpgrade) && debitReasons[0] != string(domain.StarsReasonReaction)) {
t.Fatalf("concurrent debit reasons = %+v, want exactly one isolated business reason", debitReasons)
}
soldOutEntry, err := gifts.CreateCatalogRevision(ctx, domain.StarGiftCatalogWrite{
Title: "Nova", Stars: 25, ConvertStars: 10, Enabled: true,
Document: collectibleTestDocument(baseDocumentID+100, "nova.tgs"),
Blob: collectibleTestBlob(baseDocumentID+100, "nova"), Animation: collectibleTestAnimation("nova.tgs"),
Actor: "integration", CommandID: "soldout-catalog-" + suffix,
})
if err != nil {
t.Fatalf("create sold-out catalog: %v", err)
}
soldOutRevision, err := gifts.PublishCollectibleRevision(ctx, domain.StarGiftCollectibleWrite{
GiftID: soldOutEntry.Gift.ID, UpgradeStars: 10, SupplyTotal: 1, SlugPrefix: "nova-" + suffix,
Models: []domain.StarGiftCollectibleAttribute{{
Kind: domain.StarGiftCollectibleModel, Name: "Nova", RarityPermille: 1000,
Document: collectibleTestDocumentPtr(baseDocumentID+101, "nova-model.tgs"),
Blob: collectibleTestBlobPtr(baseDocumentID+101, "nova-model"), Animation: collectibleTestAnimationPtr("nova-model.tgs"),
}},
Patterns: []domain.StarGiftCollectibleAttribute{{
Kind: domain.StarGiftCollectiblePattern, Name: "Ray", RarityPermille: 1000,
Document: collectibleTestDocumentPtr(baseDocumentID+102, "nova-pattern.tgs"),
Blob: collectibleTestBlobPtr(baseDocumentID+102, "nova-pattern"), Animation: collectibleTestAnimationPtr("nova-pattern.tgs"),
}},
Backdrops: []domain.StarGiftCollectibleAttribute{{
Kind: domain.StarGiftCollectibleBackdrop, Name: "Void", BackdropID: 2,
CenterColor: 0x101010, EdgeColor: 0x202020, PatternColor: 0x303030, TextColor: 0xffffff,
RarityPermille: 1000,
}},
Actor: "integration", CommandID: "soldout-pool-" + suffix,
})
if err != nil {
t.Fatalf("publish sold-out pool: %v", err)
}
soldOutOwner := createTestUser(t, ctx, users, "+1778"+suffix+"44", "SoldOutOwner", "")
soldOutPeer := domain.Peer{Type: domain.PeerTypeUser, ID: soldOutOwner.ID}
for index, msgID := range []int{700010, 700011} {
if _, err := gifts.Create(ctx, domain.SavedStarGift{
Owner: soldOutPeer, FromUserID: sender.ID, GiftID: soldOutEntry.Gift.ID, RevisionID: soldOutEntry.Gift.RevisionID,
MsgID: msgID, Date: 1700001020 + index, ConvertStars: 10,
}); err != nil {
t.Fatalf("create sold-out target %d: %v", msgID, err)
}
}
if _, _, err := stars.EnsureGrant(ctx, soldOutOwner.ID, 100, 1700001022); err != nil {
t.Fatalf("grant sold-out owner balance: %v", err)
}
if _, err := upgrades.UpgradeStarGift(ctx, domain.StarGiftUpgradeRequest{
UserID: soldOutOwner.ID, Ref: domain.SavedStarGiftRef{Owner: soldOutPeer, MsgID: 700010},
ChargeStars: 10, CommandKey: "soldout-first-" + suffix, Date: 1700001023,
}); err != nil {
t.Fatalf("fill collectible supply: %v", err)
}
balanceBeforeSoldOut, _ := stars.GetBalance(ctx, soldOutOwner.ID)
if _, err := upgrades.UpgradeStarGift(ctx, domain.StarGiftUpgradeRequest{
UserID: soldOutOwner.ID, Ref: domain.SavedStarGiftRef{Owner: soldOutPeer, MsgID: 700011},
ChargeStars: 10, CommandKey: "soldout-second-" + suffix, Date: 1700001024,
}); !errors.Is(err, domain.ErrStarGiftCollectibleSoldOut) {
t.Fatalf("sold-out upgrade err = %v", err)
}
balanceAfterSoldOut, _ := stars.GetBalance(ctx, soldOutOwner.ID)
var soldOutIssued int
if err := pool.QueryRow(ctx, `SELECT issued FROM star_gift_collectible_revisions WHERE id=$1`, soldOutRevision.ID).Scan(&soldOutIssued); err != nil || soldOutIssued != 1 || balanceAfterSoldOut.Balance != balanceBeforeSoldOut.Balance {
t.Fatalf("sold-out state issued=%d balance=%d->%d err=%v", soldOutIssued, balanceBeforeSoldOut.Balance, balanceAfterSoldOut.Balance, err)
}
ordinaryCollection, err := gifts.CreateCollection(ctx, ownerPeer, "Ordinary", []int64{insufficientSavedID})
if err != nil {
t.Fatalf("create ordinary collection: %v", err)
}
converted, err := gifts.MarkConverted(ctx, domain.SavedStarGiftRef{Owner: ownerPeer, MsgID: 700003})
if err != nil || !converted.Converted || converted.PinnedOrder != 0 || len(converted.CollectionIDs) != 0 {
t.Fatalf("convert collection member = %+v err %v", converted, err)
}
collections, err := gifts.ListCollections(ctx, ownerPeer)
if err != nil {
t.Fatalf("list collections after conversion: %v", err)
}
foundOrdinary := false
for _, got := range collections {
if got.CollectionID != ordinaryCollection.CollectionID {
continue
}
foundOrdinary = true
if len(got.GiftIDs) != 0 || got.Hash != domain.StarGiftCollectionHash(got.Title, nil) || got.Hash == ordinaryCollection.Hash {
t.Fatalf("ordinary collection after conversion = %+v", got)
}
}
if !foundOrdinary {
t.Fatal("ordinary collection disappeared after member conversion")
}
filteredAfterConvert, err := gifts.ListByOwnerFiltered(ctx, domain.SavedStarGiftFilter{
Owner: ownerPeer, CollectionID: ordinaryCollection.CollectionID, Limit: 10,
})
if err != nil || filteredAfterConvert.Count != 0 || len(filteredAfterConvert.Gifts) != 0 {
t.Fatalf("converted collection filter = %+v err %v, want empty", filteredAfterConvert, err)
}
}
func collectibleTestAnimation(name string) domain.StarGiftAnimation {
return domain.StarGiftAnimation{
SourceName: name, SourceFormat: domain.StarGiftAnimationTGS,
JSON: []byte(`{"v":"5.7","w":512,"h":512,"fr":30,"ip":0,"op":30,"layers":[{}]}`),
TGS: []byte("test"), SHA256: make([]byte, 32), Width: 512, Height: 512, FrameRate: 30, OutPoint: 30,
}
}
func collectibleTestAnimationPtr(name string) *domain.StarGiftAnimation {
animation := collectibleTestAnimation(name)
return &animation
}
func collectibleTestDocument(id int64, name string) domain.Document {
return domain.Document{
ID: id, AccessHash: id + 100, FileReference: []byte("collectible-test"), Date: 1700001000,
MimeType: "application/x-tgsticker", Size: 4, DCID: 2,
Attributes: []domain.DocumentAttribute{
{Kind: domain.DocAttrImageSize, W: 512, H: 512},
{Kind: domain.DocAttrSticker, Alt: "🎁"},
{Kind: domain.DocAttrFilename, FileName: name},
},
}
}
func collectibleTestDocumentPtr(id int64, name string) *domain.Document {
document := collectibleTestDocument(id, name)
return &document
}
func collectibleTestBlob(id int64, suffix string) domain.FileBlob {
return domain.FileBlob{
LocationKey: fmt.Sprintf("doc:%d", id), Backend: domain.MediaBackendLocalFS,
ObjectKey: "collectible-integration-" + suffix, Size: 4, SHA256: make([]byte, 32), MimeType: "application/x-tgsticker",
}
}
func collectibleTestBlobPtr(id int64, suffix string) *domain.FileBlob {
blob := collectibleTestBlob(id, suffix)
return &blob
}

65
internal/store/postgres/star_gift_integration_test.go
View file

@ -3,13 +3,15 @@ package postgres
import (
"context"
"errors"
"fmt"
"testing"
"time"
"telesrv/internal/domain"
)
// TestStarGiftStorePostgres 回归迁移 0011：用户收到礼物实例对真实 PG 的 CRUD
// （创建 / keyset 分页 / excludeUnsaved / 隐藏切换 / 转换幂等）。
// TestStarGiftStorePostgres 回归迁移 0089：目录不可变版本与用户收到礼物实例对真实 PG 的 CRUD
// （版本固定 / 创建 / keyset 分页 / excludeUnsaved / 隐藏切换 / 转换幂等）。
func TestStarGiftStorePostgres(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
@ -26,21 +28,72 @@ func TestStarGiftStorePostgres(t *testing.T) {
t.Fatalf("create sender: %v", err)
}
ownerPeer := domain.Peer{Type: domain.PeerTypeUser, ID: owner.ID}
docID := time.Now().UnixNano() & 0x7fffffffffffffff
documentIDs := []int64{docID}
locationKeys := []string{"doc:" + fmt.Sprint(docID)}
entry, err := st.CreateCatalogRevision(ctx, domain.StarGiftCatalogWrite{
Stars: 50, ConvertStars: 50, Enabled: true, Document: domain.Document{
ID: docID, AccessHash: docID + 1, MimeType: "application/x-tgsticker", Size: 4, DCID: 2,
Attributes: []domain.DocumentAttribute{{Kind: domain.DocAttrSticker}},
},
Blob: domain.FileBlob{LocationKey: "doc:" + fmt.Sprint(docID), Backend: domain.MediaBackendLocalFS, ObjectKey: "test-star-gift", Size: 4, SHA256: make([]byte, 32), MimeType: "application/x-tgsticker"},
Animation: domain.StarGiftAnimation{JSON: []byte(`{"v":"5.7","w":512,"h":512,"fr":30,"ip":0,"op":30,"layers":[{}]}`), SHA256: make([]byte, 32), SourceFormat: domain.StarGiftAnimationTGS, Width: 512, Height: 512},
Actor: "test", CommandID: "test-star-gift-" + suffix,
})
if err != nil {
t.Fatalf("create catalog gift: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, "DELETE FROM peer_star_gifts WHERE owner_peer_id IN ($1, $2)", owner.ID, int64(987654321))
tx, _ := pool.Begin(ctx)
if tx != nil {
_, _ = tx.Exec(ctx, "DELETE FROM star_gift_catalog WHERE gift_id=$1", entry.Gift.ID)
_, _ = tx.Exec(ctx, "DELETE FROM star_gift_catalog_revisions WHERE gift_id=$1", entry.Gift.ID)
_, _ = tx.Exec(ctx, "DELETE FROM file_blobs WHERE location_key = ANY($1::text[])", locationKeys)
_, _ = tx.Exec(ctx, "DELETE FROM documents WHERE id = ANY($1::bigint[])", documentIDs)
_ = tx.Commit(ctx)
}
_, _ = pool.Exec(ctx, "DELETE FROM users WHERE id = ANY($1::bigint[])", []int64{owner.ID, from.ID})
})
// 创建三份礼物（msg_id 递增）。
for i := 0; i < 3; i++ {
if _, err := st.Create(ctx, domain.SavedStarGift{
Owner: ownerPeer, FromUserID: from.ID, GiftID: 8001, MsgID: 100 + i,
Owner: ownerPeer, FromUserID: from.ID, GiftID: entry.Gift.ID, RevisionID: entry.Gift.RevisionID, MsgID: 100 + i,
Date: 1700000000 + i, ConvertStars: 50,
}); err != nil {
t.Fatalf("create gift #%d: %v", i, err)
}
}
// active revision 更新后，已收到的礼物必须继续固定到购买瞬间的 immutable revision。
docID2 := docID + 1
documentIDs = append(documentIDs, docID2)
locationKeys = append(locationKeys, "doc:"+fmt.Sprint(docID2))
updated, err := st.CreateCatalogRevision(ctx, domain.StarGiftCatalogWrite{
GiftID: entry.Gift.ID, Title: "Revision 2", Stars: 75, ConvertStars: 25, Enabled: true,
Document: domain.Document{
ID: docID2, AccessHash: docID2 + 1, MimeType: "application/x-tgsticker", Size: 4, DCID: 2,
Attributes: []domain.DocumentAttribute{{Kind: domain.DocAttrSticker}},
},
Blob: domain.FileBlob{LocationKey: "doc:" + fmt.Sprint(docID2), Backend: domain.MediaBackendLocalFS, ObjectKey: "test-star-gift-v2", Size: 4, SHA256: make([]byte, 32), MimeType: "application/x-tgsticker"},
Animation: domain.StarGiftAnimation{JSON: []byte(`{"v":"5.7","w":512,"h":512,"fr":30,"ip":0,"op":30,"layers":[{}]}`), SHA256: make([]byte, 32), SourceFormat: domain.StarGiftAnimationTGS, Width: 512, Height: 512},
Actor: "test", CommandID: "test-star-gift-v2-" + suffix,
})
if err != nil {
t.Fatalf("create catalog revision 2: %v", err)
}
if updated.Revision != 2 || updated.Gift.RevisionID == entry.Gift.RevisionID {
t.Fatalf("revision 2 = %+v, want a new immutable revision", updated)
}
if updated.ReceivedCount != 3 {
t.Fatalf("revision 2 received count = %d, want all 3 historical instances", updated.ReceivedCount)
}
historical, found, err := st.CatalogRevision(ctx, entry.Gift.RevisionID)
if err != nil || !found || historical.Stars != 50 || historical.Sticker.ID != docID {
t.Fatalf("historical revision = %+v found %v err %v", historical, found, err)
}
// keyset 分页：每页 2，末页省略游标。
page1, err := st.ListByOwner(ctx, ownerPeer, false, "", 2)
if err != nil {
@ -71,7 +124,7 @@ func TestStarGiftStorePostgres(t *testing.T) {
// GetByRef(user msg_id)。
g, found, err := st.GetByRef(ctx, domain.SavedStarGiftRef{Owner: ownerPeer, MsgID: 100})
if err != nil || !found || g.GiftID != 8001 || g.ConvertStars != 50 {
if err != nil || !found || g.GiftID != entry.Gift.ID || g.RevisionID != entry.Gift.RevisionID || g.ConvertStars != 50 {
t.Fatalf("get = %+v found %v err %v", g, found, err)
}
@ -101,13 +154,13 @@ func TestStarGiftStorePostgres(t *testing.T) {
// 频道礼物用 saved_id 定位，和用户 msg_id 身份键隔离。
channelPeer := domain.Peer{Type: domain.PeerTypeChannel, ID: 987654321}
if _, err := st.Create(ctx, domain.SavedStarGift{
Owner: ownerPeer, FromUserID: from.ID, GiftID: 8001, MsgID: 700,
Owner: ownerPeer, FromUserID: from.ID, GiftID: entry.Gift.ID, RevisionID: entry.Gift.RevisionID, MsgID: 700,
Date: 1700000100, ConvertStars: 50,
}); err != nil {
t.Fatalf("create user gift with same msg_id namespace: %v", err)
}
channelSavedID, err := st.Create(ctx, domain.SavedStarGift{
Owner: channelPeer, FromUserID: from.ID, GiftID: 8001, MsgID: 0, SavedID: 0,
Owner: channelPeer, FromUserID: from.ID, GiftID: entry.Gift.ID, RevisionID: entry.Gift.RevisionID, MsgID: 0, SavedID: 0,
Date: 1700000101, ConvertStars: 50,
})
if err != nil {

365
internal/store/postgres/star_gift_upgrade.go Normal file
View file

@ -0,0 +1,365 @@
package postgres
import (
"context"
"crypto/rand"
"crypto/sha256"
"encoding/binary"
"errors"
"fmt"
"math/big"
"strings"
"github.com/jackc/pgx/v5"
"telesrv/internal/domain"
"telesrv/internal/store"
"telesrv/internal/store/postgres/sqlcgen"
)
// StarGiftUpgradeStore is the PostgreSQL aggregate coordinator for collectible
// upgrades. It intentionally shares MessageStore's allocator and transaction
// machinery so Stars, issuance, the saved gift and durable updates commit once.
type StarGiftUpgradeStore struct {
db sqlcgen.DBTX
messages *MessageStore
}
func NewStarGiftUpgradeStore(db sqlcgen.DBTX, messages *MessageStore) *StarGiftUpgradeStore {
return &StarGiftUpgradeStore{db: db, messages: messages}
}
func (s *StarGiftUpgradeStore) UpgradeStarGift(ctx context.Context, req domain.StarGiftUpgradeRequest) (domain.StarGiftUpgradeResult, error) {
if s == nil || s.db == nil || s.messages == nil || req.UserID <= 0 || !req.Ref.Valid() ||
req.Ref.Owner != (domain.Peer{Type: domain.PeerTypeUser, ID: req.UserID}) ||
req.ChargeStars < 0 || req.Date <= 0 || strings.TrimSpace(req.CommandKey) == "" || len(req.CommandKey) > 256 {
return domain.StarGiftUpgradeResult{}, domain.ErrStarGiftCollectibleInvalid
}
saved, found, err := NewStarGiftStore(s.db).GetByRef(ctx, req.Ref)
if err != nil {
return domain.StarGiftUpgradeResult{}, err
}
if !found || saved.FromUserID <= 0 {
return domain.StarGiftUpgradeResult{}, domain.ErrStarGiftNotFound
}
commandKey := strings.TrimSpace(req.CommandKey)
fingerprint := sha256.Sum256([]byte(fmt.Sprintf(
"telesrv:star-gift-upgrade:v1:%s:%d:%d:%t:%d:%t",
commandKey, saved.ID, req.ChargeStars, req.RequirePrepaid, req.FormID, req.KeepOriginalDetails,
)))
randomID := starGiftUpgradeRandomID(saved.FromUserID, req.UserID, commandKey)
placeholder := &domain.MessageMedia{
Kind: domain.MessageMediaKindService,
ServiceAction: &domain.MessageServiceAction{
Kind: domain.MessageServiceActionStarGiftUnique,
StarGiftUnique: &domain.MessageStarGiftUniqueAction{Upgrade: true, Saved: true},
},
}
messageReq := domain.SendPrivateTextRequest{
SenderUserID: saved.FromUserID,
RecipientUserID: req.UserID,
RandomID: randomID,
Media: placeholder,
Date: req.Date,
OriginAuthKeyID: req.OriginAuthKeyID,
OriginSessionID: req.OriginSessionID,
OriginUserID: req.UserID,
IdempotencyFingerprint: fingerprint[:],
}
var result domain.StarGiftUpgradeResult
hooks := privateSendTxHooks{
before: func(ctx context.Context, tx pgx.Tx, messageReq *domain.SendPrivateTextRequest) error {
locked, err := lockSavedStarGiftForUpgrade(ctx, tx, req.Ref)
if err != nil {
return err
}
if locked.ID != saved.ID || locked.FromUserID != saved.FromUserID {
return domain.ErrStarGiftCollectibleInvalid
}
if locked.Converted {
return domain.ErrStarGiftAlreadyConverted
}
if locked.UniqueGiftID != 0 {
return domain.ErrStarGiftAlreadyUpgraded
}
revision, err := lockActiveCollectibleRevision(ctx, tx, locked.GiftID)
if err != nil {
return err
}
if revision.Issued >= revision.SupplyTotal {
return domain.ErrStarGiftCollectibleSoldOut
}
if req.RequirePrepaid {
// Prepayment is an entitlement captured at gift purchase time. A
// later published revision may change the current price, but must not
// retroactively invalidate that already-paid entitlement.
if req.ChargeStars != 0 || locked.PrepaidUpgradeStars <= 0 {
return domain.ErrStarGiftCollectibleUnavailable
}
} else if req.ChargeStars != revision.UpgradeStars {
return domain.ErrStarGiftCollectibleUnavailable
}
balance, err := debitStarGiftUpgrade(ctx, tx, req.UserID, req.ChargeStars, locked.Owner, req.Date)
if err != nil {
return err
}
modelID, err := chooseCollectibleAttribute(ctx, tx, "star_gift_collectible_models", revision.ID)
if err != nil {
return err
}
patternID, err := chooseCollectibleAttribute(ctx, tx, "star_gift_collectible_patterns", revision.ID)
if err != nil {
return err
}
backdropID, err := chooseCollectibleAttribute(ctx, tx, "star_gift_collectible_backdrops", revision.ID)
if err != nil {
return err
}
num := revision.Issued + 1
var uniqueID int64
if err := tx.QueryRow(ctx, `SELECT nextval('unique_star_gift_id_seq')`).Scan(&uniqueID); err != nil {
return fmt.Errorf("allocate unique star gift id: %w", err)
}
var title string
if err := tx.QueryRow(ctx, `SELECT title FROM star_gift_catalog_revisions WHERE id=$1`, locked.RevisionID).Scan(&title); err != nil {
return fmt.Errorf("load upgrade gift title: %w", err)
}
slug := fmt.Sprintf("%s-%d", revision.SlugPrefix, num)
if _, err := tx.Exec(ctx, `
INSERT INTO unique_star_gifts
(id, gift_id, collectible_revision_id, source_saved_gift_id, title, slug, num,
owner_peer_type, owner_peer_id, model_attribute_id, pattern_attribute_id,
backdrop_attribute_id, keep_original_details)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)`,
uniqueID, locked.GiftID, revision.ID, locked.ID, title, slug, num,
string(locked.Owner.Type), locked.Owner.ID, modelID, patternID, backdropID, req.KeepOriginalDetails); err != nil {
return fmt.Errorf("insert unique star gift: %w", err)
}
if _, err := tx.Exec(ctx, `UPDATE star_gift_collectible_revisions SET issued=issued+1 WHERE id=$1`, revision.ID); err != nil {
return fmt.Errorf("increment collectible issuance: %w", err)
}
if _, err := tx.Exec(ctx, `
UPDATE peer_star_gifts
SET unique_gift_id=$2, prepaid_upgrade_stars=0, convert_stars=0
WHERE id=$1 AND unique_gift_id IS NULL AND NOT converted`, locked.ID, uniqueID); err != nil {
return fmt.Errorf("upgrade saved star gift: %w", err)
}
if _, err := tx.Exec(ctx, `
INSERT INTO star_gift_upgrade_commands
(user_id, command_key, source_saved_gift_id, form_id, unique_gift_id, balance_after)
VALUES ($1,$2,$3,$4,$5,$6)`, req.UserID, commandKey, locked.ID, req.FormID, uniqueID, balance.Balance); err != nil {
return fmt.Errorf("insert star gift upgrade command: %w", err)
}
unique, found, err := NewStarGiftStore(tx).UniqueByID(ctx, uniqueID)
if err != nil {
return err
}
if !found {
return fmt.Errorf("new unique star gift %d disappeared", uniqueID)
}
locked.UniqueGiftID = uniqueID
locked.PrepaidUpgradeStars = 0
locked.ConvertStars = 0
locked.Unique = &unique
result.Saved, result.Unique, result.Balance = locked, unique, balance
messageReq.Media = &domain.MessageMedia{
Kind: domain.MessageMediaKindService,
ServiceAction: &domain.MessageServiceAction{
Kind: domain.MessageServiceActionStarGiftUnique,
StarGiftUnique: &domain.MessageStarGiftUniqueAction{
Gift: unique, FromUserID: func() int64 {
if locked.NameHidden {
return 0
}
return locked.FromUserID
}(), Peer: locked.Owner, Upgrade: true, Saved: !locked.Unsaved,
PrepaidUpgrade: req.RequirePrepaid,
},
},
}
return nil
},
after: func(ctx context.Context, tx pgx.Tx, sent domain.SendPrivateTextResult) error {
ownerMessageID := sent.RecipientMessage.ID
if saved.FromUserID == req.UserID {
ownerMessageID = sent.SenderMessage.ID
}
if ownerMessageID <= 0 {
return fmt.Errorf("upgrade service message missing owner box")
}
tag, err := tx.Exec(ctx, `UPDATE peer_star_gifts SET upgrade_msg_id=$2 WHERE id=$1 AND unique_gift_id=$3`, result.Saved.ID, ownerMessageID, result.Unique.ID)
if err != nil {
return fmt.Errorf("save star gift upgrade message id: %w", err)
}
if tag.RowsAffected() != 1 {
return fmt.Errorf("save star gift upgrade message id lost aggregate row")
}
result.Saved.UpgradeMsgID = ownerMessageID
return nil
},
}
sent, err := s.messages.sendPrivateTextWithHooks(ctx, messageReq, hooks)
if err != nil {
return domain.StarGiftUpgradeResult{}, err
}
result.Send = sent
result.Duplicate = sent.Duplicate
if sent.Duplicate {
return s.loadUpgradeReplay(ctx, req, saved, sent)
}
return result, nil
}
func lockSavedStarGiftForUpgrade(ctx context.Context, tx pgx.Tx, ref domain.SavedStarGiftRef) (domain.SavedStarGift, error) {
where, args := savedStarGiftRefWhere(ref)
row := tx.QueryRow(ctx, `
SELECT p.id, p.owner_peer_type, p.owner_peer_id, p.from_user_id, p.gift_id, p.catalog_revision_id,
p.msg_id, p.saved_id, p.gift_date, p.name_hidden, p.unsaved, p.converted, p.convert_stars, p.prepaid_upgrade_stars,
p.message, COALESCE(p.unique_gift_id, 0), p.upgrade_msg_id, p.pinned_order,
COALESCE((SELECT array_agg(i.collection_id ORDER BY c.sort_order, i.collection_id)
FROM star_gift_collection_items i
JOIN star_gift_collections c ON c.collection_id=i.collection_id
WHERE i.saved_gift_id=p.id), ARRAY[]::integer[])
FROM peer_star_gifts p WHERE `+where+` FOR UPDATE`, args...)
saved, err := scanSavedStarGift(row)
if errors.Is(err, pgx.ErrNoRows) {
return domain.SavedStarGift{}, domain.ErrStarGiftNotFound
}
return saved, err
}
func lockActiveCollectibleRevision(ctx context.Context, tx pgx.Tx, giftID int64) (domain.StarGiftCollectibleRevision, error) {
var revision domain.StarGiftCollectibleRevision
var status string
err := tx.QueryRow(ctx, `
SELECT r.id, r.gift_id, r.upgrade_stars, r.supply_total, r.issued, r.slug_prefix, r.status
FROM star_gift_catalog c
JOIN star_gift_collectible_revisions r ON r.id=c.collectible_revision_id
WHERE c.gift_id=$1 FOR UPDATE OF r`, giftID).Scan(
&revision.ID, &revision.GiftID, &revision.UpgradeStars, &revision.SupplyTotal,
&revision.Issued, &revision.SlugPrefix, &status)
if errors.Is(err, pgx.ErrNoRows) {
return domain.StarGiftCollectibleRevision{}, domain.ErrStarGiftCollectibleUnavailable
}
if err != nil {
return domain.StarGiftCollectibleRevision{}, fmt.Errorf("lock active collectible revision: %w", err)
}
if status != "published" {
return domain.StarGiftCollectibleRevision{}, domain.ErrStarGiftCollectibleUnavailable
}
return revision, nil
}
func debitStarGiftUpgrade(ctx context.Context, tx pgx.Tx, userID, amount int64, peer domain.Peer, date int) (domain.StarsBalance, error) {
result := domain.StarsBalance{UserID: userID}
var balance int64
err := tx.QueryRow(ctx, `SELECT balance, granted FROM stars_balances WHERE user_id=$1 FOR UPDATE`, userID).Scan(&balance, &result.Granted)
if amount == 0 && errors.Is(err, pgx.ErrNoRows) {
return result, nil
}
if errors.Is(err, pgx.ErrNoRows) || (err == nil && balance < amount) {
return domain.StarsBalance{}, domain.ErrStarsInsufficient
}
if err != nil {
return domain.StarsBalance{}, fmt.Errorf("lock stars balance for gift upgrade: %w", err)
}
if amount == 0 {
result.Balance = balance
return result, nil
}
if err := tx.QueryRow(ctx, `UPDATE stars_balances SET balance=balance-$2, updated_at=now() WHERE user_id=$1 RETURNING balance`, userID, amount).Scan(&result.Balance); err != nil {
return domain.StarsBalance{}, fmt.Errorf("debit star gift upgrade: %w", err)
}
if err := insertStarsTxn(ctx, tx, userID, -amount, domain.StarsReasonGiftUpgrade, peer, date, "Star gift upgrade", ""); err != nil {
return domain.StarsBalance{}, err
}
return result, nil
}
func chooseCollectibleAttribute(ctx context.Context, tx pgx.Tx, table string, revisionID int64) (int64, error) {
rows, err := tx.Query(ctx, fmt.Sprintf(`SELECT id, rarity_permille FROM %s WHERE collectible_revision_id=$1 ORDER BY sort_order, id`, table), revisionID)
if err != nil {
return 0, fmt.Errorf("list collectible attributes for issuance: %w", err)
}
defer rows.Close()
type weightedID struct {
id int64
weight int
}
items := make([]weightedID, 0)
total := 0
for rows.Next() {
var item weightedID
if err := rows.Scan(&item.id, &item.weight); err != nil {
return 0, err
}
items = append(items, item)
total += item.weight
}
if err := rows.Err(); err != nil {
return 0, err
}
if len(items) == 0 || total != 1000 {
return 0, domain.ErrStarGiftCollectibleInvalid
}
draw, err := rand.Int(rand.Reader, big.NewInt(int64(total)))
if err != nil {
return 0, fmt.Errorf("draw collectible attribute: %w", err)
}
value := int(draw.Int64())
for _, item := range items {
if value < item.weight {
return item.id, nil
}
value -= item.weight
}
return 0, domain.ErrStarGiftCollectibleInvalid
}
func starGiftUpgradeRandomID(senderID, ownerID int64, commandKey string) int64 {
sum := sha256.Sum256([]byte(fmt.Sprintf("%d:%d:%s", senderID, ownerID, commandKey)))
id := int64(binary.LittleEndian.Uint64(sum[:8]) & 0x7fffffffffffffff)
if id == 0 {
id = 1
}
return id
}
func (s *StarGiftUpgradeStore) loadUpgradeReplay(ctx context.Context, req domain.StarGiftUpgradeRequest, original domain.SavedStarGift, sent domain.SendPrivateTextResult) (domain.StarGiftUpgradeResult, error) {
saved, found, err := NewStarGiftStore(s.db).GetByRef(ctx, req.Ref)
if err != nil || !found || saved.UniqueGiftID == 0 {
if err == nil {
err = domain.ErrStarGiftCollectibleInvalid
}
return domain.StarGiftUpgradeResult{}, err
}
unique, found, err := NewStarGiftStore(s.db).UniqueByID(ctx, saved.UniqueGiftID)
if err != nil || !found {
if err == nil {
err = domain.ErrStarGiftCollectibleInvalid
}
return domain.StarGiftUpgradeResult{}, err
}
var commandUniqueID int64
var balanceAfter int64
if err := s.db.QueryRow(ctx, `SELECT unique_gift_id, balance_after FROM star_gift_upgrade_commands WHERE user_id=$1 AND command_key=$2`, req.UserID, strings.TrimSpace(req.CommandKey)).Scan(&commandUniqueID, &balanceAfter); err != nil {
return domain.StarGiftUpgradeResult{}, fmt.Errorf("load star gift upgrade replay: %w", err)
}
if commandUniqueID != unique.ID || saved.ID != original.ID {
return domain.StarGiftUpgradeResult{}, domain.ErrStarGiftCollectibleInvalid
}
uniqueCopy := unique
saved.Unique = &uniqueCopy
return domain.StarGiftUpgradeResult{
Saved: saved, Unique: unique, Balance: domain.StarsBalance{UserID: req.UserID, Balance: balanceAfter},
Send: sent, Duplicate: true,
}, nil
}
var _ store.StarGiftUpgradeStore = (*StarGiftUpgradeStore)(nil)