tipibot/scripts/sync_pb_schema.py

"""Reconcile the PocketBase economy collections with the bot's user schema.

The expected field list is derived from core.economy._default_user() - the
single source of truth for what the bot persists - so this supersedes
add_stats_fields.py / add_quest_fields.py and also covers the fields those
scripts never added (fishing, prestige, quests, last_heist, ...).

PocketBase silently drops record fields that are missing from the collection
schema, so any drift here breaks features without a single error message.

Patches BOTH profile collections regardless of BOT_PROFILE. Only ever adds
fields; existing fields are never removed or retyped (type mismatches are
reported for manual review).

    python scripts/sync_pb_schema.py           # add whatever is missing
    python scripts/sync_pb_schema.py --check   # report only; exit 1 on drift
"""

from __future__ import annotations

import asyncio
import sys
from pathlib import Path

import aiohttp
from dotenv import load_dotenv

sys.path.insert(0, str(Path(__file__).parent.parent))
load_dotenv()

import config  # noqa: E402
from core import economy  # noqa: E402

PB_URL = config.PB_URL
COLLECTIONS = sorted({config.PB_ECONOMY_COLLECTION_DEV, config.PB_ECONOMY_COLLECTION_ECONOMY})

# _default_user() keys with a None/str default that PocketBase stores as text
# (ISO date/datetime strings, plus the vanity badge id).
_TEXT_FIELDS = {
    "last_daily", "last_work", "last_beg", "last_crime", "last_rob",
    "last_heist", "last_fish", "last_streak_date", "jailed_until",
    "vanity_active", "lottery_period",
}


def _expected_fields() -> dict[str, str]:
    """Map every persisted field name to its PocketBase field type."""
    expected = {"user_id": "text"}
    for key, default in economy._default_user().items():
        if key in _TEXT_FIELDS:
            expected[key] = "text"
        elif isinstance(default, bool):
            expected[key] = "bool"
        elif isinstance(default, (int, float)):
            expected[key] = "number"
        elif isinstance(default, (list, dict)):
            expected[key] = "json"
        else:
            raise SystemExit(
                f"Cannot infer a PocketBase type for {key!r} (default {default!r}). "
                "Add it to _TEXT_FIELDS or extend the mapping."
            )
    return expected


async def main() -> int:
    check_only = "--check" in sys.argv
    expected = _expected_fields()
    drift = False

    timeout = aiohttp.ClientTimeout(total=15)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.post(
            f"{PB_URL}/api/collections/_superusers/auth-with-password",
            json={"identity": config.PB_ADMIN_EMAIL, "password": config.PB_ADMIN_PASSWORD},
        ) as resp:
            if resp.status != 200:
                print(f"Auth failed ({resp.status}): {await resp.text()}")
                return 1
            hdrs = {"Authorization": (await resp.json())["token"]}

        for collection in COLLECTIONS:
            print(f"── {collection} ──")
            async with session.get(
                f"{PB_URL}/api/collections/{collection}", headers=hdrs
            ) as resp:
                if resp.status != 200:
                    print(f"Could not fetch collection ({resp.status}): {await resp.text()}\n")
                    drift = True
                    continue
                col = await resp.json()

            existing = {f["name"]: f.get("type", "?") for f in col.get("fields", [])}

            missing = [name for name in expected if name not in existing]
            mismatched = [
                (name, existing[name], ftype)
                for name, ftype in expected.items()
                if name in existing and existing[name] != ftype
            ]

            for name, have, want in mismatched:
                drift = True
                print(f"  ! {name}: schema has type '{have}', bot expects '{want}' - fix manually")

            if not missing:
                print("  ✓ no missing fields\n" if not mismatched else "")
                continue

            drift = True
            for name in missing:
                print(f"  + {name}  ({expected[name]})")

            if check_only:
                print()
                continue

            new_fields = [{"name": name, "type": expected[name]} for name in missing]
            async with session.patch(
                f"{PB_URL}/api/collections/{collection}",
                json={"fields": col.get("fields", []) + new_fields},
                headers=hdrs,
            ) as resp:
                if resp.status != 200:
                    print(f"  Schema update failed ({resp.status}): {await resp.text()}\n")
                    continue
                print(f"  ✅ added {len(new_fields)} field(s)\n")

    return 1 if (check_only and drift) else 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))