hyapp-server/services/room-service/internal/storage/mysql/outbox_retention_test.go

package mysql

import (
	"context"
	"testing"
	"time"

	"google.golang.org/protobuf/proto"
	roomeventsv1 "hyapp.local/api/proto/events/room/v1"
	"hyapp/internal/testutil/mysqlschema"
	"hyapp/pkg/appcode"
	"hyapp/services/room-service/internal/room/outbox"
)

func TestSaveOutboxScopesBuiltRecordToContextApp(t *testing.T) {
	schema := mysqlschema.New(t, mysqlschema.Config{
		EnvVar:         "ROOM_SERVICE_MYSQL_TEST_DSN",
		InitDBPath:     mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), "..", "..", "..", "deploy", "mysql", "initdb", "001_room_service.sql"),
		DatabasePrefix: "hyapp_room_outbox_scope_test",
	})
	ctx := appcode.WithContext(context.Background(), "huwaa")
	repo, err := Open(ctx, schema.DSN, 4, 4)
	if err != nil {
		t.Fatalf("open repository: %v", err)
	}
	defer repo.Close()

	record, err := outbox.Build("huwaa-room-outbox", "RoomUserJoined", 9, time.UnixMilli(2000), &roomeventsv1.RoomUserJoined{UserId: 99})
	if err != nil {
		t.Fatalf("build room outbox: %v", err)
	}
	if err := repo.SaveOutbox(ctx, []outbox.Record{record}); err != nil {
		t.Fatalf("save scoped outbox: %v", err)
	}

	var storedApp string
	var envelopeBytes []byte
	if err := schema.DB.QueryRowContext(ctx,
		`SELECT app_code, envelope FROM room_outbox WHERE app_code = ? AND event_id = ?`,
		"huwaa", record.EventID,
	).Scan(&storedApp, &envelopeBytes); err != nil {
		t.Fatalf("read scoped outbox: %v", err)
	}
	var envelope roomeventsv1.EventEnvelope
	if err := proto.Unmarshal(envelopeBytes, &envelope); err != nil {
		t.Fatalf("decode scoped outbox envelope: %v", err)
	}
	if storedApp != "huwaa" || envelope.GetAppCode() != "huwaa" {
		t.Fatalf("SaveOutbox lost scoped app: row=%q envelope=%q", storedApp, envelope.GetAppCode())
	}
}

func retentionTestRecord(eventID string, createdAtMS int64) outbox.Record {
	return outbox.Record{
		AppCode:   appcode.Default,
		EventID:   eventID,
		EventType: "RoomGiftSent",
		RoomID:    "room-retention",
		Status:    outbox.StatusPending,
		Envelope: &roomeventsv1.EventEnvelope{
			AppCode:   appcode.Default,
			EventId:   eventID,
			RoomId:    "room-retention",
			EventType: "RoomGiftSent",
		},
		CreatedAtMS: createdAtMS,
	}
}

// 清理只允许命中 delivered 且早于 cutoff 的行；其他状态和窗口内的 delivered 必须原样保留，
// 否则统计重放和人工补偿会丢事实。
func TestDeleteDeliveredOutboxBeforeOnlyPurgesOldDelivered(t *testing.T) {
	schema := mysqlschema.New(t, mysqlschema.Config{
		EnvVar:         "ROOM_SERVICE_MYSQL_TEST_DSN",
		InitDBPath:     mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), "..", "..", "..", "deploy", "mysql", "initdb", "001_room_service.sql"),
		DatabasePrefix: "hyapp_room_retention_test",
	})
	ctx := context.Background()
	repo, err := Open(ctx, schema.DSN, 4, 4)
	if err != nil {
		t.Fatalf("open repository: %v", err)
	}
	defer repo.Close()

	records := []outbox.Record{
		retentionTestRecord("evt-old-delivered-1", 1000),
		retentionTestRecord("evt-old-delivered-2", 2000),
		retentionTestRecord("evt-old-delivered-3", 3000),
		retentionTestRecord("evt-fresh-delivered", 4000),
		retentionTestRecord("evt-old-pending", 1000),
		retentionTestRecord("evt-old-failed", 1000),
	}
	if err := repo.SaveOutbox(ctx, records); err != nil {
		t.Fatalf("save outbox: %v", err)
	}
	const cutoffMS = int64(1_000_000)
	setStatusAt := func(eventID string, status string, updatedAtMS int64) {
		t.Helper()
		if _, err := schema.DB.ExecContext(ctx, `UPDATE room_outbox SET status = ?, updated_at_ms = ? WHERE app_code = ? AND event_id = ?`, status, updatedAtMS, appcode.Default, eventID); err != nil {
			t.Fatalf("rewrite outbox row %s: %v", eventID, err)
		}
	}
	setStatusAt("evt-old-delivered-1", outbox.StatusDelivered, 1000)
	setStatusAt("evt-old-delivered-2", outbox.StatusDelivered, 2000)
	setStatusAt("evt-old-delivered-3", outbox.StatusDelivered, 3000)
	setStatusAt("evt-fresh-delivered", outbox.StatusDelivered, cutoffMS+1)
	setStatusAt("evt-old-failed", outbox.StatusFailed, 1000)

	// 批量上限必须生效：3 条可删行按 limit=2 分两批删完，第三批确认删空。
	for step, want := range []int64{2, 1, 0} {
		deleted, err := repo.DeleteDeliveredOutboxBefore(ctx, cutoffMS, 2)
		if err != nil {
			t.Fatalf("delete batch %d: %v", step, err)
		}
		if deleted != want {
			t.Fatalf("delete batch %d deleted %d rows, want %d", step, deleted, want)
		}
	}

	var survivors []string
	rows, err := schema.DB.QueryContext(ctx, `SELECT event_id FROM room_outbox WHERE app_code = ? ORDER BY event_id`, appcode.Default)
	if err != nil {
		t.Fatalf("list survivors: %v", err)
	}
	defer rows.Close()
	for rows.Next() {
		var eventID string
		if err := rows.Scan(&eventID); err != nil {
			t.Fatalf("scan survivor: %v", err)
		}
		survivors = append(survivors, eventID)
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterate survivors: %v", err)
	}
	want := []string{"evt-fresh-delivered", "evt-old-failed", "evt-old-pending"}
	if len(survivors) != len(want) {
		t.Fatalf("survivors mismatch got=%v want=%v", survivors, want)
	}
	for index := range want {
		if survivors[index] != want[index] {
			t.Fatalf("survivors mismatch got=%v want=%v", survivors, want)
		}
	}
}