diff --git a/internal/port/consumer_data_cleanup_test.go b/internal/port/consumer_data_cleanup_test.go
new file mode 100644
index 0000000..8bfb9e0
--- /dev/null
+++ b/internal/port/consumer_data_cleanup_test.go
@@ -0,0 +1,230 @@
+package port_test
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"path"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"gitea.dwysokinski.me/twhelp/corev3/internal/adapter"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/app"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/bun/buntest"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/domain"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/port"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/watermill/watermillamqptest"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/watermill/watermillmsg"
+	"gitea.dwysokinski.me/twhelp/corev3/internal/watermill/watermilltest"
+	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill-amqp/v2/pkg/amqp"
+	"github.com/brianvoe/gofakeit/v7"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDataCleanup(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skip("skipping long-running test")
+	}
+
+	ctxTimeout, cancel := context.WithTimeout(context.Background(), 150*time.Second)
+	defer cancel()
+
+	ctx, stop := signal.NotifyContext(ctxTimeout, os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	filesys := os.DirFS("./testdata/cleanup")
+
+	// bun
+	db := postgres.NewDB(t)
+	buntest.NewFixture(db).Load(t, ctx, filesys, "fixture.yml")
+
+	// watermill
+	marshaler := watermillmsg.JSONMarshaler{NewUUID: watermill.NewUUID}
+	generateExchangeAndRoutingKeyName := func(topic string) string {
+		return topic + "_data_cleanup"
+	}
+	rmqConn := rabbitMQ.NewConnection(t)
+	nopLogger := watermill.NopLogger{}
+	tribePub, tribeSub := watermillamqptest.NewPubSub(
+		t,
+		rmqConn,
+		amqp.GenerateQueueNameTopicNameWithSuffix("_data_cleanup_tribe"),
+		generateExchangeAndRoutingKeyName,
+	)
+	_, playerSub := watermillamqptest.NewPubSub(
+		t,
+		rmqConn,
+		amqp.GenerateQueueNameTopicNameWithSuffix("_data_cleanup_player"),
+		generateExchangeAndRoutingKeyName,
+	)
+	_, ennoblementSub := watermillamqptest.NewPubSub(
+		t,
+		rmqConn,
+		amqp.GenerateQueueNameTopicNameWithSuffix("_data_cleanup_ennoblement"),
+		generateExchangeAndRoutingKeyName,
+	)
+
+	// events/commands
+	cmdCleanUpTopic := gofakeit.UUID()
+
+	// adapters
+	serverRepo := adapter.NewServerBunRepository(db)
+	ennoblementRepo := adapter.NewEnnoblementBunRepository(db)
+	tribeSnapshotRepo := adapter.NewTribeSnapshotBunRepository(db)
+	playerSnapshotRepo := adapter.NewPlayerSnapshotBunRepository(db)
+	dataCleanupPublisher := adapter.NewDataCleanupPublisher(tribePub, marshaler, cmdCleanUpTopic)
+
+	// services
+	serverSvc := app.NewServerService(serverRepo, nil, nil)
+	ennoblementSvc := app.NewEnnoblementService(ennoblementRepo, nil, nil)
+	tribeSnapshotSvc := app.NewTribeSnapshotService(tribeSnapshotRepo, nil, nil)
+	playerSnapshotSvc := app.NewPlayerSnapshotService(playerSnapshotRepo, nil, nil)
+	dataCleanupSvc := app.NewDataCleanupService(serverSvc, dataCleanupPublisher)
+
+	watermilltest.RunRouterWithContext(
+		t,
+		ctx,
+		port.NewTribeWatermillConsumer(
+			nil,
+			tribeSnapshotSvc,
+			tribeSub,
+			nopLogger,
+			marshaler,
+			"",
+			"",
+			"",
+			cmdCleanUpTopic,
+		),
+		port.NewPlayerWatermillConsumer(
+			nil,
+			playerSnapshotSvc,
+			playerSub,
+			nopLogger,
+			marshaler,
+			"",
+			"",
+			cmdCleanUpTopic,
+		),
+		port.NewEnnoblementWatermillConsumer(ennoblementSvc, ennoblementSub, nopLogger, marshaler, "", cmdCleanUpTopic),
+	)
+
+	require.NoError(t, dataCleanupSvc.CleanUp(ctx))
+
+	var wg sync.WaitGroup
+
+	var expectedEnnoblements []int
+	readJSONFile(t, filesys, path.Join("expected", "ennoblements.json"), &expectedEnnoblements)
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		assert.EventuallyWithTf(t, func(collect *assert.CollectT) {
+			var ids []int
+
+			params := domain.NewListEnnoblementsParams()
+			require.NoError(collect, params.SetSort([]domain.EnnoblementSort{
+				domain.EnnoblementSortServerKeyASC,
+				domain.EnnoblementSortCreatedAtASC,
+				domain.EnnoblementSortIDASC,
+			}))
+
+			for {
+				res, err := ennoblementRepo.List(ctx, params)
+				require.NoError(collect, err)
+
+				for _, e := range res.Ennoblements() {
+					ids = append(ids, e.ID())
+				}
+
+				if res.Next().IsZero() {
+					break
+				}
+
+				require.NoError(collect, params.SetCursor(res.Next()))
+			}
+
+			assert.Equal(collect, expectedEnnoblements, ids)
+		}, 30*time.Second, 500*time.Millisecond, "ennoblements")
+	}()
+
+	var expectedTribeSnapshots []int
+	readJSONFile(t, filesys, path.Join("expected", "tribe-snapshots.json"), &expectedTribeSnapshots)
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		assert.EventuallyWithTf(t, func(collect *assert.CollectT) {
+			var ids []int
+
+			params := domain.NewListTribeSnapshotsParams()
+			require.NoError(collect, params.SetSort([]domain.TribeSnapshotSort{
+				domain.TribeSnapshotSortServerKeyASC,
+				domain.TribeSnapshotSortDateASC,
+				domain.TribeSnapshotSortIDASC,
+			}))
+
+			for {
+				res, err := tribeSnapshotRepo.List(ctx, params)
+				require.NoError(collect, err)
+
+				for _, ts := range res.TribeSnapshots() {
+					ids = append(ids, ts.ID())
+				}
+
+				if res.Next().IsZero() {
+					break
+				}
+
+				require.NoError(collect, params.SetCursor(res.Next()))
+			}
+
+			assert.Equal(collect, expectedTribeSnapshots, ids)
+		}, 30*time.Second, 500*time.Millisecond, "tribe snapshots")
+	}()
+
+	var expectedPlayerSnapshots []int
+	readJSONFile(t, filesys, path.Join("expected", "player-snapshots.json"), &expectedPlayerSnapshots)
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		assert.EventuallyWithTf(t, func(collect *assert.CollectT) {
+			var ids []int
+
+			params := domain.NewListPlayerSnapshotsParams()
+			require.NoError(collect, params.SetSort([]domain.PlayerSnapshotSort{
+				domain.PlayerSnapshotSortServerKeyASC,
+				domain.PlayerSnapshotSortDateASC,
+				domain.PlayerSnapshotSortIDASC,
+			}))
+
+			for {
+				res, err := playerSnapshotRepo.List(ctx, params)
+				require.NoError(collect, err)
+
+				for _, ps := range res.PlayerSnapshots() {
+					ids = append(ids, ps.ID())
+				}
+
+				if res.Next().IsZero() {
+					break
+				}
+
+				require.NoError(collect, params.SetCursor(res.Next()))
+			}
+
+			assert.Equal(collect, expectedPlayerSnapshots, ids)
+		}, 30*time.Second, 500*time.Millisecond, "player snapshots")
+	}()
+
+	wg.Wait()
+}
diff --git a/internal/port/testdata/cleanup/expected/ennoblements.json b/internal/port/testdata/cleanup/expected/ennoblements.json
new file mode 100644
index 0000000..d542682
--- /dev/null
+++ b/internal/port/testdata/cleanup/expected/ennoblements.json
@@ -0,0 +1 @@
+[220041,2200407,2200408,22004089]
\ No newline at end of file
diff --git a/internal/port/testdata/cleanup/expected/player-snapshots.json b/internal/port/testdata/cleanup/expected/player-snapshots.json
new file mode 100644
index 0000000..ac992f4
--- /dev/null
+++ b/internal/port/testdata/cleanup/expected/player-snapshots.json
@@ -0,0 +1 @@
+[100001,1000000,1000001,10000000]
\ No newline at end of file
diff --git a/internal/port/testdata/cleanup/expected/tribe-snapshots.json b/internal/port/testdata/cleanup/expected/tribe-snapshots.json
new file mode 100644
index 0000000..c85837e
--- /dev/null
+++ b/internal/port/testdata/cleanup/expected/tribe-snapshots.json
@@ -0,0 +1 @@
+[10001,100000,100001,1000000]
\ No newline at end of file
diff --git a/internal/port/testdata/cleanup/fixture.yml b/internal/port/testdata/cleanup/fixture.yml
new file mode 100644
index 0000000..8922b3a
--- /dev/null
+++ b/internal/port/testdata/cleanup/fixture.yml
@@ -0,0 +1,405 @@
+- model: Server
+  rows:
+    - key: pl181 # data from this server should be cleaned up
+      url: https://pl181.plemiona.pl/
+      open: false
+      special: false
+      num_players: 144
+      num_tribes: 26
+      num_villages: 59581
+      num_player_villages: 58558
+      num_barbarian_villages: 1023
+      num_bonus_villages: 3715
+      created_at: 2022-09-30T08:00:04.504556Z
+      version_code: pl
+    - key: pl182 # data from this server shouldn't be cleaned up because open=true
+      url: https://pl182.plemiona.pl/
+      open: true
+      special: false
+      num_players: 144
+      num_tribes: 26
+      num_villages: 39702
+      num_player_villages: 39697
+      num_barbarian_villages: 5
+      num_bonus_villages: 2086
+      created_at: 2022-09-30T08:00:04.504556Z
+      version_code: pl
+    - key: pl185 # data from this server should not be cleaned because of a recent data update
+      url: https://pl185.plemiona.pl/
+      open: false
+      special: false
+      num_players: 458
+      num_tribes: 88
+      num_villages: 84069
+      num_player_villages: 83057
+      num_barbarian_villages: 1012
+      num_bonus_villages: 13379
+      created_at:
+      version_code: pl
+      player_data_updated_at: "{{ now }}"
+      player_snapshots_created_at: "{{ now }}"
+      tribe_data_updated_at: "{{ now }}"
+      tribe_snapshots_created_at: "{{ now }}"
+      village_data_updated_at: "{{ now }}"
+      ennoblement_data_updated_at: "{{ now }}"
+- model: Ennoblement
+  rows:
+    # pl181
+    - id: 220040
+      server_key: pl181
+      village_id: 1112
+      new_owner_id: 848928624
+      new_tribe_id: 0
+      old_owner_id: 0
+      old_tribe_id: 0
+      points: 9505
+      created_at: 2023-03-25T05:43:43Z
+    - id: 220041
+      server_key: pl181
+      village_id: 1112
+      new_owner_id: 699396429
+      new_tribe_id: 1312
+      old_owner_id: 0
+      old_tribe_id: 0
+      points: 9392
+      created_at: "{{ now }}"
+    # pl182
+    - id: 2200407
+      server_key: pl182
+      village_id: 1112
+      new_owner_id: 1179
+      new_tribe_id: 181
+      old_owner_id: 0
+      old_tribe_id: 0
+      points: 123
+      created_at: 2022-09-20T06:25:55Z
+    - id: 2200408
+      server_key: pl182
+      village_id: 1112
+      new_owner_id: 698496770
+      new_tribe_id: 145
+      old_owner_id: 698593426
+      old_tribe_id: 145
+      points: 4754
+      created_at: "{{ now }}"
+    # pl185
+    - id: 22004089
+      server_key: pl185
+      village_id: 1112
+      new_owner_id: 698321605
+      new_tribe_id: 489
+      old_owner_id: 699779452
+      old_tribe_id: 489
+      points: 3999
+      created_at: 2023-03-22T07:43:02Z
+- model: Tribe
+  rows:
+    # pl181
+    - rank_att: 1
+      score_att: 730488339
+      rank_def: 1
+      score_def: 303300128
+      rank_sup: 0
+      score_sup: 0
+      rank_total: 1
+      score_total: 1033788467
+      id: 1
+      server_key: pl181
+      name: Obdachlose Otter
+      tag: ObOtt
+      num_members: 20
+      num_villages: 8419
+      points: 86151161
+      all_points: 86151161
+      rank: 1
+      dominance: 58.367997781475324
+      created_at: 2021-03-07T11:00:51.000Z
+      profile_url: https://de188.die-staemme.de/game.php?screen=info_ally&id=772
+    # pl182
+    - rank_att: 1
+      score_att: 730488339
+      rank_def: 1
+      score_def: 303300128
+      rank_sup: 0
+      score_sup: 0
+      rank_total: 1
+      score_total: 1033788467
+      id: 1
+      server_key: pl182
+      name: Obdachlose Otter
+      tag: ObOtt
+      num_members: 20
+      num_villages: 8419
+      points: 86151161
+      all_points: 86151161
+      rank: 1
+      dominance: 58.367997781475324
+      created_at: 2021-03-07T11:00:51.000Z
+      profile_url: https://de188.die-staemme.de/game.php?screen=info_ally&id=772
+    # pl185
+    - rank_att: 1
+      score_att: 730488339
+      rank_def: 1
+      score_def: 303300128
+      rank_sup: 0
+      score_sup: 0
+      rank_total: 1
+      score_total: 1033788467
+      id: 1
+      server_key: pl185
+      name: Obdachlose Otter
+      tag: ObOtt
+      num_members: 20
+      num_villages: 8419
+      points: 86151161
+      all_points: 86151161
+      rank: 1
+      dominance: 58.367997781475324
+      created_at: 2021-03-07T11:00:51.000Z
+      profile_url: https://de188.die-staemme.de/game.php?screen=info_ally&id=772
+- model: Player
+  rows:
+    # pl181
+    - rank_att: 10
+      score_att: 27638462
+      rank_def: 35
+      score_def: 5555868
+      rank_sup: 3
+      score_sup: 9815277
+      rank_total: 16
+      score_total: 43009607
+      id: 698152377
+      server_key: pl181
+      name: 1scoo
+      num_villages: 878
+      points: 9199775
+      rank: 1
+      tribe_id: 772
+      created_at: 2021-06-25T11:00:53Z
+      profile_url: https://www.internalbenchmark.com/e-markets/proactive/user-centric
+    # pl182
+    - rank_att: 10
+      score_att: 27638462
+      rank_def: 35
+      score_def: 5555868
+      rank_sup: 3
+      score_sup: 9815277
+      rank_total: 16
+      score_total: 43009607
+      id: 2483601
+      server_key: pl182
+      name: 1scoo
+      num_villages: 878
+      points: 9199775
+      rank: 1
+      tribe_id: 772
+      created_at: 2021-06-25T11:00:53Z
+      profile_url: https://www.internalbenchmark.com/e-markets/proactive/user-centric
+    # pl185
+    - rank_att: 10
+      score_att: 27638462
+      rank_def: 35
+      score_def: 5555868
+      rank_sup: 3
+      score_sup: 9815277
+      rank_total: 16
+      score_total: 43009607
+      id: 7523589
+      server_key: pl185
+      name: 1scoo
+      num_villages: 878
+      points: 9199775
+      rank: 1
+      tribe_id: 772
+      created_at: 2021-06-25T11:00:53Z
+      profile_url: https://www.internalbenchmark.com/e-markets/proactive/user-centric
+- model: PlayerSnapshot
+  rows:
+    # pl181
+    - rank_att: 5
+      score_att: 38213157
+      rank_def: 2
+      score_def: 51137731
+      rank_sup: 104
+      score_sup: 2282683
+      rank_total: 1
+      score_total: 91633571
+      id: 100000
+      player_id: 698152377
+      server_key: pl181
+      num_villages: 665
+      points: 7298055
+      rank: 1
+      tribe_id: 47
+      date: 2023-09-01T00:00:00.000Z
+      created_at: 2022-09-17T06:03:40Z
+    - rank_att: 5
+      score_att: 38213157
+      rank_def: 2
+      score_def: 51137731
+      rank_sup: 104
+      score_sup: 2282683
+      rank_total: 1
+      score_total: 91633571
+      id: 100001
+      player_id: 698152377
+      server_key: pl181
+      num_villages: 665
+      points: 7298055
+      rank: 1
+      tribe_id: 47
+      date: "{{ now }}"
+      created_at: "{{ now }}"
+    # pl182
+    - rank_att: 5
+      score_att: 38213157
+      rank_def: 2
+      score_def: 51137731
+      rank_sup: 104
+      score_sup: 2282683
+      rank_total: 1
+      score_total: 91633571
+      id: 1000000
+      player_id: 2483601
+      server_key: pl182
+      num_villages: 665
+      points: 7298055
+      rank: 1
+      tribe_id: 65
+      date: 2023-09-01T00:00:00.000Z
+      created_at: 2022-09-17T06:03:40Z
+    - rank_att: 5
+      score_att: 38213157
+      rank_def: 2
+      score_def: 51137731
+      rank_sup: 104
+      score_sup: 2282683
+      rank_total: 1
+      score_total: 91633571
+      id: 1000001
+      player_id: 2483601
+      server_key: pl182
+      num_villages: 665
+      points: 7298055
+      rank: 1
+      tribe_id: 65
+      date: "{{ now }}"
+      created_at: "{{ now }}"
+    # pl185
+    - rank_att: 5
+      score_att: 38213157
+      rank_def: 2
+      score_def: 51137731
+      rank_sup: 104
+      score_sup: 2282683
+      rank_total: 1
+      score_total: 91633571
+      id: 10000000
+      player_id: 7523589
+      server_key: pl185
+      num_villages: 665
+      points: 7298055
+      rank: 1
+      tribe_id: 1768
+      date: "{{ now }}"
+      created_at: "{{ now }}"
+- model: TribeSnapshot
+  rows:
+    # pl181
+    - rank_att: 1
+      score_att: 669292167
+      rank_def: 5
+      score_def: 199124128
+      rank_sup: 0
+      score_sup: 0
+      rank_total: 2
+      score_total: 868416295
+      id: 10000
+      tribe_id: 1
+      server_key: pl181
+      num_members: 66
+      num_villages: 14480
+      points: 114277979
+      all_points: 143350058
+      rank: 1
+      dominance: 29.76912481240106
+      date: 2023-09-01T00:00:00.000Z
+      created_at: 2022-09-17T06:03:40Z
+    - rank_att: 1
+      score_att: 669292167
+      rank_def: 5
+      score_def: 199124128
+      rank_sup: 0
+      score_sup: 0
+      rank_total: 2
+      score_total: 868416295
+      id: 10001
+      tribe_id: 1
+      server_key: pl181
+      num_members: 66
+      num_villages: 14480
+      points: 114277979
+      all_points: 143350058
+      rank: 1
+      dominance: 29.76912481240106
+      date: "{{ now }}"
+      created_at: "{{ now }}"
+    # pl182
+    - rank_att: 1
+      score_att: 669292167
+      rank_def: 5
+      score_def: 199124128
+      rank_sup: 0
+      score_sup: 0
+      rank_total: 2
+      score_total: 868416295
+      id: 100000
+      tribe_id: 1
+      server_key: pl182
+      num_members: 66
+      num_villages: 14480
+      points: 114277979
+      all_points: 143350058
+      rank: 1
+      dominance: 29.76912481240106
+      date: 2023-09-01T00:00:00.000Z
+      created_at: 2022-09-17T06:03:40Z
+    - rank_att: 1
+      score_att: 669292167
+      rank_def: 5
+      score_def: 199124128
+      rank_sup: 0
+      score_sup: 0
+      rank_total: 2
+      score_total: 868416295
+      id: 100001
+      tribe_id: 1
+      server_key: pl182
+      num_members: 66
+      num_villages: 14480
+      points: 114277979
+      all_points: 143350058
+      rank: 1
+      dominance: 29.76912481240106
+      date: "{{ now }}"
+      created_at: "{{ now }}"
+      # pl185
+    - rank_att: 1
+      score_att: 669292167
+      rank_def: 5
+      score_def: 199124128
+      rank_sup: 0
+      score_sup: 0
+      rank_total: 2
+      score_total: 868416295
+      id: 1000000
+      tribe_id: 1
+      server_key: pl185
+      num_members: 66
+      num_villages: 14480
+      points: 114277979
+      all_points: 143350058
+      rank: 1
+      dominance: 29.76912481240106
+      date: 2023-09-01T00:00:00.000Z
+      created_at: 2022-09-17T06:03:40Z