internal/store/postgres/broadcast_integration_test.go
Rama main · réplica de lectura del repositorio original.
package postgres
import (
"context"
"errors"
"reflect"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
broadcastapp "telesrv/internal/app/broadcast"
"telesrv/internal/domain"
"telesrv/internal/store"
)
func TestBroadcastDeliveryPreservesAutoEntitiesInHistoryAndUpdates(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
recipient := createLoginCodeDeliveryTestUser(t, ctx, pool, "BroadcastEntities")
broadcasts := NewBroadcastStore(pool)
messages := NewMessageStore(pool)
service := broadcastapp.NewService(broadcasts, messages, nil)
const text = "اعلان 🚀\n\n@matrixG"
want := []domain.MessageEntity{{Type: domain.MessageEntityMention, Offset: 10, Length: 8}}
ptsBefore := broadcastWatermark(t, ctx, pool, recipient.ID)
campaign, err := service.Create(ctx, text, domain.BroadcastTargetSelected, []int64{recipient.ID}, "integration-test")
if err != nil {
t.Fatalf("Create broadcast: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(), `DELETE FROM broadcasts WHERE id = $1`, campaign.ID)
})
if campaign.Message != text || !reflect.DeepEqual(campaign.Entities, want) {
t.Fatalf("created campaign message/entities = %q/%+v, want %q/%+v", campaign.Message, campaign.Entities, text, want)
}
claims, err := broadcasts.ClaimBroadcastRecipients(ctx, "entity-worker", 1, time.Minute)
if err != nil {
t.Fatalf("ClaimBroadcastRecipients: %v", err)
}
if len(claims) != 1 || claims[0].BroadcastID != campaign.ID || claims[0].UserID != recipient.ID {
t.Fatalf("claims = %+v, want campaign %d recipient %d", claims, campaign.ID, recipient.ID)
}
msg, err := messages.DeliverBroadcastRecipient(ctx, claims[0])
if err != nil {
t.Fatalf("DeliverBroadcastRecipient: %v", err)
}
if msg.Body != text || !reflect.DeepEqual(msg.Entities, want) {
t.Fatalf("delivered message body/entities = %q/%+v, want %q/%+v", msg.Body, msg.Entities, text, want)
}
if msg.Pts != ptsBefore+1 {
t.Fatalf("delivered pts = %d, want %d", msg.Pts, ptsBefore+1)
}
entityRows := []struct {
name string
query string
args []any
}{
{name: "campaign", query: `SELECT entities::text FROM broadcasts WHERE id = $1`, args: []any{campaign.ID}},
{name: "logical message", query: `SELECT entities::text FROM private_messages WHERE sender_user_id = $1 AND id = $2`, args: []any{domain.OfficialSystemUserID, msg.UID}},
{name: "recipient box", query: `SELECT entities::text FROM message_boxes WHERE owner_user_id = $1 AND box_id = $2`, args: []any{recipient.ID, msg.ID}},
}
for _, row := range entityRows {
var raw string
if err := pool.QueryRow(ctx, row.query, row.args...).Scan(&raw); err != nil {
t.Fatalf("load %s entities: %v", row.name, err)
}
got, err := decodeMessageEntities(raw)
if err != nil {
t.Fatalf("decode %s entities: %v", row.name, err)
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("%s entities = %+v, want %+v", row.name, got, want)
}
}
history, err := messages.ListByUser(ctx, recipient.ID, domain.MessageFilter{
HasPeer: true,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID},
Limit: 10,
})
if err != nil {
t.Fatalf("ListByUser: %v", err)
}
if len(history.Messages) != 1 || history.Messages[0].ID != msg.ID || !reflect.DeepEqual(history.Messages[0].Entities, want) {
t.Fatalf("history messages = %+v, want message %d with entities %+v", history.Messages, msg.ID, want)
}
updates := NewUpdateEventStore(pool)
events, err := updates.ListAfter(ctx, recipient.ID, ptsBefore, 10)
if err != nil {
t.Fatalf("ListAfter: %v", err)
}
if len(events) != 1 || events[0].Type != domain.UpdateEventNewMessage || events[0].Pts != msg.Pts ||
events[0].Message.ID != msg.ID || !reflect.DeepEqual(events[0].Message.Entities, want) {
t.Fatalf("difference events = %+v, want new message %d/%d with entities %+v", events, msg.ID, msg.Pts, want)
}
batch, err := updates.BatchByCursor(ctx, []store.EventCursor{{UserID: recipient.ID, Pts: msg.Pts}})
if err != nil {
t.Fatalf("BatchByCursor: %v", err)
}
if len(batch) != 1 || batch[0].Message.ID != msg.ID || !reflect.DeepEqual(batch[0].Message.Entities, want) {
t.Fatalf("outbox events = %+v, want message %d with entities %+v", batch, msg.ID, want)
}
}
func TestBroadcastSelectedClaimsAreDisjointAndDeliveryIsIncomingOnly(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
firstUser := createLoginCodeDeliveryTestUser(t, ctx, pool, "BroadcastFirst")
secondUser := createLoginCodeDeliveryTestUser(t, ctx, pool, "BroadcastSecond")
broadcasts := NewBroadcastStore(pool)
campaign, err := broadcasts.CreateBroadcast(ctx, "maintenance complete", nil, domain.BroadcastTargetSelected, []int64{firstUser.ID, secondUser.ID}, "integration-test")
if err != nil {
t.Fatalf("CreateBroadcast: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(), `DELETE FROM broadcasts WHERE id = $1`, campaign.ID)
})
firstClaims, err := broadcasts.ClaimBroadcastRecipients(ctx, "worker-one", 1, time.Minute)
if err != nil {
t.Fatalf("ClaimBroadcastRecipients worker one: %v", err)
}
secondClaims, err := broadcasts.ClaimBroadcastRecipients(ctx, "worker-two", 10, time.Minute)
if err != nil {
t.Fatalf("ClaimBroadcastRecipients worker two: %v", err)
}
if len(firstClaims) != 1 || len(secondClaims) != 1 {
t.Fatalf("claim sizes = %d/%d, want 1/1", len(firstClaims), len(secondClaims))
}
if firstClaims[0].RecipientID == secondClaims[0].RecipientID || firstClaims[0].UserID == secondClaims[0].UserID {
t.Fatalf("claims overlap: first=%+v second=%+v", firstClaims[0], secondClaims[0])
}
recipientPtsBefore := broadcastWatermark(t, ctx, pool, firstClaims[0].UserID)
systemPtsBefore := broadcastWatermark(t, ctx, pool, domain.OfficialSystemUserID)
systemBoxesBefore := broadcastFactCount(t, ctx, pool, `SELECT count(*)::int FROM message_boxes WHERE owner_user_id = $1`, domain.OfficialSystemUserID)
systemEventsBefore := broadcastFactCount(t, ctx, pool, `SELECT count(*)::int FROM user_update_events WHERE user_id = $1`, domain.OfficialSystemUserID)
systemOutboxBefore := broadcastFactCount(t, ctx, pool, `SELECT count(*)::int FROM dispatch_outbox WHERE target_user_id = $1`, domain.OfficialSystemUserID)
msg, err := NewMessageStore(pool).DeliverBroadcastRecipient(ctx, firstClaims[0])
if err != nil {
t.Fatalf("DeliverBroadcastRecipient: %v", err)
}
if msg.OwnerUserID != firstClaims[0].UserID || msg.Out || msg.Peer.ID != domain.OfficialSystemUserID || msg.From.ID != domain.OfficialSystemUserID {
t.Fatalf("delivered message viewpoint = %+v", msg)
}
if msg.Pts != recipientPtsBefore+1 {
t.Fatalf("recipient pts = %d, want %d", msg.Pts, recipientPtsBefore+1)
}
if got := broadcastWatermark(t, ctx, pool, domain.OfficialSystemUserID); got != systemPtsBefore {
t.Fatalf("system user pts = %d, want unchanged %d", got, systemPtsBefore)
}
if got := broadcastFactCount(t, ctx, pool, `SELECT count(*)::int FROM message_boxes WHERE owner_user_id = $1`, domain.OfficialSystemUserID); got != systemBoxesBefore {
t.Fatalf("system user box count = %d, want unchanged %d", got, systemBoxesBefore)
}
if got := broadcastFactCount(t, ctx, pool, `SELECT count(*)::int FROM user_update_events WHERE user_id = $1`, domain.OfficialSystemUserID); got != systemEventsBefore {
t.Fatalf("system user event count = %d, want unchanged %d", got, systemEventsBefore)
}
if got := broadcastFactCount(t, ctx, pool, `SELECT count(*)::int FROM dispatch_outbox WHERE target_user_id = $1`, domain.OfficialSystemUserID); got != systemOutboxBefore {
t.Fatalf("system user outbox count = %d, want unchanged %d", got, systemOutboxBefore)
}
assertBroadcastDeliveryFacts(t, ctx, pool, firstClaims[0], msg)
if _, err := NewMessageStore(pool).DeliverBroadcastRecipient(ctx, firstClaims[0]); !errors.Is(err, domain.ErrBroadcastLeaseLost) {
t.Fatalf("replay delivery error = %v, want lease lost", err)
}
if got := broadcastWatermark(t, ctx, pool, firstClaims[0].UserID); got != msg.Pts {
t.Fatalf("recipient pts after replay = %d, want %d", got, msg.Pts)
}
if _, err := pool.Exec(ctx, `UPDATE broadcast_recipients SET lease_until = now() - interval '1 second' WHERE id = $1`, secondClaims[0].RecipientID); err != nil {
t.Fatalf("expire second recipient lease: %v", err)
}
reclaimed, err := broadcasts.ClaimBroadcastRecipients(ctx, "worker-three", 10, time.Minute)
if err != nil {
t.Fatalf("reclaim stale recipient: %v", err)
}
if len(reclaimed) != 1 || reclaimed[0].RecipientID != secondClaims[0].RecipientID || reclaimed[0].Attempts != 2 {
t.Fatalf("reclaimed claims = %+v, want second recipient attempt 2", reclaimed)
}
if _, err := NewMessageStore(pool).DeliverBroadcastRecipient(ctx, secondClaims[0]); !errors.Is(err, domain.ErrBroadcastLeaseLost) {
t.Fatalf("stale claimant delivery error = %v, want lease lost", err)
}
if _, err := NewMessageStore(pool).DeliverBroadcastRecipient(ctx, reclaimed[0]); err != nil {
t.Fatalf("deliver second recipient: %v", err)
}
var sentCount int64
if err := pool.QueryRow(ctx, `SELECT sent_count FROM broadcasts WHERE id = $1`, campaign.ID).Scan(&sentCount); err != nil {
t.Fatalf("load sent_count: %v", err)
}
if sentCount != 2 {
t.Fatalf("sent_count = %d, want 2", sentCount)
}
}
func TestBroadcastAllMaterializationUsesCreationHighWaterAndBoundedKeysets(t *testing.T) {
pool := testPool(t)
ctx := context.Background()
firstUser := createLoginCodeDeliveryTestUser(t, ctx, pool, "BroadcastAllFirst")
secondUser := createLoginCodeDeliveryTestUser(t, ctx, pool, "BroadcastAllSecond")
broadcasts := NewBroadcastStore(pool)
campaign, err := broadcasts.CreateBroadcast(ctx, "all users", nil, domain.BroadcastTargetAll, nil, "integration-test")
if err != nil {
t.Fatalf("CreateBroadcast all: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(), `DELETE FROM broadcasts WHERE id = $1`, campaign.ID)
})
lateUser := createLoginCodeDeliveryTestUser(t, ctx, pool, "BroadcastAllLate")
for i := 0; i < 10000; i++ {
inserted, err := broadcasts.MaterializeBroadcastRecipients(ctx, 1)
if err != nil {
t.Fatalf("MaterializeBroadcastRecipients iteration %d: %v", i, err)
}
if inserted < 0 || inserted > 1 {
t.Fatalf("materialized batch = %d, want 0..1", inserted)
}
var done bool
if err := pool.QueryRow(ctx, `SELECT enumeration_done FROM broadcasts WHERE id = $1`, campaign.ID).Scan(&done); err != nil {
t.Fatalf("load enumeration state: %v", err)
}
if done {
break
}
if i == 9999 {
t.Fatal("materialization did not finish")
}
}
for _, userID := range []int64{firstUser.ID, secondUser.ID} {
var count int
if err := pool.QueryRow(ctx, `SELECT count(*)::int FROM broadcast_recipients WHERE broadcast_id = $1 AND user_id = $2`, campaign.ID, userID).Scan(&count); err != nil {
t.Fatalf("count materialized user %d: %v", userID, err)
}
if count != 1 {
t.Fatalf("materialized user %d count = %d, want 1", userID, count)
}
}
for _, excludedID := range []int64{lateUser.ID, domain.OfficialSystemUserID} {
var count int
if err := pool.QueryRow(ctx, `SELECT count(*)::int FROM broadcast_recipients WHERE broadcast_id = $1 AND user_id = $2`, campaign.ID, excludedID).Scan(&count); err != nil {
t.Fatalf("count excluded user %d: %v", excludedID, err)
}
if count != 0 {
t.Fatalf("excluded user %d was materialized", excludedID)
}
}
var targetCount, materializedCount, actualCount int64
if err := pool.QueryRow(ctx, `
SELECT b.target_count, b.materialized_count, count(r.id)
FROM broadcasts b
LEFT JOIN broadcast_recipients r ON r.broadcast_id = b.id
WHERE b.id = $1
GROUP BY b.id`, campaign.ID).Scan(&targetCount, &materializedCount, &actualCount); err != nil {
t.Fatalf("load final materialization counters: %v", err)
}
if targetCount != materializedCount || materializedCount != actualCount {
t.Fatalf("materialization counters target/materialized/actual = %d/%d/%d", targetCount, materializedCount, actualCount)
}
}
func broadcastWatermark(t *testing.T, ctx context.Context, pool *pgxpool.Pool, userID int64) int {
t.Helper()
var pts int
if err := pool.QueryRow(ctx, `
SELECT COALESCE((SELECT contiguous_pts FROM user_update_watermarks WHERE user_id = $1), 0)`, userID).Scan(&pts); err != nil {
t.Fatalf("load user %d watermark: %v", userID, err)
}
return pts
}
func broadcastFactCount(t *testing.T, ctx context.Context, pool *pgxpool.Pool, query string, args ...any) int {
t.Helper()
var count int
if err := pool.QueryRow(ctx, query, args...).Scan(&count); err != nil {
t.Fatalf("count broadcast fact: %v", err)
}
return count
}
func assertBroadcastDeliveryFacts(t *testing.T, ctx context.Context, pool *pgxpool.Pool, claim store.BroadcastRecipientClaim, msg domain.Message) {
t.Helper()
var status string
var privateMessageID int64
var boxID, recipientPts int
if err := pool.QueryRow(ctx, `
SELECT status, private_message_id, message_box_id, pts
FROM broadcast_recipients
WHERE id = $1`, claim.RecipientID).Scan(&status, &privateMessageID, &boxID, &recipientPts); err != nil {
t.Fatalf("load broadcast recipient receipt: %v", err)
}
if status != string(domain.BroadcastRecipientSent) || privateMessageID != msg.UID || boxID != msg.ID || recipientPts != msg.Pts {
t.Fatalf("broadcast receipt = %s/%d/%d/%d, message=%+v", status, privateMessageID, boxID, recipientPts, msg)
}
queries := []struct {
name string
sql string
args []any
}{
{
name: "recipient message box",
sql: `SELECT count(*)::int FROM message_boxes WHERE owner_user_id = $1 AND box_id = $2 AND NOT outgoing AND from_user_id = $3`,
args: []any{claim.UserID, msg.ID, domain.OfficialSystemUserID},
},
{
name: "recipient update event",
sql: `SELECT count(*)::int FROM user_update_events WHERE user_id = $1 AND pts = $2 AND event_type = 'new_message' AND message_box_id = $3`,
args: []any{claim.UserID, msg.Pts, msg.ID},
},
{
name: "recipient dispatch outbox",
sql: `SELECT count(*)::int FROM dispatch_outbox WHERE target_user_id = $1 AND pts = $2 AND event_type = 'new_message'`,
args: []any{claim.UserID, msg.Pts},
},
}
for _, query := range queries {
var count int
if err := pool.QueryRow(ctx, query.sql, query.args...).Scan(&count); err != nil {
t.Fatalf("count %s: %v", query.name, err)
}
if count != 1 {
t.Fatalf("%s count = %d, want 1", query.name, count)
}
}
}