Move redaction of requests to a Cron

Dockerfile

@@ -5,6 +5,12 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     build-essential g++ make cmake python3-dev pkg-config libgomp1 curl ca-certificates gnupg && \
     rm -rf /var/lib/apt/lists/*
 
+# Install supercronic for container-friendly cron scheduling
+ARG SUPERCRONIC_VERSION=v0.2.38
+RUN curl -fsSLo /usr/local/bin/supercronic \
+    "https://github.com/aptible/supercronic/releases/download/${SUPERCRONIC_VERSION}/supercronic-linux-amd64" && \
+    chmod +x /usr/local/bin/supercronic
+
 # Setup the app in workspace
 WORKDIR /workspace
 
@@ -28,10 +34,8 @@ COPY pyproject.toml uv.lock ./
 RUN uv sync --locked
 
 COPY wikidatasearch ./wikidatasearch
+COPY jobs ./jobs
+COPY start.sh ./start.sh
 
 # Container start script
-CMD [ "uv", "run", "gunicorn", "wikidatasearch:app", "--bind", "0.0.0.0:8080", \
-    "-k", "uvicorn.workers.UvicornWorker", "-w", "4", \
-    "--timeout", "30", "--graceful-timeout", "15", "--keep-alive", "5", \
-    "--max-requests", "1000", "--max-requests-jitter", "100", \
-    "--access-logfile", "-", "--error-logfile", "-" ]
+CMD [ "sh", "/workspace/start.sh" ]

docker-compose.yml

@@ -47,9 +47,42 @@ services:
     ports:
       - "8080:8080"
     healthcheck:
-      test: ["CMD-SHELL", "curl -fsS http://127.0.0.1:8080/health/ready || exit 1"]
+      test: ["CMD-SHELL", "curl -fsS http://127.0.0.1:8080/health/live || exit 1"]
       interval: 15s
       timeout: 3s
       retries: 3
       start_period: 30s
     restart: unless-stopped
+
+  redaction-scheduler:
+    build: .
+    depends_on:
+      db:
+        condition: service_healthy
+    environment:
+    # Search credentials are required while importing application configuration
+      ASTRA_DB_APPLICATION_TOKEN: ${ASTRA_DB_APPLICATION_TOKEN}
+      ASTRA_DB_API_ENDPOINT: ${ASTRA_DB_API_ENDPOINT}
+      ASTRA_DB_DATABASE_ID: ${ASTRA_DB_DATABASE_ID}
+      ASTRA_DB_KEYSPACE: ${ASTRA_DB_KEYSPACE}
+      ASTRA_DB_COLLECTION: ${ASTRA_DB_COLLECTION}
+      JINA_API_KEY: ${JINA_API_KEY}
+    # DB + redaction settings
+      DB_NAME: ${DB_NAME}
+      DB_USER: ${DB_USER}
+      DB_PASS: ${DB_PASS}
+      DB_HOST: requestsDB
+      DB_PORT: ${DB_PORT:-3306}
+      LOG_DB_POOL_SIZE: ${LOG_DB_POOL_SIZE:-2}
+      LOG_DB_MAX_OVERFLOW: ${LOG_DB_MAX_OVERFLOW:-1}
+      LOG_DB_POOL_TIMEOUT: ${LOG_DB_POOL_TIMEOUT:-10}
+      LOG_DB_POOL_RECYCLE: ${LOG_DB_POOL_RECYCLE:-1800}
+      REDACTION_DAYS: ${REDACTION_DAYS:-90}
+      REDACTION_BATCH_SIZE: ${REDACTION_BATCH_SIZE:-2000}
+    container_name: redaction-scheduler
+    command: [ "supercronic", "/workspace/jobs/redact.cron" ]
+    volumes:
+      - "./wikidatasearch:/workspace/wikidatasearch"
+      - "./jobs:/workspace/jobs"
+      - "./data:/workspace/data"
+    restart: unless-stopped

jobs/__init__.py

@@ -0,0 +1 @@
+"""Background and maintenance jobs."""

jobs/archive_redacted_logs.py

@@ -0,0 +1,118 @@
+"""Archive redacted request logs to a SQL dump, then delete them."""
+
+import argparse
+import json
+import os
+from datetime import UTC, datetime
+from pathlib import Path
+
+from sqlalchemy import String, delete, func, insert, literal, select
+
+from wikidatasearch.services.logger import Logger, engine
+
+ARCHIVE_BATCH_SIZE = 1000
+
+
+def archive_redacted_requests(output_path: Path, archive_engine=engine) -> int:
+    """Archive all currently redacted requests and delete the archived rows.
+
+    The dump is fully written before rows are deleted. If the database transaction
+    fails, the dump remains available and the deletion is rolled back.
+
+    Args:
+        output_path: Destination SQL dump path. It must not already exist.
+        archive_engine: SQLAlchemy engine used for the archive transaction.
+
+    Returns:
+        Number of archived and deleted request rows.
+    """
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    if output_path.exists():
+        raise FileExistsError(f"archive already exists: {output_path}")
+
+    archived = 0
+    dump_file = None
+    try:
+        with archive_engine.begin() as connection:
+            max_id = connection.scalar(select(func.max(Logger.id)).where(Logger.is_redacted.is_(True)))
+        if max_id is None:
+            return 0
+
+        while True:
+            with archive_engine.begin() as connection:
+                rows = list(
+                    connection.execute(
+                        select(Logger.__table__)
+                        .where(Logger.is_redacted.is_(True), Logger.id <= max_id)
+                        .order_by(Logger.id)
+                        .limit(ARCHIVE_BATCH_SIZE)
+                        .with_for_update()
+                    ).mappings()
+                )
+                if not rows:
+                    break
+
+                batch = []
+                for row in rows:
+                    values = {}
+                    for column in Logger.__table__.columns:
+                        value = row[column.name]
+                        if isinstance(value, (dict, list)):
+                            value = json.dumps(value, ensure_ascii=False, separators=(",", ":"))
+                            values[column.name] = literal(value, type_=String())
+                        else:
+                            values[column.name] = literal(value, type_=column.type)
+                    batch.append(values)
+
+                statement = insert(Logger.__table__).values(batch)
+                dump = (
+                    str(
+                        statement.compile(
+                            dialect=archive_engine.dialect,
+                            compile_kwargs={"literal_binds": True},
+                        )
+                    )
+                    + ";\n"
+                ).encode("utf-8")
+
+                if dump_file is None:
+                    dump_file = output_path.open(mode="xb")
+                dump_position = dump_file.tell()
+                try:
+                    dump_file.write(dump)
+                    dump_file.flush()
+                    os.fsync(dump_file.fileno())
+                except Exception:
+                    dump_file.seek(dump_position)
+                    dump_file.truncate()
+                    raise
+
+                archived_ids = [row["id"] for row in rows]
+                connection.execute(delete(Logger.__table__).where(Logger.id.in_(archived_ids)))
+                archived += len(rows)
+    finally:
+        if dump_file is not None:
+            dump_file.close()
+
+    return archived
+
+
+def main() -> None:
+    """Archive and delete redacted request logs."""
+    timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--output",
+        type=Path,
+        default=Path("data/archives") / f"requests-redacted-{timestamp}.sql",
+        help="SQL dump destination; defaults to data/archives/requests-redacted-<timestamp>.sql",
+    )
+    args = parser.parse_args()
+
+    archived = archive_redacted_requests(args.output)
+    print(f"archive complete: archived_rows={archived} output={args.output}")
+
+
+if __name__ == "__main__":
+    main()

jobs/initialize_database.py

@@ -0,0 +1,59 @@
+"""Initialize database tables before starting the web server."""
+
+from sqlalchemy import inspect as sqlalchemy_inspect
+from sqlalchemy import text
+
+from wikidatasearch.services.logger.database import Base, Logger, UserAgents, engine
+
+
+def sync_indexes() -> None:
+    """Create model indexes and drop obsolete managed indexes."""
+    inspector = sqlalchemy_inspect(engine)
+    managed_tables = (Logger.__table__, UserAgents.__table__)
+
+    # Create missing indexes
+    for table in managed_tables:
+        if not inspector.has_table(table.name):
+            continue
+        for index in table.indexes:
+            index.create(bind=engine, checkfirst=True)
+
+    # Drop obsolete indexes
+    with engine.begin() as conn:
+        for table in managed_tables:
+            if not inspector.has_table(table.name):
+                continue
+
+            existing_indexes = {index["name"] for index in inspector.get_indexes(table.name)}
+            declared_index_names = {index.name for index in table.indexes}
+
+            obsolete_indexes = sorted(
+                index_name
+                for index_name in existing_indexes
+                if index_name.startswith(f"ix_{table.name}_") and index_name not in declared_index_names
+            )
+            for index_name in obsolete_indexes:
+                conn.execute(text(f"DROP INDEX {index_name} ON {table.name}"))
+
+
+def initialize_database():
+    """Create tables if they do not already exist."""
+    try:
+        user_agent_history_exists = sqlalchemy_inspect(engine).has_table(UserAgents.__tablename__)
+
+        Base.metadata.create_all(engine)
+
+        if not user_agent_history_exists:
+            print("Building user agent history from existing request logs...")
+            UserAgents.build_from_requests()
+
+        sync_indexes()
+        return True
+    except Exception as e:
+        print(f"Error while initializing labels database: {e}")
+        return False
+
+
+if __name__ == "__main__":
+    """Run database initialization as a standalone startup step."""
+    raise SystemExit(0 if initialize_database() else 1)

jobs/redact.cron

@@ -0,0 +1,2 @@
+# Run once daily at 03:17 UTC.
+17 3 * * * cd /workspace && uv run python -m jobs.redact_logs

jobs/redact_logs.py

@@ -0,0 +1,18 @@
+"""Job entrypoint to redact old request logs."""
+
+import os
+
+from wikidatasearch.services.logger import Logger
+
+
+def main() -> None:
+    """Run one redaction cycle and print the number of redacted rows."""
+    days = int(os.getenv("REDACTION_DAYS", str(90)))
+    batch_size = int(os.getenv("REDACTION_BATCH_SIZE", str(2000)))
+
+    redacted = Logger.redact_old_requests(days=days, batch_size=batch_size)
+    print(f"redaction complete: redacted_rows={redacted} days={days} batch_size={batch_size}")
+
+
+if __name__ == "__main__":
+    main()

start.sh

@@ -1,14 +1,16 @@
-#!/bin/bash
-
+#!/bin/sh
 set -e
 
-set -a
-source .env
-set +a
-
-echo "API_SECRET set to ${API_SECRET}"
-
-cd /workspace
+uv run python -m jobs.initialize_database
 
-echo "Starting api"
-exec uvicorn wikidatasearch:app --reload --host 0.0.0.0 --port 8000
+exec uv run gunicorn wikidatasearch:app \
+    --bind 0.0.0.0:8080 \
+    -k uvicorn.workers.UvicornWorker \
+    -w 6 \
+    --timeout 120 \
+    --graceful-timeout 30 \
+    --keep-alive 10 \
+    --max-requests 1000 \
+    --max-requests-jitter 200 \
+    --access-logfile - \
+    --error-logfile -

tests/unit/test_analytics_queries.py

@@ -49,11 +49,9 @@ def _fake_read_sql(sql, *_args, **_kwargs):
     return captured
 
 
-def _assert_vector_routes_and_status_filter(sql_text: str) -> None:
-    """Assert vector-route queries exclude 400 and 422 statuses."""
+def _assert_vector_routes_filter(sql_text: str) -> None:
+    """Assert analytics queries stay scoped to vector-query routes."""
     assert f"route IN {AnalyticsQueryService.VECTOR_QUERY_ROUTES_SQL}" in sql_text
-    assert "status NOT IN (400, 422)" in sql_text
-    assert "status <> 422" not in sql_text
 
 
 def test_get_total_user_agents_prefers_original_user_agent(monkeypatch):
@@ -191,8 +189,53 @@ def test_get_consistent_user_agents_count_only_uses_total_query(monkeypatch):
         ),
     ],
 )
-def test_vector_route_queries_exclude_400_and_422(monkeypatch, call):
-    """Ensure all vector-route analytics queries exclude 400 and 422."""
+def test_vector_route_queries_filter_to_vector_routes(monkeypatch, call):
+    """Ensure vector-route analytics queries filter to vector routes."""
     captured = _capture_sql(monkeypatch, pd.DataFrame())
     call()
-    _assert_vector_routes_and_status_filter(captured["query"])
+    _assert_vector_routes_filter(captured["query"])
+
+
+@pytest.mark.parametrize(
+    ("call", "history_filter"),
+    [
+        (
+            lambda: AnalyticsQueryService.get_new_user_agents(
+                datetime(2026, 4, 1),
+                datetime(2026, 4, 23),
+                include_user_agents=True,
+            ),
+            "h.query_first_seen BETWEEN :start AND :end",
+        ),
+        (
+            lambda: AnalyticsQueryService.get_new_user_agents(
+                datetime(2026, 4, 1),
+                datetime(2026, 4, 23),
+                include_user_agents=False,
+            ),
+            "h.query_first_seen BETWEEN :start AND :end",
+        ),
+        (
+            lambda: AnalyticsQueryService.get_consistent_user_agents(
+                datetime(2026, 4, 1),
+                datetime(2026, 4, 23),
+                include_user_agents=True,
+            ),
+            "h.query_distinct_days >= :min_days",
+        ),
+        (
+            lambda: AnalyticsQueryService.get_consistent_user_agents(
+                datetime(2026, 4, 1),
+                datetime(2026, 4, 23),
+                include_user_agents=False,
+            ),
+            "h.query_distinct_days >= :min_days",
+        ),
+    ],
+)
+def test_cross_history_user_agent_analytics_use_history_table(monkeypatch, call, history_filter):
+    """Use query-scoped user-agent history for first-seen and consistency checks."""
+    captured = _capture_sql(monkeypatch, pd.DataFrame())
+    call()
+    assert "JOIN user_agent_history AS h" in captured["query"]
+    assert history_filter in captured["query"]