Workflow-App/Workflow/scripts/init_pb_schema.py

#!/usr/bin/env python3
"""
init_pb_schema.py
=================
Create the `tasks` collection in PocketBase if it doesn't already exist.

Idempotent — safe to re-run. If the collection exists, it logs the
current state and exits without modifying anything. If anything looks
off (missing fields, wrong type), it prints a warning.

Usage:
  python3 scripts/init_pb_schema.py
  python3 scripts/init_pb_schema.py --pb-url https://pb... --token <jwt>

Reads PB_URL and PB_TOKEN from .env.local in the project root if
not provided.
"""

import argparse
import json
import os
import sys
import urllib.request
import urllib.error
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
ENV_FILE = ROOT / ".env.local"

COLLECTION_NAME = "tasks"
COLLECTION_ID = "pbc_tasks_ysterbaard"  # deterministic name — easier to debug

# Schema for the tasks collection. Field names match the wire format
# the Next.js side and the Python worker expect.
TASKS_SCHEMA = {
    "name": COLLECTION_NAME,
    "type": "base",
    "listRule": "",          # public read — internal tool, no auth
    "viewRule": "",
    "createRule": "",
    "updateRule": "",
    "deleteRule": "",
    "fields": [
        # ── Core ──────────────────────────────────────────────────
        {"name": "title", "type": "text", "required": True, "min": 1, "max": 200},
        {"name": "description", "type": "text", "required": False, "max": 5000},
        {"name": "task_type", "type": "select", "required": True, "maxSelect": 1,
         "values": ["bizportal_search", "bizportal_reserve", "pdf_generate", "email_send", "custom"]},
        {"name": "payload", "type": "json", "required": False, "maxSize": 65535},

        # ── Lifecycle ─────────────────────────────────────────────
        {"name": "status", "type": "select", "required": True, "maxSelect": 1,
         "values": ["backlog", "ready", "in_progress", "blocked", "done", "failed"]},
        {"name": "priority", "type": "select", "required": True, "maxSelect": 1,
         "values": ["low", "normal", "high", "urgent"]},
        {"name": "assignee", "type": "text", "required": False, "max": 100},

        # ── Agent bookkeeping ─────────────────────────────────────
        {"name": "claimed_by", "type": "text", "required": False, "max": 100},
        {"name": "claimed_at", "type": "date", "required": False, "max": "", "min": ""},
        {"name": "started_at", "type": "date", "required": False, "max": "", "min": ""},
        {"name": "completed_at", "type": "date", "required": False, "max": "", "min": ""},
        {"name": "duration_ms", "type": "number", "required": False, "min": 0, "noDecimal": True},

        # ── Output ────────────────────────────────────────────────
        {"name": "result", "type": "json", "required": False, "maxSize": 65535},
        {"name": "error", "type": "text", "required": False, "max": 5000},
        {"name": "log", "type": "json", "required": False, "maxSize": 65535},
        {"name": "retry_count", "type": "number", "required": False, "min": 0, "max": 10, "noDecimal": True},

        # ── Producer ──────────────────────────────────────────────
        {"name": "created_by", "type": "text", "required": False, "max": 100},
    ],
    "indexes": [
        "CREATE INDEX idx_tasks_status ON tasks (status)",
        "CREATE INDEX idx_tasks_priority ON tasks (priority)",
    ],
}


def load_env() -> dict:
    env = {}
    if ENV_FILE.exists():
        for line in ENV_FILE.read_text().splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            if "=" in line:
                k, v = line.split("=", 1)
                env[k.strip()] = v.strip().strip('"').strip("'")
    for k in ("PB_URL", "PB_TOKEN"):
        if k in os.environ:
            env[k] = os.environ[k]
    return env


def http_request(url: str, method: str = "GET", headers: dict = None, body: dict = None):
    data = None
    if body is not None:
        data = json.dumps(body).encode("utf-8")
    headers = dict(headers or {})
    if data is not None and "Content-Type" not in headers:
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return resp.status, json.loads(resp.read().decode("utf-8") or "null")
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="ignore")
        try:
            return e.code, json.loads(body)
        except Exception:
            return e.code, body


def get_collection_by_name(pb_url: str, token: str, name: str):
    url = f"{pb_url}/api/collections/{name}"
    code, body = http_request(url, headers={"Authorization": token})
    if code == 200:
        return body
    if code == 404:
        return None
    sys.exit(f"GET {url} returned {code}: {body}")


def create_collection(pb_url: str, token: str, schema: dict):
    url = f"{pb_url}/api/collections"
    code, body = http_request(url, method="POST", headers={"Authorization": token}, body=schema)
    if code in (200, 201):
        return body
    sys.exit(f"POST {url} returned {code}: {body}")


def verify_schema(collection: dict):
    """Print the current schema and warn about missing/wrong fields."""
    existing = {f["name"]: f for f in collection.get("fields", [])}
    ok = True
    for want in TASKS_SCHEMA["fields"]:
        f = existing.get(want["name"])
        if not f:
            print(f"  ⚠ MISSING field: {want['name']} ({want['type']})")
            ok = False
            continue
        if f["type"] != want["type"]:
            # Special case: dates stored as text in this PB version
            if want["type"] == "date" and f["type"] == "text":
                continue
            print(f"  ⚠ WRONG TYPE field: {want['name']} (want {want['type']}, got {f['type']})")
            ok = False
    if ok:
        print("  schema looks good")
    return ok


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--pb-url")
    parser.add_argument("--token")
    parser.add_argument("--force", action="store_true",
                        help="Recreate the collection if it exists (destructive — wipes data)")
    args = parser.parse_args()

    env = load_env()
    pb_url = (args.pb_url or env.get("PB_URL") or "").rstrip("/")
    token = args.token or env.get("PB_TOKEN") or ""

    if not pb_url or not token:
        sys.exit("PB_URL and PB_TOKEN required (env or .env.local or --pb-url/--token)")

    print(f"PB: {pb_url}")
    print(f"Collection: {COLLECTION_NAME}")
    print()

    existing = get_collection_by_name(pb_url, token, COLLECTION_NAME)
    if existing:
        print(f"✓ Collection '{COLLECTION_NAME}' already exists (id={existing['id']})")
        verify_schema(existing)
        return

    print(f"Creating collection '{COLLECTION_NAME}'…")
    created = create_collection(pb_url, token, TASKS_SCHEMA)
    print(f"✓ Created (id={created['id']})")
    # Add indexes — PocketBase requires a separate PUT
    # http_request(
    #     f"{pb_url}/api/collections/{created['id']}",
    #     method="PATCH",
    #     headers={"Authorization": token},
    #     body={"indexes": TASKS_SCHEMA["indexes"]},
    # )
    print("✓ Done")


if __name__ == "__main__":
    main()