feat(TribeConsumer): update dominance

internal/bundb/tribe.go

@@ -72,6 +72,10 @@ func (t *Tribe) List(ctx context.Context, params domain.ListTribesParams) ([]dom
 	return result, int64(count), nil
 }
 
+func (t *Tribe) UpdateDominance(ctx context.Context, serverKey string, numVillages int64) error {
+	return nil
+}
+
 type listTribesParamsApplier struct {
 	params domain.ListTribesParams
 }

internal/msg/consumer_tribe.go

@@ -12,6 +12,7 @@ import (
 //counterfeiter:generate -o internal/mock/tribe_service.gen.go . TribeService
 type TribeService interface {
 	Refresh(ctx context.Context, key, url string) (int64, error)
+	UpdateDominance(ctx context.Context, serverKey string, numVillages int64) error
 }
 
 type TribeConsumer struct {
@@ -83,6 +84,15 @@ func (t *TribeConsumer) refresh(msg *message.Message) ([]*message.Message, error
 }
 
 func (t *TribeConsumer) updateDominance(msg *message.Message) error {
-	// TODO: implement this method
+	var payload domain.EventVillagesRefreshedPayload
+	if err := t.marshaler.Unmarshal(msg, &payload); err != nil {
+		// drop message with invalid payload
+		return nil
+	}
+
+	if err := t.svc.UpdateDominance(msg.Context(), payload.Key, payload.NumVillages); err != nil {
+		return fmt.Errorf("TribeService.UpdateDominance: %w", err)
+	}
+
 	return nil
 }

internal/msg/consumer_tribe_test.go

@@ -94,3 +94,58 @@ func TestTribeConsumer_refresh(t *testing.T) {
 		assert.Equal(t, 0, tribeSvc.RefreshCallCount())
 	})
 }
+
+func TestTribeConsumer_updateDominance(t *testing.T) {
+	t.Parallel()
+
+	marshaler := msg.GobMarshaler{}
+	pubSub := newPubSub(t)
+
+	tribeSvc := &mock.FakeTribeService{}
+	tribeSvc.UpdateDominanceReturns(nil)
+
+	wait := make(chan struct{})
+	runRouter(
+		t,
+		middlewareRegisterer{
+			h: func(h message.HandlerFunc) message.HandlerFunc {
+				return func(msg *message.Message) ([]*message.Message, error) {
+					if message.HandlerNameFromCtx(msg.Context()) != "TribeConsumer.updateDominance" {
+						return h(msg)
+					}
+
+					result, err := h(msg)
+					if len(result) == 0 && err == nil {
+						close(wait)
+					}
+					return result, err
+				}
+			},
+		},
+		msg.NewTribeConsumer(marshaler, pubSub, pubSub, tribeSvc),
+	)
+
+	payload := domain.EventVillagesRefreshedPayload{
+		Key:                  "pl151",
+		URL:                  "https://pl151.plemiona.pl",
+		VersionCode:          "pl",
+		NumVillages:          1500,
+		NumPlayerVillages:    1400,
+		NumBarbarianVillages: 100,
+		NumBonusVillages:     300,
+	}
+	ev, err := marshaler.Marshal(payload)
+	require.NoError(t, err)
+	require.NoError(t, pubSub.Publish("villages.event.refreshed", ev))
+
+	select {
+	case <-wait:
+	case <-time.After(time.Second):
+		t.Fatal("timeout")
+	}
+
+	require.Equal(t, 1, tribeSvc.UpdateDominanceCallCount())
+	_, serverKey, numVillages := tribeSvc.UpdateDominanceArgsForCall(0)
+	assert.Equal(t, payload.Key, serverKey)
+	assert.Equal(t, payload.NumVillages, numVillages)
+}

internal/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, numVillages int64) error
 }
 
 type TribesGetter interface {
@@ -45,7 +46,9 @@ func (t *Tribe) Refresh(ctx context.Context, key, url string) (int64, error) {
 	return int64(len(tribes)), nil
 }
 
-func (t *Tribe) UpdateDominance(ctx context.Context, key string, numVillages int64) error {
-	// TODO: implement this method
+func (t *Tribe) UpdateDominance(ctx context.Context, serverKey string, numVillages int64) error {
+	if err := t.repo.UpdateDominance(ctx, serverKey, numVillages); err != nil {
+		return fmt.Errorf("TribeRepository.UpdateDominance: %w", err)
+	}
 	return nil
 }