internal/store/postgres/album_group_integration_test.go

Rama main · réplica de lectura del repositorio original.

package postgres

import (
	"context"
	"crypto/sha256"
	"errors"
	"sync"
	"testing"

	"telesrv/internal/domain"
)

func pgAlbumItem(randomID int64, label string) domain.AlbumGroupReservationItem {
	sum := sha256.Sum256([]byte(label))
	return domain.AlbumGroupReservationItem{RandomID: randomID, IntentHash: sum[:]}
}

func pgAlbumReq(sender int64, peer domain.Peer, group int64, items ...domain.AlbumGroupReservationItem) domain.AlbumGroupReservationRequest {
	return domain.AlbumGroupReservationRequest{
		SenderUserID:      sender,
		Peer:              peer,
		Items:             items,
		ProposedGroupedID: group,
	}
}

func TestAlbumGroupReservationConvergesAcrossPostgresInstances(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	suffix := randomSuffix(t)
	users := NewUserStore(pool)
	sender, err := users.Create(ctx, domain.User{AccessHash: 7601, Phone: "+1760" + suffix + "01", FirstName: "AlbumSender"})
	if err != nil {
		t.Fatalf("create sender: %v", err)
	}
	recipient, err := users.Create(ctx, domain.User{AccessHash: 7602, Phone: "+1760" + suffix + "02", FirstName: "AlbumRecipient"})
	if err != nil {
		t.Fatalf("create recipient: %v", err)
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(ctx, "DELETE FROM album_group_reservations WHERE sender_user_id = $1", sender.ID)
		_, _ = pool.Exec(ctx, "DELETE FROM users WHERE id = ANY($1::bigint[])", []int64{sender.ID, recipient.ID})
	})

	privatePeer := domain.Peer{Type: domain.PeerTypeUser, ID: recipient.ID}
	full := []domain.AlbumGroupReservationItem{
		pgAlbumItem(76001, "one"),
		pgAlbumItem(76002, "two"),
		pgAlbumItem(76003, "three"),
	}
	firstStore := NewMessageStore(pool)
	groupedID, err := firstStore.ReserveAlbumGroup(ctx, pgAlbumReq(sender.ID, privatePeer, 761, full...))
	if err != nil || groupedID != 761 {
		t.Fatalf("reserve full = %d err=%v, want 761", groupedID, err)
	}
	// A fresh store instance simulates another process; a failed subset must recover the group of the first full batch.
	replayed, err := NewMessageStore(pool).ReserveAlbumGroup(ctx, pgAlbumReq(sender.ID, privatePeer, 762, full[1:]...))
	if err != nil || replayed != groupedID {
		t.Fatalf("reserve subset = %d err=%v, want %d", replayed, err, groupedID)
	}
	if _, err := NewMessageStore(pool).ReserveAlbumGroup(ctx, pgAlbumReq(sender.ID, privatePeer, 763, pgAlbumItem(76002, "changed"))); !errors.Is(err, domain.ErrMessageRandomIDDuplicate) {
		t.Fatalf("changed intent err=%v, want ErrMessageRandomIDDuplicate", err)
	}

	// The same random_id is independent per peer scope and does not wrongly merge albums.
	channelPeer := domain.Peer{Type: domain.PeerTypeChannel, ID: recipient.ID}
	channelGroup, err := firstStore.ReserveAlbumGroup(ctx, pgAlbumReq(sender.ID, channelPeer, 764, full[0]))
	if err != nil || channelGroup != 764 {
		t.Fatalf("channel peer isolated group = %d err=%v, want 764", channelGroup, err)
	}

	// Two instances reserve partially overlapping batches concurrently; the advisory lock
	// on the shared random_id must make both converge serially; finally items 4/5/6 all share one group.
	left := pgAlbumReq(sender.ID, privatePeer, 765, pgAlbumItem(76004, "four"), pgAlbumItem(76005, "shared"))
	right := pgAlbumReq(sender.ID, privatePeer, 766, pgAlbumItem(76005, "shared"), pgAlbumItem(76006, "six"))
	requests := []domain.AlbumGroupReservationRequest{left, right}
	results := make([]int64, 2)
	errs := make([]error, 2)
	start := make(chan struct{})
	var wg sync.WaitGroup
	for i := range requests {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			<-start
			results[i], errs[i] = NewMessageStore(pool).ReserveAlbumGroup(ctx, requests[i])
		}(i)
	}
	close(start)
	wg.Wait()
	if errs[0] != nil || errs[1] != nil || results[0] == 0 || results[0] != results[1] {
		t.Fatalf("concurrent groups=%v errs=%v, want one non-zero group", results, errs)
	}
	for _, item := range []domain.AlbumGroupReservationItem{left.Items[0], left.Items[1], right.Items[1]} {
		got, err := NewMessageStore(pool).ReserveAlbumGroup(ctx, pgAlbumReq(sender.ID, privatePeer, 767, item))
		if err != nil || got != results[0] {
			t.Fatalf("verify random_id %d = %d err=%v, want %d", item.RandomID, got, err, results[0])
		}
	}

	// A request hitting two historical groups at once must fail as a whole and must not bind the new item into either.
	oldA := pgAlbumItem(76007, "old-a")
	oldB := pgAlbumItem(76008, "old-b")
	newItem := pgAlbumItem(76009, "new")
	if _, err := firstStore.ReserveAlbumGroup(ctx, pgAlbumReq(sender.ID, privatePeer, 768, oldA)); err != nil {
		t.Fatal(err)
	}
	if _, err := firstStore.ReserveAlbumGroup(ctx, pgAlbumReq(sender.ID, privatePeer, 769, oldB)); err != nil {
		t.Fatal(err)
	}
	if _, err := firstStore.ReserveAlbumGroup(ctx, pgAlbumReq(sender.ID, privatePeer, 770, oldA, oldB, newItem)); !errors.Is(err, domain.ErrMessageRandomIDDuplicate) {
		t.Fatalf("mixed old groups err=%v, want ErrMessageRandomIDDuplicate", err)
	}
	fresh, err := firstStore.ReserveAlbumGroup(ctx, pgAlbumReq(sender.ID, privatePeer, 771, newItem))
	if err != nil || fresh != 771 {
		t.Fatalf("post-conflict fresh item = %d err=%v, want 771", fresh, err)
	}
}