feat: tribes - update dominance after village sync (#25)

Reviewed-on: twhelp/corev3#25

cmd/twhelp/cmd_consumer.go

@@ -106,6 +106,7 @@ var cmdConsumer = &cli.Command{
 							logger,
 							marshaler,
 							c.String(rmqFlagTopicServerSyncedEvent.Name),
+							c.String(rmqFlagTopicVillagesSyncedEvent.Name),
 						)
 						consumer.Register(router)
 

internal/adapter/repository_bun_tribe.go

@@ -90,6 +90,26 @@ func (repo *TribeBunRepository) CreateOrUpdate(ctx context.Context, params ...do
 	return nil
 }
 
+func (repo *TribeBunRepository) UpdateDominance(ctx context.Context, serverKey string, numPlayerVillages int) error {
+	q := repo.db.NewUpdate().
+		Model((*bunmodel.Tribe)(nil)).
+		Returning("NULL").
+		Where("server_key = ?", serverKey).
+		Where("deleted_at IS NULL")
+
+	if numPlayerVillages > 0 {
+		q = q.Set("dominance = CAST(num_villages as double precision) / ? * 100", numPlayerVillages)
+	} else {
+		q = q.Set("dominance = 0")
+	}
+
+	if _, err := q.Exec(ctx); err != nil {
+		return fmt.Errorf("%s: couldn't update dominance: %w", serverKey, err)
+	}
+
+	return nil
+}
+
 func (repo *TribeBunRepository) List(ctx context.Context, params domain.ListTribesParams) (domain.Tribes, error) {
 	var tribes bunmodel.Tribes
 

internal/adapter/repository_test.go

@@ -27,6 +27,7 @@ type serverRepository interface {
 
 type tribeRepository interface {
 	CreateOrUpdate(ctx context.Context, params ...domain.CreateTribeParams) error
+	UpdateDominance(ctx context.Context, serverKey string, numPlayerVillages int) error
 	List(ctx context.Context, params domain.ListTribesParams) (domain.Tribes, error)
 	Delete(ctx context.Context, serverKey string, ids ...int) error
 }

internal/adapter/repository_tribe_test.go

@@ -99,6 +99,72 @@ func testTribeRepository(t *testing.T, newRepos func(t *testing.T) repositories)
 		})
 	})
 
+	t.Run("UpdateDominance", func(t *testing.T) {
+		t.Parallel()
+
+		repos := newRepos(t)
+
+		listServersParams := domain.NewListServersParams()
+		require.NoError(t, listServersParams.SetOpen(domain.NullBool{
+			Value: true,
+			Valid: true,
+		}))
+		require.NoError(t, listServersParams.SetSpecial(domain.NullBool{
+			Value: false,
+			Valid: true,
+		}))
+
+		servers, listServersErr := repos.server.List(ctx, listServersParams)
+		require.NoError(t, listServersErr)
+		require.GreaterOrEqual(t, len(servers), 2)
+
+		tests := []struct {
+			name              string
+			serverKey         string
+			numPlayerVillages int
+		}{
+			{
+				name:              "numPlayerVillages=0",
+				serverKey:         servers[0].Key(),
+				numPlayerVillages: 0,
+			},
+			{
+				name:              "numPlayerVillages=35000",
+				serverKey:         servers[1].Key(),
+				numPlayerVillages: 35000,
+			},
+		}
+
+		for _, tt := range tests {
+			tt := tt
+
+			t.Run(tt.name, func(t *testing.T) {
+				t.Parallel()
+
+				require.NoError(t, repos.tribe.UpdateDominance(ctx, tt.serverKey, tt.numPlayerVillages))
+
+				listTribesParams := domain.NewListTribesParams()
+				require.NoError(t, listTribesParams.SetDeleted(domain.NullBool{
+					Value: false,
+					Valid: true,
+				}))
+				require.NoError(t, listTribesParams.SetServerKeys([]string{tt.serverKey}))
+
+				tribes, err := repos.tribe.List(ctx, listTribesParams)
+				require.NoError(t, err)
+				assert.NotEmpty(t, tribes)
+				for _, tr := range tribes {
+					if tt.numPlayerVillages == 0 {
+						assert.InDelta(t, 0.0, tr.Dominance(), 0.001)
+						continue
+					}
+
+					assert.InDelta(t, float64(tr.NumVillages())/float64(tt.numPlayerVillages)*100, tr.Dominance(), 0.01)
+				}
+			})
+		}
+	})
+
 	t.Run("List & ListCount", func(t *testing.T) {
 		t.Parallel()
 

internal/app/service_tribe.go

@@ -9,6 +9,7 @@ import (
 
 type TribeRepository interface {
 	CreateOrUpdate(ctx context.Context, params ...domain.CreateTribeParams) error
+	UpdateDominance(ctx context.Context, serverKey string, numPlayerVillages int) error
 	List(ctx context.Context, params domain.ListTribesParams) (domain.Tribes, error)
 	// Delete marks players with the given serverKey and ids as deleted (sets deleted at to now).
 	//
@@ -152,3 +153,7 @@ func (svc *TribeService) delete(ctx context.Context, serverKey string, tribes do
 
 	return svc.repo.Delete(ctx, serverKey, toDelete...)
 }
+
+func (svc *TribeService) UpdateDominance(ctx context.Context, payload domain.VillagesSyncedEventPayload) error {
+	return svc.repo.UpdateDominance(ctx, payload.ServerKey(), payload.NumPlayerVillages())
+}

internal/port/consumer_watermill_tribe.go

@@ -9,11 +9,12 @@ import (
 )
 
 type TribeWatermillConsumer struct {
-	svc                    *app.TribeService
+	svc                      *app.TribeService
-	subscriber             message.Subscriber
+	subscriber               message.Subscriber
-	logger                 watermill.LoggerAdapter
+	logger                   watermill.LoggerAdapter
-	marshaler              watermillmsg.Marshaler
+	marshaler                watermillmsg.Marshaler
-	eventServerSyncedTopic string
+	eventServerSyncedTopic   string
+	eventVillagesSyncedTopic string
 }
 
 func NewTribeWatermillConsumer(
@@ -22,13 +23,15 @@ func NewTribeWatermillConsumer(
 	logger watermill.LoggerAdapter,
 	marshaler watermillmsg.Marshaler,
 	eventServerSyncedTopic string,
+	eventVillagesSyncedTopic string,
 ) *TribeWatermillConsumer {
 	return &TribeWatermillConsumer{
-		svc:                    svc,
+		svc:                      svc,
-		subscriber:             subscriber,
+		subscriber:               subscriber,
-		logger:                 logger,
+		logger:                   logger,
-		marshaler:              marshaler,
+		marshaler:                marshaler,
-		eventServerSyncedTopic: eventServerSyncedTopic,
+		eventServerSyncedTopic:   eventServerSyncedTopic,
+		eventVillagesSyncedTopic: eventVillagesSyncedTopic,
 	}
 }
 
@@ -39,6 +42,12 @@ func (c *TribeWatermillConsumer) Register(router *message.Router) {
 		c.subscriber,
 		c.sync,
 	)
+	router.AddNoPublisherHandler(
+		"TribeConsumer.updateDominance",
+		c.eventVillagesSyncedTopic,
+		c.subscriber,
+		c.updateDominance,
+	)
 }
 
 func (c *TribeWatermillConsumer) sync(msg *message.Message) error {
@@ -61,3 +70,32 @@ func (c *TribeWatermillConsumer) sync(msg *message.Message) error {
 
 	return c.svc.Sync(msg.Context(), payload)
 }
+
+func (c *TribeWatermillConsumer) updateDominance(msg *message.Message) error {
+	var rawPayload watermillmsg.VillagesSyncedEventPayload
+
+	if err := c.marshaler.Unmarshal(msg, &rawPayload); err != nil {
+		c.logger.Error("couldn't unmarshal payload", err, watermill.LogFields{
+			"handler": message.HandlerNameFromCtx(msg.Context()),
+		})
+		return nil
+	}
+
+	payload, err := domain.NewVillagesSyncedEventPayload(
+		rawPayload.ServerKey,
+		rawPayload.ServerURL,
+		rawPayload.VersionCode,
+		rawPayload.NumVillages,
+		rawPayload.NumPlayerVillages,
+		rawPayload.NumBarbarianVillages,
+		rawPayload.NumBonusVillages,
+	)
+	if err != nil {
+		c.logger.Error("couldn't construct domain.VillagesSyncedEventPayload", err, watermill.LogFields{
+			"handler": message.HandlerNameFromCtx(msg.Context()),
+		})
+		return nil
+	}
+
+	return c.svc.UpdateDominance(msg.Context(), payload)
+}