feat: worker_ban_migration + players.banned event infra (ban archival)

_infra/_kafka/src/config.py

@@ -9,6 +9,7 @@ class KafkaSeederConfig(BaseSettings):
     SEED_PLAYERS: int = 100
     SEED_SCRAPES_PER_PLAYER: int = 30
     SEED_REPORTS: int = 100_000
+    SEED_BANNED: int = 0
 
     RESET_TOPICS: bool = False
     RANDOM_SEED: int = 42

_infra/_kafka/src/core.py

@@ -7,6 +7,7 @@
 sys.path.insert(0, "/app/_shared")
 
 from config import KafkaSeederConfig, load_names
+from seeders.players_banned import create_players_banned
 from seeders.players_scraped import create_players_scraped
 from seeders.players_to_scrape import create_players_to_scrape
 from seeders.reports_to_insert import create_reports_to_insert
@@ -89,6 +90,27 @@ def seed_reports_to_insert(
     print("Done seeding reports.to_insert")
 
 
+def seed_players_banned(
+    producer: KafkaProducer,
+    names: list[str],
+    count: int,
+):
+    print(f"Seeding {count} banned players to players.banned...")
+
+    players = []
+    for to_scrape in create_players_to_scrape(names=names, count=count):
+        players.append(to_scrape.player_data)
+
+    for banned in create_players_banned(players, count):
+        producer.send(
+            topic="players.banned",
+            value=banned.model_dump(mode="json"),
+        )
+        print(f"  -> {banned.name} (id={banned.player_id})")
+    producer.flush()
+    print("Done seeding players.banned")
+
+
 def main():
     random.seed(config.RANDOM_SEED)
 
@@ -122,6 +144,13 @@ def main():
             reports_per_player=config.SEED_REPORTS,
         )
 
+    if config.SEED_BANNED > 0:
+        seed_players_banned(
+            producer=producer,
+            names=names,
+            count=config.SEED_BANNED,
+        )
+
 
 if __name__ == "__main__":
     main()

_infra/_kafka/src/seeders/players_banned.py

@@ -0,0 +1,28 @@
+import random
+from typing import Generator
+
+from pydantic import BaseModel
+
+from seeders.players_to_scrape import PlayerStruct
+
+
+class MetaData(BaseModel):
+    version: int
+    source: str
+
+
+class PlayerBannedStruct(BaseModel):
+    metadata: MetaData
+    player_id: int
+    name: str
+
+
+def create_players_banned(
+    players: list[PlayerStruct], count: int
+) -> Generator[PlayerBannedStruct, None, None]:
+    for player in random.sample(players, min(count, len(players))):
+        yield PlayerBannedStruct(
+            metadata=MetaData(version=1, source="init"),
+            player_id=player.id,
+            name=player.name,
+        )

_infra/_kafka/src/topics.py

@@ -21,6 +21,11 @@
         num_partitions=4,
         replication_factor=1,
     ),
+    NewTopic(
+        name="players.banned",
+        num_partitions=4,
+        replication_factor=1,
+    ),
     NewTopic(
         name="reports.to_insert",
         num_partitions=4,

_infra/_mysql/docker-entrypoint-initdb.d/00_init.sql

@@ -6,4 +6,5 @@ CREATE USER `ml-worker`@`%` IDENTIFIED BY 'ml_worker_pw';
 CREATE USER `job-prune-hs`@`%` IDENTIFIED BY 'job_prune_hs_pw';
 CREATE USER `job-hs-migration`@`%` IDENTIFIED BY 'job_hs_migration_pw';
 CREATE USER `api-public`@`%` IDENTIFIED BY 'api_public_pw';
-CREATE USER `job-prune-reports`@`%` IDENTIFIED BY 'job_prune_reports_pw';
+CREATE USER `job-prune-reports`@`%` IDENTIFIED BY 'job_prune_reports_pw';
+CREATE USER `ban-migration`@`%` IDENTIFIED BY 'ban_migration_pw';

_infra/_mysql/docker-entrypoint-initdb.d/01_tables.sql

@@ -250,6 +250,33 @@ CREATE TABLE report (
   )
 );
 
+CREATE TABLE report_archive (
+  /* denormalized sighting */
+  reported_id INT UNSIGNED NOT NULL,
+  /* denormalized report row */
+  world_number SMALLINT UNSIGNED DEFAULT NULL,
+  on_members_world TINYINT DEFAULT NULL,
+  on_pvp_world TINYINT DEFAULT NULL,
+  reported_at TIMESTAMP NOT NULL,
+  /* denormalized location */
+  region_id MEDIUMINT UNSIGNED NOT NULL,
+  x_coord MEDIUMINT UNSIGNED NOT NULL,
+  y_coord MEDIUMINT UNSIGNED NOT NULL,
+  z_coord MEDIUMINT UNSIGNED NOT NULL,
+  /* denormalized gear */
+  equip_head_id SMALLINT UNSIGNED DEFAULT NULL,
+  equip_amulet_id SMALLINT UNSIGNED DEFAULT NULL,
+  equip_torso_id SMALLINT UNSIGNED DEFAULT NULL,
+  equip_legs_id SMALLINT UNSIGNED DEFAULT NULL,
+  equip_boots_id SMALLINT UNSIGNED DEFAULT NULL,
+  equip_cape_id SMALLINT UNSIGNED DEFAULT NULL,
+  equip_hands_id SMALLINT UNSIGNED DEFAULT NULL,
+  equip_weapon_id SMALLINT UNSIGNED DEFAULT NULL,
+  equip_shield_id SMALLINT UNSIGNED DEFAULT NULL,
+  PRIMARY KEY (reported_id, reported_at),
+  KEY idx_region_id (region_id)
+);
+
 CREATE TABLE `Labels` (
   `id` int NOT NULL AUTO_INCREMENT,
   `label` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL,

_infra/_mysql/docker-entrypoint-initdb.d/02_grants.sql

@@ -52,4 +52,12 @@ GRANT SELECT, INSERT ON playerdata.apiUsage TO `api-public`@`%`;
 GRANT CREATE TEMPORARY TABLES ON *.* TO `job-prune-reports`@`%`;
 GRANT SELECT, DELETE ON playerdata.report TO `job-prune-reports`@`%`;
 
+/*ban-migration*/
+GRANT SELECT ON playerdata.report TO `ban-migration`@`%`;
+GRANT SELECT ON playerdata.report_sighting TO `ban-migration`@`%`;
+GRANT SELECT ON playerdata.report_gear TO `ban-migration`@`%`;
+GRANT SELECT ON playerdata.report_location TO `ban-migration`@`%`;
+GRANT SELECT ON playerdata.Players TO `ban-migration`@`%`;
+GRANT INSERT ON playerdata.report_archive TO `ban-migration`@`%`;
+
 FLUSH PRIVILEGES;

_infra/_mysql_data/src/config.py

@@ -10,6 +10,9 @@ class MySQLSeederConfig(BaseSettings):
     SEED_PLAYERS: int = 100
     SEED_REPORTS: int = 0
     SEED_PREDICTIONS: int = 0
+    SEED_BANNED: int = 0
+    SEED_AGED_REPORTS: int = 0
+    SEED_RETENTION_DAYS: int = 90
 
     SKIP_IF_EXISTING: bool = True
     SKIP_THRESHOLD: int = 100

_infra/_mysql_data/src/core.py

@@ -2,6 +2,7 @@
 import json
 import random
 import sys
+from datetime import datetime, timedelta
 
 import sqlalchemy
 from database.database import Session, wait_for_db
@@ -12,6 +13,7 @@
 from seeders.players import create_players
 from seeders.predictions import create_predictions
 from seeders.reports import (
+    Report,
     create_report_gear,
     create_report_locations,
     create_report_sightings,
@@ -146,6 +148,109 @@ async def insert_reports(player_ids: list[int], count: int) -> None:
     print(f"Seeded {count} reports")
 
 
+async def insert_aged_reports(
+    player_ids: list[int], retention_days: int, count: int
+) -> None:
+    """Seed report rows that straddle the prune retention cutoff.
+
+    Roughly half are inserted with reported_at older than the cutoff (so the
+    prune job has rows to delete) and half younger (retained).
+    """
+    sighting_ids: list[int] = []
+    gear_ids: list[int] = []
+    location_ids: list[int] = []
+
+    sighting_sql = sqlalchemy.text("""
+    INSERT IGNORE INTO report_sighting (reporting_id, reported_id, manual_detect)
+    VALUES (:reporting_id, :reported_id, :manual_detect)
+    """)
+    gear_sql = sqlalchemy.text("""
+    INSERT IGNORE INTO report_gear (
+        equip_head_id, equip_amulet_id, equip_torso_id, equip_legs_id,
+        equip_boots_id, equip_cape_id, equip_hands_id, equip_weapon_id, equip_shield_id
+    ) VALUES (
+        :equip_head_id, :equip_amulet_id, :equip_torso_id, :equip_legs_id,
+        :equip_boots_id, :equip_cape_id, :equip_hands_id, :equip_weapon_id, :equip_shield_id
+    )
+    """)
+    location_sql = sqlalchemy.text("""
+    INSERT IGNORE INTO report_location (region_id, x_coord, y_coord, z_coord)
+    VALUES (:region_id, :x_coord, :y_coord, :z_coord)
+    """)
+    report_sql = sqlalchemy.text("""
+    INSERT INTO report (report_sighting_id, report_location_id, report_gear_id,
+        reported_at, on_members_world, on_pvp_world, world_number, region_id)
+    VALUES (:report_sighting_id, :report_location_id, :report_gear_id,
+        :reported_at, :on_members_world, :on_pvp_world, :world_number, :region_id)
+    """)
+
+    for sighting in create_report_sightings(player_ids=player_ids, count=count):
+        async with Session.begin() as session:
+            result = await session.execute(
+                sighting_sql, sighting.model_dump(mode="json")
+            )
+            sighting_ids.append(result.lastrowid)
+
+    for gear in create_report_gear(count=count):
+        async with Session.begin() as session:
+            result = await session.execute(gear_sql, gear.model_dump(mode="json"))
+            gear_ids.append(result.lastrowid)
+
+    for location in create_report_locations(count=count):
+        async with Session.begin() as session:
+            result = await session.execute(
+                location_sql, location.model_dump(mode="json")
+            )
+            location_ids.append(result.lastrowid)
+
+    cutoff = datetime.now() - timedelta(days=retention_days)
+    old_count = count // 2
+
+    for i in range(count):
+        if i < old_count:
+            reported_at = cutoff - timedelta(days=random.randint(1, 30))
+        else:
+            reported_at = cutoff + timedelta(days=random.randint(1, 30))
+
+        sighting_id = (
+            sighting_ids[i] if i < len(sighting_ids) else random.choice(sighting_ids)
+        )
+        gear_id = gear_ids[i] if i < len(gear_ids) else random.choice(gear_ids)
+        location_id = (
+            location_ids[i] if i < len(location_ids) else random.choice(location_ids)
+        )
+
+        report = Report(
+            report_sighting_id=sighting_id,
+            report_location_id=location_id,
+            report_gear_id=gear_id,
+            reported_at=reported_at,
+            on_members_world=random.choice([0, 1]),
+            on_pvp_world=random.choice([0, 0, 0, 1]),
+            world_number=random.randint(300, 500),
+            region_id=random.randint(1, 15000),
+        )
+        async with Session.begin() as session:
+            await session.execute(report_sql, report.model_dump(mode="json"))
+
+    print(
+        f"Seeded {count} aged reports "
+        f"({old_count} older than {retention_days}d retention cutoff)"
+    )
+
+
+async def mark_banned_players(player_ids: list[int], count: int) -> None:
+    sql = sqlalchemy.text("""
+    UPDATE Players SET label_jagex = 2, confirmed_ban = 1
+    WHERE id = :id
+    """)
+    banned_ids = random.sample(player_ids, min(count, len(player_ids)))
+    for pid in banned_ids:
+        async with Session.begin() as session:
+            await session.execute(sql, {"id": pid})
+    print(f"Marked {len(banned_ids)} players as banned (label_jagex=2)")
+
+
 async def main():
     await wait_for_db()
     random.seed(config.RANDOM_SEED)
@@ -171,6 +276,16 @@ async def main():
     if config.SEED_REPORTS > 0:
         await insert_reports(player_ids=player_ids, count=config.SEED_REPORTS)
 
+    if config.SEED_AGED_REPORTS > 0:
+        await insert_aged_reports(
+            player_ids=player_ids,
+            retention_days=config.SEED_RETENTION_DAYS,
+            count=config.SEED_AGED_REPORTS,
+        )
+
+    if config.SEED_BANNED > 0:
+        await mark_banned_players(player_ids=player_ids, count=config.SEED_BANNED)
+
 
 if __name__ == "__main__":
     asyncio.run(main())

bases/bot_detector/worker_ban_migration/__init__.py

@@ -0,0 +1,3 @@
+from bot_detector import logfmt
+
+__all__ = ["logfmt"]

bases/bot_detector/worker_ban_migration/adapter.py

@@ -0,0 +1,12 @@
+import logging
+
+from bot_detector.event_queue.structs import PlayerBannedStruct
+
+logger = logging.getLogger(__name__)
+
+
+def transform_player_banned(record: PlayerBannedStruct) -> int | None:
+    if record.metadata.version != 1:
+        logger.warning(f"Unsupported player_banned version: {record.metadata.version}")
+        return None
+    return record.player_id

bases/bot_detector/worker_ban_migration/main.py

@@ -0,0 +1,69 @@
+import asyncio
+import logging
+
+from bot_detector import database as db
+from bot_detector.database import Settings as DBSettings
+from bot_detector.event_queue.adapters.kafka import (
+    KafkaConfig,
+    KafkaConsumerConfig,
+    KafkaProducerConfig,
+    KafkaSettings,
+)
+from bot_detector.event_queue.structs import PlayerBannedStruct
+from bot_detector.worker.core import WorkerRunner
+from bot_detector.worker_ban_migration.settings import Settings
+from bot_detector.worker_ban_migration.worker import BanMigrationWorker
+
+logger = logging.getLogger(__name__)
+
+SETTINGS = Settings()
+KAFKA_SETTINGS = KafkaSettings()
+
+
+async def main():
+    session_factory, async_engine = db.get_session_factory(SETTINGS=DBSettings())
+    stop_event = asyncio.Event()
+    tasks = []
+    for worker_id in range(SETTINGS.N_WORKERS):
+        worker = BanMigrationWorker(
+            worker_id=worker_id,
+            session_factory=session_factory,
+        )
+        kafka_config = KafkaConfig(
+            topic="players.banned",
+            bootstrap_servers=KAFKA_SETTINGS.bootstrap_servers,
+            producer=True,
+            consumer=True,
+            producer_config=KafkaProducerConfig(partition_key_fn=None),
+            consumer_config=KafkaConsumerConfig(
+                group_id="ban_migration_worker",
+                consume_timeout_ms=SETTINGS.MAX_INTERVAL_MS,
+            ),
+        )
+        runner = WorkerRunner(
+            worker=worker,
+            config=kafka_config,
+            model=PlayerBannedStruct,
+            batch_size=SETTINGS.MAX_BATCH_SIZE,
+            stop_event=stop_event,
+        )
+        task = asyncio.create_task(runner.run())
+        tasks.append(task)
+
+    try:
+        await asyncio.gather(*tasks)
+    finally:
+        stop_event.set()
+        await async_engine.dispose()
+
+
+async def run_async():
+    await main()
+
+
+def run():
+    asyncio.run(run_async())
+
+
+if __name__ == "__main__":
+    run()