feat: ennoblement sync (#28)

Reviewed-on: twhelp/corev3#28

cmd/twhelp/cmd_consumer.go

@@ -64,6 +64,7 @@ var cmdConsumer = &cli.Command{
 							c.String(rmqFlagTopicTribesSyncedEvent.Name),
 							c.String(rmqFlagTopicPlayersSyncedEvent.Name),
 							c.String(rmqFlagTopicVillagesSyncedEvent.Name),
+							c.String(rmqFlagTopicEnnoblementsSyncedEvent.Name),
 						)
 						consumer.Register(router)
 
@@ -194,6 +195,49 @@ var cmdConsumer = &cli.Command{
 						)
 						consumer.Register(router)
 
+						return nil
+					},
+				)
+			},
+		},
+		{
+			Name:  "ennoblement",
+			Usage: "Run the worker responsible for consuming ennoblement-related messages",
+			Flags: concatSlices(dbFlags, rmqFlags, twSvcFlags),
+			Action: func(c *cli.Context) error {
+				return runConsumer(
+					c,
+					"EnnoblementConsumer",
+					func(
+						c *cli.Context,
+						router *message.Router,
+						logger watermill.LoggerAdapter,
+						publisher *amqp.Publisher,
+						subscriber *amqp.Subscriber,
+						marshaler watermillmsg.Marshaler,
+						db *bun.DB,
+					) error {
+						twSvc, err := newTWServiceFromFlags(c)
+						if err != nil {
+							return err
+						}
+
+						ennoblementPublisher := adapter.NewEnnoblementWatermillPublisher(
+							publisher,
+							marshaler,
+							c.String(rmqFlagTopicSyncEnnoblementsCmd.Name),
+							c.String(rmqFlagTopicEnnoblementsSyncedEvent.Name),
+						)
+
+						consumer := port.NewEnnoblementWatermillConsumer(
+							app.NewEnnoblementService(adapter.NewEnnoblementBunRepository(db), twSvc, ennoblementPublisher),
+							subscriber,
+							logger,
+							marshaler,
+							c.String(rmqFlagTopicSyncEnnoblementsCmd.Name),
+						)
+						consumer.Register(router)
+
 						return nil
 					},
 				)

internal/adapter/adaptertest/fixture.go

@@ -22,6 +22,7 @@ func NewFixture(bunDB *bun.DB) *Fixture {
 		(*bunmodel.Tribe)(nil),
 		(*bunmodel.Player)(nil),
 		(*bunmodel.Village)(nil),
+		(*bunmodel.Ennoblement)(nil),
 	)
 	return &Fixture{
 		f: dbfixture.New(bunDB),

internal/adapter/http_tw.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+	"time"
 
 	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
 	"gitea.dwysokinski.me/twhelp/corev3/internal/tw"
@@ -282,3 +283,39 @@ func (t *TWHTTP) convertVillagesToDomain(villages []tw.Village) (domain.BaseVill
 
 	return res, nil
 }
+
+func (t *TWHTTP) GetEnnoblements(
+	ctx context.Context,
+	baseURL *url.URL,
+	since time.Time,
+) (domain.BaseEnnoblements, error) {
+	ennoblements, err := t.client.GetEnnoblements(ctx, baseURL, since)
+	if err != nil {
+		return nil, err
+	}
+
+	return t.convertEnnoblementsToDomain(ennoblements)
+}
+
+func (t *TWHTTP) convertEnnoblementsToDomain(ennoblements []tw.Ennoblement) (domain.BaseEnnoblements, error) {
+	res := make(domain.BaseEnnoblements, 0, len(ennoblements))
+
+	for _, e := range ennoblements {
+		converted, err := domain.NewBaseEnnoblement(
+			e.VillageID,
+			e.NewOwnerID,
+			e.NewTribeID,
+			e.OldOwnerID,
+			e.OldTribeID,
+			e.Points,
+			e.CreatedAt,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("couldn't construct domain.BaseEnnoblement: %w", err)
+		}
+
+		res = append(res, converted)
+	}
+
+	return res, nil
+}

internal/adapter/internal/bunmodel/ennoblement.go

@@ -0,0 +1,69 @@
+package bunmodel
+
+import (
+	"fmt"
+	"time"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"github.com/uptrace/bun"
+)
+
+type Ennoblement struct {
+	bun.BaseModel `bun:"table:ennoblements,alias:ennoblement"`
+
+	ID         int       `bun:"id,pk,autoincrement,identity"`
+	ServerKey  string    `bun:"server_key,nullzero"`
+	VillageID  int       `bun:"village_id,nullzero"`
+	Village    Village   `bun:"village,rel:belongs-to,join:village_id=id,join:server_key=server_key"`
+	NewOwnerID int       `bun:"new_owner_id,nullzero"`
+	NewOwner   Player    `bun:"new_owner,rel:belongs-to,join:new_owner_id=id,join:server_key=server_key"`
+	NewTribeID int       `bun:"new_tribe_id,nullzero"`
+	NewTribe   Tribe     `bun:"new_tribe,rel:belongs-to,join:new_tribe_id=id,join:server_key=server_key"`
+	OldOwnerID int       `bun:"old_owner_id,nullzero"`
+	OldOwner   Player    `bun:"old_owner,rel:belongs-to,join:old_owner_id=id,join:server_key=server_key"`
+	OldTribeID int       `bun:"old_tribe_id,nullzero"`
+	OldTribe   Tribe     `bun:"old_tribe,rel:belongs-to,join:old_tribe_id=id,join:server_key=server_key"`
+	Points     int       `bun:"points"`
+	CreatedAt  time.Time `bun:"created_at,nullzero"`
+}
+
+func (e Ennoblement) ToDomain() (domain.Ennoblement, error) {
+	converted, err := domain.UnmarshalEnnoblementFromDatabase(
+		e.ID,
+		e.ServerKey,
+		e.VillageID,
+		e.NewOwnerID,
+		e.NewTribeID,
+		e.OldOwnerID,
+		e.OldTribeID,
+		e.Points,
+		e.CreatedAt,
+	)
+	if err != nil {
+		return domain.Ennoblement{}, fmt.Errorf(
+			"couldn't construct domain.Ennoblement (id=%d,serverKey=%s): %w",
+			e.ID,
+			e.ServerKey,
+			err,
+		)
+	}
+
+	return converted, nil
+}
+
+type Ennoblements []Ennoblement
+
+func (es Ennoblements) ToDomain() (domain.Ennoblements, error) {
+	res := make(domain.Ennoblements, 0, len(es))
+
+	for _, e := range es {
+		converted, err := e.ToDomain()
+		if err != nil {
+			return nil, err
+		}
+
+		res = append(res, converted)
+	}
+
+	return res, nil
+}

internal/adapter/repository_bun_ennoblement.go

@@ -0,0 +1,96 @@
+package adapter
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/adapter/internal/bunmodel"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"github.com/uptrace/bun"
+)
+
+type EnnoblementBunRepository struct {
+	db bun.IDB
+}
+
+func NewEnnoblementBunRepository(db bun.IDB) *EnnoblementBunRepository {
+	return &EnnoblementBunRepository{db: db}
+}
+
+func (repo *EnnoblementBunRepository) Create(ctx context.Context, params ...domain.CreateEnnoblementParams) error {
+	if len(params) == 0 {
+		return nil
+	}
+
+	ennoblements := make(bunmodel.Ennoblements, 0, len(params))
+
+	for _, p := range params {
+		base := p.Base()
+		ennoblements = append(ennoblements, bunmodel.Ennoblement{
+			ServerKey:  p.ServerKey(),
+			VillageID:  base.VillageID(),
+			NewOwnerID: base.NewOwnerID(),
+			NewTribeID: base.NewTribeID(),
+			OldOwnerID: base.OldOwnerID(),
+			OldTribeID: base.OldTribeID(),
+			Points:     base.Points(),
+			CreatedAt:  base.CreatedAt(),
+		})
+	}
+
+	if _, err := repo.db.NewInsert().
+		Model(&ennoblements).
+		Ignore().
+		Returning("").
+		Exec(ctx); err != nil {
+		return fmt.Errorf("something went wrong while inserting ennoblements into the db: %w", err)
+	}
+
+	return nil
+}
+
+func (repo *EnnoblementBunRepository) List(
+	ctx context.Context,
+	params domain.ListEnnoblementsParams,
+) (domain.Ennoblements, error) {
+	var ennoblements bunmodel.Ennoblements
+
+	if err := repo.db.NewSelect().
+		Model(&ennoblements).
+		Apply(listEnnoblementsParamsApplier{params: params}.apply).
+		Scan(ctx); err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return nil, fmt.Errorf("couldn't select ennoblements from the db: %w", err)
+	}
+
+	return ennoblements.ToDomain()
+}
+
+type listEnnoblementsParamsApplier struct {
+	params domain.ListEnnoblementsParams
+}
+
+//nolint:gocyclo
+func (a listEnnoblementsParamsApplier) apply(q *bun.SelectQuery) *bun.SelectQuery {
+	if serverKeys := a.params.ServerKeys(); len(serverKeys) > 0 {
+		q = q.Where("ennoblement.server_key IN (?)", bun.In(serverKeys))
+	}
+
+	for _, s := range a.params.Sort() {
+		switch s {
+		case domain.EnnoblementSortCreatedAtASC:
+			q = q.Order("ennoblement.created_at ASC")
+		case domain.EnnoblementSortCreatedAtDESC:
+			q = q.Order("ennoblement.created_at DESC")
+		case domain.EnnoblementSortServerKeyASC:
+			q = q.Order("ennoblement.server_key ASC")
+		case domain.EnnoblementSortServerKeyDESC:
+			q = q.Order("ennoblement.server_key DESC")
+		default:
+			return q.Err(errors.New("unsupported sort value"))
+		}
+	}
+
+	return q.Limit(a.params.Limit()).Offset(a.params.Offset())
+}

internal/adapter/repository_bun_ennoblement_test.go

@@ -0,0 +1,29 @@
+package adapter_test
+
+import (
+	"testing"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/adapter/adaptertest"
+)
+
+func TestEnnoblementBunRepository_Postgres(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skip("skipping long-running test")
+	}
+
+	testEnnoblementRepository(t, func(t *testing.T) repositories {
+		t.Helper()
+		return newBunDBRepositories(t, postgres.NewBunDB(t))
+	})
+}
+
+func TestEnnoblementBunRepository_SQLite(t *testing.T) {
+	t.Parallel()
+
+	testEnnoblementRepository(t, func(t *testing.T) repositories {
+		t.Helper()
+		return newBunDBRepositories(t, adaptertest.NewBunDBSQLite(t))
+	})
+}

internal/adapter/repository_ennoblement_test.go

@@ -0,0 +1,255 @@
+package adapter_test
+
+import (
+	"cmp"
+	"context"
+	"fmt"
+	"slices"
+	"testing"
+	"time"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain/domaintest"
+	gocmp "github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func testEnnoblementRepository(t *testing.T, newRepos func(t *testing.T) repositories) {
+	t.Helper()
+
+	ctx := context.Background()
+
+	t.Run("Create", func(t *testing.T) {
+		t.Parallel()
+
+		repos := newRepos(t)
+
+		assertCreated := func(t *testing.T, params []domain.CreateEnnoblementParams) {
+			t.Helper()
+
+			require.NotEmpty(t, params)
+
+			listParams := domain.NewListEnnoblementsParams()
+			require.NoError(t, listParams.SetServerKeys([]string{params[0].ServerKey()}))
+
+			ennoblements, err := repos.ennoblement.List(ctx, listParams)
+			require.NoError(t, err)
+			for i, p := range params {
+				idx := slices.IndexFunc(ennoblements, func(ennoblement domain.Ennoblement) bool {
+					return ennoblement.VillageID() == p.Base().VillageID() && ennoblement.ServerKey() == p.ServerKey()
+				})
+				require.GreaterOrEqualf(t, idx, 0, "params[%d] not found", i)
+				ennoblement := ennoblements[idx]
+
+				assert.Emptyf(t, gocmp.Diff(
+					p.Base(),
+					ennoblement.Base(),
+					cmpopts.EquateApproxTime(time.Minute),
+					gocmp.AllowUnexported(domain.BaseEnnoblement{}),
+				), "params[%d]", i)
+				assert.Equalf(t, p.ServerKey(), ennoblement.ServerKey(), "params[%d]", i)
+			}
+		}
+
+		assertNoDuplicates := func(t *testing.T, params []domain.CreateEnnoblementParams) {
+			t.Helper()
+
+			require.NotEmpty(t, params)
+
+			listParams := domain.NewListEnnoblementsParams()
+			require.NoError(t, listParams.SetServerKeys([]string{params[0].ServerKey()}))
+
+			ennoblements, err := repos.ennoblement.List(ctx, listParams)
+			require.NoError(t, err)
+
+			res := make(map[string][]int)
+
+			for _, p := range params {
+				key := fmt.Sprintf("%s-%d", p.ServerKey(), p.Base().VillageID())
+
+				for i, e := range ennoblements {
+					if e.ServerKey() == p.ServerKey() && gocmp.Equal(
+						p.Base(),
+						e.Base(),
+						cmpopts.EquateApproxTime(time.Minute),
+						gocmp.AllowUnexported(domain.BaseEnnoblement{}),
+					) {
+						res[key] = append(res[key], i)
+					}
+				}
+			}
+
+			for key, indexes := range res {
+				assert.Len(t, indexes, 1, key)
+			}
+		}
+
+		t.Run("OK", func(t *testing.T) {
+			t.Parallel()
+
+			servers, err := repos.server.List(ctx, domain.NewListServersParams())
+			require.NoError(t, err)
+			require.NotEmpty(t, servers)
+			server := servers[0]
+
+			ennoblementsToCreate := domain.BaseEnnoblements{
+				domaintest.NewBaseEnnoblement(t),
+				domaintest.NewBaseEnnoblement(t),
+			}
+
+			createParams, err := domain.NewCreateEnnoblementParams(server.Key(), ennoblementsToCreate)
+			require.NoError(t, err)
+
+			require.NoError(t, repos.ennoblement.Create(ctx, createParams...))
+			assertCreated(t, createParams)
+
+			require.NoError(t, repos.ennoblement.Create(ctx, createParams...))
+			assertNoDuplicates(t, createParams)
+		})
+
+		t.Run("OK: len(params) == 0", func(t *testing.T) {
+			t.Parallel()
+
+			require.NoError(t, repos.ennoblement.Create(ctx))
+		})
+	})
+
+	t.Run("List & ListCount", func(t *testing.T) {
+		t.Parallel()
+
+		repos := newRepos(t)
+
+		ennoblements, listEnnoblementsErr := repos.ennoblement.List(ctx, domain.NewListEnnoblementsParams())
+		require.NoError(t, listEnnoblementsErr)
+		require.NotEmpty(t, ennoblements)
+		randEnnoblement := ennoblements[0]
+
+		tests := []struct {
+			name               string
+			params             func(t *testing.T) domain.ListEnnoblementsParams
+			assertEnnoblements func(t *testing.T, params domain.ListEnnoblementsParams, ennoblements domain.Ennoblements)
+			assertError        func(t *testing.T, err error)
+			assertTotal        func(t *testing.T, params domain.ListEnnoblementsParams, total int)
+		}{
+			{
+				name: "OK: default params",
+				params: func(t *testing.T) domain.ListEnnoblementsParams {
+					t.Helper()
+					return domain.NewListEnnoblementsParams()
+				},
+				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 {
+						if x := cmp.Compare(a.ServerKey(), b.ServerKey()); x != 0 {
+							return x
+						}
+						return a.CreatedAt().Compare(b.CreatedAt())
+					}))
+				},
+				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=[serverKey DESC, createdAt DESC]",
+				params: func(t *testing.T) domain.ListEnnoblementsParams {
+					t.Helper()
+					params := domain.NewListEnnoblementsParams()
+					require.NoError(t, params.SetSort([]domain.EnnoblementSort{
+						domain.EnnoblementSortServerKeyDESC,
+						domain.EnnoblementSortCreatedAtDESC,
+					}))
+					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 {
+						if x := cmp.Compare(a.ServerKey(), b.ServerKey()) * -1; x != 0 {
+							return x
+						}
+						return a.CreatedAt().Compare(b.CreatedAt()) * -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 {
+					t.Helper()
+					params := domain.NewListEnnoblementsParams()
+					require.NoError(t, params.SetServerKeys([]string{randEnnoblement.ServerKey()}))
+					return params
+				},
+				assertEnnoblements: func(t *testing.T, params domain.ListEnnoblementsParams, ennoblements domain.Ennoblements) {
+					t.Helper()
+
+					serverKeys := params.ServerKeys()
+
+					for _, v := range ennoblements {
+						assert.True(t, slices.Contains(serverKeys, v.ServerKey()))
+					}
+				},
+				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: offset=1 limit=2",
+				params: func(t *testing.T) domain.ListEnnoblementsParams {
+					t.Helper()
+					params := domain.NewListEnnoblementsParams()
+					require.NoError(t, params.SetOffset(1))
+					require.NoError(t, params.SetLimit(2))
+					return params
+				},
+				assertEnnoblements: func(t *testing.T, params domain.ListEnnoblementsParams, ennoblements domain.Ennoblements) {
+					t.Helper()
+					assert.Len(t, ennoblements, params.Limit())
+				},
+				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)
+				},
+			},
+		}
+
+		for _, tt := range tests {
+			tt := tt
+
+			t.Run(tt.name, func(t *testing.T) {
+				t.Parallel()
+
+				params := tt.params(t)
+
+				res, err := repos.ennoblement.List(ctx, params)
+				tt.assertError(t, err)
+				tt.assertEnnoblements(t, params, res)
+			})
+		}
+	})
+}

internal/adapter/repository_test.go

@@ -44,12 +44,18 @@ type villageRepository interface {
 	Delete(ctx context.Context, serverKey string, ids ...int) error
 }
 
+type ennoblementRepository interface {
+	Create(ctx context.Context, params ...domain.CreateEnnoblementParams) error
+	List(ctx context.Context, params domain.ListEnnoblementsParams) (domain.Ennoblements, error)
+}
+
 type repositories struct {
-	version versionRepository
-	server  serverRepository
-	tribe   tribeRepository
-	player  playerRepository
-	village villageRepository
+	version     versionRepository
+	server      serverRepository
+	tribe       tribeRepository
+	player      playerRepository
+	village     villageRepository
+	ennoblement ennoblementRepository
 }
 
 func newBunDBRepositories(tb testing.TB, bunDB *bun.DB) repositories {
@@ -58,10 +64,11 @@ func newBunDBRepositories(tb testing.TB, bunDB *bun.DB) repositories {
 	adaptertest.NewFixture(bunDB).Load(tb, context.Background(), os.DirFS("testdata"), "fixture.yml")
 
 	return repositories{
-		version: adapter.NewVersionBunRepository(bunDB),
-		server:  adapter.NewServerBunRepository(bunDB),
-		tribe:   adapter.NewTribeBunRepository(bunDB),
-		player:  adapter.NewPlayerBunRepository(bunDB),
-		village: adapter.NewVillageBunRepository(bunDB),
+		version:     adapter.NewVersionBunRepository(bunDB),
+		server:      adapter.NewServerBunRepository(bunDB),
+		tribe:       adapter.NewTribeBunRepository(bunDB),
+		player:      adapter.NewPlayerBunRepository(bunDB),
+		village:     adapter.NewVillageBunRepository(bunDB),
+		ennoblement: adapter.NewEnnoblementBunRepository(bunDB),
 	}
 }

internal/adapter/testdata/fixture.yml

@@ -7288,3 +7288,45 @@
       player_id: 0
       created_at: 2022-02-25T15:00:10.000Z
       profile_url: https://it70.tribals.it/game.php?screen=info_village&id=10024
+- model: Ennoblement
+  rows:
+    - _id: pl169-village-2-1
+      id: 10000
+      server_key: pl169
+      village_id: 1112
+      new_owner_id: 699513260
+      new_tribe_id: 27
+      old_owner_id: 0
+      old_tribe_id: 0
+      points: 2500
+      created_at: 2021-09-30T09:01:00.000Z
+    - _id: pl169-village-2-2
+      id: 10001
+      server_key: pl169
+      village_id: 1112
+      new_owner_id: 699783765
+      new_tribe_id: 28
+      old_owner_id: 699513260
+      old_tribe_id: 27
+      points: 5000
+      created_at: 2021-10-30T09:01:00.000Z
+    - _id: it70-village-2-1
+      id: 20000
+      server_key: it70
+      village_id: 10023
+      new_owner_id: 848881282
+      new_tribe_id: 31
+      old_owner_id: 0
+      old_tribe_id: 0
+      points: 2500
+      created_at: 2022-03-15T15:00:10.000Z
+    - _id: it70-village-2-2
+      id: 20001
+      server_key: it70
+      village_id: 10023
+      new_owner_id: 578014
+      new_tribe_id: 1
+      old_owner_id: 848881282
+      old_tribe_id: 31
+      points: 5123
+      created_at: 2022-04-22T15:00:10.000Z

internal/app/service_ennoblement.go

@@ -0,0 +1,118 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+)
+
+type EnnoblementRepository interface {
+	Create(ctx context.Context, params ...domain.CreateEnnoblementParams) error
+	List(ctx context.Context, params domain.ListEnnoblementsParams) (domain.Ennoblements, error)
+}
+
+type EnnoblementService struct {
+	repo  EnnoblementRepository
+	twSvc TWService
+	pub   EnnoblementPublisher
+}
+
+func NewEnnoblementService(repo EnnoblementRepository, twSvc TWService, pub EnnoblementPublisher) *EnnoblementService {
+	return &EnnoblementService{repo: repo, twSvc: twSvc, pub: pub}
+}
+
+func (svc *EnnoblementService) Sync(
+	ctx context.Context,
+	syncEnnoblementsCmdPayload domain.SyncEnnoblementsCmdPayload,
+) error {
+	serverURL := syncEnnoblementsCmdPayload.ServerURL()
+	serverKey := syncEnnoblementsCmdPayload.ServerKey()
+
+	latestEnnoblement, err := svc.getLatestEnnoblement(ctx, serverKey)
+	if err != nil && !errors.Is(err, errEnnoblementNotFound) {
+		return fmt.Errorf("%s: %w", serverKey, err)
+	}
+
+	var since time.Time
+	if err == nil {
+		since = latestEnnoblement.CreatedAt()
+	}
+
+	ennoblements, err := svc.twSvc.GetEnnoblements(ctx, serverURL, since)
+	if err != nil {
+		return fmt.Errorf("%s: couldn't get ennoblements: %w", serverKey, err)
+	}
+
+	if err = svc.create(ctx, serverKey, ennoblements); err != nil {
+		return fmt.Errorf("%s: couldn't create ennoblements: %w", serverKey, err)
+	}
+
+	payload, err := domain.NewEnnoblementsSyncedEventPayload(
+		serverKey,
+		serverURL,
+		syncEnnoblementsCmdPayload.VersionCode(),
+	)
+	if err != nil {
+		return fmt.Errorf("%s: couldn't construct domain.EnnoblementsSyncedEventPayload %w", serverKey, err)
+	}
+
+	if err = svc.pub.EventSynced(ctx, payload); err != nil {
+		return fmt.Errorf("%s: %w", serverKey, err)
+	}
+
+	return nil
+}
+
+var errEnnoblementNotFound = errors.New("ennoblement not found")
+
+func (svc *EnnoblementService) getLatestEnnoblement(ctx context.Context, serverKey string) (domain.Ennoblement, error) {
+	params := domain.NewListEnnoblementsParams()
+	if err := params.SetLimit(1); err != nil {
+		return domain.Ennoblement{}, err
+	}
+	if err := params.SetServerKeys([]string{serverKey}); err != nil {
+		return domain.Ennoblement{}, err
+	}
+	if err := params.SetSort([]domain.EnnoblementSort{domain.EnnoblementSortCreatedAtDESC}); err != nil {
+		return domain.Ennoblement{}, err
+	}
+
+	ennoblements, err := svc.repo.List(ctx, params)
+	if err != nil {
+		return domain.Ennoblement{}, err
+	}
+	if len(ennoblements) == 0 {
+		return domain.Ennoblement{}, errEnnoblementNotFound
+	}
+
+	return ennoblements[0], nil
+}
+
+const ennoblementCreateChunkSize = 500
+
+func (svc *EnnoblementService) create(
+	ctx context.Context,
+	serverKey string,
+	ennoblements domain.BaseEnnoblements,
+) error {
+	for i := 0; i < len(ennoblements); i += ennoblementCreateChunkSize {
+		end := i + ennoblementCreateChunkSize
+		if end > len(ennoblements) {
+			end = len(ennoblements)
+		}
+
+		createParams, err := domain.NewCreateEnnoblementParams(serverKey, ennoblements[i:end])
+		if err != nil {
+			return err
+		}
+
+		if err = svc.repo.Create(ctx, createParams...); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

internal/app/service_server.go

@@ -60,10 +60,14 @@ func (svc *ServerService) Sync(ctx context.Context, payload domain.SyncServersCm
 
 	payloads, err := domain.NewServerSyncedEventPayloads(openServersWithoutSpecial, versionCode)
 	if err != nil {
-		return fmt.Errorf("%s: couldn't construct server synced event payloads: %w", versionCode, err)
+		return fmt.Errorf("%s: couldn't construct domain.ServerSyncedEventPayload: %w", versionCode, err)
 	}
 
-	return svc.pub.EventSynced(ctx, payloads...)
+	if err = svc.pub.EventSynced(ctx, payloads...); err != nil {
+		return fmt.Errorf("%s: %w", versionCode, err)
+	}
+
+	return nil
 }
 
 func (svc *ServerService) listAllSpecial(ctx context.Context, versionCode string) (domain.Servers, error) {
@@ -257,3 +261,20 @@ func (svc *ServerService) UpdateNumVillages(ctx context.Context, payload domain.
 
 	return svc.repo.Update(ctx, key, updateParams)
 }
+
+func (svc *ServerService) UpdateEnnoblementDataSyncedAt(
+	ctx context.Context,
+	payload domain.EnnoblementsSyncedEventPayload,
+) error {
+	key := payload.ServerKey()
+
+	var updateParams domain.UpdateServerParams
+	if err := updateParams.SetEnnoblementDataSyncedAt(domain.NullTime{
+		Value: time.Now(),
+		Valid: true,
+	}); err != nil {
+		return fmt.Errorf("%s: %w", key, err)
+	}
+
+	return svc.repo.Update(ctx, key, updateParams)
+}

internal/app/service_tw.go

@@ -3,6 +3,7 @@ package app
 import (
 	"context"
 	"net/url"
+	"time"
 
 	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
 )
@@ -15,4 +16,5 @@ type TWService interface {
 	GetTribes(ctx context.Context, baseURL *url.URL) (domain.BaseTribes, error)
 	GetPlayers(ctx context.Context, baseURL *url.URL) (domain.BasePlayers, error)
 	GetVillages(ctx context.Context, baseURL *url.URL) (domain.BaseVillages, error)
+	GetEnnoblements(ctx context.Context, baseURL *url.URL, since time.Time) (domain.BaseEnnoblements, error)
 }

internal/app/service_village.go

@@ -66,7 +66,12 @@ func (svc *VillageService) createOrUpdate(ctx context.Context, serverKey string,
 			end = len(villages)
 		}
 
-		if err := svc.createOrUpdateChunk(ctx, serverKey, villages[i:end]); err != nil {
+		createParams, err := domain.NewCreateVillageParams(serverKey, villages[i:end])
+		if err != nil {
+			return err
+		}
+
+		if err = svc.repo.CreateOrUpdate(ctx, createParams...); err != nil {
 			return err
 		}
 	}
@@ -74,19 +79,6 @@ func (svc *VillageService) createOrUpdate(ctx context.Context, serverKey string,
 	return nil
 }
 
-func (svc *VillageService) createOrUpdateChunk(
-	ctx context.Context,
-	serverKey string,
-	villages domain.BaseVillages,
-) error {
-	createParams, err := domain.NewCreateVillageParams(serverKey, villages)
-	if err != nil {
-		return err
-	}
-
-	return svc.repo.CreateOrUpdate(ctx, createParams...)
-}
-
 func (svc *VillageService) delete(ctx context.Context, serverKey string, villages domain.BaseVillages) error {
 	listParams := domain.NewListVillagesParams()
 	if err := listParams.SetServerKeys([]string{serverKey}); err != nil {

internal/domain/base_ennoblement.go

@@ -0,0 +1,120 @@
+package domain
+
+import (
+	"math"
+	"time"
+)
+
+type BaseEnnoblement struct {
+	villageID  int
+	newOwnerID int
+	newTribeID int
+	oldOwnerID int
+	oldTribeID int
+	points     int
+	createdAt  time.Time
+}
+
+const baseEnnoblementModelName = "BaseEnnoblement"
+
+func NewBaseEnnoblement(
+	villageID int,
+	newOwnerID int,
+	newTribeID int,
+	oldOwnerID int,
+	oldTribeID int,
+	points int,
+	createdAt time.Time,
+) (BaseEnnoblement, error) {
+	if err := validateIntInRange(villageID, 1, math.MaxInt); err != nil {
+		return BaseEnnoblement{}, ValidationError{
+			Model: baseEnnoblementModelName,
+			Field: "villageID",
+			Err:   err,
+		}
+	}
+
+	if err := validateIntInRange(newOwnerID, 0, math.MaxInt); err != nil {
+		return BaseEnnoblement{}, ValidationError{
+			Model: baseEnnoblementModelName,
+			Field: "newOwnerID",
+			Err:   err,
+		}
+	}
+
+	if err := validateIntInRange(newTribeID, 0, math.MaxInt); err != nil {
+		return BaseEnnoblement{}, ValidationError{
+			Model: baseEnnoblementModelName,
+			Field: "newTribeID",
+			Err:   err,
+		}
+	}
+
+	if err := validateIntInRange(oldOwnerID, 0, math.MaxInt); err != nil {
+		return BaseEnnoblement{}, ValidationError{
+			Model: baseEnnoblementModelName,
+			Field: "oldOwnerID",
+			Err:   err,
+		}
+	}
+
+	if err := validateIntInRange(oldTribeID, 0, math.MaxInt); err != nil {
+		return BaseEnnoblement{}, ValidationError{
+			Model: baseEnnoblementModelName,
+			Field: "oldTribeID",
+			Err:   err,
+		}
+	}
+
+	if err := validateIntInRange(points, 0, math.MaxInt); err != nil {
+		return BaseEnnoblement{}, ValidationError{
+			Model: baseEnnoblementModelName,
+			Field: "points",
+			Err:   err,
+		}
+	}
+
+	return BaseEnnoblement{
+		villageID:  villageID,
+		newOwnerID: newOwnerID,
+		newTribeID: newTribeID,
+		oldOwnerID: oldOwnerID,
+		oldTribeID: oldTribeID,
+		points:     points,
+		createdAt:  createdAt,
+	}, nil
+}
+
+func (e BaseEnnoblement) VillageID() int {
+	return e.villageID
+}
+
+func (e BaseEnnoblement) NewOwnerID() int {
+	return e.newOwnerID
+}
+
+func (e BaseEnnoblement) NewTribeID() int {
+	return e.newTribeID
+}
+
+func (e BaseEnnoblement) OldOwnerID() int {
+	return e.oldOwnerID
+}
+
+func (e BaseEnnoblement) OldTribeID() int {
+	return e.oldTribeID
+}
+
+func (e BaseEnnoblement) Points() int {
+	return e.points
+}
+
+func (e BaseEnnoblement) CreatedAt() time.Time {
+	return e.createdAt
+}
+
+func (e BaseEnnoblement) IsZero() bool {
+	return e == BaseEnnoblement{}
+}
+
+type BaseEnnoblements []BaseEnnoblement

internal/domain/base_ennoblement_test.go

@@ -0,0 +1,195 @@
+package domain_test
+
+import (
+	"testing"
+	"time"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain/domaintest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewBaseEnnoblement(t *testing.T) {
+	t.Parallel()
+
+	validBaseEnnoblement := domaintest.NewBaseEnnoblement(t)
+
+	type args struct {
+		villageID  int
+		newOwnerID int
+		newTribeID int
+		oldOwnerID int
+		oldTribeID int
+		points     int
+		createdAt  time.Time
+	}
+
+	tests := []struct {
+		name        string
+		args        args
+		expectedErr error
+	}{
+		{
+			name: "OK",
+			args: args{
+				villageID:  validBaseEnnoblement.VillageID(),
+				newOwnerID: validBaseEnnoblement.NewOwnerID(),
+				newTribeID: validBaseEnnoblement.NewTribeID(),
+				oldOwnerID: validBaseEnnoblement.OldOwnerID(),
+				oldTribeID: validBaseEnnoblement.OldTribeID(),
+				points:     validBaseEnnoblement.Points(),
+				createdAt:  validBaseEnnoblement.CreatedAt(),
+			},
+		},
+		{
+			name: "ERR: villageID < 1",
+			args: args{
+				villageID:  0,
+				newOwnerID: validBaseEnnoblement.NewOwnerID(),
+				newTribeID: validBaseEnnoblement.NewTribeID(),
+				oldOwnerID: validBaseEnnoblement.OldOwnerID(),
+				oldTribeID: validBaseEnnoblement.OldTribeID(),
+				points:     validBaseEnnoblement.Points(),
+				createdAt:  validBaseEnnoblement.CreatedAt(),
+			},
+			expectedErr: domain.ValidationError{
+				Model: "BaseEnnoblement",
+				Field: "villageID",
+				Err: domain.MinGreaterEqualError{
+					Min:     1,
+					Current: 0,
+				},
+			},
+		},
+		{
+			name: "ERR: newOwnerID < 0",
+			args: args{
+				villageID:  validBaseEnnoblement.VillageID(),
+				newOwnerID: -1,
+				newTribeID: validBaseEnnoblement.NewTribeID(),
+				oldOwnerID: validBaseEnnoblement.OldOwnerID(),
+				oldTribeID: validBaseEnnoblement.OldTribeID(),
+				points:     validBaseEnnoblement.Points(),
+				createdAt:  validBaseEnnoblement.CreatedAt(),
+			},
+			expectedErr: domain.ValidationError{
+				Model: "BaseEnnoblement",
+				Field: "newOwnerID",
+				Err: domain.MinGreaterEqualError{
+					Min:     0,
+					Current: -1,
+				},
+			},
+		},
+		{
+			name: "ERR: newTribeID < 0",
+			args: args{
+				villageID:  validBaseEnnoblement.VillageID(),
+				newOwnerID: validBaseEnnoblement.NewOwnerID(),
+				newTribeID: -1,
+				oldOwnerID: validBaseEnnoblement.OldOwnerID(),
+				oldTribeID: validBaseEnnoblement.OldTribeID(),
+				points:     validBaseEnnoblement.Points(),
+				createdAt:  validBaseEnnoblement.CreatedAt(),
+			},
+			expectedErr: domain.ValidationError{
+				Model: "BaseEnnoblement",
+				Field: "newTribeID",
+				Err: domain.MinGreaterEqualError{
+					Min:     0,
+					Current: -1,
+				},
+			},
+		},
+		{
+			name: "ERR: oldOwnerID < 0",
+			args: args{
+				villageID:  validBaseEnnoblement.VillageID(),
+				newOwnerID: validBaseEnnoblement.NewOwnerID(),
+				newTribeID: validBaseEnnoblement.NewTribeID(),
+				oldOwnerID: -1,
+				oldTribeID: validBaseEnnoblement.OldTribeID(),
+				points:     validBaseEnnoblement.Points(),
+				createdAt:  validBaseEnnoblement.CreatedAt(),
+			},
+			expectedErr: domain.ValidationError{
+				Model: "BaseEnnoblement",
+				Field: "oldOwnerID",
+				Err: domain.MinGreaterEqualError{
+					Min:     0,
+					Current: -1,
+				},
+			},
+		},
+		{
+			name: "ERR: oldTribeID < 0",
+			args: args{
+				villageID:  validBaseEnnoblement.VillageID(),
+				newOwnerID: validBaseEnnoblement.NewOwnerID(),
+				newTribeID: validBaseEnnoblement.NewTribeID(),
+				oldOwnerID: validBaseEnnoblement.OldOwnerID(),
+				oldTribeID: -1,
+				points:     validBaseEnnoblement.Points(),
+				createdAt:  validBaseEnnoblement.CreatedAt(),
+			},
+			expectedErr: domain.ValidationError{
+				Model: "BaseEnnoblement",
+				Field: "oldTribeID",
+				Err: domain.MinGreaterEqualError{
+					Min:     0,
+					Current: -1,
+				},
+			},
+		},
+		{
+			name: "ERR: points < 0",
+			args: args{
+				villageID:  validBaseEnnoblement.VillageID(),
+				newOwnerID: validBaseEnnoblement.NewOwnerID(),
+				newTribeID: validBaseEnnoblement.NewTribeID(),
+				oldOwnerID: validBaseEnnoblement.OldOwnerID(),
+				oldTribeID: validBaseEnnoblement.OldTribeID(),
+				points:     -1,
+				createdAt:  validBaseEnnoblement.CreatedAt(),
+			},
+			expectedErr: domain.ValidationError{
+				Model: "BaseEnnoblement",
+				Field: "points",
+				Err: domain.MinGreaterEqualError{
+					Min:     0,
+					Current: -1,
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			res, err := domain.NewBaseEnnoblement(
+				tt.args.villageID,
+				tt.args.newOwnerID,
+				tt.args.newTribeID,
+				tt.args.oldOwnerID,
+				tt.args.oldTribeID,
+				tt.args.points,
+				tt.args.createdAt,
+			)
+			require.ErrorIs(t, err, tt.expectedErr)
+			if tt.expectedErr != nil {
+				return
+			}
+			assert.Equal(t, tt.args.villageID, res.VillageID())
+			assert.Equal(t, tt.args.newOwnerID, res.NewOwnerID())
+			assert.Equal(t, tt.args.newTribeID, res.NewTribeID())
+			assert.Equal(t, tt.args.oldOwnerID, res.OldOwnerID())
+			assert.Equal(t, tt.args.oldTribeID, res.OldTribeID())
+			assert.Equal(t, tt.args.points, res.Points())
+			assert.Equal(t, tt.args.createdAt, res.CreatedAt())
+		})
+	}
+}

internal/domain/base_village.go

@@ -45,7 +45,7 @@ func NewBaseVillage(
 		}
 	}
 
-	if err := validateIntInRange(points, 1, math.MaxInt); err != nil {
+	if err := validateIntInRange(points, 0, math.MaxInt); err != nil {
 		return BaseVillage{}, ValidationError{
 			Model: baseVillageModelName,
 			Field: "points",

internal/domain/base_village_test.go

@@ -116,11 +116,11 @@ func TestNewBaseVillage(t *testing.T) {
 			},
 		},
 		{
-			name: "ERR: points < 1",
+			name: "ERR: points < 0",
 			args: args{
 				id:         validBaseVillage.ID(),
 				name:       validBaseVillage.Name(),
-				points:     0,
+				points:     -1,
 				x:          validBaseVillage.X(),
 				y:          validBaseVillage.Y(),
 				continent:  validBaseVillage.Continent(),
@@ -132,8 +132,8 @@ func TestNewBaseVillage(t *testing.T) {
 				Model: "BaseVillage",
 				Field: "points",
 				Err: domain.MinGreaterEqualError{
-					Min:     1,
-					Current: 0,
+					Min:     0,
+					Current: -1,
 				},
 			},
 		},

internal/domain/domaintest/base_ennoblement.go

@@ -0,0 +1,38 @@
+package domaintest
+
+import (
+	"time"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"github.com/brianvoe/gofakeit/v6"
+	"github.com/stretchr/testify/require"
+)
+
+type BaseEnnoblementConfig struct {
+	VillageID int
+}
+
+func NewBaseEnnoblement(tb TestingTB, opts ...func(cfg *BaseEnnoblementConfig)) domain.BaseEnnoblement {
+	tb.Helper()
+
+	cfg := &BaseEnnoblementConfig{
+		VillageID: RandID(),
+	}
+
+	for _, opt := range opts {
+		opt(cfg)
+	}
+
+	e, err := domain.NewBaseEnnoblement(
+		cfg.VillageID,
+		RandID(),
+		RandID(),
+		RandID(),
+		RandID(),
+		gofakeit.IntRange(1, 10000),
+		time.Now(),
+	)
+	require.NoError(tb, err)
+
+	return e
+}

internal/domain/domaintest/base_player.go

@@ -26,7 +26,7 @@ func NewBasePlayer(tb TestingTB, opts ...func(cfg *BasePlayerConfig)) domain.Bas
 		NumVillages: gofakeit.IntRange(1, 10000),
 		AllPoints:   gofakeit.IntRange(1, 10000),
 		Rank:        gofakeit.IntRange(1, 10000),
-		TribeID:     gofakeit.IntRange(0, 10000),
+		TribeID:     RandID(),
 	}
 
 	for _, opt := range opts {

internal/domain/domaintest/base_village.go

@@ -19,7 +19,7 @@ func NewBaseVillage(tb TestingTB, opts ...func(cfg *BaseVillageConfig)) domain.B
 
 	cfg := &BaseVillageConfig{
 		ID:       RandID(),
-		PlayerID: gofakeit.IntRange(0, 10000),
+		PlayerID: RandID(),
 		Bonus:    0,
 	}
 

internal/domain/domaintest/ennoblement.go

@@ -0,0 +1,44 @@
+package domaintest
+
+import (
+	"time"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"github.com/brianvoe/gofakeit/v6"
+	"github.com/stretchr/testify/require"
+)
+
+type EnnoblementConfig struct {
+	ID        int
+	ServerKey string
+	VillageID int
+}
+
+func NewEnnoblement(tb TestingTB, opts ...func(cfg *EnnoblementConfig)) domain.Ennoblement {
+	tb.Helper()
+
+	cfg := &EnnoblementConfig{
+		ID:        RandID(),
+		ServerKey: RandServerKey(),
+		VillageID: RandID(),
+	}
+
+	for _, opt := range opts {
+		opt(cfg)
+	}
+
+	e, err := domain.UnmarshalEnnoblementFromDatabase(
+		cfg.ID,
+		cfg.ServerKey,
+		cfg.VillageID,
+		RandID(),
+		RandID(),
+		RandID(),
+		RandID(),
+		gofakeit.IntRange(1, 10000),
+		time.Now(),
+	)
+	require.NoError(tb, err)
+
+	return e
+}

internal/domain/ennoblement.go

@@ -0,0 +1,253 @@
+package domain
+
+import (
+	"fmt"
+	"math"
+	"time"
+)
+
+type Ennoblement struct {
+	id         int
+	serverKey  string
+	villageID  int
+	newOwnerID int
+	newTribeID int
+	oldOwnerID int
+	oldTribeID int
+	points     int
+	createdAt  time.Time
+}
+
+const ennoblementModelName = "Ennoblement"
+
+// UnmarshalEnnoblementFromDatabase unmarshals Ennoblement from the database.
+//
+// It should be used only for unmarshalling from the database!
+// You can't use UnmarshalEnnoblementFromDatabase as constructor - It may put domain into the invalid state!
+func UnmarshalEnnoblementFromDatabase(
+	id int,
+	serverKey string,
+	villageID int,
+	newOwnerID int,
+	newTribeID int,
+	oldOwnerID int,
+	oldTribeID int,
+	points int,
+	createdAt time.Time,
+) (Ennoblement, error) {
+	if err := validateIntInRange(id, 1, math.MaxInt); err != nil {
+		return Ennoblement{}, ValidationError{
+			Model: ennoblementModelName,
+			Field: "id",
+			Err:   err,
+		}
+	}
+
+	if err := validateServerKey(serverKey); err != nil {
+		return Ennoblement{}, ValidationError{
+			Model: ennoblementModelName,
+			Field: "serverKey",
+			Err:   err,
+		}
+	}
+
+	return Ennoblement{
+		id:         id,
+		serverKey:  serverKey,
+		villageID:  villageID,
+		newOwnerID: newOwnerID,
+		newTribeID: newTribeID,
+		oldOwnerID: oldOwnerID,
+		oldTribeID: oldTribeID,
+		points:     points,
+		createdAt:  createdAt,
+	}, nil
+}
+
+func (e Ennoblement) ID() int {
+	return e.id
+}
+
+func (e Ennoblement) ServerKey() string {
+	return e.serverKey
+}
+
+func (e Ennoblement) VillageID() int {
+	return e.villageID
+}
+
+func (e Ennoblement) NewOwnerID() int {
+	return e.newOwnerID
+}
+
+func (e Ennoblement) NewTribeID() int {
+	return e.newTribeID
+}
+
+func (e Ennoblement) OldOwnerID() int {
+	return e.oldOwnerID
+}
+
+func (e Ennoblement) OldTribeID() int {
+	return e.oldTribeID
+}
+
+func (e Ennoblement) Points() int {
+	return e.points
+}
+
+func (e Ennoblement) CreatedAt() time.Time {
+	return e.createdAt
+}
+
+func (e Ennoblement) Base() BaseEnnoblement {
+	return BaseEnnoblement{
+		villageID:  e.villageID,
+		newOwnerID: e.newOwnerID,
+		newTribeID: e.newTribeID,
+		oldOwnerID: e.oldOwnerID,
+		oldTribeID: e.oldTribeID,
+		points:     e.points,
+		createdAt:  e.createdAt,
+	}
+}
+
+type Ennoblements []Ennoblement
+
+type CreateEnnoblementParams struct {
+	base      BaseEnnoblement
+	serverKey string
+}
+
+const createEnnoblementParamsModelName = "CreateEnnoblementParams"
+
+func NewCreateEnnoblementParams(serverKey string, ennoblements BaseEnnoblements) ([]CreateEnnoblementParams, error) {
+	if err := validateServerKey(serverKey); err != nil {
+		return nil, ValidationError{
+			Model: createEnnoblementParamsModelName,
+			Field: "serverKey",
+			Err:   err,
+		}
+	}
+
+	params := make([]CreateEnnoblementParams, 0, len(ennoblements))
+
+	for i, e := range ennoblements {
+		if e.IsZero() {
+			return nil, fmt.Errorf("ennoblements[%d] is an empty struct", i)
+		}
+
+		params = append(params, CreateEnnoblementParams{
+			base:      e,
+			serverKey: serverKey,
+		})
+	}
+
+	return params, nil
+}
+
+func (params CreateEnnoblementParams) Base() BaseEnnoblement {
+	return params.base
+}
+
+func (params CreateEnnoblementParams) ServerKey() string {
+	return params.serverKey
+}
+
+type EnnoblementSort uint8
+
+const (
+	EnnoblementSortCreatedAtASC EnnoblementSort = iota + 1
+	EnnoblementSortCreatedAtDESC
+	EnnoblementSortServerKeyASC
+	EnnoblementSortServerKeyDESC
+)
+
+const EnnoblementListMaxLimit = 200
+
+type ListEnnoblementsParams struct {
+	serverKeys []string
+	sort       []EnnoblementSort
+	limit      int
+	offset     int
+}
+
+const listEnnoblementsParamsModelName = "ListEnnoblementsParams"
+
+func NewListEnnoblementsParams() ListEnnoblementsParams {
+	return ListEnnoblementsParams{
+		sort: []EnnoblementSort{
+			EnnoblementSortServerKeyASC,
+			EnnoblementSortCreatedAtASC,
+		},
+		limit: EnnoblementListMaxLimit,
+	}
+}
+
+func (params *ListEnnoblementsParams) ServerKeys() []string {
+	return params.serverKeys
+}
+
+func (params *ListEnnoblementsParams) SetServerKeys(serverKeys []string) error {
+	params.serverKeys = serverKeys
+	return nil
+}
+
+func (params *ListEnnoblementsParams) Sort() []EnnoblementSort {
+	return params.sort
+}
+
+const (
+	ennoblementSortMinLength = 1
+	ennoblementSortMaxLength = 2
+)
+
+func (params *ListEnnoblementsParams) SetSort(sort []EnnoblementSort) error {
+	if err := validateSliceLen(sort, ennoblementSortMinLength, ennoblementSortMaxLength); err != nil {
+		return ValidationError{
+			Model: listEnnoblementsParamsModelName,
+			Field: "sort",
+			Err:   err,
+		}
+	}
+
+	params.sort = sort
+
+	return nil
+}
+
+func (params *ListEnnoblementsParams) Limit() int {
+	return params.limit
+}
+
+func (params *ListEnnoblementsParams) SetLimit(limit int) error {
+	if err := validateIntInRange(limit, 1, EnnoblementListMaxLimit); err != nil {
+		return ValidationError{
+			Model: listEnnoblementsParamsModelName,
+			Field: "limit",
+			Err:   err,
+		}
+	}
+
+	params.limit = limit
+
+	return nil
+}
+
+func (params *ListEnnoblementsParams) Offset() int {
+	return params.offset
+}
+
+func (params *ListEnnoblementsParams) SetOffset(offset int) error {
+	if err := validateIntInRange(offset, 0, math.MaxInt); err != nil {
+		return ValidationError{
+			Model: listEnnoblementsParamsModelName,
+			Field: "offset",
+			Err:   err,
+		}
+	}
+
+	params.offset = offset
+
+	return nil
+}

internal/domain/ennoblement_test.go

@@ -0,0 +1,227 @@
+package domain_test
+
+import (
+	"fmt"
+	"slices"
+	"testing"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain/domaintest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewCreateEnnoblementParams(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewServer(t)
+
+	ennoblements := domain.BaseEnnoblements{
+		domaintest.NewBaseEnnoblement(t),
+		domaintest.NewBaseEnnoblement(t),
+		domaintest.NewBaseEnnoblement(t),
+	}
+
+	res, err := domain.NewCreateEnnoblementParams(server.Key(), ennoblements)
+	require.NoError(t, err)
+	for i, e := range ennoblements {
+		idx := slices.IndexFunc(res, func(params domain.CreateEnnoblementParams) bool {
+			return params.Base().VillageID() == e.VillageID() && params.ServerKey() == server.Key()
+		})
+		require.GreaterOrEqualf(t, idx, 0, "ennoblements[%d] not found", i)
+
+		params := res[idx]
+
+		assert.Equalf(t, e, params.Base(), "ennoblements[%d]", i)
+	}
+}
+
+func TestListEnnoblementsParams_SetSort(t *testing.T) {
+	t.Parallel()
+
+	type args struct {
+		sort []domain.EnnoblementSort
+	}
+
+	tests := []struct {
+		name        string
+		args        args
+		expectedErr error
+	}{
+		{
+			name: "OK",
+			args: args{
+				sort: []domain.EnnoblementSort{
+					domain.EnnoblementSortCreatedAtASC,
+					domain.EnnoblementSortServerKeyASC,
+				},
+			},
+		},
+		{
+			name: "ERR: len(sort) < 1",
+			args: args{
+				sort: nil,
+			},
+			expectedErr: domain.ValidationError{
+				Model: "ListEnnoblementsParams",
+				Field: "sort",
+				Err: domain.LenOutOfRangeError{
+					Min:     1,
+					Max:     2,
+					Current: 0,
+				},
+			},
+		},
+		{
+			name: "ERR: len(sort) > 2",
+			args: args{
+				sort: []domain.EnnoblementSort{
+					domain.EnnoblementSortCreatedAtASC,
+					domain.EnnoblementSortServerKeyASC,
+					domain.EnnoblementSortServerKeyDESC,
+				},
+			},
+			expectedErr: domain.ValidationError{
+				Model: "ListEnnoblementsParams",
+				Field: "sort",
+				Err: domain.LenOutOfRangeError{
+					Min:     1,
+					Max:     2,
+					Current: 3,
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			params := domain.NewListEnnoblementsParams()
+
+			require.ErrorIs(t, params.SetSort(tt.args.sort), tt.expectedErr)
+			if tt.expectedErr != nil {
+				return
+			}
+			assert.Equal(t, tt.args.sort, params.Sort())
+		})
+	}
+}
+
+func TestListEnnoblementsParams_SetLimit(t *testing.T) {
+	t.Parallel()
+
+	type args struct {
+		limit int
+	}
+
+	tests := []struct {
+		name        string
+		args        args
+		expectedErr error
+	}{
+		{
+			name: "OK",
+			args: args{
+				limit: domain.EnnoblementListMaxLimit,
+			},
+		},
+		{
+			name: "ERR: limit < 1",
+			args: args{
+				limit: 0,
+			},
+			expectedErr: domain.ValidationError{
+				Model: "ListEnnoblementsParams",
+				Field: "limit",
+				Err: domain.MinGreaterEqualError{
+					Min:     1,
+					Current: 0,
+				},
+			},
+		},
+		{
+			name: fmt.Sprintf("ERR: limit > %d", domain.EnnoblementListMaxLimit),
+			args: args{
+				limit: domain.EnnoblementListMaxLimit + 1,
+			},
+			expectedErr: domain.ValidationError{
+				Model: "ListEnnoblementsParams",
+				Field: "limit",
+				Err: domain.MaxLessEqualError{
+					Max:     domain.EnnoblementListMaxLimit,
+					Current: domain.EnnoblementListMaxLimit + 1,
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			params := domain.NewListEnnoblementsParams()
+
+			require.ErrorIs(t, params.SetLimit(tt.args.limit), tt.expectedErr)
+			if tt.expectedErr != nil {
+				return
+			}
+			assert.Equal(t, tt.args.limit, params.Limit())
+		})
+	}
+}
+
+func TestListEnnoblementsParams_SetOffset(t *testing.T) {
+	t.Parallel()
+
+	type args struct {
+		offset int
+	}
+
+	tests := []struct {
+		name        string
+		args        args
+		expectedErr error
+	}{
+		{
+			name: "OK",
+			args: args{
+				offset: 100,
+			},
+		},
+		{
+			name: "ERR: offset < 0",
+			args: args{
+				offset: -1,
+			},
+			expectedErr: domain.ValidationError{
+				Model: "ListEnnoblementsParams",
+				Field: "offset",
+				Err: domain.MinGreaterEqualError{
+					Min:     0,
+					Current: -1,
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			params := domain.NewListEnnoblementsParams()
+
+			require.ErrorIs(t, params.SetOffset(tt.args.offset), tt.expectedErr)
+			if tt.expectedErr != nil {
+				return
+			}
+			assert.Equal(t, tt.args.offset, params.Offset())
+		})
+	}
+}

internal/domain/player_test.go

@@ -3,14 +3,12 @@ package domain_test
 import (
 	"cmp"
 	"fmt"
-	"math"
 	"slices"
 	"testing"
 	"time"
 
 	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
 	"gitea.dwysokinski.me/twhelp/corev3/internal/domain/domaintest"
-	"github.com/brianvoe/gofakeit/v6"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -195,9 +193,9 @@ func TestListPlayersParams_SetIDs(t *testing.T) {
 			name: "OK",
 			args: args{
 				ids: []int{
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
+					domaintest.RandID(),
+					domaintest.RandID(),
+					domaintest.RandID(),
 				},
 			},
 		},
@@ -205,11 +203,11 @@ func TestListPlayersParams_SetIDs(t *testing.T) {
 			name: "ERR: value < 0",
 			args: args{
 				ids: []int{
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
+					domaintest.RandID(),
+					domaintest.RandID(),
+					domaintest.RandID(),
 					-1,
-					gofakeit.IntRange(0, math.MaxInt),
+					domaintest.RandID(),
 				},
 			},
 			expectedErr: domain.SliceElementValidationError{
@@ -257,7 +255,7 @@ func TestListPlayersParams_SetIDGT(t *testing.T) {
 			name: "OK",
 			args: args{
 				idGT: domain.NullInt{
-					Value: gofakeit.IntRange(0, math.MaxInt),
+					Value: domaintest.RandID(),
 					Valid: true,
 				},
 			},

internal/domain/tribe_test.go

@@ -3,14 +3,12 @@ package domain_test
 import (
 	"cmp"
 	"fmt"
-	"math"
 	"slices"
 	"testing"
 	"time"
 
 	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
 	"gitea.dwysokinski.me/twhelp/corev3/internal/domain/domaintest"
-	"github.com/brianvoe/gofakeit/v6"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -178,9 +176,9 @@ func TestListTribesParams_SetIDs(t *testing.T) {
 			name: "OK",
 			args: args{
 				ids: []int{
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
+					domaintest.RandID(),
+					domaintest.RandID(),
+					domaintest.RandID(),
 				},
 			},
 		},
@@ -188,11 +186,11 @@ func TestListTribesParams_SetIDs(t *testing.T) {
 			name: "ERR: value < 0",
 			args: args{
 				ids: []int{
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
+					domaintest.RandID(),
+					domaintest.RandID(),
+					domaintest.RandID(),
 					-1,
-					gofakeit.IntRange(0, math.MaxInt),
+					domaintest.RandID(),
 				},
 			},
 			expectedErr: domain.SliceElementValidationError{
@@ -240,7 +238,7 @@ func TestListTribesParams_SetIDGT(t *testing.T) {
 			name: "OK",
 			args: args{
 				idGT: domain.NullInt{
-					Value: gofakeit.IntRange(0, math.MaxInt),
+					Value: domaintest.RandID(),
 					Valid: true,
 				},
 			},

internal/domain/village_message_payloads_test.go

@@ -57,11 +57,11 @@ func TestNewVillagesSyncedEventPayloadFromVillages(t *testing.T) {
 			cfg.Bonus = 0
 		}),
 		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
-			cfg.PlayerID = gofakeit.IntRange(1, 10000)
+			cfg.PlayerID = domaintest.RandID()
 			cfg.Bonus = 0
 		}),
 		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
-			cfg.PlayerID = gofakeit.IntRange(1, 10000)
+			cfg.PlayerID = domaintest.RandID()
 			cfg.Bonus = 1
 		}),
 	}

internal/domain/village_test.go

@@ -3,13 +3,11 @@ package domain_test
 import (
 	"cmp"
 	"fmt"
-	"math"
 	"slices"
 	"testing"
 
 	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
 	"gitea.dwysokinski.me/twhelp/corev3/internal/domain/domaintest"
-	"github.com/brianvoe/gofakeit/v6"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -72,7 +70,7 @@ func TestNewCreateVillageParams(t *testing.T) {
 		idx := slices.IndexFunc(res, func(params domain.CreateVillageParams) bool {
 			return params.Base().ID() == v.ID() && params.ServerKey() == server.Key()
 		})
-		require.GreaterOrEqualf(t, idx, 0, "village[%d] not found", i)
+		require.GreaterOrEqualf(t, idx, 0, "villages[%d] not found", i)
 
 		params := res[idx]
 
@@ -96,9 +94,9 @@ func TestListVillagesParams_SetIDs(t *testing.T) {
 			name: "OK",
 			args: args{
 				ids: []int{
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
+					domaintest.RandID(),
+					domaintest.RandID(),
+					domaintest.RandID(),
 				},
 			},
 		},
@@ -106,11 +104,11 @@ func TestListVillagesParams_SetIDs(t *testing.T) {
 			name: "ERR: value < 0",
 			args: args{
 				ids: []int{
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
-					gofakeit.IntRange(0, math.MaxInt),
+					domaintest.RandID(),
+					domaintest.RandID(),
+					domaintest.RandID(),
 					-1,
-					gofakeit.IntRange(0, math.MaxInt),
+					domaintest.RandID(),
 				},
 			},
 			expectedErr: domain.SliceElementValidationError{
@@ -158,7 +156,7 @@ func TestListVillagesParams_SetIDGT(t *testing.T) {
 			name: "OK",
 			args: args{
 				idGT: domain.NullInt{
-					Value: gofakeit.IntRange(0, math.MaxInt),
+					Value: domaintest.RandID(),
 					Valid: true,
 				},
 			},

internal/migrations/20231220051322_create_ennoblements_table.go

@@ -11,8 +11,7 @@ func init() {
 		_, err := db.ExecContext(ctx, `
 create table if not exists ennoblements
 (
-    id           bigint ?
-        primary key,
+    ?,
     server_key   varchar(100)                                       not null
         references servers,
     village_id   bigint                                             not null,
@@ -41,7 +40,7 @@ create index if not exists ennoblements_server_key_new_tribe_id_idx
 
 create index if not exists ennoblements_server_key_old_tribe_id_idx
     on ennoblements (server_key, old_tribe_id);
-`, bun.Safe(autoIncrement(db)))
+`, bun.Safe(autoincrementIDColumn(db)))
 		return err
 	}, func(ctx context.Context, db *bun.DB) error {
 		_, err := db.ExecContext(ctx, "drop table if exists ennoblements cascade;")

internal/migrations/20231220051411_create_player_snapshots_table.go

@@ -11,8 +11,7 @@ func init() {
 		_, err := db.ExecContext(ctx, `
 create table if not exists player_snapshots
 (
-    id           bigint ?
-        primary key,
+    ?,
     player_id    bigint                                             not null,
     num_villages bigint                   default 0,
     points       bigint                   default 0,
@@ -33,7 +32,7 @@ create table if not exists player_snapshots
     unique (player_id, server_key, date),
     foreign key (player_id, server_key) references players
 );
-`, bun.Safe(autoIncrement(db)))
+`, bun.Safe(autoincrementIDColumn(db)))
 		return err
 	}, func(ctx context.Context, db *bun.DB) error {
 		_, err := db.ExecContext(ctx, "drop table if exists player_snapshots cascade;")

internal/migrations/20231220052323_create_tribe_snapshots_table.go

@@ -11,7 +11,7 @@ func init() {
 		_, err := db.ExecContext(ctx, `
 create table if not exists tribe_snapshots
 (
-    id           bigint ? primary key,
+    ?,
     tribe_id     bigint                                             not null,
     server_key   varchar(100)                                       not null
         references servers,
@@ -34,7 +34,7 @@ create table if not exists tribe_snapshots
     unique (tribe_id, server_key, date),
     foreign key (tribe_id, server_key) references tribes
 );
-`, bun.Safe(autoIncrement(db)))
+`, bun.Safe(autoincrementIDColumn(db)))
 		return err
 	}, func(ctx context.Context, db *bun.DB) error {
 		_, err := db.ExecContext(ctx, "drop table if exists tribe_snapshots cascade;")

internal/migrations/20231220052428_create_tribe_changes_table.go

@@ -11,7 +11,7 @@ func init() {
 		_, err := db.ExecContext(ctx, `
 create table if not exists tribe_changes
 (
-    id           bigint ? primary key,
+    ?,
     player_id    bigint                                             not null,
     new_tribe_id bigint,
     old_tribe_id bigint,
@@ -29,7 +29,7 @@ create index if not exists tribe_changes_server_key_new_tribe_id_idx
 
 create index if not exists tribe_changes_server_key_old_tribe_id_idx
     on tribe_changes (server_key, old_tribe_id);
-`, bun.Safe(autoIncrement(db)))
+`, bun.Safe(autoincrementIDColumn(db)))
 		return err
 	}, func(ctx context.Context, db *bun.DB) error {
 		_, err := db.ExecContext(ctx, "drop table if exists tribe_changes cascade;")

internal/migrations/20231220052526_create_index_ennoblements_hash_key.go

@@ -9,26 +9,30 @@ import (
 
 //nolint:lll
 func init() {
-	// this index is for Postgres only
 	migrations.MustRegister(func(ctx context.Context, db *bun.DB) error {
-		if db.Dialect().Name() != dialect.PG {
-			return nil
-		}
+		var err error
 
-		_, err := db.ExecContext(
-			ctx,
-			`create unique index concurrently if not exists ennoblements_hash_key
+		if db.Dialect().Name() == dialect.PG {
+			// hash_record_extended is Postgres specific
+			// https://dba.stackexchange.com/questions/299098/why-doesnt-my-unique-constraint-trigger/299107#299107
+			_, err = db.ExecContext(
+				ctx,
+				`create unique index concurrently if not exists ennoblements_hash_key
     				on ennoblements (hash_record_extended(
                              ROW (server_key, village_id, new_owner_id, new_tribe_id, old_owner_id, old_tribe_id, points, created_at),
                              0::bigint));`,
-		)
-		return err
-	}, func(ctx context.Context, db *bun.DB) error {
-		if db.Dialect().Name() != dialect.PG {
-			return nil
+			)
+		} else {
+			_, err = db.ExecContext(
+				ctx,
+				`create unique index if not exists ennoblements_hash_key
+    				on ennoblements (server_key, coalesce(village_id, 0), coalesce(new_owner_id, 0), coalesce(new_tribe_id, 0), coalesce(old_owner_id, 0), coalesce(old_tribe_id, 0), points, created_at);`,
+			)
 		}
 
-		_, err := db.ExecContext(ctx, "DROP INDEX CONCURRENTLY ennoblements_hash_key")
+		return err
+	}, func(ctx context.Context, db *bun.DB) error {
+		_, err := db.NewDropIndex().IfExists().Index("ennoblements_hash_key").Concurrently().Exec(ctx)
 		return err
 	})
 }

internal/migrations/20231220052547_create_index_tribe_changes_hash_key.go

@@ -24,11 +24,7 @@ func init() {
 		)
 		return err
 	}, func(ctx context.Context, db *bun.DB) error {
-		if db.Dialect().Name() != dialect.PG {
-			return nil
-		}
-
-		_, err := db.ExecContext(ctx, "DROP INDEX CONCURRENTLY tribe_changes_hash_key")
+		_, err := db.NewDropIndex().IfExists().Index("tribe_changes_hash_key").Concurrently().Exec(ctx)
 		return err
 	})
 }

internal/migrations/sql_utils.go

@@ -1,17 +1,19 @@
 package migrations
 
-import "github.com/uptrace/bun/dialect/feature"
+import (
+	"github.com/uptrace/bun/dialect/feature"
+)
 
 type hasFeaturer interface {
 	HasFeature(feat feature.Feature) bool
 }
 
-func autoIncrement(f hasFeaturer) string {
+func autoincrementIDColumn(f hasFeaturer) string {
 	if f.HasFeature(feature.GeneratedIdentity) {
-		return "GENERATED BY DEFAULT AS IDENTITY"
+		// postgres
+		return "id bigint GENERATED BY DEFAULT AS IDENTITY primary key"
 	}
-	if f.HasFeature(feature.AutoIncrement) {
-		return "AUTO_INCREMENT"
-	}
-	return ""
+
+	// sqlite
+	return "id INTEGER PRIMARY KEY"
 }

internal/port/consumer_watermill_ennoblement.go

@@ -0,0 +1,67 @@
+package port
+
+import (
+	"gitea.dwysokinski.me/twhelp/corev3/internal/app"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/watermillmsg"
+	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill/message"
+)
+
+type EnnoblementWatermillConsumer struct {
+	svc          *app.EnnoblementService
+	subscriber   message.Subscriber
+	logger       watermill.LoggerAdapter
+	marshaler    watermillmsg.Marshaler
+	cmdSyncTopic string
+}
+
+func NewEnnoblementWatermillConsumer(
+	svc *app.EnnoblementService,
+	subscriber message.Subscriber,
+	logger watermill.LoggerAdapter,
+	marshaler watermillmsg.Marshaler,
+	cmdSyncTopic string,
+) *EnnoblementWatermillConsumer {
+	return &EnnoblementWatermillConsumer{
+		svc:          svc,
+		subscriber:   subscriber,
+		logger:       logger,
+		marshaler:    marshaler,
+		cmdSyncTopic: cmdSyncTopic,
+	}
+}
+
+func (c *EnnoblementWatermillConsumer) Register(router *message.Router) {
+	router.AddNoPublisherHandler(
+		"EnnoblementConsumer.sync",
+		c.cmdSyncTopic,
+		c.subscriber,
+		c.sync,
+	)
+}
+
+func (c *EnnoblementWatermillConsumer) sync(msg *message.Message) error {
+	var rawPayload watermillmsg.SyncEnnoblementsCmdPayload
+
+	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.NewSyncEnnoblementsCmdPayload(
+		rawPayload.ServerKey,
+		rawPayload.ServerURL,
+		rawPayload.VersionCode,
+	)
+	if err != nil {
+		c.logger.Error("couldn't construct domain.SyncEnnoblementsCmdPayload", err, watermill.LogFields{
+			"handler": message.HandlerNameFromCtx(msg.Context()),
+		})
+		return nil
+	}
+
+	return c.svc.Sync(msg.Context(), payload)
+}

internal/port/consumer_watermill_server.go

@@ -9,15 +9,16 @@ import (
 )
 
 type ServerWatermillConsumer struct {
-	svc                      *app.ServerService
-	subscriber               message.Subscriber
-	logger                   watermill.LoggerAdapter
-	marshaler                watermillmsg.Marshaler
-	cmdSyncTopic             string
-	eventServerSyncedTopic   string
-	eventTribesSyncedTopic   string
-	eventPlayersSyncedTopic  string
-	eventVillagesSyncedTopic string
+	svc                          *app.ServerService
+	subscriber                   message.Subscriber
+	logger                       watermill.LoggerAdapter
+	marshaler                    watermillmsg.Marshaler
+	cmdSyncTopic                 string
+	eventServerSyncedTopic       string
+	eventTribesSyncedTopic       string
+	eventPlayersSyncedTopic      string
+	eventVillagesSyncedTopic     string
+	eventEnnoblementsSyncedTopic string
 }
 
 func NewServerWatermillConsumer(
@@ -30,17 +31,19 @@ func NewServerWatermillConsumer(
 	eventTribesSyncedTopic string,
 	eventPlayersSyncedTopic string,
 	eventVillagesSyncedTopic string,
+	eventEnnoblementsSyncedTopic string,
 ) *ServerWatermillConsumer {
 	return &ServerWatermillConsumer{
-		svc:                      svc,
-		subscriber:               subscriber,
-		logger:                   logger,
-		marshaler:                marshaler,
-		cmdSyncTopic:             cmdSyncTopic,
-		eventServerSyncedTopic:   eventServerSyncedTopic,
-		eventTribesSyncedTopic:   eventTribesSyncedTopic,
-		eventPlayersSyncedTopic:  eventPlayersSyncedTopic,
-		eventVillagesSyncedTopic: eventVillagesSyncedTopic,
+		svc:                          svc,
+		subscriber:                   subscriber,
+		logger:                       logger,
+		marshaler:                    marshaler,
+		cmdSyncTopic:                 cmdSyncTopic,
+		eventServerSyncedTopic:       eventServerSyncedTopic,
+		eventTribesSyncedTopic:       eventTribesSyncedTopic,
+		eventPlayersSyncedTopic:      eventPlayersSyncedTopic,
+		eventVillagesSyncedTopic:     eventVillagesSyncedTopic,
+		eventEnnoblementsSyncedTopic: eventEnnoblementsSyncedTopic,
 	}
 }
 
@@ -70,6 +73,12 @@ func (c *ServerWatermillConsumer) Register(router *message.Router) {
 		c.subscriber,
 		c.updateNumVillages,
 	)
+	router.AddNoPublisherHandler(
+		"ServerConsumer.updateEnnoblementDataSyncedAt",
+		c.eventEnnoblementsSyncedTopic,
+		c.subscriber,
+		c.updateEnnoblementDataSyncedAt,
+	)
 }
 
 func (c *ServerWatermillConsumer) sync(msg *message.Message) error {
@@ -194,3 +203,28 @@ func (c *ServerWatermillConsumer) updateNumVillages(msg *message.Message) error
 
 	return c.svc.UpdateNumVillages(msg.Context(), payload)
 }
+
+func (c *ServerWatermillConsumer) updateEnnoblementDataSyncedAt(msg *message.Message) error {
+	var rawPayload watermillmsg.EnnoblementsSyncedEventPayload
+
+	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.NewEnnoblementsSyncedEventPayload(
+		rawPayload.ServerKey,
+		rawPayload.ServerURL,
+		rawPayload.VersionCode,
+	)
+	if err != nil {
+		c.logger.Error("couldn't construct domain.EnnoblementsSyncedEventPayload", err, watermill.LogFields{
+			"handler": message.HandlerNameFromCtx(msg.Context()),
+		})
+		return nil
+	}
+
+	return c.svc.UpdateEnnoblementDataSyncedAt(msg.Context(), payload)
+}

k8s/base/ennoblement-consumer.yml

@@ -0,0 +1,44 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: twhelp-ennoblement-consumer-deployment
+spec:
+  selector:
+    matchLabels:
+      app: twhelp-ennoblement-consumer
+  template:
+    metadata:
+      labels:
+        app: twhelp-ennoblement-consumer
+    spec:
+      containers:
+        - name: twhelp-ennoblement-consumer
+          image: twhelp
+          args: [consumer, ennoblement]
+          env:
+            - name: APP_MODE
+              value: development
+            - name: LOG_LEVEL
+              value: debug
+            - name: DB_CONNECTION_STRING
+              valueFrom:
+                secretKeyRef:
+                  name: twhelp-secret
+                  key: db-connection-string
+            - name: RABBITMQ_CONNECTION_STRING
+              valueFrom:
+                secretKeyRef:
+                  name: twhelp-secret
+                  key: rabbitmq-connection-string
+          livenessProbe:
+            exec:
+              command: [cat, /tmp/live]
+            initialDelaySeconds: 5
+            periodSeconds: 10
+          resources:
+            requests:
+              cpu: 100m
+              memory: 100Mi
+            limits:
+              cpu: 300m
+              memory: 300Mi

k8s/base/kustomization.yml

@@ -6,6 +6,7 @@ resources:
   - tribe-consumer.yml
   - player-consumer.yml
   - village-consumer.yml
+  - ennoblement-consumer.yml
 images:
   - name: twhelp
     newName: twhelp