internal/store/postgres/groupcall_integration_test.go

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

package postgres

import (
	"context"
	"sync"
	"testing"
	"time"

	"telesrv/internal/domain"
	"telesrv/internal/store"
	"telesrv/internal/store/storetest"
)

// TestGroupCallStoreContractPostgres runs the same contract as the memory implementation (gated on TELESRV_TEST_POSTGRES_DSN).
func TestGroupCallStoreContractPostgres(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	// A run-unique namespace keeps filtered subtest runs independent from stale
	// rows left by an interrupted older run. Per-subtest cleanup below still makes
	// successful runs leave no state behind.
	var nextChannel = int64(1_000_000_000 + time.Now().UnixNano()%100_000_000)
	storetest.RunGroupCallStoreContract(t, func(t *testing.T) (store.GroupCallStore, int64) {
		nextChannel++
		channelID := nextChannel
		// Conference contract rows have channel_id=0 and derive their call IDs from
		// this synthetic channel namespace. Clean both shapes before and after each
		// subtest so a previous failed run cannot feed discarded calls into the next
		// run (and so conference invite/chain rows cascade away as well).
		cleanup := func() {
			_, _ = pool.Exec(ctx, `
DELETE FROM group_calls
WHERE channel_id = $1
   OR (call_id >= $1 * 100 AND call_id < $1 * 100 + 100)`, channelID)
		}
		cleanup()
		t.Cleanup(cleanup)
		return NewGroupCallStore(pool), channelID
	})
}

// TestGroupCallStoreM2ContractPostgres runs the overrides/raise-hand contract (DSN-gated).
func TestGroupCallStoreM2ContractPostgres(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	var nextChannel int64 = 920_000_000
	storetest.RunGroupCallStoreM2Contract(t, func(t *testing.T) (store.GroupCallStore, int64) {
		nextChannel++
		channelID := nextChannel
		t.Cleanup(func() {
			_, _ = pool.Exec(ctx, "DELETE FROM group_calls WHERE channel_id = $1", channelID)
		})
		return NewGroupCallStore(pool), channelID
	})
}

// TestGroupCallStoreConcurrentJoinVersions verifies that version has no skips under row-lock serialization.
func TestGroupCallStoreConcurrentJoinVersions(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	const channelID = 910_900_001
	t.Cleanup(func() {
		_, _ = pool.Exec(ctx, "DELETE FROM group_calls WHERE channel_id = $1", channelID)
	})
	st := NewGroupCallStore(pool)
	// Use the run time as the time basis: the test database may be shared with a running dev
	// server, where an ancient last_check_date would be swept away as a ghost by the online
	// sweeper and pollute the version/count assertions.
	now := int(time.Now().Unix())
	call, err := st.CreateGroupCall(ctx, domain.GroupCall{ID: 910_900_100, AccessHash: 5, ChannelID: channelID, CreatorUserID: 1, CreatedAt: now})
	if err != nil {
		t.Fatalf("create: %v", err)
	}
	const joiners = 8
	var wg sync.WaitGroup
	versions := make(chan int, joiners)
	for i := int64(0); i < joiners; i++ {
		wg.Add(1)
		go func(userID int64) {
			defer wg.Done()
			mut, err := st.JoinGroupCall(ctx, domain.JoinGroupCallRequest{CallID: call.ID, UserID: userID, SSRC: 7000 + userID, Now: now})
			if err != nil {
				t.Errorf("join %d: %v", userID, err)
				return
			}
			versions <- mut.Call.Version
		}(100 + i)
	}
	wg.Wait()
	close(versions)
	seen := map[int]bool{}
	for v := range versions {
		if seen[v] {
			t.Fatalf("duplicate version %d under concurrency", v)
		}
		seen[v] = true
	}
	final, _, err := st.GetGroupCall(ctx, call.ID)
	if err != nil || final.Version != 1+joiners || final.ParticipantsCount != joiners {
		t.Fatalf("final call = %+v err=%v", final, err)
	}
}