internal/msg/msg_test.go

@@ -162,6 +162,10 @@ func TestUpdateData(t *testing.T) {
 					Valid: true,
 				},
 			})
+			if err != nil {
+				return false
+			}
+
 			for _, s := range servers {
 				if s.PlayerDataUpdatedAt.IsZero() ||
 					s.TribeDataUpdatedAt.IsZero() ||
@@ -169,7 +173,8 @@ func TestUpdateData(t *testing.T) {
 					return false
 				}
 			}
-			return err == nil && cmp.Equal(servers, expectedServers, cmpopts.IgnoreTypes(time.Time{}))
+
+			return cmp.Equal(servers, expectedServers, cmpopts.IgnoreTypes(time.Time{}))
 		}, 70*time.Second, time.Second, "servers - stage %d", stage)
 
 		require.NoError(t, ctx.Err())
@@ -181,9 +186,10 @@ func TestUpdateData(t *testing.T) {
 					{By: domain.PlayerSortByID, Direction: domain.SortDirectionASC},
 				},
 			})
-			if len(players) != len(expectedPlayers) {
+			if err != nil && len(players) != len(expectedPlayers) {
 				return false
 			}
+
 			for i, p := range players {
 				if p.CreatedAt.IsZero() ||
 					p.BestRankAt.IsZero() ||
@@ -194,7 +200,8 @@ func TestUpdateData(t *testing.T) {
 					return false
 				}
 			}
-			return err == nil && cmp.Equal(players, expectedPlayers, cmpopts.IgnoreTypes(time.Time{}))
+
+			return cmp.Equal(players, expectedPlayers, cmpopts.IgnoreTypes(time.Time{}))
 		}, 70*time.Second, time.Second, "players - stage %d", stage)
 
 		require.NoError(t, ctx.Err())
@@ -206,9 +213,10 @@ func TestUpdateData(t *testing.T) {
 					{By: domain.TribeSortByID, Direction: domain.SortDirectionASC},
 				},
 			})
-			if len(tribes) != len(expectedTribes) {
+			if err != nil || len(tribes) != len(expectedTribes) {
 				return false
 			}
+
 			for i, trb := range tribes {
 				if trb.CreatedAt.IsZero() ||
 					trb.BestRankAt.IsZero() ||
@@ -219,7 +227,8 @@ func TestUpdateData(t *testing.T) {
 					return false
 				}
 			}
-			return err == nil && cmp.Equal(tribes, expectedTribes, cmpopts.IgnoreTypes(time.Time{}))
+
+			return cmp.Equal(tribes, expectedTribes, cmpopts.IgnoreTypes(time.Time{}))
 		}, 70*time.Second, time.Second, "tribes - stage %d", stage)
 
 		require.NoError(t, ctx.Err())
@@ -227,12 +236,17 @@ func TestUpdateData(t *testing.T) {
 		expectedVillages := loadJSONFile[[]domain.Village](t, filesys, path.Join("expected", "villages.json"))
 		assert.Eventuallyf(t, func() bool {
 			villages, err := villageRepo.List(ctx, domain.ListVillagesParams{})
+			if err != nil {
+				return false
+			}
+
 			for _, v := range villages {
 				if v.CreatedAt.IsZero() {
 					return false
 				}
 			}
-			return err == nil && cmp.Equal(villages, expectedVillages, cmpopts.IgnoreTypes(time.Time{}))
+
+			return cmp.Equal(villages, expectedVillages, cmpopts.IgnoreTypes(time.Time{}))
 		}, 70*time.Second, time.Second, "villages - stage %d", stage)
 
 		require.NoError(t, ctx.Err())
@@ -244,12 +258,17 @@ func TestUpdateData(t *testing.T) {
 					{By: domain.TribeChangeSortByID, Direction: domain.SortDirectionASC},
 				},
 			})
+			if err != nil {
+				return false
+			}
+
 			for _, tc := range tcs {
 				if tc.CreatedAt.IsZero() {
 					return false
 				}
 			}
-			return err == nil && cmp.Equal(tcs, expectedTribeChanges, cmpopts.IgnoreTypes(time.Time{}))
+
+			return cmp.Equal(tcs, expectedTribeChanges, cmpopts.IgnoreTypes(time.Time{}))
 		}, 70*time.Second, time.Second, "tribe changes - stage %d", stage)
 	}
 
@@ -257,6 +276,141 @@ func TestUpdateData(t *testing.T) {
 	runUpdateData(2)
 }
 
+func TestUpdateEnnoblements(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skip("skipping long-running test")
+	}
+
+	ctxTimeout, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	ctx, stop := signal.NotifyContext(ctxTimeout, os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	generateExchangeAndRoutingKeyName := func(topic string) string {
+		return topic + "_update_ennoblements"
+	}
+	marshaler := msg.GobMarshaler{}
+	rmqConn := newRabbitMQConnection(t)
+	nopLogger := watermill.NopLogger{}
+	serverPub, serverSub := newRabbitMQPubSub(
+		t,
+		rmqConn,
+		amqp.GenerateQueueNameTopicNameWithSuffix("_update_ennoblements_server"),
+		generateExchangeAndRoutingKeyName,
+	)
+	ennoblementPub, ennoblementSub := newRabbitMQPubSub(
+		t,
+		rmqConn,
+		amqp.GenerateQueueNameTopicNameWithSuffix("_update_ennoblements_ennoblement"),
+		generateExchangeAndRoutingKeyName,
+	)
+	ennoblementPublisher := msg.NewEnnoblementPublisher(ennoblementPub, marshaler)
+
+	// filesys is set later in this test to os.DirFS("./testdata/updateennoblements/stage{n}")
+	var filesys fs.FS
+
+	// client
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		url, err := urlpkg.Parse(r.URL.Query().Get("url"))
+		if err != nil {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		f, err := filesys.Open(path.Join(url.Host, url.Path+url.Query().Get("func")+url.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)
+
+	client := tw.NewClient(tw.WithHTTPClient(&http.Client{
+		Transport: &urlChangerTransport{
+			url:       srv.URL,
+			transport: srv.Client().Transport,
+		},
+	}))
+
+	// db
+	db := bundbtest.NewDB(t)
+	require.NoError(t, bundbtest.NewFixture(db).Load(ctx, os.DirFS("./testdata/updateennoblements"), "fixture.yml"))
+
+	// repos
+	versionRepo := bundb.NewVersion(db)
+	serverRepo := bundb.NewServer(db)
+	ennoblementRepo := bundb.NewEnnoblement(db)
+
+	// services
+	versionSvc := service.NewVersion(versionRepo)
+	serverSvc := service.NewServer(serverRepo, client)
+	ennoblementSvc := service.NewEnnoblement(ennoblementRepo, client)
+	jobSvc := service.NewJob(versionSvc, serverSvc, nil, ennoblementPublisher, nil, nil)
+
+	runRouterWithContext(
+		t,
+		ctx,
+		msg.NewServerConsumer(marshaler, serverPub, serverSub, nopLogger, serverSvc),
+		msg.NewEnnoblementConsumer(marshaler, ennoblementPub, ennoblementSub, nopLogger, ennoblementSvc),
+	)
+
+	runUpdateEnnoblements := func(stage int64) {
+		filesys = os.DirFS("./testdata/updateennoblements/stage" + strconv.FormatInt(stage, 10))
+
+		require.NoError(t, jobSvc.UpdateEnnoblements(ctx))
+
+		expectedEnnoblements := loadJSONFile[[]domain.Ennoblement](t, filesys, path.Join("expected", "ennoblements.json"))
+		assert.Eventuallyf(t, func() bool {
+			ennoblements, err := ennoblementRepo.List(ctx, domain.ListEnnoblementsParams{
+				Sort: []domain.EnnoblementSort{
+					{By: domain.EnnoblementSortByCreatedAt, Direction: domain.SortDirectionASC},
+				},
+			})
+			return err == nil && cmp.Equal(ennoblements, expectedEnnoblements)
+		}, 60*time.Second, time.Second, "ennoblements - stage %d", stage)
+
+		require.NoError(t, ctx.Err())
+
+		now := time.Now()
+		assert.Eventuallyf(t, func() bool {
+			servers, err := serverRepo.List(ctx, domain.ListServersParams{
+				Special: domain.NullBool{
+					Bool:  false,
+					Valid: true,
+				},
+				Open: domain.NullBool{
+					Bool:  true,
+					Valid: true,
+				},
+			})
+			if err != nil {
+				return false
+			}
+
+			for _, s := range servers {
+				dt := now.Sub(s.EnnoblementDataUpdatedAt)
+				if dt < -time.Minute || dt > time.Minute {
+					return false
+				}
+			}
+
+			return true
+		}, 60*time.Second, time.Second, "servers - stage %d", stage)
+	}
+
+	runUpdateEnnoblements(1)
+	runUpdateEnnoblements(2)
+}
+
 type urlChangerTransport struct {
 	url       string
 	transport http.RoundTripper

internal/msg/testdata/updateennoblements/fixture.yml

@@ -0,0 +1,15 @@
+- model: Server
+  rows:
+    - _id: us63
+      key: us63
+      url: 'https://us63.tribalwars.us'
+      open: true
+      special: false
+      num_players: 1438
+      num_tribes: 125
+      num_villages: 10048
+      num_player_villages: 5386
+      num_barbarian_villages: 4662
+      num_bonus_villages: 3155
+      created_at: 2023-01-04T22:00:08.748927Z
+      version_code: us

internal/msg/testdata/updateennoblements/stage2/us63.tribalwars.us/interface.phpget_conquer_extended1676958836

@@ -0,0 +1,5 @@
+9158,1676958836,848955563,848955598,0,185,778
+8655,1676959184,282832,0,0,167,111
+8610,1676959781,750685,0,0,76,103
+1111,1676959935,848927565,0,0,138,964
+5288,1676960486,848914361,0,0,20,273