feat: add a new job responsible for triggering data cleanup (#33)

Reviewed-on: twhelp/corev3#33

Makefile

@@ -50,3 +50,7 @@ create-job-sync-ennoblements:
 .PHONY: create-job-create-snapshots
 create-job-create-snapshots:
 	@kubectl create job --from=cronjob/twhelp-job-create-snapshots-dev "twhelp-job-create-snapshots-$(shell openssl rand -hex 10)"
+
+.PHONY: create-job-cleanup
+create-job-cleanup:
+	@kubectl create job --from=cronjob/twhelp-job-cleanup-dev "twhelp-job-cleanup-$(shell openssl rand -hex 10)"

cmd/twhelp/cmd_job.go

@@ -203,6 +203,61 @@ var (
 					},
 				},
 			},
+			{
+				Name:        "cleanup",
+				Description: "Trigger data cleanup",
+				Flags:       slices.Concat(dbFlags, rmqFlags),
+				Action: func(c *cli.Context) error {
+					logger := loggerFromCtx(c.Context)
+					watermillLogger := newWatermillLogger(logger)
+
+					amqpConn, err := newAMQPConnectionFromFlags(c, watermillLogger)
+					if err != nil {
+						return err
+					}
+					defer func() {
+						if closeErr := amqpConn.Close(); closeErr != nil {
+							logger.Warn("couldn't close amqp connection", slog.Any("error", err))
+						}
+					}()
+
+					publisher, err := newAMQPPublisher(amqpConn, watermillLogger)
+					if err != nil {
+						return err
+					}
+					defer func() {
+						if closeErr := publisher.Close(); closeErr != nil {
+							logger.Warn("couldn't close amqp publisher", slog.Any("error", err))
+						}
+					}()
+
+					bunDB, err := newBunDBFromFlags(c)
+					if err != nil {
+						return err
+					}
+					defer closeBunDB(bunDB, logger)
+
+					dataCleanupPublisher := adapter.NewDataCleanupPublisher(
+						publisher,
+						newWatermillMarshaler(),
+						c.String(rmqFlagTopicCleanUpDataCmd.Name),
+					)
+
+					serverSvc := app.NewServerService(adapter.NewServerBunRepository(bunDB), nil, nil)
+					dataCleanupSvc := app.NewDataCleanupService(serverSvc, dataCleanupPublisher)
+
+					shutdownSignalCtx, stop := newShutdownSignalContext(c.Context)
+					defer stop()
+
+					if err = dataCleanupSvc.CleanUp(shutdownSignalCtx); err != nil {
+						return fmt.Errorf("couldn't trigger data cleanup: %w", err)
+					}
+
+					logger.Info("data cleanup triggered")
+
+					return nil
+				},
+			},
 		},
 	}
 )

cmd/twhelp/rabbitmq.go

@@ -71,6 +71,11 @@ var (
 		Value:   "ennoblements.event.synced",
 		EnvVars: []string{"RABBITMQ_TOPIC_ENNOBLEMENTS_SYNCED_EVENT"},
 	}
+	rmqFlagTopicCleanUpDataCmd = &cli.StringFlag{
+		Name:    "rabbitmq.topic.cleanUpDataCmd",
+		Value:   "all.cmd.clean_up",
+		EnvVars: []string{"RABBITMQ_TOPIC_CLEAN_UP_DATA_CMD"},
+	}
 	rmqFlags = []cli.Flag{
 		rmqFlagConnectionString,
 		rmqFlagTopicSyncServersCmd,
@@ -84,6 +89,7 @@ var (
 		rmqFlagTopicVillagesSyncedEvent,
 		rmqFlagTopicSyncEnnoblementsCmd,
 		rmqFlagTopicEnnoblementsSyncedEvent,
+		rmqFlagTopicCleanUpDataCmd,
 	}
 )
 

internal/adapter/publisher_watermill_data_cleanup.go

@@ -0,0 +1,63 @@
+package adapter
+
+import (
+	"context"
+	"fmt"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/watermill/watermillmsg"
+	"github.com/ThreeDotsLabs/watermill/message"
+)
+
+type DataCleanupPublisher struct {
+	marshaler           watermillmsg.Marshaler
+	publisher           message.Publisher
+	cmdCleanUpDataTopic string
+}
+
+func NewDataCleanupPublisher(
+	publisher message.Publisher,
+	marshaler watermillmsg.Marshaler,
+	cmdCleanUpDataTopic string,
+) *DataCleanupPublisher {
+	return &DataCleanupPublisher{
+		publisher:           publisher,
+		marshaler:           marshaler,
+		cmdCleanUpDataTopic: cmdCleanUpDataTopic,
+	}
+}
+
+func (pub *DataCleanupPublisher) CmdCleanUp(
+	ctx context.Context,
+	payloads ...domain.CleanUpDataCmdPayload,
+) error {
+	msgs := make([]*message.Message, 0, len(payloads))
+
+	for _, p := range payloads {
+		msg, err := pub.marshaler.Marshal(ctx, watermillmsg.CleanUpDataCmdPayload{
+			Server: watermillmsg.CleanUpDataCmdPayloadServer{
+				Key:                      p.Server().Key(),
+				VersionCode:              p.Server().VersionCode(),
+				Open:                     p.Server().Open(),
+				Special:                  p.Server().Special(),
+				PlayerDataSyncedAt:       p.Server().PlayerDataSyncedAt(),
+				PlayerSnapshotsCreatedAt: p.Server().PlayerSnapshotsCreatedAt(),
+				TribeDataSyncedAt:        p.Server().TribeDataSyncedAt(),
+				TribeSnapshotsCreatedAt:  p.Server().TribeSnapshotsCreatedAt(),
+				VillageDataSyncedAt:      p.Server().VillageDataSyncedAt(),
+				EnnoblementDataSyncedAt:  p.Server().EnnoblementDataSyncedAt(),
+			},
+		})
+		if err != nil {
+			return fmt.Errorf("%s: couldn't marshal CleanUpDataCmdPayload: %w", p.Server().Key(), err)
+		}
+
+		msgs = append(msgs, msg)
+	}
+
+	if err := pub.publisher.Publish(pub.cmdCleanUpDataTopic, msgs...); err != nil {
+		return fmt.Errorf("couldn't publish messages to topic '%s': %w", pub.cmdCleanUpDataTopic, err)
+	}
+
+	return nil
+}

internal/app/publishers.go

@@ -32,3 +32,7 @@ type SnapshotPublisher interface {
 	CmdCreate(ctx context.Context, payloads ...domain.CreateSnapshotsCmdPayload) error
 	EventCreated(ctx context.Context, payloads ...domain.SnapshotsCreatedEventPayload) error
 }
+
+type DataCleanupPublisher interface {
+	CmdCleanUp(ctx context.Context, payloads ...domain.CleanUpDataCmdPayload) error
+}

internal/app/service_data_cleanup.go

@@ -0,0 +1,44 @@
+package app
+
+import (
+	"context"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+)
+
+type DataCleanupService struct {
+	serverSvc *ServerService
+	pub       DataCleanupPublisher
+}
+
+func NewDataCleanupService(serverSvc *ServerService, pub DataCleanupPublisher) *DataCleanupService {
+	return &DataCleanupService{serverSvc: serverSvc, pub: pub}
+}
+
+func (svc *DataCleanupService) CleanUp(ctx context.Context) error {
+	params := domain.NewListServersParams()
+	if err := params.SetOpen(domain.NullBool{
+		V:     false,
+		Valid: true,
+	}); err != nil {
+		return err
+	}
+	if err := params.SetSpecial(domain.NullBool{
+		V:     false,
+		Valid: true,
+	}); err != nil {
+		return err
+	}
+
+	servers, err := svc.serverSvc.ListAll(ctx, params)
+	if err != nil {
+		return err
+	}
+
+	payloads, err := servers.CleanUpData()
+	if err != nil {
+		return err
+	}
+
+	return svc.pub.CmdCleanUp(ctx, payloads...)
+}

internal/app/service_data_sync.go

@@ -50,23 +50,21 @@ func (svc *DataSyncService) Sync(ctx context.Context) error {
 }
 
 func (svc *DataSyncService) SyncEnnoblements(ctx context.Context) error {
-	listVersionsRes, err := svc.versionSvc.List(ctx, domain.NewListVersionsParams())
-	if err != nil {
+	params := domain.NewListServersParams()
+	if err := params.SetOpen(domain.NullBool{
+		V:     true,
+		Valid: true,
+	}); err != nil {
 		return err
 	}
-	versions := listVersionsRes.Versions()
-
-	for _, v := range versions {
-		if err = svc.syncEnnoblementsForVersion(ctx, v); err != nil {
-			return fmt.Errorf("%s: %w", v.Code(), err)
-		}
+	if err := params.SetSpecial(domain.NullBool{
+		V:     false,
+		Valid: true,
+	}); err != nil {
+		return err
 	}
 
-	return nil
-}
-
-func (svc *DataSyncService) syncEnnoblementsForVersion(ctx context.Context, v domain.Version) error {
-	servers, err := svc.serverSvc.ListAllOpen(ctx, v.Code())
+	servers, err := svc.serverSvc.ListAll(ctx, params)
 	if err != nil {
 		return err
 	}

internal/app/service_snapshot.go

@@ -10,10 +10,10 @@ import (
 )
 
 type SnapshotService struct {
-	versionSvc              *VersionService
-	serverSvc               *ServerService
-	tribeSnapshotPublisher  SnapshotPublisher
-	playerSnapshotPublisher SnapshotPublisher
+	versionSvc        *VersionService
+	serverSvc         *ServerService
+	tribeSnapshotPub  SnapshotPublisher
+	playerSnapshotPub SnapshotPublisher
 }
 
 func NewSnapshotService(
@@ -23,10 +23,10 @@ func NewSnapshotService(
 	playerSnapshotPublisher SnapshotPublisher,
 ) *SnapshotService {
 	return &SnapshotService{
-		versionSvc:              versionSvc,
-		serverSvc:               serverSvc,
-		tribeSnapshotPublisher:  tribeSnapshotPublisher,
-		playerSnapshotPublisher: playerSnapshotPublisher,
+		versionSvc:        versionSvc,
+		serverSvc:         serverSvc,
+		tribeSnapshotPub:  tribeSnapshotPublisher,
+		playerSnapshotPub: playerSnapshotPublisher,
 	}
 }
 
@@ -47,11 +47,10 @@ func (svc *SnapshotService) Create(ctx context.Context) error {
 		snapshotsCreatedAtLT := time.Date(year, month, day, 0, 0, 0, 0, loc)
 		date := time.Date(year, month, day, 0, 0, 0, 0, time.UTC)
 
-		loopErr = errors.Join(
+		if loopErr = errors.Join(
 			svc.publishTribe(ctx, v, snapshotsCreatedAtLT, date),
 			svc.publishPlayer(ctx, v, snapshotsCreatedAtLT, date),
-		)
-		if loopErr != nil {
+		); loopErr != nil {
 			return loopErr
 		}
 	}
@@ -92,7 +91,7 @@ func (svc *SnapshotService) publishTribe(
 		return err
 	}
 
-	return svc.tribeSnapshotPublisher.CmdCreate(ctx, payloads...)
+	return svc.tribeSnapshotPub.CmdCreate(ctx, payloads...)
 }
 
 func (svc *SnapshotService) publishPlayer(
@@ -128,7 +127,7 @@ func (svc *SnapshotService) publishPlayer(
 		return err
 	}
 
-	return svc.playerSnapshotPublisher.CmdCreate(ctx, payloads...)
+	return svc.playerSnapshotPub.CmdCreate(ctx, payloads...)
 }
 
 func (svc *SnapshotService) toPayload(

internal/domain/domaintest/message_payloads.go

@@ -0,0 +1,41 @@
+package domaintest
+
+import (
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"github.com/stretchr/testify/require"
+)
+
+type CleanUpDataCmdPayloadServerConfig struct {
+	ServerOptions []func(cfg *ServerConfig)
+}
+
+func NewCleanUpDataCmdPayloadServer(
+	tb TestingTB,
+	opts ...func(cfg *CleanUpDataCmdPayloadServerConfig),
+) domain.CleanUpDataCmdPayloadServer {
+	tb.Helper()
+
+	cfg := &CleanUpDataCmdPayloadServerConfig{}
+
+	for _, opt := range opts {
+		opt(cfg)
+	}
+
+	server := NewServer(tb, cfg.ServerOptions...)
+
+	payloadServer, err := domain.NewCleanUpDataCmdPayloadServer(
+		server.Key(),
+		server.VersionCode(),
+		server.Open(),
+		server.Special(),
+		server.PlayerDataSyncedAt(),
+		server.PlayerSnapshotsCreatedAt(),
+		server.TribeDataSyncedAt(),
+		server.TribeSnapshotsCreatedAt(),
+		server.VillageDataSyncedAt(),
+		server.EnnoblementDataSyncedAt(),
+	)
+	require.NoError(tb, err)
+
+	return payloadServer
+}

internal/domain/ennoblement_message_payloads.go

@@ -1,117 +0,0 @@
-package domain
-
-import (
-	"net/url"
-)
-
-type SyncEnnoblementsCmdPayload struct {
-	serverKey   string
-	serverURL   *url.URL
-	versionCode string
-}
-
-const syncEnnoblementsCmdPayloadModelName = "SyncEnnoblementsCmdPayload"
-
-func NewSyncEnnoblementsCmdPayload(
-	serverKey string,
-	serverURL *url.URL,
-	versionCode string,
-) (SyncEnnoblementsCmdPayload, error) {
-	if serverKey == "" {
-		return SyncEnnoblementsCmdPayload{}, ValidationError{
-			Model: syncEnnoblementsCmdPayloadModelName,
-			Field: "serverKey",
-			Err:   ErrRequired,
-		}
-	}
-
-	if serverURL == nil {
-		return SyncEnnoblementsCmdPayload{}, ValidationError{
-			Model: syncEnnoblementsCmdPayloadModelName,
-			Field: "serverURL",
-			Err:   ErrNil,
-		}
-	}
-
-	if versionCode == "" {
-		return SyncEnnoblementsCmdPayload{}, ValidationError{
-			Model: syncEnnoblementsCmdPayloadModelName,
-			Field: "versionCode",
-			Err:   ErrRequired,
-		}
-	}
-
-	return SyncEnnoblementsCmdPayload{
-		serverKey:   serverKey,
-		serverURL:   serverURL,
-		versionCode: versionCode,
-	}, nil
-}
-
-func (p SyncEnnoblementsCmdPayload) ServerKey() string {
-	return p.serverKey
-}
-
-func (p SyncEnnoblementsCmdPayload) ServerURL() *url.URL {
-	return p.serverURL
-}
-
-func (p SyncEnnoblementsCmdPayload) VersionCode() string {
-	return p.versionCode
-}
-
-type EnnoblementsSyncedEventPayload struct {
-	serverKey   string
-	serverURL   *url.URL
-	versionCode string
-}
-
-const ennoblementsSyncedEventPayloadModelName = "EnnoblementsSyncedEventPayload"
-
-func NewEnnoblementsSyncedEventPayload(
-	serverKey string,
-	serverURL *url.URL,
-	versionCode string,
-) (EnnoblementsSyncedEventPayload, error) {
-	if serverKey == "" {
-		return EnnoblementsSyncedEventPayload{}, ValidationError{
-			Model: ennoblementsSyncedEventPayloadModelName,
-			Field: "serverKey",
-			Err:   ErrRequired,
-		}
-	}
-
-	if serverURL == nil {
-		return EnnoblementsSyncedEventPayload{}, ValidationError{
-			Model: ennoblementsSyncedEventPayloadModelName,
-			Field: "serverURL",
-			Err:   ErrNil,
-		}
-	}
-
-	if versionCode == "" {
-		return EnnoblementsSyncedEventPayload{}, ValidationError{
-			Model: ennoblementsSyncedEventPayloadModelName,
-			Field: "versionCode",
-			Err:   ErrRequired,
-		}
-	}
-
-	return EnnoblementsSyncedEventPayload{
-		serverKey:   serverKey,
-		serverURL:   serverURL,
-		versionCode: versionCode,
-	}, nil
-}
-
-func (p EnnoblementsSyncedEventPayload) ServerKey() string {
-	return p.serverKey
-}
-
-func (p EnnoblementsSyncedEventPayload) ServerURL() *url.URL {
-	return p.serverURL
-}
-
-func (p EnnoblementsSyncedEventPayload) VersionCode() string {
-	return p.versionCode
-}

internal/domain/ennoblement_message_payloads_test.go

@@ -1,42 +0,0 @@
-package domain_test
-
-import (
-	"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 TestNewSyncEnnoblementsCmdPayload(t *testing.T) {
-	t.Parallel()
-
-	server := domaintest.NewServer(t)
-
-	payload, err := domain.NewSyncEnnoblementsCmdPayload(
-		server.Key(),
-		server.URL(),
-		server.VersionCode(),
-	)
-	require.NoError(t, err)
-	assert.Equal(t, server.Key(), payload.ServerKey())
-	assert.Equal(t, server.URL(), payload.ServerURL())
-	assert.Equal(t, server.VersionCode(), payload.VersionCode())
-}
-
-func TestNewEnnoblementsSyncedEventPayload(t *testing.T) {
-	t.Parallel()
-
-	server := domaintest.NewServer(t)
-
-	payload, err := domain.NewEnnoblementsSyncedEventPayload(
-		server.Key(),
-		server.URL(),
-		server.VersionCode(),
-	)
-	require.NoError(t, err)
-	assert.Equal(t, server.Key(), payload.ServerKey())
-	assert.Equal(t, server.URL(), payload.ServerURL())
-	assert.Equal(t, server.VersionCode(), payload.VersionCode())
-}

internal/domain/message_payloads.go

@@ -0,0 +1,733 @@
+package domain
+
+import (
+	"math"
+	"net/url"
+	"time"
+)
+
+type SyncServersCmdPayload struct {
+	versionCode string
+	url         *url.URL
+}
+
+const syncServersCmdPayloadModelName = "SyncServersCmdPayload"
+
+func NewSyncServersCmdPayload(versionCode string, u *url.URL) (SyncServersCmdPayload, error) {
+	if versionCode == "" {
+		return SyncServersCmdPayload{}, ValidationError{
+			Model: syncServersCmdPayloadModelName,
+			Field: "versionCode",
+			Err:   ErrRequired,
+		}
+	}
+
+	if u == nil {
+		return SyncServersCmdPayload{}, ValidationError{
+			Model: syncServersCmdPayloadModelName,
+			Field: "url",
+			Err:   ErrNil,
+		}
+	}
+
+	return SyncServersCmdPayload{versionCode: versionCode, url: u}, nil
+}
+
+func (p SyncServersCmdPayload) VersionCode() string {
+	return p.versionCode
+}
+
+func (p SyncServersCmdPayload) URL() *url.URL {
+	return p.url
+}
+
+type ServerSyncedEventPayload struct {
+	key         string
+	url         *url.URL
+	versionCode string
+}
+
+const serverSyncedEventPayloadModelName = "ServerSyncedEventPayload"
+
+func NewServerSyncedEventPayload(key string, u *url.URL, versionCode string) (ServerSyncedEventPayload, error) {
+	if key == "" {
+		return ServerSyncedEventPayload{}, ValidationError{
+			Model: serverSyncedEventPayloadModelName,
+			Field: "key",
+			Err:   ErrRequired,
+		}
+	}
+
+	if versionCode == "" {
+		return ServerSyncedEventPayload{}, ValidationError{
+			Model: serverSyncedEventPayloadModelName,
+			Field: "versionCode",
+			Err:   ErrRequired,
+		}
+	}
+
+	if u == nil {
+		return ServerSyncedEventPayload{}, ValidationError{
+			Model: serverSyncedEventPayloadModelName,
+			Field: "url",
+			Err:   ErrNil,
+		}
+	}
+
+	return ServerSyncedEventPayload{
+		key:         key,
+		url:         u,
+		versionCode: versionCode,
+	}, nil
+}
+
+func NewServerSyncedEventPayloads(servers BaseServers, versionCode string) ([]ServerSyncedEventPayload, error) {
+	res := make([]ServerSyncedEventPayload, 0, len(servers))
+
+	for _, s := range servers {
+		payload, err := NewServerSyncedEventPayload(s.Key(), s.URL(), versionCode)
+		if err != nil {
+			return nil, err
+		}
+
+		res = append(res, payload)
+	}
+
+	return res, nil
+}
+
+func (p ServerSyncedEventPayload) Key() string {
+	return p.key
+}
+
+func (p ServerSyncedEventPayload) URL() *url.URL {
+	return p.url
+}
+
+func (p ServerSyncedEventPayload) VersionCode() string {
+	return p.versionCode
+}
+
+type TribesSyncedEventPayload struct {
+	serverKey   string
+	serverURL   *url.URL
+	versionCode string
+	numTribes   int
+}
+
+const tribesSyncedEventPayloadModelName = "TribesSyncedEventPayload"
+
+func NewTribesSyncedEventPayload(
+	serverKey string,
+	serverURL *url.URL,
+	versionCode string,
+	numTribes int,
+) (TribesSyncedEventPayload, error) {
+	if serverKey == "" {
+		return TribesSyncedEventPayload{}, ValidationError{
+			Model: tribesSyncedEventPayloadModelName,
+			Field: "serverKey",
+			Err:   ErrRequired,
+		}
+	}
+
+	if serverURL == nil {
+		return TribesSyncedEventPayload{}, ValidationError{
+			Model: tribesSyncedEventPayloadModelName,
+			Field: "serverURL",
+			Err:   ErrNil,
+		}
+	}
+
+	if versionCode == "" {
+		return TribesSyncedEventPayload{}, ValidationError{
+			Model: tribesSyncedEventPayloadModelName,
+			Field: "versionCode",
+			Err:   ErrRequired,
+		}
+	}
+
+	if err := validateIntInRange(numTribes, 0, math.MaxInt); err != nil {
+		return TribesSyncedEventPayload{}, ValidationError{
+			Model: tribesSyncedEventPayloadModelName,
+			Field: "numTribes",
+			Err:   err,
+		}
+	}
+
+	return TribesSyncedEventPayload{
+		serverKey:   serverKey,
+		serverURL:   serverURL,
+		versionCode: versionCode,
+		numTribes:   numTribes,
+	}, nil
+}
+
+func (p TribesSyncedEventPayload) ServerKey() string {
+	return p.serverKey
+}
+
+func (p TribesSyncedEventPayload) ServerURL() *url.URL {
+	return p.serverURL
+}
+
+func (p TribesSyncedEventPayload) VersionCode() string {
+	return p.versionCode
+}
+
+func (p TribesSyncedEventPayload) NumTribes() int {
+	return p.numTribes
+}
+
+type PlayersSyncedEventPayload struct {
+	serverKey   string
+	serverURL   *url.URL
+	versionCode string
+	numPlayers  int
+}
+
+const playersSyncedEventPayloadModelName = "PlayersSyncedEventPayload"
+
+func NewPlayersSyncedEventPayload(
+	serverKey string,
+	serverURL *url.URL,
+	versionCode string,
+	numPlayers int,
+) (PlayersSyncedEventPayload, error) {
+	if serverKey == "" {
+		return PlayersSyncedEventPayload{}, ValidationError{
+			Model: playersSyncedEventPayloadModelName,
+			Field: "serverKey",
+			Err:   ErrRequired,
+		}
+	}
+
+	if serverURL == nil {
+		return PlayersSyncedEventPayload{}, ValidationError{
+			Model: playersSyncedEventPayloadModelName,
+			Field: "serverURL",
+			Err:   ErrNil,
+		}
+	}
+
+	if versionCode == "" {
+		return PlayersSyncedEventPayload{}, ValidationError{
+			Model: playersSyncedEventPayloadModelName,
+			Field: "versionCode",
+			Err:   ErrRequired,
+		}
+	}
+
+	if err := validateIntInRange(numPlayers, 0, math.MaxInt); err != nil {
+		return PlayersSyncedEventPayload{}, ValidationError{
+			Model: playersSyncedEventPayloadModelName,
+			Field: "numPlayers",
+			Err:   err,
+		}
+	}
+
+	return PlayersSyncedEventPayload{
+		serverKey:   serverKey,
+		serverURL:   serverURL,
+		versionCode: versionCode,
+		numPlayers:  numPlayers,
+	}, nil
+}
+
+func (p PlayersSyncedEventPayload) ServerKey() string {
+	return p.serverKey
+}
+
+func (p PlayersSyncedEventPayload) ServerURL() *url.URL {
+	return p.serverURL
+}
+
+func (p PlayersSyncedEventPayload) VersionCode() string {
+	return p.versionCode
+}
+
+func (p PlayersSyncedEventPayload) NumPlayers() int {
+	return p.numPlayers
+}
+
+type VillagesSyncedEventPayload struct {
+	serverKey            string
+	serverURL            *url.URL
+	versionCode          string
+	numVillages          int
+	numPlayerVillages    int
+	numBarbarianVillages int
+	numBonusVillages     int
+}
+
+const villagesSyncedEventPayloadModelName = "VillagesSyncedEventPayload"
+
+func NewVillagesSyncedEventPayload(
+	serverKey string,
+	serverURL *url.URL,
+	versionCode string,
+	numVillages int,
+	numPlayerVillages int,
+	numBarbarianVillages int,
+	numBonusVillages int,
+) (VillagesSyncedEventPayload, error) {
+	if serverKey == "" {
+		return VillagesSyncedEventPayload{}, ValidationError{
+			Model: villagesSyncedEventPayloadModelName,
+			Field: "serverKey",
+			Err:   ErrRequired,
+		}
+	}
+
+	if serverURL == nil {
+		return VillagesSyncedEventPayload{}, ValidationError{
+			Model: villagesSyncedEventPayloadModelName,
+			Field: "serverURL",
+			Err:   ErrNil,
+		}
+	}
+
+	if versionCode == "" {
+		return VillagesSyncedEventPayload{}, ValidationError{
+			Model: villagesSyncedEventPayloadModelName,
+			Field: "versionCode",
+			Err:   ErrRequired,
+		}
+	}
+
+	if err := validateIntInRange(numVillages, 0, math.MaxInt); err != nil {
+		return VillagesSyncedEventPayload{}, ValidationError{
+			Model: villagesSyncedEventPayloadModelName,
+			Field: "numVillages",
+			Err:   err,
+		}
+	}
+
+	if err := validateIntInRange(numPlayerVillages, 0, math.MaxInt); err != nil {
+		return VillagesSyncedEventPayload{}, ValidationError{
+			Model: villagesSyncedEventPayloadModelName,
+			Field: "numPlayerVillages",
+			Err:   err,
+		}
+	}
+
+	if err := validateIntInRange(numBarbarianVillages, 0, math.MaxInt); err != nil {
+		return VillagesSyncedEventPayload{}, ValidationError{
+			Model: villagesSyncedEventPayloadModelName,
+			Field: "numBarbarianVillages",
+			Err:   err,
+		}
+	}
+
+	if err := validateIntInRange(numBonusVillages, 0, math.MaxInt); err != nil {
+		return VillagesSyncedEventPayload{}, ValidationError{
+			Model: villagesSyncedEventPayloadModelName,
+			Field: "numBonusVillages",
+			Err:   err,
+		}
+	}
+
+	return VillagesSyncedEventPayload{
+		serverKey:            serverKey,
+		serverURL:            serverURL,
+		versionCode:          versionCode,
+		numVillages:          numVillages,
+		numPlayerVillages:    numPlayerVillages,
+		numBarbarianVillages: numBarbarianVillages,
+		numBonusVillages:     numBonusVillages,
+	}, nil
+}
+
+func NewVillagesSyncedEventPayloadFromVillages(
+	serverKey string,
+	serverURL *url.URL,
+	versionCode string,
+	villages BaseVillages,
+) (VillagesSyncedEventPayload, error) {
+	numPlayerVillages := 0
+	numBarbarianVillages := 0
+	numBonusVillages := 0
+
+	for _, v := range villages {
+		if v.PlayerID() > 0 {
+			numPlayerVillages++
+		} else {
+			numBarbarianVillages++
+		}
+
+		if v.Bonus() > 0 {
+			numBonusVillages++
+		}
+	}
+
+	return NewVillagesSyncedEventPayload(
+		serverKey,
+		serverURL,
+		versionCode,
+		len(villages),
+		numPlayerVillages,
+		numBarbarianVillages,
+		numBonusVillages,
+	)
+}
+
+func (p VillagesSyncedEventPayload) ServerKey() string {
+	return p.serverKey
+}
+
+func (p VillagesSyncedEventPayload) ServerURL() *url.URL {
+	return p.serverURL
+}
+
+func (p VillagesSyncedEventPayload) VersionCode() string {
+	return p.versionCode
+}
+
+func (p VillagesSyncedEventPayload) NumVillages() int {
+	return p.numVillages
+}
+
+func (p VillagesSyncedEventPayload) NumPlayerVillages() int {
+	return p.numPlayerVillages
+}
+
+func (p VillagesSyncedEventPayload) NumBarbarianVillages() int {
+	return p.numBarbarianVillages
+}
+
+func (p VillagesSyncedEventPayload) NumBonusVillages() int {
+	return p.numBonusVillages
+}
+
+type SyncEnnoblementsCmdPayload struct {
+	serverKey   string
+	serverURL   *url.URL
+	versionCode string
+}
+
+const syncEnnoblementsCmdPayloadModelName = "SyncEnnoblementsCmdPayload"
+
+func NewSyncEnnoblementsCmdPayload(
+	serverKey string,
+	serverURL *url.URL,
+	versionCode string,
+) (SyncEnnoblementsCmdPayload, error) {
+	if serverKey == "" {
+		return SyncEnnoblementsCmdPayload{}, ValidationError{
+			Model: syncEnnoblementsCmdPayloadModelName,
+			Field: "serverKey",
+			Err:   ErrRequired,
+		}
+	}
+
+	if serverURL == nil {
+		return SyncEnnoblementsCmdPayload{}, ValidationError{
+			Model: syncEnnoblementsCmdPayloadModelName,
+			Field: "serverURL",
+			Err:   ErrNil,
+		}
+	}
+
+	if versionCode == "" {
+		return SyncEnnoblementsCmdPayload{}, ValidationError{
+			Model: syncEnnoblementsCmdPayloadModelName,
+			Field: "versionCode",
+			Err:   ErrRequired,
+		}
+	}
+
+	return SyncEnnoblementsCmdPayload{
+		serverKey:   serverKey,
+		serverURL:   serverURL,
+		versionCode: versionCode,
+	}, nil
+}
+
+func (p SyncEnnoblementsCmdPayload) ServerKey() string {
+	return p.serverKey
+}
+
+func (p SyncEnnoblementsCmdPayload) ServerURL() *url.URL {
+	return p.serverURL
+}
+
+func (p SyncEnnoblementsCmdPayload) VersionCode() string {
+	return p.versionCode
+}
+
+type EnnoblementsSyncedEventPayload struct {
+	serverKey   string
+	serverURL   *url.URL
+	versionCode string
+}
+
+const ennoblementsSyncedEventPayloadModelName = "EnnoblementsSyncedEventPayload"
+
+func NewEnnoblementsSyncedEventPayload(
+	serverKey string,
+	serverURL *url.URL,
+	versionCode string,
+) (EnnoblementsSyncedEventPayload, error) {
+	if serverKey == "" {
+		return EnnoblementsSyncedEventPayload{}, ValidationError{
+			Model: ennoblementsSyncedEventPayloadModelName,
+			Field: "serverKey",
+			Err:   ErrRequired,
+		}
+	}
+
+	if serverURL == nil {
+		return EnnoblementsSyncedEventPayload{}, ValidationError{
+			Model: ennoblementsSyncedEventPayloadModelName,
+			Field: "serverURL",
+			Err:   ErrNil,
+		}
+	}
+
+	if versionCode == "" {
+		return EnnoblementsSyncedEventPayload{}, ValidationError{
+			Model: ennoblementsSyncedEventPayloadModelName,
+			Field: "versionCode",
+			Err:   ErrRequired,
+		}
+	}
+
+	return EnnoblementsSyncedEventPayload{
+		serverKey:   serverKey,
+		serverURL:   serverURL,
+		versionCode: versionCode,
+	}, nil
+}
+
+func (p EnnoblementsSyncedEventPayload) ServerKey() string {
+	return p.serverKey
+}
+
+func (p EnnoblementsSyncedEventPayload) ServerURL() *url.URL {
+	return p.serverURL
+}
+
+func (p EnnoblementsSyncedEventPayload) VersionCode() string {
+	return p.versionCode
+}
+
+type CreateSnapshotsCmdPayload struct {
+	serverKey       string
+	versionCode     string
+	versionTimezone string
+	date            time.Time
+}
+
+const createSnapshotsCmdPayloadModelName = "CreateSnapshotsCmdPayload"
+
+func NewCreateSnapshotsCmdPayload(
+	serverKey string,
+	versionCode string,
+	versionTimezone string,
+	date time.Time,
+) (CreateSnapshotsCmdPayload, error) {
+	if serverKey == "" {
+		return CreateSnapshotsCmdPayload{}, ValidationError{
+			Model: createSnapshotsCmdPayloadModelName,
+			Field: "serverKey",
+			Err:   ErrRequired,
+		}
+	}
+
+	if versionCode == "" {
+		return CreateSnapshotsCmdPayload{}, ValidationError{
+			Model: createSnapshotsCmdPayloadModelName,
+			Field: "versionCode",
+			Err:   ErrRequired,
+		}
+	}
+
+	if versionTimezone == "" {
+		return CreateSnapshotsCmdPayload{}, ValidationError{
+			Model: createSnapshotsCmdPayloadModelName,
+			Field: "versionTimezone",
+			Err:   ErrRequired,
+		}
+	}
+
+	return CreateSnapshotsCmdPayload{
+		serverKey:       serverKey,
+		versionCode:     versionCode,
+		versionTimezone: versionTimezone,
+		date:            date,
+	}, nil
+}
+
+func (p CreateSnapshotsCmdPayload) ServerKey() string {
+	return p.serverKey
+}
+
+func (p CreateSnapshotsCmdPayload) VersionCode() string {
+	return p.versionCode
+}
+
+func (p CreateSnapshotsCmdPayload) VersionTimezone() string {
+	return p.versionTimezone
+}
+
+func (p CreateSnapshotsCmdPayload) Date() time.Time {
+	return p.date
+}
+
+type SnapshotsCreatedEventPayload struct {
+	serverKey   string
+	versionCode string
+}
+
+const snapshotsCreatedEventPayloadModelName = "SnapshotsCreatedEventPayload"
+
+func NewSnapshotsCreatedEventPayload(serverKey string, versionCode string) (SnapshotsCreatedEventPayload, error) {
+	if serverKey == "" {
+		return SnapshotsCreatedEventPayload{}, ValidationError{
+			Model: snapshotsCreatedEventPayloadModelName,
+			Field: "serverKey",
+			Err:   ErrRequired,
+		}
+	}
+
+	if versionCode == "" {
+		return SnapshotsCreatedEventPayload{}, ValidationError{
+			Model: snapshotsCreatedEventPayloadModelName,
+			Field: "versionCode",
+			Err:   ErrRequired,
+		}
+	}
+
+	return SnapshotsCreatedEventPayload{serverKey: serverKey, versionCode: versionCode}, nil
+}
+
+func (p SnapshotsCreatedEventPayload) ServerKey() string {
+	return p.serverKey
+}
+
+func (p SnapshotsCreatedEventPayload) VersionCode() string {
+	return p.versionCode
+}
+
+type CleanUpDataCmdPayloadServer struct {
+	key                      string
+	versionCode              string
+	open                     bool
+	special                  bool
+	playerDataSyncedAt       time.Time
+	playerSnapshotsCreatedAt time.Time
+	tribeDataSyncedAt        time.Time
+	tribeSnapshotsCreatedAt  time.Time
+	villageDataSyncedAt      time.Time
+	ennoblementDataSyncedAt  time.Time
+}
+
+const cleanUpDataCmdPayloadServerModelName = "CleanUpDataCmdPayloadServer"
+
+func NewCleanUpDataCmdPayloadServer(
+	key string,
+	versionCode string,
+	open bool,
+	special bool,
+	playerDataSyncedAt time.Time,
+	playerSnapshotsCreatedAt time.Time,
+	tribeDataSyncedAt time.Time,
+	tribeSnapshotsCreatedAt time.Time,
+	villageDataSyncedAt time.Time,
+	ennoblementDataSyncedAt time.Time,
+) (CleanUpDataCmdPayloadServer, error) {
+	if key == "" {
+		return CleanUpDataCmdPayloadServer{}, ValidationError{
+			Model: cleanUpDataCmdPayloadServerModelName,
+			Field: "key",
+			Err:   ErrRequired,
+		}
+	}
+
+	if versionCode == "" {
+		return CleanUpDataCmdPayloadServer{}, ValidationError{
+			Model: cleanUpDataCmdPayloadServerModelName,
+			Field: "versionCode",
+			Err:   ErrRequired,
+		}
+	}
+
+	return CleanUpDataCmdPayloadServer{
+		key:                      key,
+		versionCode:              versionCode,
+		open:                     open,
+		special:                  special,
+		playerDataSyncedAt:       playerDataSyncedAt,
+		playerSnapshotsCreatedAt: playerSnapshotsCreatedAt,
+		tribeDataSyncedAt:        tribeDataSyncedAt,
+		tribeSnapshotsCreatedAt:  tribeSnapshotsCreatedAt,
+		villageDataSyncedAt:      villageDataSyncedAt,
+		ennoblementDataSyncedAt:  ennoblementDataSyncedAt,
+	}, nil
+}
+
+func (s CleanUpDataCmdPayloadServer) Key() string {
+	return s.key
+}
+
+func (s CleanUpDataCmdPayloadServer) VersionCode() string {
+	return s.versionCode
+}
+
+func (s CleanUpDataCmdPayloadServer) Open() bool {
+	return s.open
+}
+
+func (s CleanUpDataCmdPayloadServer) Special() bool {
+	return s.special
+}
+
+func (s CleanUpDataCmdPayloadServer) PlayerDataSyncedAt() time.Time {
+	return s.playerDataSyncedAt
+}
+
+func (s CleanUpDataCmdPayloadServer) PlayerSnapshotsCreatedAt() time.Time {
+	return s.playerSnapshotsCreatedAt
+}
+
+func (s CleanUpDataCmdPayloadServer) TribeDataSyncedAt() time.Time {
+	return s.tribeDataSyncedAt
+}
+
+func (s CleanUpDataCmdPayloadServer) TribeSnapshotsCreatedAt() time.Time {
+	return s.tribeSnapshotsCreatedAt
+}
+
+func (s CleanUpDataCmdPayloadServer) VillageDataSyncedAt() time.Time {
+	return s.villageDataSyncedAt
+}
+
+func (s CleanUpDataCmdPayloadServer) EnnoblementDataSyncedAt() time.Time {
+	return s.ennoblementDataSyncedAt
+}
+
+func (s CleanUpDataCmdPayloadServer) IsZero() bool {
+	return s == CleanUpDataCmdPayloadServer{}
+}
+
+type CleanUpDataCmdPayload struct {
+	server CleanUpDataCmdPayloadServer
+}
+
+const cleanUpDataCmdPayloadModelName = "CleanUpDataCmdPayload"
+
+func NewCleanUpDataCmdPayload(server CleanUpDataCmdPayloadServer) (CleanUpDataCmdPayload, error) {
+	if server.IsZero() {
+		return CleanUpDataCmdPayload{}, ValidationError{
+			Model: cleanUpDataCmdPayloadModelName,
+			Field: "server",
+			Err:   ErrRequired,
+		}
+	}
+
+	return CleanUpDataCmdPayload{server: server}, nil
+}
+
+func (p CleanUpDataCmdPayload) Server() CleanUpDataCmdPayloadServer {
+	return p.server
+}

internal/domain/message_payloads_test.go

@@ -0,0 +1,264 @@
+package domain_test
+
+import (
+	"math"
+	"slices"
+	"testing"
+	"time"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain/domaintest"
+	"github.com/brianvoe/gofakeit/v7"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewSyncServersCmdPayload(t *testing.T) {
+	t.Parallel()
+
+	version := domaintest.NewVersion(t)
+
+	payload, err := domain.NewSyncServersCmdPayload(version.Code(), version.URL())
+	require.NoError(t, err)
+	assert.Equal(t, version.Code(), payload.VersionCode())
+	assert.Equal(t, version.URL(), payload.URL())
+}
+
+func TestNewServerSyncedEventPayload(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewServer(t)
+
+	payload, err := domain.NewServerSyncedEventPayload(server.Key(), server.URL(), server.VersionCode())
+	require.NoError(t, err)
+	assert.Equal(t, server.Key(), payload.Key())
+	assert.Equal(t, server.URL(), payload.URL())
+	assert.Equal(t, server.VersionCode(), payload.VersionCode())
+}
+
+func TestNewServerSyncedEventPayloads(t *testing.T) {
+	t.Parallel()
+
+	versionCode := domaintest.RandVersionCode()
+	servers := domain.BaseServers{
+		domaintest.NewBaseServer(t),
+		domaintest.NewBaseServer(t),
+	}
+
+	payloads, err := domain.NewServerSyncedEventPayloads(servers, versionCode)
+	require.NoError(t, err)
+	for _, s := range servers {
+		assert.True(t, slices.ContainsFunc(payloads, func(payload domain.ServerSyncedEventPayload) bool {
+			return payload.Key() == s.Key() && payload.VersionCode() == versionCode && payload.URL() == s.URL()
+		}))
+	}
+}
+
+func TestNewTribesSyncedEventPayload(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewServer(t)
+	numTribes := gofakeit.IntRange(0, math.MaxInt)
+
+	payload, err := domain.NewTribesSyncedEventPayload(
+		server.Key(),
+		server.URL(),
+		server.VersionCode(),
+		numTribes,
+	)
+	require.NoError(t, err)
+	assert.Equal(t, server.Key(), payload.ServerKey())
+	assert.Equal(t, server.URL(), payload.ServerURL())
+	assert.Equal(t, server.VersionCode(), payload.VersionCode())
+	assert.Equal(t, numTribes, payload.NumTribes())
+}
+
+func TestNewPlayersSyncedEventPayload(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewServer(t)
+	numPlayers := gofakeit.IntRange(0, math.MaxInt)
+
+	payload, err := domain.NewPlayersSyncedEventPayload(
+		server.Key(),
+		server.URL(),
+		server.VersionCode(),
+		numPlayers,
+	)
+	require.NoError(t, err)
+	assert.Equal(t, server.Key(), payload.ServerKey())
+	assert.Equal(t, server.URL(), payload.ServerURL())
+	assert.Equal(t, server.VersionCode(), payload.VersionCode())
+	assert.Equal(t, numPlayers, payload.NumPlayers())
+}
+
+func TestNewVillagesSyncedEventPayload(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewServer(t)
+	numVillages := gofakeit.IntRange(0, math.MaxInt)
+	numPlayerVillages := gofakeit.IntRange(0, math.MaxInt)
+	numBarbarianVillages := gofakeit.IntRange(0, math.MaxInt)
+	numBonusVillages := gofakeit.IntRange(0, math.MaxInt)
+
+	payload, err := domain.NewVillagesSyncedEventPayload(
+		server.Key(),
+		server.URL(),
+		server.VersionCode(),
+		numVillages,
+		numPlayerVillages,
+		numBarbarianVillages,
+		numBonusVillages,
+	)
+	require.NoError(t, err)
+	assert.Equal(t, server.Key(), payload.ServerKey())
+	assert.Equal(t, server.URL(), payload.ServerURL())
+	assert.Equal(t, server.VersionCode(), payload.VersionCode())
+	assert.Equal(t, numVillages, payload.NumVillages())
+	assert.Equal(t, numPlayerVillages, payload.NumPlayerVillages())
+	assert.Equal(t, numBarbarianVillages, payload.NumBarbarianVillages())
+	assert.Equal(t, numBonusVillages, payload.NumBonusVillages())
+}
+
+func TestNewVillagesSyncedEventPayloadFromVillages(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewServer(t)
+	villages := domain.BaseVillages{
+		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
+			cfg.PlayerID = 0
+			cfg.Bonus = 1
+		}),
+		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
+			cfg.PlayerID = 0
+			cfg.Bonus = 0
+		}),
+		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
+			cfg.PlayerID = 0
+			cfg.Bonus = 0
+		}),
+		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
+			cfg.PlayerID = domaintest.RandID()
+			cfg.Bonus = 0
+		}),
+		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
+			cfg.PlayerID = domaintest.RandID()
+			cfg.Bonus = 1
+		}),
+	}
+
+	payload, err := domain.NewVillagesSyncedEventPayloadFromVillages(
+		server.Key(),
+		server.URL(),
+		server.VersionCode(),
+		villages,
+	)
+	require.NoError(t, err)
+	assert.Equal(t, server.Key(), payload.ServerKey())
+	assert.Equal(t, server.URL(), payload.ServerURL())
+	assert.Equal(t, server.VersionCode(), payload.VersionCode())
+	assert.Equal(t, len(villages), payload.NumVillages()) //nolint:testifylint
+	assert.Equal(t, 2, payload.NumPlayerVillages())
+	assert.Equal(t, 2, payload.NumBonusVillages())
+	assert.Equal(t, 3, payload.NumBarbarianVillages())
+}
+
+func TestNewSyncEnnoblementsCmdPayload(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewServer(t)
+
+	payload, err := domain.NewSyncEnnoblementsCmdPayload(
+		server.Key(),
+		server.URL(),
+		server.VersionCode(),
+	)
+	require.NoError(t, err)
+	assert.Equal(t, server.Key(), payload.ServerKey())
+	assert.Equal(t, server.URL(), payload.ServerURL())
+	assert.Equal(t, server.VersionCode(), payload.VersionCode())
+}
+
+func TestNewEnnoblementsSyncedEventPayload(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewServer(t)
+
+	payload, err := domain.NewEnnoblementsSyncedEventPayload(
+		server.Key(),
+		server.URL(),
+		server.VersionCode(),
+	)
+	require.NoError(t, err)
+	assert.Equal(t, server.Key(), payload.ServerKey())
+	assert.Equal(t, server.URL(), payload.ServerURL())
+	assert.Equal(t, server.VersionCode(), payload.VersionCode())
+}
+
+func TestNewCreateSnapshotsCmdPayload(t *testing.T) {
+	t.Parallel()
+
+	version := domaintest.NewVersion(t)
+	server := domaintest.NewServer(t, func(cfg *domaintest.ServerConfig) {
+		cfg.VersionCode = version.Code()
+	})
+	date := time.Now()
+
+	payload, err := domain.NewCreateSnapshotsCmdPayload(server.Key(), version.Code(), version.Timezone(), date)
+	require.NoError(t, err)
+	assert.Equal(t, server.Key(), payload.ServerKey())
+	assert.Equal(t, version.Code(), payload.VersionCode())
+	assert.Equal(t, version.Timezone(), payload.VersionTimezone())
+	assert.Equal(t, date, payload.Date())
+}
+
+func TestNewSnapshotsCreatedEventPayload(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewServer(t)
+
+	payload, err := domain.NewSnapshotsCreatedEventPayload(server.Key(), server.VersionCode())
+	require.NoError(t, err)
+	assert.Equal(t, server.Key(), payload.ServerKey())
+	assert.Equal(t, server.VersionCode(), payload.VersionCode())
+}
+
+func TestNewCleanUpDataCmdPayloadServer(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewServer(t)
+
+	payloadServer, err := domain.NewCleanUpDataCmdPayloadServer(
+		server.Key(),
+		server.VersionCode(),
+		server.Open(),
+		server.Special(),
+		server.PlayerDataSyncedAt(),
+		server.PlayerSnapshotsCreatedAt(),
+		server.TribeDataSyncedAt(),
+		server.TribeSnapshotsCreatedAt(),
+		server.VillageDataSyncedAt(),
+		server.EnnoblementDataSyncedAt(),
+	)
+	require.NoError(t, err)
+	assert.Equal(t, server.Key(), payloadServer.Key())
+	assert.Equal(t, server.VersionCode(), payloadServer.VersionCode())
+	assert.Equal(t, server.Open(), payloadServer.Open())
+	assert.Equal(t, server.Special(), payloadServer.Special())
+	assert.Equal(t, server.PlayerDataSyncedAt(), payloadServer.PlayerDataSyncedAt())
+	assert.Equal(t, server.PlayerSnapshotsCreatedAt(), payloadServer.PlayerSnapshotsCreatedAt())
+	assert.Equal(t, server.TribeDataSyncedAt(), payloadServer.TribeDataSyncedAt())
+	assert.Equal(t, server.TribeSnapshotsCreatedAt(), payloadServer.TribeSnapshotsCreatedAt())
+	assert.Equal(t, server.VillageDataSyncedAt(), payloadServer.VillageDataSyncedAt())
+	assert.Equal(t, server.EnnoblementDataSyncedAt(), payloadServer.EnnoblementDataSyncedAt())
+}
+
+func TestNewCleanUpDataCmdPayload(t *testing.T) {
+	t.Parallel()
+
+	server := domaintest.NewCleanUpDataCmdPayloadServer(t)
+
+	payload, err := domain.NewCleanUpDataCmdPayload(server)
+	require.NoError(t, err)
+	assert.Equal(t, server, payload.Server())
+}

internal/domain/player.go

@@ -232,6 +232,23 @@ func (p Player) IsZero() bool {
 	return p == Player{}
 }
 
+// active players must be sorted in ascending order by ID
+func (p Player) canBeDeleted(serverKey string, active BasePlayers) bool {
+	if p.IsDeleted() {
+		return false
+	}
+
+	if p.serverKey != serverKey {
+		return false
+	}
+
+	_, found := slices.BinarySearchFunc(active, p, func(a BasePlayer, b Player) int {
+		return cmp.Compare(a.ID(), b.id)
+	})
+
+	return !found
+}
+
 type Players []Player
 
 // Delete finds all players with the given serverKey that are not in the given slice with active players
@@ -245,26 +262,19 @@ func (ps Players) Delete(serverKey string, active BasePlayers) ([]int, []CreateT
 	var params []CreateTribeChangeParams
 
 	for _, p := range ps {
-		if p.IsDeleted() || p.ServerKey() != serverKey {
-			continue
-		}
-
-		_, found := slices.BinarySearchFunc(active, p, func(a BasePlayer, b Player) int {
-			return cmp.Compare(a.ID(), b.ID())
-		})
-		if found {
+		if !p.canBeDeleted(serverKey, active) {
 			continue
 		}
 
 		toDelete = append(toDelete, p.ID())
 
 		if p.TribeID() > 0 {
-			p, err := NewCreateTribeChangeParams(serverKey, p.ID(), p.TribeID(), 0)
+			ctcParams, err := NewCreateTribeChangeParams(serverKey, p.ID(), p.TribeID(), 0)
 			if err != nil {
 				return nil, nil, err
 			}
 
-			params = append(params, p)
+			params = append(params, ctcParams)
 		}
 	}
 

internal/domain/player_message_payloads.go

@@ -1,77 +0,0 @@
-package domain
-
-import (
-	"math"
-	"net/url"
-)
-
-type PlayersSyncedEventPayload struct {
-	serverKey   string
-	serverURL   *url.URL
-	versionCode string
-	numPlayers  int
-}
-
-const playersSyncedEventPayloadModelName = "PlayersSyncedEventPayload"
-
-func NewPlayersSyncedEventPayload(
-	serverKey string,
-	serverURL *url.URL,
-	versionCode string,
-	numPlayers int,
-) (PlayersSyncedEventPayload, error) {
-	if serverKey == "" {
-		return PlayersSyncedEventPayload{}, ValidationError{
-			Model: playersSyncedEventPayloadModelName,
-			Field: "serverKey",
-			Err:   ErrRequired,
-		}
-	}
-
-	if serverURL == nil {
-		return PlayersSyncedEventPayload{}, ValidationError{
-			Model: playersSyncedEventPayloadModelName,
-			Field: "serverURL",
-			Err:   ErrNil,
-		}
-	}
-
-	if versionCode == "" {
-		return PlayersSyncedEventPayload{}, ValidationError{
-			Model: playersSyncedEventPayloadModelName,
-			Field: "versionCode",
-			Err:   ErrRequired,
-		}
-	}
-
-	if err := validateIntInRange(numPlayers, 0, math.MaxInt); err != nil {
-		return PlayersSyncedEventPayload{}, ValidationError{
-			Model: playersSyncedEventPayloadModelName,
-			Field: "numPlayers",
-			Err:   err,
-		}
-	}
-
-	return PlayersSyncedEventPayload{
-		serverKey:   serverKey,
-		serverURL:   serverURL,
-		versionCode: versionCode,
-		numPlayers:  numPlayers,
-	}, nil
-}
-
-func (p PlayersSyncedEventPayload) ServerKey() string {
-	return p.serverKey
-}
-
-func (p PlayersSyncedEventPayload) ServerURL() *url.URL {
-	return p.serverURL
-}
-
-func (p PlayersSyncedEventPayload) VersionCode() string {
-	return p.versionCode
-}
-
-func (p PlayersSyncedEventPayload) NumPlayers() int {
-	return p.numPlayers
-}

internal/domain/player_message_payloads_test.go

@@ -1,31 +0,0 @@
-package domain_test
-
-import (
-	"math"
-	"testing"
-
-	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
-	"gitea.dwysokinski.me/twhelp/corev3/internal/domain/domaintest"
-	"github.com/brianvoe/gofakeit/v7"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestNewPlayersSyncedEventPayload(t *testing.T) {
-	t.Parallel()
-
-	server := domaintest.NewServer(t)
-	numPlayers := gofakeit.IntRange(0, math.MaxInt)
-
-	payload, err := domain.NewPlayersSyncedEventPayload(
-		server.Key(),
-		server.URL(),
-		server.VersionCode(),
-		numPlayers,
-	)
-	require.NoError(t, err)
-	assert.Equal(t, server.Key(), payload.ServerKey())
-	assert.Equal(t, server.URL(), payload.ServerURL())
-	assert.Equal(t, server.VersionCode(), payload.VersionCode())
-	assert.Equal(t, numPlayers, payload.NumPlayers())
-}

internal/domain/server.go

@@ -207,6 +207,16 @@ func (s Server) Base() BaseServer {
 	}
 }
 
+func (s Server) canBeClosed(open BaseServers) bool {
+	return s.open && !slices.ContainsFunc(open, func(openServer BaseServer) bool {
+		return openServer.Key() == s.key && openServer.Open() == s.open
+	})
+}
+
+func (s Server) canCleanUpData() bool {
+	return !s.open && !s.special
+}
+
 func (s Server) IsZero() bool {
 	return s == Server{}
 }
@@ -219,9 +229,7 @@ func (ss Servers) Close(open BaseServers) (BaseServers, error) {
 	res := make(BaseServers, 0, len(ss))
 
 	for _, s := range ss {
-		if !s.Open() || slices.ContainsFunc(open, func(openServer BaseServer) bool {
-			return openServer.Key() == s.Key() && openServer.Open() == s.Open()
-		}) {
+		if !s.canBeClosed(open) {
 			continue
 		}
 
@@ -236,6 +244,42 @@ func (ss Servers) Close(open BaseServers) (BaseServers, error) {
 	return res, nil
 }
 
+// CleanUpData finds all servers for which old data (ennoblements, snapshots etc.) can be deleted.
+func (ss Servers) CleanUpData() ([]CleanUpDataCmdPayload, error) {
+	res := make([]CleanUpDataCmdPayload, 0, len(ss))
+
+	for _, s := range ss {
+		if !s.canCleanUpData() {
+			continue
+		}
+
+		payloadServer, err := NewCleanUpDataCmdPayloadServer(
+			s.Key(),
+			s.VersionCode(),
+			s.Open(),
+			s.Special(),
+			s.PlayerDataSyncedAt(),
+			s.PlayerSnapshotsCreatedAt(),
+			s.TribeDataSyncedAt(),
+			s.TribeSnapshotsCreatedAt(),
+			s.VillageDataSyncedAt(),
+			s.EnnoblementDataSyncedAt(),
+		)
+		if err != nil {
+			return nil, fmt.Errorf("couldn't construct CleanUpDataCmdPayloadServer for server with key '%s': %w", s.Key(), err)
+		}
+
+		payload, err := NewCleanUpDataCmdPayload(payloadServer)
+		if err != nil {
+			return nil, fmt.Errorf("couldn't construct CleanUpDataCmdPayload for server with key '%s': %w", s.Key(), err)
+		}
+
+		res = append(res, payload)
+	}
+
+	return res, nil
+}
+
 type CreateServerParams struct {
 	base        BaseServer
 	versionCode string

internal/domain/server_message_payloads.go

@@ -1,107 +0,0 @@
-package domain
-
-import (
-	"net/url"
-)
-
-type SyncServersCmdPayload struct {
-	versionCode string
-	url         *url.URL
-}
-
-const syncServersCmdPayloadModelName = "SyncServersCmdPayload"
-
-func NewSyncServersCmdPayload(versionCode string, u *url.URL) (SyncServersCmdPayload, error) {
-	if versionCode == "" {
-		return SyncServersCmdPayload{}, ValidationError{
-			Model: syncServersCmdPayloadModelName,
-			Field: "versionCode",
-			Err:   ErrRequired,
-		}
-	}
-
-	if u == nil {
-		return SyncServersCmdPayload{}, ValidationError{
-			Model: syncServersCmdPayloadModelName,
-			Field: "url",
-			Err:   ErrNil,
-		}
-	}
-
-	return SyncServersCmdPayload{versionCode: versionCode, url: u}, nil
-}
-
-func (p SyncServersCmdPayload) VersionCode() string {
-	return p.versionCode
-}
-
-func (p SyncServersCmdPayload) URL() *url.URL {
-	return p.url
-}
-
-type ServerSyncedEventPayload struct {
-	key         string
-	url         *url.URL
-	versionCode string
-}
-
-const serverSyncedEventPayloadModelName = "ServerSyncedEventPayload"
-
-func NewServerSyncedEventPayload(key string, u *url.URL, versionCode string) (ServerSyncedEventPayload, error) {
-	if key == "" {
-		return ServerSyncedEventPayload{}, ValidationError{
-			Model: serverSyncedEventPayloadModelName,
-			Field: "key",
-			Err:   ErrRequired,
-		}
-	}
-
-	if versionCode == "" {
-		return ServerSyncedEventPayload{}, ValidationError{
-			Model: serverSyncedEventPayloadModelName,
-			Field: "versionCode",
-			Err:   ErrRequired,
-		}
-	}
-
-	if u == nil {
-		return ServerSyncedEventPayload{}, ValidationError{
-			Model: serverSyncedEventPayloadModelName,
-			Field: "url",
-			Err:   ErrNil,
-		}
-	}
-
-	return ServerSyncedEventPayload{
-		key:         key,
-		url:         u,
-		versionCode: versionCode,
-	}, nil
-}
-
-func NewServerSyncedEventPayloads(servers BaseServers, versionCode string) ([]ServerSyncedEventPayload, error) {
-	res := make([]ServerSyncedEventPayload, 0, len(servers))
-
-	for _, s := range servers {
-		payload, err := NewServerSyncedEventPayload(s.Key(), s.URL(), versionCode)
-		if err != nil {
-			return nil, err
-		}
-
-		res = append(res, payload)
-	}
-
-	return res, nil
-}
-
-func (p ServerSyncedEventPayload) Key() string {
-	return p.key
-}
-
-func (p ServerSyncedEventPayload) URL() *url.URL {
-	return p.url
-}
-
-func (p ServerSyncedEventPayload) VersionCode() string {
-	return p.versionCode
-}

internal/domain/server_message_payloads_test.go

@@ -1,52 +0,0 @@
-package domain_test
-
-import (
-	"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 TestNewSyncServersCmdPayload(t *testing.T) {
-	t.Parallel()
-
-	version := domaintest.NewVersion(t)
-
-	payload, err := domain.NewSyncServersCmdPayload(version.Code(), version.URL())
-	require.NoError(t, err)
-	assert.Equal(t, version.Code(), payload.VersionCode())
-	assert.Equal(t, version.URL(), payload.URL())
-}
-
-func TestNewServerSyncedEventPayload(t *testing.T) {
-	t.Parallel()
-
-	server := domaintest.NewServer(t)
-
-	payload, err := domain.NewServerSyncedEventPayload(server.Key(), server.URL(), server.VersionCode())
-	require.NoError(t, err)
-	assert.Equal(t, server.Key(), payload.Key())
-	assert.Equal(t, server.URL(), payload.URL())
-	assert.Equal(t, server.VersionCode(), payload.VersionCode())
-}
-
-func TestNewServerSyncedEventPayloads(t *testing.T) {
-	t.Parallel()
-
-	versionCode := domaintest.RandVersionCode()
-	servers := domain.BaseServers{
-		domaintest.NewBaseServer(t),
-		domaintest.NewBaseServer(t),
-	}
-
-	payloads, err := domain.NewServerSyncedEventPayloads(servers, versionCode)
-	require.NoError(t, err)
-	for _, s := range servers {
-		assert.True(t, slices.ContainsFunc(payloads, func(payload domain.ServerSyncedEventPayload) bool {
-			return payload.Key() == s.Key() && payload.VersionCode() == versionCode && payload.URL() == s.URL()
-		}))
-	}
-}

internal/domain/server_test.go

@@ -56,6 +56,47 @@ func TestServers_Close(t *testing.T) {
 	}
 }
 
+func TestServers_CleanUpData(t *testing.T) {
+	t.Parallel()
+
+	servers := domain.Servers{
+		domaintest.NewServer(t, func(cfg *domaintest.ServerConfig) {
+			cfg.Special = true
+		}),
+		domaintest.NewServer(t, func(cfg *domaintest.ServerConfig) {
+			cfg.Open = true
+			cfg.Special = false
+		}),
+		domaintest.NewServer(t, func(cfg *domaintest.ServerConfig) {
+			cfg.Open = false
+			cfg.Special = false
+		}),
+	}
+
+	expectedServers := domain.Servers{servers[2]}
+
+	res, err := servers.CleanUpData()
+	require.NoError(t, err)
+	assert.Len(t, res, len(expectedServers))
+	for i, expected := range expectedServers {
+		idx := slices.IndexFunc(res, func(payload domain.CleanUpDataCmdPayload) bool {
+			return payload.Server().Key() == expected.Key()
+		})
+		require.GreaterOrEqualf(t, idx, 0, "expected[%d] not found", i)
+
+		assert.Equal(t, expected.Key(), res[idx].Server().Key())
+		assert.Equal(t, expected.VersionCode(), res[idx].Server().VersionCode())
+		assert.Equal(t, expected.Open(), res[idx].Server().Open())
+		assert.Equal(t, expected.Special(), res[idx].Server().Special())
+		assert.Equal(t, expected.PlayerDataSyncedAt(), res[idx].Server().PlayerDataSyncedAt())
+		assert.Equal(t, expected.PlayerSnapshotsCreatedAt(), res[idx].Server().PlayerSnapshotsCreatedAt())
+		assert.Equal(t, expected.TribeDataSyncedAt(), res[idx].Server().TribeDataSyncedAt())
+		assert.Equal(t, expected.TribeSnapshotsCreatedAt(), res[idx].Server().TribeSnapshotsCreatedAt())
+		assert.Equal(t, expected.VillageDataSyncedAt(), res[idx].Server().VillageDataSyncedAt())
+		assert.Equal(t, expected.EnnoblementDataSyncedAt(), res[idx].Server().EnnoblementDataSyncedAt())
+	}
+}
+
 func TestNewCreateServerParams(t *testing.T) {
 	t.Parallel()
 

internal/domain/snapshot_message_payloads.go

@@ -1,101 +0,0 @@
-package domain
-
-import "time"
-
-type CreateSnapshotsCmdPayload struct {
-	serverKey       string
-	versionCode     string
-	versionTimezone string
-	date            time.Time
-}
-
-const createSnapshotsCmdPayloadModelName = "CreateSnapshotsCmdPayload"
-
-func NewCreateSnapshotsCmdPayload(
-	serverKey string,
-	versionCode string,
-	versionTimezone string,
-	date time.Time,
-) (CreateSnapshotsCmdPayload, error) {
-	if serverKey == "" {
-		return CreateSnapshotsCmdPayload{}, ValidationError{
-			Model: createSnapshotsCmdPayloadModelName,
-			Field: "serverKey",
-			Err:   ErrRequired,
-		}
-	}
-
-	if versionCode == "" {
-		return CreateSnapshotsCmdPayload{}, ValidationError{
-			Model: createSnapshotsCmdPayloadModelName,
-			Field: "versionCode",
-			Err:   ErrRequired,
-		}
-	}
-
-	if versionTimezone == "" {
-		return CreateSnapshotsCmdPayload{}, ValidationError{
-			Model: createSnapshotsCmdPayloadModelName,
-			Field: "versionTimezone",
-			Err:   ErrRequired,
-		}
-	}
-
-	return CreateSnapshotsCmdPayload{
-		serverKey:       serverKey,
-		versionCode:     versionCode,
-		versionTimezone: versionTimezone,
-		date:            date,
-	}, nil
-}
-
-func (p CreateSnapshotsCmdPayload) ServerKey() string {
-	return p.serverKey
-}
-
-func (p CreateSnapshotsCmdPayload) VersionCode() string {
-	return p.versionCode
-}
-
-func (p CreateSnapshotsCmdPayload) VersionTimezone() string {
-	return p.versionTimezone
-}
-
-func (p CreateSnapshotsCmdPayload) Date() time.Time {
-	return p.date
-}
-
-type SnapshotsCreatedEventPayload struct {
-	serverKey   string
-	versionCode string
-}
-
-const snapshotsCreatedEventPayloadModelName = "SnapshotsCreatedEventPayload"
-
-func NewSnapshotsCreatedEventPayload(serverKey string, versionCode string) (SnapshotsCreatedEventPayload, error) {
-	if serverKey == "" {
-		return SnapshotsCreatedEventPayload{}, ValidationError{
-			Model: snapshotsCreatedEventPayloadModelName,
-			Field: "serverKey",
-			Err:   ErrRequired,
-		}
-	}
-
-	if versionCode == "" {
-		return SnapshotsCreatedEventPayload{}, ValidationError{
-			Model: snapshotsCreatedEventPayloadModelName,
-			Field: "versionCode",
-			Err:   ErrRequired,
-		}
-	}
-
-	return SnapshotsCreatedEventPayload{serverKey: serverKey, versionCode: versionCode}, nil
-}
-
-func (p SnapshotsCreatedEventPayload) ServerKey() string {
-	return p.serverKey
-}
-
-func (p SnapshotsCreatedEventPayload) VersionCode() string {
-	return p.versionCode
-}

internal/domain/snapshot_message_payloads_test.go

@@ -1,39 +0,0 @@
-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 TestNewCreateSnapshotsCmdPayload(t *testing.T) {
-	t.Parallel()
-
-	version := domaintest.NewVersion(t)
-	server := domaintest.NewServer(t, func(cfg *domaintest.ServerConfig) {
-		cfg.VersionCode = version.Code()
-	})
-	date := time.Now()
-
-	payload, err := domain.NewCreateSnapshotsCmdPayload(server.Key(), version.Code(), version.Timezone(), date)
-	require.NoError(t, err)
-	assert.Equal(t, server.Key(), payload.ServerKey())
-	assert.Equal(t, version.Code(), payload.VersionCode())
-	assert.Equal(t, version.Timezone(), payload.VersionTimezone())
-	assert.Equal(t, date, payload.Date())
-}
-
-func TestNewSnapshotsCreatedEventPayload(t *testing.T) {
-	t.Parallel()
-
-	server := domaintest.NewServer(t)
-
-	payload, err := domain.NewSnapshotsCreatedEventPayload(server.Key(), server.VersionCode())
-	require.NoError(t, err)
-	assert.Equal(t, server.Key(), payload.ServerKey())
-	assert.Equal(t, server.VersionCode(), payload.VersionCode())
-}

internal/domain/tribe.go

@@ -244,6 +244,23 @@ func (t Tribe) IsZero() bool {
 	return t == Tribe{}
 }
 
+// active tribes must be sorted in ascending order by ID
+func (t Tribe) canBeDeleted(serverKey string, active BaseTribes) bool {
+	if t.IsDeleted() {
+		return false
+	}
+
+	if t.serverKey != serverKey {
+		return false
+	}
+
+	_, found := slices.BinarySearchFunc(active, t, func(a BaseTribe, b Tribe) int {
+		return cmp.Compare(a.ID(), b.id)
+	})
+
+	return !found
+}
+
 type Tribes []Tribe
 
 // Delete finds all tribes with the given serverKey that are not in the given slice with active tribes
@@ -254,14 +271,7 @@ func (ts Tribes) Delete(serverKey string, active BaseTribes) []int {
 	var toDelete []int
 
 	for _, t := range ts {
-		if t.IsDeleted() || t.ServerKey() != serverKey {
-			continue
-		}
-
-		_, found := slices.BinarySearchFunc(active, t, func(a BaseTribe, b Tribe) int {
-			return cmp.Compare(a.ID(), b.ID())
-		})
-		if found {
+		if !t.canBeDeleted(serverKey, active) {
 			continue
 		}
 

internal/domain/tribe_change.go

@@ -216,7 +216,7 @@ func NewCreateTribeChangeParamsFromPlayers(
 			old = storedPlayers[idx]
 		}
 
-		if (old.ID() > 0 && old.TribeID() == player.TribeID()) || (old.ID() == 0 && player.TribeID() == 0) {
+		if (old.ID() > 0 && old.TribeID() == player.TribeID()) || (old.IsZero() && player.TribeID() == 0) {
 			continue
 		}
 

internal/domain/tribe_message_payloads.go

@@ -1,77 +0,0 @@
-package domain
-
-import (
-	"math"
-	"net/url"
-)
-
-type TribesSyncedEventPayload struct {
-	serverKey   string
-	serverURL   *url.URL
-	versionCode string
-	numTribes   int
-}
-
-const tribesSyncedEventPayloadModelName = "TribesSyncedEventPayload"
-
-func NewTribesSyncedEventPayload(
-	serverKey string,
-	serverURL *url.URL,
-	versionCode string,
-	numTribes int,
-) (TribesSyncedEventPayload, error) {
-	if serverKey == "" {
-		return TribesSyncedEventPayload{}, ValidationError{
-			Model: tribesSyncedEventPayloadModelName,
-			Field: "serverKey",
-			Err:   ErrRequired,
-		}
-	}
-
-	if serverURL == nil {
-		return TribesSyncedEventPayload{}, ValidationError{
-			Model: tribesSyncedEventPayloadModelName,
-			Field: "serverURL",
-			Err:   ErrNil,
-		}
-	}
-
-	if versionCode == "" {
-		return TribesSyncedEventPayload{}, ValidationError{
-			Model: tribesSyncedEventPayloadModelName,
-			Field: "versionCode",
-			Err:   ErrRequired,
-		}
-	}
-
-	if err := validateIntInRange(numTribes, 0, math.MaxInt); err != nil {
-		return TribesSyncedEventPayload{}, ValidationError{
-			Model: tribesSyncedEventPayloadModelName,
-			Field: "numTribes",
-			Err:   err,
-		}
-	}
-
-	return TribesSyncedEventPayload{
-		serverKey:   serverKey,
-		serverURL:   serverURL,
-		versionCode: versionCode,
-		numTribes:   numTribes,
-	}, nil
-}
-
-func (p TribesSyncedEventPayload) ServerKey() string {
-	return p.serverKey
-}
-
-func (p TribesSyncedEventPayload) ServerURL() *url.URL {
-	return p.serverURL
-}
-
-func (p TribesSyncedEventPayload) VersionCode() string {
-	return p.versionCode
-}
-
-func (p TribesSyncedEventPayload) NumTribes() int {
-	return p.numTribes
-}

internal/domain/tribe_message_payloads_test.go

@@ -1,31 +0,0 @@
-package domain_test
-
-import (
-	"math"
-	"testing"
-
-	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
-	"gitea.dwysokinski.me/twhelp/corev3/internal/domain/domaintest"
-	"github.com/brianvoe/gofakeit/v7"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestNewTribesSyncedEventPayload(t *testing.T) {
-	t.Parallel()
-
-	server := domaintest.NewServer(t)
-	numTribes := gofakeit.IntRange(0, math.MaxInt)
-
-	payload, err := domain.NewTribesSyncedEventPayload(
-		server.Key(),
-		server.URL(),
-		server.VersionCode(),
-		numTribes,
-	)
-	require.NoError(t, err)
-	assert.Equal(t, server.Key(), payload.ServerKey())
-	assert.Equal(t, server.URL(), payload.ServerURL())
-	assert.Equal(t, server.VersionCode(), payload.VersionCode())
-	assert.Equal(t, numTribes, payload.NumTribes())
-}

internal/domain/village.go

@@ -187,25 +187,31 @@ func (v Village) IsZero() bool {
 	return v == Village{}
 }
 
+// active villages must be sorted in ascending order by ID
+func (v Village) canBeDeleted(serverKey string, active BaseVillages) bool {
+	if v.serverKey != serverKey {
+		return false
+	}
+
+	_, found := slices.BinarySearchFunc(active, v, func(a BaseVillage, b Village) int {
+		return cmp.Compare(a.ID(), b.id)
+	})
+
+	return !found
+}
+
 type Villages []Village
 
 // Delete finds all villages with the given serverKey that are not in the given slice with active villages
 // and returns their ids. Both slices must be sorted in ascending order by ID
 // + if vs contains villages from different servers. they must be sorted in ascending order by server key.
 func (vs Villages) Delete(serverKey string, active BaseVillages) []int {
-	// villages are deleted now and then, there is no point in prereallocating these slices
+	// villages are deleted now and then, there is no point in prereallocating this slice
 	//nolint:prealloc
 	var toDelete []int
 
 	for _, v := range vs {
-		if v.ServerKey() != serverKey {
-			continue
-		}
-
-		_, found := slices.BinarySearchFunc(active, v, func(a BaseVillage, b Village) int {
-			return cmp.Compare(a.ID(), b.ID())
-		})
-		if found {
+		if !v.canBeDeleted(serverKey, active) {
 			continue
 		}
 

internal/domain/village_message_payloads.go

@@ -1,155 +0,0 @@
-package domain
-
-import (
-	"math"
-	"net/url"
-)
-
-type VillagesSyncedEventPayload struct {
-	serverKey            string
-	serverURL            *url.URL
-	versionCode          string
-	numVillages          int
-	numPlayerVillages    int
-	numBarbarianVillages int
-	numBonusVillages     int
-}
-
-const villagesSyncedEventPayloadModelName = "VillagesSyncedEventPayload"
-
-func NewVillagesSyncedEventPayload(
-	serverKey string,
-	serverURL *url.URL,
-	versionCode string,
-	numVillages int,
-	numPlayerVillages int,
-	numBarbarianVillages int,
-	numBonusVillages int,
-) (VillagesSyncedEventPayload, error) {
-	if serverKey == "" {
-		return VillagesSyncedEventPayload{}, ValidationError{
-			Model: villagesSyncedEventPayloadModelName,
-			Field: "serverKey",
-			Err:   ErrRequired,
-		}
-	}
-
-	if serverURL == nil {
-		return VillagesSyncedEventPayload{}, ValidationError{
-			Model: villagesSyncedEventPayloadModelName,
-			Field: "serverURL",
-			Err:   ErrNil,
-		}
-	}
-
-	if versionCode == "" {
-		return VillagesSyncedEventPayload{}, ValidationError{
-			Model: villagesSyncedEventPayloadModelName,
-			Field: "versionCode",
-			Err:   ErrRequired,
-		}
-	}
-
-	if err := validateIntInRange(numVillages, 0, math.MaxInt); err != nil {
-		return VillagesSyncedEventPayload{}, ValidationError{
-			Model: villagesSyncedEventPayloadModelName,
-			Field: "numVillages",
-			Err:   err,
-		}
-	}
-
-	if err := validateIntInRange(numPlayerVillages, 0, math.MaxInt); err != nil {
-		return VillagesSyncedEventPayload{}, ValidationError{
-			Model: villagesSyncedEventPayloadModelName,
-			Field: "numPlayerVillages",
-			Err:   err,
-		}
-	}
-
-	if err := validateIntInRange(numBarbarianVillages, 0, math.MaxInt); err != nil {
-		return VillagesSyncedEventPayload{}, ValidationError{
-			Model: villagesSyncedEventPayloadModelName,
-			Field: "numBarbarianVillages",
-			Err:   err,
-		}
-	}
-
-	if err := validateIntInRange(numBonusVillages, 0, math.MaxInt); err != nil {
-		return VillagesSyncedEventPayload{}, ValidationError{
-			Model: villagesSyncedEventPayloadModelName,
-			Field: "numBonusVillages",
-			Err:   err,
-		}
-	}
-
-	return VillagesSyncedEventPayload{
-		serverKey:            serverKey,
-		serverURL:            serverURL,
-		versionCode:          versionCode,
-		numVillages:          numVillages,
-		numPlayerVillages:    numPlayerVillages,
-		numBarbarianVillages: numBarbarianVillages,
-		numBonusVillages:     numBonusVillages,
-	}, nil
-}
-
-func NewVillagesSyncedEventPayloadFromVillages(
-	serverKey string,
-	serverURL *url.URL,
-	versionCode string,
-	villages BaseVillages,
-) (VillagesSyncedEventPayload, error) {
-	numPlayerVillages := 0
-	numBarbarianVillages := 0
-	numBonusVillages := 0
-
-	for _, v := range villages {
-		if v.PlayerID() > 0 {
-			numPlayerVillages++
-		} else {
-			numBarbarianVillages++
-		}
-
-		if v.Bonus() > 0 {
-			numBonusVillages++
-		}
-	}
-
-	return NewVillagesSyncedEventPayload(
-		serverKey,
-		serverURL,
-		versionCode,
-		len(villages),
-		numPlayerVillages,
-		numBarbarianVillages,
-		numBonusVillages,
-	)
-}
-
-func (p VillagesSyncedEventPayload) ServerKey() string {
-	return p.serverKey
-}
-
-func (p VillagesSyncedEventPayload) ServerURL() *url.URL {
-	return p.serverURL
-}
-
-func (p VillagesSyncedEventPayload) VersionCode() string {
-	return p.versionCode
-}
-
-func (p VillagesSyncedEventPayload) NumVillages() int {
-	return p.numVillages
-}
-
-func (p VillagesSyncedEventPayload) NumPlayerVillages() int {
-	return p.numPlayerVillages
-}
-
-func (p VillagesSyncedEventPayload) NumBarbarianVillages() int {
-	return p.numBarbarianVillages
-}
-
-func (p VillagesSyncedEventPayload) NumBonusVillages() int {
-	return p.numBonusVillages
-}

internal/domain/village_message_payloads_test.go

@@ -1,83 +0,0 @@
-package domain_test
-
-import (
-	"math"
-	"testing"
-
-	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
-	"gitea.dwysokinski.me/twhelp/corev3/internal/domain/domaintest"
-	"github.com/brianvoe/gofakeit/v7"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestNewVillagesSyncedEventPayload(t *testing.T) {
-	t.Parallel()
-
-	server := domaintest.NewServer(t)
-	numVillages := gofakeit.IntRange(0, math.MaxInt)
-	numPlayerVillages := gofakeit.IntRange(0, math.MaxInt)
-	numBarbarianVillages := gofakeit.IntRange(0, math.MaxInt)
-	numBonusVillages := gofakeit.IntRange(0, math.MaxInt)
-
-	payload, err := domain.NewVillagesSyncedEventPayload(
-		server.Key(),
-		server.URL(),
-		server.VersionCode(),
-		numVillages,
-		numPlayerVillages,
-		numBarbarianVillages,
-		numBonusVillages,
-	)
-	require.NoError(t, err)
-	assert.Equal(t, server.Key(), payload.ServerKey())
-	assert.Equal(t, server.URL(), payload.ServerURL())
-	assert.Equal(t, server.VersionCode(), payload.VersionCode())
-	assert.Equal(t, numVillages, payload.NumVillages())
-	assert.Equal(t, numPlayerVillages, payload.NumPlayerVillages())
-	assert.Equal(t, numBarbarianVillages, payload.NumBarbarianVillages())
-	assert.Equal(t, numBonusVillages, payload.NumBonusVillages())
-}
-
-func TestNewVillagesSyncedEventPayloadFromVillages(t *testing.T) {
-	t.Parallel()
-
-	server := domaintest.NewServer(t)
-	villages := domain.BaseVillages{
-		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
-			cfg.PlayerID = 0
-			cfg.Bonus = 1
-		}),
-		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
-			cfg.PlayerID = 0
-			cfg.Bonus = 0
-		}),
-		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
-			cfg.PlayerID = 0
-			cfg.Bonus = 0
-		}),
-		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
-			cfg.PlayerID = domaintest.RandID()
-			cfg.Bonus = 0
-		}),
-		domaintest.NewBaseVillage(t, func(cfg *domaintest.BaseVillageConfig) {
-			cfg.PlayerID = domaintest.RandID()
-			cfg.Bonus = 1
-		}),
-	}
-
-	payload, err := domain.NewVillagesSyncedEventPayloadFromVillages(
-		server.Key(),
-		server.URL(),
-		server.VersionCode(),
-		villages,
-	)
-	require.NoError(t, err)
-	assert.Equal(t, server.Key(), payload.ServerKey())
-	assert.Equal(t, server.URL(), payload.ServerURL())
-	assert.Equal(t, server.VersionCode(), payload.VersionCode())
-	assert.Equal(t, len(villages), payload.NumVillages()) //nolint:testifylint
-	assert.Equal(t, 2, payload.NumPlayerVillages())
-	assert.Equal(t, 2, payload.NumBonusVillages())
-	assert.Equal(t, 3, payload.NumBarbarianVillages())
-}

internal/watermill/watermillmsg/data_cleanup.go

@@ -0,0 +1,20 @@
+package watermillmsg
+
+import "time"
+
+type CleanUpDataCmdPayloadServer struct {
+	Key                      string    `json:"key"`
+	VersionCode              string    `json:"versionCode"`
+	Open                     bool      `json:"open"`
+	Special                  bool      `json:"special"`
+	PlayerDataSyncedAt       time.Time `json:"playerDataSyncedAt"`
+	PlayerSnapshotsCreatedAt time.Time `json:"playerSnapshotsCreatedAt"`
+	TribeDataSyncedAt        time.Time `json:"tribeDataSyncedAt"`
+	TribeSnapshotsCreatedAt  time.Time `json:"tribeSnapshotsCreatedAt"`
+	VillageDataSyncedAt      time.Time `json:"villageDataSyncedAt"`
+	EnnoblementDataSyncedAt  time.Time `json:"ennoblementDataSyncedAt"`
+}
+
+type CleanUpDataCmdPayload struct {
+	Server CleanUpDataCmdPayloadServer `json:"server"`
+}

k8s/base/jobs.yml

@@ -42,7 +42,7 @@ spec:
                   memory: 64Mi
                 limits:
                   cpu: 100m
-                  memory: 128Mi
+                  memory: 64Mi
 ---
 apiVersion: batch/v1
 kind: CronJob
@@ -88,7 +88,7 @@ spec:
                   memory: 64Mi
                 limits:
                   cpu: 100m
-                  memory: 128Mi
+                  memory: 64Mi
 ---
 apiVersion: batch/v1
 kind: CronJob
@@ -134,7 +134,53 @@ spec:
                   memory: 64Mi
                 limits:
                   cpu: 100m
-                  memory: 128Mi
+                  memory: 64Mi
+---
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+  name: twhelp-job-cleanup
+spec:
+  schedule: 35 0 * * *
+  successfulJobsHistoryLimit: 3
+  failedJobsHistoryLimit: 3
+  concurrencyPolicy: Forbid
+  jobTemplate:
+    spec:
+      parallelism: 1
+      template:
+        spec:
+          restartPolicy: Never
+          containers:
+            - name: twhelp-job-cleanup
+              image: twhelp
+              args: [job, cleanup]
+              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: DB_MAX_OPEN_CONNS
+                  value: "1"
+                - name: DB_MAX_IDLE_CONNS
+                  value: "1"
+                - name: RABBITMQ_CONNECTION_STRING
+                  valueFrom:
+                    secretKeyRef:
+                      name: twhelp-secret
+                      key: rabbitmq-connection-string
+              resources:
+                requests:
+                  cpu: 50m
+                  memory: 64Mi
+                limits:
+                  cpu: 100m
+                  memory: 64Mi
 ---
 apiVersion: batch/v1
 kind: Job