feat: job_backfill_banned (one-time historical ban archival)

_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

@@ -5,4 +5,5 @@ CREATE USER `hiscore-worker`@`%` IDENTIFIED BY 'hiscore_worker_pw';
 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 `api-public`@`%` IDENTIFIED BY 'api_public_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

@@ -48,4 +48,12 @@ GRANT SELECT ON playerdata.apiPermissions TO `api-public`@`%`;
 GRANT SELECT ON playerdata.apiUserPerms TO `api-public`@`%`;
 GRANT SELECT, INSERT ON playerdata.apiUsage TO `api-public`@`%`;
 
+/*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;

bases/bot_detector/job_backfill_banned/__init__.py

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

bases/bot_detector/job_backfill_banned/core.py

@@ -0,0 +1,106 @@
+import asyncio
+import logging
+
+import sqlalchemy as sqla
+from bot_detector.database import Settings as DBSettings
+from bot_detector.database import get_session_factory
+from bot_detector.database.report import migrate_banned_player_reports
+from pydantic_settings import BaseSettings
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+logger = logging.getLogger(__name__)
+
+
+class Settings(BaseSettings):
+    BATCH_SIZE: int = 10_000
+
+
+def _select_banned_player_ids() -> sqla.TextClause:
+    """Page through Players flagged as Jagex-banned (label_jagex = 2)."""
+    return sqla.text(
+        """
+        SELECT id FROM Players
+        WHERE label_jagex = 2 AND id > :player_id
+        ORDER BY id ASC
+        LIMIT :limit
+        """
+    )
+
+
+async def backfill(
+    session_factory: async_sessionmaker[AsyncSession],
+    batch_size: int = 10_000,
+) -> int:
+    """Archive location history for every Jagex-banned player.
+
+    Paginates Players where label_jagex = 2 and runs the migration function per
+    player. Each player is migrated in its own transaction; a failure for one
+    player is logged and skipped so the backfill can complete. Idempotent:
+    INSERT IGNORE makes a re-run safe if the one-time execution is interrupted.
+
+    Args:
+        session_factory: SQLAlchemy async session factory.
+        batch_size: Number of player ids fetched per page.
+
+    Returns:
+        Number of banned players processed.
+    """
+    sql = _select_banned_player_ids()
+    player_id = 0
+    processed = 0
+
+    while True:
+        async with session_factory() as session:
+            result = await session.execute(
+                sql,
+                params={"player_id": player_id, "limit": batch_size},
+            )
+            ids = [row for row in result.scalars().all()]
+
+        if not ids:
+            logger.info("backfill: no more banned players, exiting")
+            break
+
+        for reported_id in ids:
+            try:
+                await migrate_banned_player_reports(
+                    session_factory=session_factory,
+                    reported_id=reported_id,
+                )
+                processed += 1
+            except Exception as e:
+                logger.error(
+                    f"backfill: failed to archive reported_id={reported_id}: {e}"
+                )
+
+        player_id = ids[-1]
+        logger.info(
+            f"backfill: processed up to player_id={player_id} ({processed} total)"
+        )
+
+    logger.info(f"backfill: finished, {processed} banned players processed")
+    return processed
+
+
+async def main():
+    session_factory, async_engine = get_session_factory(SETTINGS=DBSettings())
+    settings = Settings()
+    try:
+        await backfill(
+            session_factory=session_factory,
+            batch_size=settings.BATCH_SIZE,
+        )
+    finally:
+        await async_engine.dispose()
+
+
+async def run_async():
+    await main()
+
+
+def run():
+    asyncio.run(run_async())
+
+
+if __name__ == "__main__":
+    run()

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()

bases/bot_detector/worker_ban_migration/settings.py

@@ -0,0 +1,7 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    N_WORKERS: int = 1
+    MAX_BATCH_SIZE: int = 10_000
+    MAX_INTERVAL_MS: int = 1_000

bases/bot_detector/worker_ban_migration/worker.py

@@ -0,0 +1,34 @@
+import logging
+
+from bot_detector.database.report import migrate_banned_player_reports
+from bot_detector.event_queue.structs import PlayerBannedStruct
+from bot_detector.worker.core import Worker
+from bot_detector.worker_ban_migration.adapter import transform_player_banned
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+logger = logging.getLogger(__name__)
+
+
+class BanMigrationWorker(Worker[PlayerBannedStruct]):
+    def __init__(
+        self,
+        worker_id: int,
+        session_factory: async_sessionmaker[AsyncSession],
+    ) -> None:
+        self._id = worker_id
+        self._session_factory = session_factory
+
+    async def handle(self, batch: list[PlayerBannedStruct]) -> None:
+        logger.info(f"[{self._id}] consumed {len(batch)} ban events")
+        for record in batch:
+            reported_id = transform_player_banned(record)
+            if reported_id is None:
+                logger.info(f"[{self._id}] skipping invalid ban event")
+                continue
+            inserted = await migrate_banned_player_reports(
+                session_factory=self._session_factory,
+                reported_id=reported_id,
+            )
+            logger.info(
+                f"[{self._id}] archived {inserted} rows for reported_id={reported_id}"
+            )

components/bot_detector/database/report/__init__.py

@@ -1,4 +1,5 @@
 from .interface import ReportInterface
+from .migration import migrate_banned_player_reports
 from .repository import ReportRepo
 
-__all__ = ["ReportInterface", "ReportRepo"]
+__all__ = ["ReportInterface", "ReportRepo", "migrate_banned_player_reports"]