feat: ennoblement sync integration test (#33)

Reviewed-on: twhelp/corev3#33

internal/adapter/repository_bun_ennoblement.go

@@ -83,6 +83,10 @@ func (a listEnnoblementsParamsApplier) apply(q *bun.SelectQuery) *bun.SelectQuer
 			q = q.Order("ennoblement.created_at ASC")
 		case domain.EnnoblementSortCreatedAtDESC:
 			q = q.Order("ennoblement.created_at DESC")
+		case domain.EnnoblementSortIDASC:
+			q = q.Order("ennoblement.id ASC")
+		case domain.EnnoblementSortIDDESC:
+			q = q.Order("ennoblement.id DESC")
 		case domain.EnnoblementSortServerKeyASC:
 			q = q.Order("ennoblement.server_key ASC")
 		case domain.EnnoblementSortServerKeyDESC:

internal/adapter/repository_ennoblement_test.go

@@ -146,7 +146,10 @@ func testEnnoblementRepository(t *testing.T, newRepos func(t *testing.T) reposit
 						if x := cmp.Compare(a.ServerKey(), b.ServerKey()); x != 0 {
 							return x
 						}
-						return a.CreatedAt().Compare(b.CreatedAt())
+						if x := a.CreatedAt().Compare(b.CreatedAt()); x != 0 {
+							return x
+						}
+						return cmp.Compare(a.ID(), b.ID())
 					}))
 				},
 				assertError: func(t *testing.T, err error) {
@@ -188,6 +191,58 @@ func testEnnoblementRepository(t *testing.T, newRepos func(t *testing.T) reposit
 					assert.NotEmpty(t, total)
 				},
 			},
+			{
+				name: "OK: sort=[id ASC]",
+				params: func(t *testing.T) domain.ListEnnoblementsParams {
+					t.Helper()
+					params := domain.NewListEnnoblementsParams()
+					require.NoError(t, params.SetSort([]domain.EnnoblementSort{
+						domain.EnnoblementSortIDASC,
+					}))
+					return params
+				},
+				assertEnnoblements: func(t *testing.T, params domain.ListEnnoblementsParams, ennoblements domain.Ennoblements) {
+					t.Helper()
+					assert.NotEmpty(t, len(ennoblements))
+					assert.True(t, slices.IsSortedFunc(ennoblements, func(a, b domain.Ennoblement) int {
+						return cmp.Compare(a.ID(), b.ID())
+					}))
+				},
+				assertError: func(t *testing.T, err error) {
+					t.Helper()
+					require.NoError(t, err)
+				},
+				assertTotal: func(t *testing.T, params domain.ListEnnoblementsParams, total int) {
+					t.Helper()
+					assert.NotEmpty(t, total)
+				},
+			},
+			{
+				name: "OK: sort=[id DESC]",
+				params: func(t *testing.T) domain.ListEnnoblementsParams {
+					t.Helper()
+					params := domain.NewListEnnoblementsParams()
+					require.NoError(t, params.SetSort([]domain.EnnoblementSort{
+						domain.EnnoblementSortIDDESC,
+					}))
+					return params
+				},
+				assertEnnoblements: func(t *testing.T, params domain.ListEnnoblementsParams, ennoblements domain.Ennoblements) {
+					t.Helper()
+					assert.NotEmpty(t, len(ennoblements))
+					assert.True(t, slices.IsSortedFunc(ennoblements, func(a, b domain.Ennoblement) int {
+						return cmp.Compare(a.ID(), b.ID()) * -1
+					}))
+				},
+				assertError: func(t *testing.T, err error) {
+					t.Helper()
+					require.NoError(t, err)
+				},
+				assertTotal: func(t *testing.T, params domain.ListEnnoblementsParams, total int) {
+					t.Helper()
+					assert.NotEmpty(t, total)
+				},
+			},
 			{
 				name: fmt.Sprintf("OK: serverKeys=[%s]", randEnnoblement.ServerKey()),
 				params: func(t *testing.T) domain.ListEnnoblementsParams {

internal/adapter/repository_tribe_change_test.go

@@ -163,7 +163,10 @@ func testTribeChangeRepository(t *testing.T, newRepos func(t *testing.T) reposit
 						if x := cmp.Compare(a.ServerKey(), b.ServerKey()); x != 0 {
 							return x
 						}
-						return a.CreatedAt().Compare(b.CreatedAt())
+						if x := a.CreatedAt().Compare(b.CreatedAt()); x != 0 {
+							return x
+						}
+						return cmp.Compare(a.ID(), b.ID())
 					}))
 				},
 				assertError: func(t *testing.T, err error) {

internal/app/service_ennoblement.go

@@ -31,7 +31,7 @@ func (svc *EnnoblementService) Sync(
 	serverURL := syncEnnoblementsCmdPayload.ServerURL()
 	serverKey := syncEnnoblementsCmdPayload.ServerKey()
 
-	latestEnnoblement, err := svc.getLatestEnnoblement(ctx, serverKey)
+	latestEnnoblement, err := svc.getLatest(ctx, serverKey)
 	if err != nil && !errors.Is(err, errEnnoblementNotFound) {
 		return fmt.Errorf("%s: %w", serverKey, err)
 	}
@@ -68,7 +68,7 @@ func (svc *EnnoblementService) Sync(
 
 var errEnnoblementNotFound = errors.New("ennoblement not found")
 
-func (svc *EnnoblementService) getLatestEnnoblement(ctx context.Context, serverKey string) (domain.Ennoblement, error) {
+func (svc *EnnoblementService) getLatest(ctx context.Context, serverKey string) (domain.Ennoblement, error) {
 	params := domain.NewListEnnoblementsParams()
 	if err := params.SetLimit(1); err != nil {
 		return domain.Ennoblement{}, err

internal/domain/ennoblement.go

@@ -159,6 +159,8 @@ type EnnoblementSort uint8
 const (
 	EnnoblementSortCreatedAtASC EnnoblementSort = iota + 1
 	EnnoblementSortCreatedAtDESC
+	EnnoblementSortIDASC
+	EnnoblementSortIDDESC
 	EnnoblementSortServerKeyASC
 	EnnoblementSortServerKeyDESC
 )
@@ -179,6 +181,7 @@ func NewListEnnoblementsParams() ListEnnoblementsParams {
 		sort: []EnnoblementSort{
 			EnnoblementSortServerKeyASC,
 			EnnoblementSortCreatedAtASC,
+			EnnoblementSortIDASC,
 		},
 		limit: EnnoblementListMaxLimit,
 	}
@@ -199,7 +202,7 @@ func (params *ListEnnoblementsParams) Sort() []EnnoblementSort {
 
 const (
 	ennoblementSortMinLength = 1
-	ennoblementSortMaxLength = 2
+	ennoblementSortMaxLength = 3
 )
 
 func (params *ListEnnoblementsParams) SetSort(sort []EnnoblementSort) error {

internal/domain/ennoblement_test.go

@@ -55,6 +55,7 @@ func TestListEnnoblementsParams_SetSort(t *testing.T) {
 				sort: []domain.EnnoblementSort{
 					domain.EnnoblementSortCreatedAtASC,
 					domain.EnnoblementSortServerKeyASC,
+					domain.EnnoblementSortIDASC,
 				},
 			},
 		},
@@ -68,17 +69,18 @@ func TestListEnnoblementsParams_SetSort(t *testing.T) {
 				Field: "sort",
 				Err: domain.LenOutOfRangeError{
 					Min:     1,
-					Max:     2,
+					Max:     3,
 					Current: 0,
 				},
 			},
 		},
 		{
-			name: "ERR: len(sort) > 2",
+			name: "ERR: len(sort) > 3",
 			args: args{
 				sort: []domain.EnnoblementSort{
 					domain.EnnoblementSortCreatedAtASC,
 					domain.EnnoblementSortServerKeyASC,
+					domain.EnnoblementSortIDASC,
 					domain.EnnoblementSortServerKeyDESC,
 				},
 			},
@@ -87,8 +89,8 @@ func TestListEnnoblementsParams_SetSort(t *testing.T) {
 				Field: "sort",
 				Err: domain.LenOutOfRangeError{
 					Min:     1,
-					Max:     2,
-					Current: 3,
+					Max:     3,
+					Current: 4,
 				},
 			},
 		},

internal/domain/tribe_change.go

@@ -229,6 +229,7 @@ func NewListTribeChangesParams() ListTribeChangesParams {
 		sort: []TribeChangeSort{
 			TribeChangeSortServerKeyASC,
 			TribeChangeSortCreatedAtASC,
+			TribeChangeSortIDASC,
 		},
 		limit: TribeChangeListMaxLimit,
 	}
@@ -249,7 +250,7 @@ func (params *ListTribeChangesParams) Sort() []TribeChangeSort {
 
 const (
 	tribeChangeSortMinLength = 1
-	tribeChangeSortMaxLength = 2
+	tribeChangeSortMaxLength = 3
 )
 
 func (params *ListTribeChangesParams) SetSort(sort []TribeChangeSort) error {

internal/domain/tribe_change_test.go

@@ -247,18 +247,19 @@ func TestListTribeChangesParams_SetSort(t *testing.T) {
 				Field: "sort",
 				Err: domain.LenOutOfRangeError{
 					Min:     1,
-					Max:     2,
+					Max:     3,
 					Current: 0,
 				},
 			},
 		},
 		{
-			name: "ERR: len(sort) > 2",
+			name: "ERR: len(sort) > 3",
 			args: args{
 				sort: []domain.TribeChangeSort{
 					domain.TribeChangeSortCreatedAtASC,
 					domain.TribeChangeSortServerKeyASC,
-					domain.TribeChangeSortServerKeyDESC,
+					domain.TribeChangeSortIDASC,
+					domain.TribeChangeSortIDDESC,
 				},
 			},
 			expectedErr: domain.ValidationError{
@@ -266,8 +267,8 @@ func TestListTribeChangesParams_SetSort(t *testing.T) {
 				Field: "sort",
 				Err: domain.LenOutOfRangeError{
 					Min:     1,
-					Max:     2,
-					Current: 3,
+					Max:     3,
+					Current: 4,
 				},
 			},
 		},

internal/service/data_sync_test.go

@@ -53,32 +53,32 @@ func TestDataSync(t *testing.T) {
 	// watermill
 	marshaler := watermillmsg.JSONMarshaler{NewUUID: watermill.NewUUID}
 	generateExchangeAndRoutingKeyName := func(topic string) string {
-		return topic + "_sync_data"
+		return topic + "_data_sync"
 	}
 	rmqConn := rabbitMQ.NewConnection(t)
 	nopLogger := watermill.NopLogger{}
 	serverPub, serverSub := watermillamqptest.NewPubSub(
 		t,
 		rmqConn,
-		amqp.GenerateQueueNameTopicNameWithSuffix("_sync_data_server"),
+		amqp.GenerateQueueNameTopicNameWithSuffix("_data_sync_server"),
 		generateExchangeAndRoutingKeyName,
 	)
 	playerPub, playerSub := watermillamqptest.NewPubSub(
 		t,
 		rmqConn,
-		amqp.GenerateQueueNameTopicNameWithSuffix("_sync_data_player"),
+		amqp.GenerateQueueNameTopicNameWithSuffix("_data_sync_player"),
 		generateExchangeAndRoutingKeyName,
 	)
 	tribePub, tribeSub := watermillamqptest.NewPubSub(
 		t,
 		rmqConn,
-		amqp.GenerateQueueNameTopicNameWithSuffix("_sync_data_tribe"),
+		amqp.GenerateQueueNameTopicNameWithSuffix("_data_sync_tribe"),
 		generateExchangeAndRoutingKeyName,
 	)
 	villagePub, villageSub := watermillamqptest.NewPubSub(
 		t,
 		rmqConn,
-		amqp.GenerateQueueNameTopicNameWithSuffix("_sync_data_village"),
+		amqp.GenerateQueueNameTopicNameWithSuffix("_data_sync_village"),
 		generateExchangeAndRoutingKeyName,
 	)
 
@@ -91,7 +91,7 @@ func TestDataSync(t *testing.T) {
 
 	// TW service
 
-	// filesys is set later in this test to os.DirFS("./testdata/syncdata/stage{n}")
+	// filesys is set later in this test to os.DirFS("./testdata/datasync/stage{n}")
 	var filesys fs.FS
 
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -173,10 +173,8 @@ func TestDataSync(t *testing.T) {
 	)
 
 	for _, stage := range []uint{1, 2} {
-		stage := stage
-
 		t.Run(fmt.Sprintf("stage %d", stage), func(t *testing.T) {
-			filesys = os.DirFS(fmt.Sprintf("./testdata/syncdata/stage%d", stage))
+			filesys = os.DirFS(fmt.Sprintf("./testdata/datasync/stage%d", stage))
 
 			require.NoError(t, dataSyncSvc.TriggerDataSync(ctx))
 

internal/service/ennoblement_sync_test.go

@@ -0,0 +1,239 @@
+package service_test
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"io/fs"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"os/signal"
+	"path"
+	"syscall"
+	"testing"
+	"time"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/adapter"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/adapter/adaptertest"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/app"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/port"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/tw"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/watermill/watermillamqptest"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/watermill/watermillmsg"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/watermill/watermilltest"
+	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill-amqp/v2/pkg/amqp"
+	"github.com/brianvoe/gofakeit/v6"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// we can't run the subtests in parallel
+
+//nolint:tparallel
+func TestEnnoblementSync(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skip("skipping long-running test")
+	}
+
+	ctxTimeout, cancel := context.WithTimeout(context.Background(), 150*time.Second)
+	defer cancel()
+
+	ctx, stop := signal.NotifyContext(ctxTimeout, os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	// bun
+	db := postgres.NewBunDB(t)
+	adaptertest.NewFixture(db).Load(t, ctx, os.DirFS("./testdata/ennoblementsync"), "fixture.yml")
+
+	// watermill
+	marshaler := watermillmsg.JSONMarshaler{NewUUID: watermill.NewUUID}
+	generateExchangeAndRoutingKeyName := func(topic string) string {
+		return topic + "_ennoblement_sync"
+	}
+	rmqConn := rabbitMQ.NewConnection(t)
+	nopLogger := watermill.NopLogger{}
+	_, serverSub := watermillamqptest.NewPubSub(
+		t,
+		rmqConn,
+		amqp.GenerateQueueNameTopicNameWithSuffix("_ennoblement_sync_server"),
+		generateExchangeAndRoutingKeyName,
+	)
+	ennoblementPub, ennoblementSub := watermillamqptest.NewPubSub(
+		t,
+		rmqConn,
+		amqp.GenerateQueueNameTopicNameWithSuffix("_ennoblement_sync_ennoblement"),
+		generateExchangeAndRoutingKeyName,
+	)
+
+	// events/commands
+	ennoblementCmdSync := gofakeit.UUID()
+	ennoblementEventSynced := gofakeit.UUID()
+
+	// filesys is set later in this test to os.DirFS("./testdata/ennoblementsync/stage{n}")
+	var filesys fs.FS
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodGet {
+			w.WriteHeader(http.StatusMethodNotAllowed)
+			return
+		}
+
+		u, err := url.Parse(r.URL.Query().Get("url"))
+		if err != nil {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		f, err := filesys.Open(path.Join(u.Host, u.Path+u.Query().Get("func")+u.Query().Get("since")))
+		if err != nil {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		defer func() {
+			_ = f.Close()
+		}()
+
+		w.WriteHeader(http.StatusOK)
+		_, _ = io.Copy(w, f)
+	}))
+	t.Cleanup(srv.Close)
+
+	twSvc := adapter.NewTWHTTP(tw.NewClient(
+		tw.WithHTTPClient(&http.Client{
+			Transport: &urlChangerTransport{
+				url:       srv.URL,
+				transport: srv.Client().Transport,
+			},
+		}),
+		tw.WithEnnoblementsUseInterfaceFunc(func(since time.Time) bool {
+			return !since.IsZero()
+		}),
+	))
+
+	// adapters
+	versionRepo := adapter.NewVersionBunRepository(db)
+	serverRepo := adapter.NewServerBunRepository(db)
+	ennoblementRepo := adapter.NewEnnoblementBunRepository(db)
+	ennoblementPublisher := adapter.NewEnnoblementWatermillPublisher(
+		ennoblementPub,
+		marshaler,
+		ennoblementCmdSync,
+		ennoblementEventSynced,
+	)
+
+	// services
+	versionSvc := app.NewVersionService(versionRepo)
+	serverSvc := app.NewServerService(serverRepo, twSvc, nil)
+	ennoblementSvc := app.NewEnnoblementService(ennoblementRepo, twSvc, ennoblementPublisher)
+	dataSyncSvc := app.NewDataSyncService(versionSvc, serverSvc, nil, ennoblementPublisher)
+
+	watermilltest.RunRouterWithContext(
+		t,
+		ctx,
+		port.NewServerWatermillConsumer(
+			serverSvc,
+			serverSub,
+			nopLogger,
+			marshaler,
+			"",
+			"",
+			"",
+			"",
+			"",
+			ennoblementEventSynced,
+		),
+		port.NewEnnoblementWatermillConsumer(ennoblementSvc, ennoblementSub, nopLogger, marshaler, ennoblementCmdSync),
+	)
+
+	for _, stage := range []uint{1, 2} {
+		t.Run(fmt.Sprintf("stage %d", stage), func(t *testing.T) {
+			filesys = os.DirFS(fmt.Sprintf("./testdata/ennoblementsync/stage%d", stage))
+
+			require.NoError(t, dataSyncSvc.TriggerEnnoblementSync(ctx))
+
+			assert.EventuallyWithTf(t, func(collect *assert.CollectT) {
+				require.NoError(collect, ctx.Err())
+
+				listParams := domain.NewListServersParams()
+				require.NoError(collect, listParams.SetSort([]domain.ServerSort{
+					domain.ServerSortKeyASC,
+				}))
+				require.NoError(collect, listParams.SetSpecial(domain.NullBool{
+					Value: false,
+					Valid: true,
+				}))
+				require.NoError(collect, listParams.SetLimit(domain.ServerListMaxLimit))
+
+				for {
+					servers, err := serverRepo.List(ctx, listParams)
+					require.NoError(collect, err)
+
+					if len(servers) == 0 {
+						break
+					}
+
+					for _, s := range servers {
+						assert.WithinDuration(collect, time.Now(), s.EnnoblementDataSyncedAt(), time.Minute)
+					}
+
+					require.NoError(collect, listParams.SetKeyGT(domain.NullString{
+						Value: servers[len(servers)-1].Key(),
+						Valid: true,
+					}))
+				}
+			}, 30*time.Second, time.Second, "servers")
+
+			var expectedEnnoblements []map[string]any
+			readJSONFile(t, filesys, path.Join("expected", "ennoblements.json"), &expectedEnnoblements)
+			assert.EventuallyWithT(t, func(collect *assert.CollectT) {
+				require.NoError(collect, ctx.Err())
+
+				listParams := domain.NewListEnnoblementsParams()
+				require.NoError(collect, listParams.SetSort([]domain.EnnoblementSort{
+					domain.EnnoblementSortServerKeyASC,
+					domain.EnnoblementSortCreatedAtASC,
+					domain.EnnoblementSortIDASC,
+				}))
+				require.NoError(collect, listParams.SetLimit(domain.EnnoblementListMaxLimit))
+
+				allEnnoblements := make(domain.Ennoblements, 0, len(expectedEnnoblements))
+
+				for {
+					ennoblements, err := ennoblementRepo.List(ctx, listParams)
+					require.NoError(collect, err)
+
+					if len(ennoblements) == 0 {
+						break
+					}
+
+					allEnnoblements = append(allEnnoblements, ennoblements...)
+
+					require.NoError(collect, listParams.SetOffset(listParams.Offset()+domain.EnnoblementListMaxLimit))
+				}
+
+				if !assert.Len(collect, allEnnoblements, len(expectedEnnoblements)) {
+					return
+				}
+				for i, expected := range expectedEnnoblements {
+					actual := allEnnoblements[i]
+					msg := fmt.Sprintf("ID=%.0f,ServerKey=%s", expected["ID"], expected["ServerKey"])
+
+					assert.EqualValues(collect, expected["ID"], actual.ID(), msg)
+					assert.Equal(collect, expected["ServerKey"], actual.ServerKey(), msg)
+					assert.EqualValues(collect, expected["OldOwnerID"], actual.OldOwnerID(), msg)
+					assert.EqualValues(collect, expected["OldTribeID"], actual.OldTribeID(), msg)
+					assert.EqualValues(collect, expected["NewOwnerID"], actual.NewOwnerID(), msg)
+					assert.EqualValues(collect, expected["NewTribeID"], actual.NewTribeID(), msg)
+					assert.EqualValues(collect, expected["Points"], actual.Points(), msg)
+					assert.Equal(collect, expected["CreatedAt"], actual.CreatedAt().Format(time.RFC3339), msg)
+				}
+			}, 30*time.Second, time.Second, "ennoblements")
+		})
+	}
+}