internal/store/postgres/channel_difference_cursor_integration_test.go

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

package postgres

import (
	"context"
	"errors"
	"testing"

	"telesrv/internal/domain"
)

func TestChannelDifferenceValidCursorBeforeRowCacheInvalidationPostgres(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	owner, err := NewUserStore(pool).Create(ctx, domain.User{
		AccessHash: 731, Phone: "+1994" + randomSuffix(t) + "01", FirstName: "CursorOwner",
	})
	if err != nil {
		t.Fatal(err)
	}
	var channelID int64
	t.Cleanup(func() {
		if channelID != 0 {
			_, _ = pool.Exec(ctx, "DELETE FROM channels WHERE id = $1", channelID)
		}
		_, _ = pool.Exec(ctx, "DELETE FROM users WHERE id = $1", owner.ID)
	})
	cache := NewChannelRowCache(16)
	channels := NewChannelStore(pool, WithChannelRowCache(cache))
	created, err := channels.CreateChannel(ctx, domain.CreateChannelRequest{
		CreatorUserID: owner.ID, Title: "Cursor notification window", Megagroup: true, Date: 1701000300,
	})
	if err != nil {
		t.Fatal(err)
	}
	channelID = created.Channel.ID
	// Populate the real read-through cache, then commit a real send. Keeping
	// the listener paused models the interval between commit/push and NOTIFY
	// consumption; no persisted cursor or event is fabricated.
	if _, _, _, err := channels.getChannelForViewer(ctx, pool, owner.ID, channelID); err != nil {
		t.Fatal(err)
	}
	sent, err := channels.SendChannelMessage(ctx, domain.SendChannelMessageRequest{
		UserID: owner.ID, ChannelID: channelID, RandomID: 1701000301, Message: "committed cursor", Date: 1701000301,
	})
	if err != nil {
		t.Fatal(err)
	}
	if cached, ok := cache.get(channelID); !ok || cached.Pts != created.Channel.Pts {
		t.Fatalf("expected pre-notification cached pts %d, got %d (cached=%v)", created.Channel.Pts, cached.Pts, ok)
	}
	for _, pts := range []int{-1, sent.Event.Pts + 1} {
		if _, err := channels.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{
			UserID: owner.ID, ChannelID: channelID, Pts: pts, Limit: 100,
		}); !errors.Is(err, domain.ErrPersistentTimestamp) {
			t.Fatalf("invalid cursor %d error = %v, want ErrPersistentTimestamp", pts, err)
		}
	}
	diff, err := channels.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{
		UserID: owner.ID, ChannelID: channelID, Pts: sent.Event.Pts, Limit: 100,
	})
	if err != nil {
		t.Fatalf("valid committed cursor rejected before cache invalidation: %v", err)
	}
	if !diff.Final || diff.Pts != sent.Event.Pts || len(diff.Events) != 0 || diff.TooLong {
		t.Fatalf("current committed cursor difference = %+v", diff)
	}
	// A retry must still load the durable tail, not turn every now-valid
	// cursor into an empty response at the latest PTS.
	second, err := channels.SendChannelMessage(ctx, domain.SendChannelMessageRequest{
		UserID: owner.ID, ChannelID: channelID, RandomID: 1701000302, Message: "second cursor", Date: 1701000302,
	})
	if err != nil {
		t.Fatal(err)
	}
	third, err := channels.SendChannelMessage(ctx, domain.SendChannelMessageRequest{
		UserID: owner.ID, ChannelID: channelID, RandomID: 1701000303, Message: "durable tail", Date: 1701000303,
	})
	if err != nil {
		t.Fatal(err)
	}
	diff, err = channels.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{
		UserID: owner.ID, ChannelID: channelID, Pts: second.Event.Pts, Limit: 100,
	})
	if err != nil {
		t.Fatal(err)
	}
	if !diff.Final || diff.Pts != third.Event.Pts || len(diff.NewMessages) != 1 || diff.NewMessages[0].ID != third.Message.ID {
		t.Fatalf("committed cursor retry lost durable tail: %+v", diff)
	}
}