e6772bfb1f
ask_threads/ask_messages/ask_log shipped with created/updated typed as plain `date` fields, which PocketBase never populates — every row's created has been empty since. pb-add-ask-collections.py now emits autodate fields (onCreate, plus onUpdate for updated) and migrates the three live collections in place under --apply, re-adding idx_ask_messages_thread_created if PocketBase drops it with the field. --backfill (apply-only) just reports affected row counts: PocketBase can only set an autodate field to *now*, never a historical value, so there's nothing safe to backfill. Since old rows keep an empty created either way, the read side sorts ask_messages by `created,id` instead of `created` alone, so tied (all-empty) rows land in a stable, deterministic order rather than whatever order PocketBase happens to return them in. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
425 lines
16 KiB
Python
425 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Add the three PocketBase collections behind the Ask agent.
|
|
|
|
``ask_threads`` (one row per conversation), ``ask_messages`` (append-only:
|
|
user/assistant/attachment/image rows) and ``ask_log`` (changes for Undo,
|
|
long-term memories, and developer change requests — one collection,
|
|
distinguished by ``type``).
|
|
|
|
Dry run unless ``--apply --confirm-host <exact configured PB host>`` is
|
|
given. Additive and idempotent, following the same pattern as
|
|
scripts/pb-add-dn-staged-ingestion.py: read the live field/index list before
|
|
appending, never delete or replace unrelated schema.
|
|
|
|
All rules are left null (superuser only) — the Next.js server is the only
|
|
client of these three collections; there is no browser-direct access.
|
|
|
|
``created``/``updated`` must be PocketBase ``autodate`` fields (they were
|
|
first shipped as plain ``date`` fields, which PocketBase never populates on
|
|
its own — every row's ``created`` has been empty since). ``--apply`` also
|
|
migrates ``ask_threads``, ``ask_messages`` and ``ask_log`` in place: any
|
|
``created``/``updated`` field still typed ``date`` on those three
|
|
collections is replaced with an ``autodate`` field of the same name.
|
|
Existing rows keep an empty ``created`` either way — PocketBase can't
|
|
backfill an autodate field to a historical value — so ``--backfill``
|
|
(only meaningful with ``--apply``) just reports how many rows are affected
|
|
per collection rather than attempting a write that would do nothing.
|
|
"""
|
|
import sys
|
|
import urllib.parse
|
|
|
|
from pb_env import PB, http, superuser_token
|
|
|
|
DRY_RUN = "--dry-run" in sys.argv or "--apply" not in sys.argv
|
|
APPLY = "--apply" in sys.argv
|
|
BACKFILL = "--backfill" in sys.argv
|
|
|
|
MIGRATED_COLLECTIONS = ("ask_threads", "ask_messages", "ask_log")
|
|
ASK_MESSAGES_THREAD_CREATED_INDEX = (
|
|
"CREATE INDEX `idx_ask_messages_thread_created` "
|
|
"ON `ask_messages` (`thread`, `created`)"
|
|
)
|
|
|
|
|
|
def confirmation_host():
|
|
for index, argument in enumerate(sys.argv):
|
|
if argument.startswith("--confirm-host="):
|
|
return argument.split("=", 1)[1]
|
|
if argument == "--confirm-host" and index + 1 < len(sys.argv):
|
|
return sys.argv[index + 1]
|
|
return None
|
|
|
|
|
|
def get_collection(token, name):
|
|
code, response = http("GET", f"/api/collections/{name}", token=token)
|
|
return response if code == 200 else None
|
|
|
|
|
|
def require_collection(token, name):
|
|
collection = get_collection(token, name)
|
|
if collection is None:
|
|
sys.exit(f"collection {name} not found")
|
|
return collection
|
|
|
|
|
|
def relation(name, collection_id, required=False, cascade_delete=False, max_select=1):
|
|
return {
|
|
"name": name,
|
|
"type": "relation",
|
|
"required": required,
|
|
"collectionId": collection_id,
|
|
"maxSelect": max_select,
|
|
"cascadeDelete": cascade_delete,
|
|
}
|
|
|
|
|
|
def field(name, field_type="text", **options):
|
|
return {"name": name, "type": field_type, "required": False, **options}
|
|
|
|
|
|
def select(name, values, required=False, max_select=1):
|
|
return {
|
|
"name": name,
|
|
"type": "select",
|
|
"required": required,
|
|
"maxSelect": max_select,
|
|
"values": values,
|
|
}
|
|
|
|
|
|
def autodate_field(name):
|
|
options = {"onCreate": True}
|
|
if name == "updated":
|
|
options["onUpdate"] = True
|
|
return field(name, "autodate", **options)
|
|
|
|
|
|
def system_fields():
|
|
return [autodate_field("created"), autodate_field("updated")]
|
|
|
|
|
|
def add_fields(token, name, wanted):
|
|
collection = require_collection(token, name)
|
|
existing = list(collection.get("fields", []))
|
|
names = {item.get("name") for item in existing}
|
|
additions = [item for item in wanted if item["name"] not in names]
|
|
if not additions:
|
|
print(f" {name}: fields already present")
|
|
return
|
|
print(f" {name}: add fields {[item['name'] for item in additions]}")
|
|
if not DRY_RUN:
|
|
code, response = http(
|
|
"PATCH", f"/api/collections/{collection['id']}",
|
|
{"fields": existing + additions}, token,
|
|
)
|
|
if code != 200:
|
|
sys.exit(f"PATCH {name} failed: {code} {response}")
|
|
|
|
|
|
def add_indexes(token, name, wanted):
|
|
collection = require_collection(token, name)
|
|
existing = list(collection.get("indexes", []) or [])
|
|
# Simple union by exact string match — these are new indexes on new
|
|
# collections, so there is nothing pre-existing to reconcile against.
|
|
additions = [i for i in wanted if i not in existing]
|
|
if not additions:
|
|
print(f" {name}: indexes already present")
|
|
return
|
|
print(f" {name}: add indexes {additions}")
|
|
if not DRY_RUN:
|
|
code, response = http(
|
|
"PATCH", f"/api/collections/{collection['id']}",
|
|
{"indexes": existing + additions}, token,
|
|
)
|
|
if code != 200:
|
|
sys.exit(f"PATCH {name} indexes failed: {code} {response}")
|
|
|
|
|
|
def stale_date_field_names(fields):
|
|
"""Names of ``created``/``updated`` fields still typed ``date``. Pure."""
|
|
return sorted(
|
|
f["name"] for f in fields
|
|
if f.get("name") in ("created", "updated") and f.get("type") == "date"
|
|
)
|
|
|
|
|
|
def migration_fields(existing_fields, stale_names):
|
|
"""The full ``fields`` list for the migration PATCH: every field that
|
|
isn't one of ``stale_names`` kept as-is, plus a fresh ``autodate``
|
|
field for each stale name. Omitting the old field's ``id`` here is
|
|
what makes PocketBase drop it — this list represents the *entire*
|
|
desired field set. Pure — no network calls, easy to unit test.
|
|
"""
|
|
kept = [f for f in existing_fields if f.get("name") not in stale_names]
|
|
return kept + [autodate_field(n) for n in sorted(stale_names)]
|
|
|
|
|
|
def migrate_system_fields(token, name):
|
|
"""On one of the three Ask collections, replace any ``created``/
|
|
``updated`` field still typed ``date`` with an ``autodate`` field of
|
|
the same name. Idempotent — a second run finds nothing stale. Existing
|
|
rows keep an empty ``created``/``updated``; see ``--backfill``.
|
|
"""
|
|
collection = require_collection(token, name)
|
|
fields = list(collection.get("fields", []))
|
|
stale = stale_date_field_names(fields)
|
|
if not stale:
|
|
print(f" {name}: created/updated already autodate")
|
|
return
|
|
|
|
print(f" {name}: migrate {stale} from date -> autodate")
|
|
replaced = migration_fields(fields, set(stale))
|
|
if DRY_RUN:
|
|
print(f" {name}: would PATCH fields (delete+recreate {stale})")
|
|
return
|
|
|
|
code, response = http(
|
|
"PATCH", f"/api/collections/{collection['id']}", {"fields": replaced}, token,
|
|
)
|
|
if code != 200:
|
|
# Some PocketBase versions refuse to delete and add a field with
|
|
# the same name in a single PATCH — fall back to two PATCHes:
|
|
# delete the stale fields first, then add the autodate ones.
|
|
print(f" {name}: combined PATCH failed ({code}: {response}); trying delete then add")
|
|
kept = [f for f in fields if f.get("name") not in set(stale)]
|
|
code, response = http(
|
|
"PATCH", f"/api/collections/{collection['id']}", {"fields": kept}, token,
|
|
)
|
|
if code != 200:
|
|
sys.exit(f" {name}: delete PATCH failed: {code} {response}")
|
|
code, current = http("GET", f"/api/collections/{name}", token=token)
|
|
if code != 200:
|
|
sys.exit(f" {name}: re-fetch after delete failed: {code} {current}")
|
|
current_fields = list(current.get("fields", []))
|
|
code, response = http(
|
|
"PATCH", f"/api/collections/{current['id']}",
|
|
{"fields": current_fields + [autodate_field(n) for n in stale]}, token,
|
|
)
|
|
if code != 200:
|
|
sys.exit(f" {name}: add PATCH failed: {code} {response}")
|
|
|
|
# The thread/created index is defined on the field we just replaced —
|
|
# re-add it if PocketBase dropped it along with the old field.
|
|
if name == "ask_messages":
|
|
code, current = http("GET", f"/api/collections/{name}", token=token)
|
|
if code != 200:
|
|
sys.exit(f" {name}: re-fetch failed: {code} {current}")
|
|
current_indexes = list(current.get("indexes", []) or [])
|
|
if ASK_MESSAGES_THREAD_CREATED_INDEX not in current_indexes:
|
|
print(f" {name}: re-adding idx_ask_messages_thread_created (dropped with the field)")
|
|
code, response = http(
|
|
"PATCH", f"/api/collections/{current['id']}",
|
|
{"indexes": current_indexes + [ASK_MESSAGES_THREAD_CREATED_INDEX]}, token,
|
|
)
|
|
if code != 200:
|
|
sys.exit(f" {name}: index PATCH failed: {code} {response}")
|
|
print(f" {name}: migrated")
|
|
|
|
|
|
def count_empty_created(token, name):
|
|
query = urllib.parse.quote("created = ''")
|
|
code, response = http(
|
|
"GET", f"/api/collections/{name}/records?perPage=1&filter={query}", token=token,
|
|
)
|
|
if code != 200:
|
|
sys.exit(f" {name}: could not count empty-created rows: {code} {response}")
|
|
return response.get("totalItems", 0)
|
|
|
|
|
|
def backfill_report(token):
|
|
"""``--backfill``: PocketBase only ever sets an ``autodate`` field to
|
|
*now*, on create (and on update, if ``onUpdate`` is set) — never to an
|
|
arbitrary historical value, so there is nothing a script can write
|
|
into an already-empty ``created`` that would be correct. This just
|
|
counts the affected rows per collection; it never issues a write.
|
|
"""
|
|
print("\nbackfill (--backfill): reporting only, autodate can't be set to a past value")
|
|
for name in MIGRATED_COLLECTIONS:
|
|
total = count_empty_created(token, name)
|
|
if total == 0:
|
|
print(f" {name}: no rows with empty created")
|
|
else:
|
|
print(
|
|
f" {name}: {total} row(s) have empty created; left as-is. "
|
|
"The read side orders these by id (see threads.ts)."
|
|
)
|
|
|
|
|
|
def create_collection(token, name, fields, indexes):
|
|
existing = get_collection(token, name)
|
|
if existing is not None:
|
|
print(f" {name}: already exists")
|
|
add_fields(token, name, fields)
|
|
add_indexes(token, name, indexes)
|
|
return existing["id"]
|
|
print(f" {name}: create collection")
|
|
if DRY_RUN:
|
|
return name
|
|
code, response = http(
|
|
"POST", "/api/collections",
|
|
{"name": name, "type": "base", "fields": fields, "indexes": indexes},
|
|
token,
|
|
)
|
|
if code != 200:
|
|
sys.exit(f"CREATE {name} failed: {code} {response}")
|
|
return response["id"]
|
|
|
|
|
|
def self_test():
|
|
r = relation("owner", "users123", required=True)
|
|
assert r == {
|
|
"name": "owner", "type": "relation", "required": True,
|
|
"collectionId": "users123", "maxSelect": 1, "cascadeDelete": False,
|
|
}
|
|
f = field("text", "text", max=100000)
|
|
assert f == {"name": "text", "type": "text", "required": False, "max": 100000}
|
|
s = select("kind", ["user", "assistant", "attachment", "image"])
|
|
assert s["values"] == ["user", "assistant", "attachment", "image"]
|
|
|
|
assert system_fields() == [
|
|
{"name": "created", "type": "autodate", "required": False, "onCreate": True},
|
|
{"name": "updated", "type": "autodate", "required": False, "onCreate": True, "onUpdate": True},
|
|
]
|
|
|
|
stale_fields = [
|
|
{"id": "f1", "name": "created", "type": "date", "onCreate": True},
|
|
{"id": "f2", "name": "updated", "type": "date", "onCreate": True, "onUpdate": True},
|
|
{"id": "f3", "name": "title", "type": "text"},
|
|
]
|
|
assert stale_date_field_names(stale_fields) == ["created", "updated"]
|
|
assert stale_date_field_names([{"name": "created", "type": "autodate"}]) == []
|
|
|
|
migrated = migration_fields(stale_fields, {"created", "updated"})
|
|
assert migrated == [
|
|
{"id": "f3", "name": "title", "type": "text"},
|
|
{"name": "created", "type": "autodate", "required": False, "onCreate": True},
|
|
{"name": "updated", "type": "autodate", "required": False, "onCreate": True, "onUpdate": True},
|
|
]
|
|
# No stale "id" leaks through on the replacement fields — that is what
|
|
# tells PocketBase to drop the old field and create a new one.
|
|
assert "id" not in migrated[1] and "id" not in migrated[2]
|
|
|
|
print("self-test: field builders and migration helpers passed")
|
|
return 0
|
|
|
|
|
|
def main():
|
|
if "--self-test" in sys.argv:
|
|
return self_test()
|
|
if BACKFILL and not APPLY:
|
|
sys.exit("--backfill requires --apply")
|
|
if APPLY and confirmation_host() != PB:
|
|
sys.exit(
|
|
"REFUSING --apply: pass --confirm-host with the exact configured "
|
|
f"PocketBase host ({PB})"
|
|
)
|
|
try:
|
|
token = superuser_token()
|
|
except SystemExit as error:
|
|
if not DRY_RUN:
|
|
raise
|
|
print(f"offline preview: {error}")
|
|
print("would create: ask_threads, ask_messages, ask_log")
|
|
print("\nDRY RUN. No schema changes written.")
|
|
return 0
|
|
|
|
print(f"target: {PB} mode: {'DRY RUN' if DRY_RUN else 'APPLY'}")
|
|
|
|
print("\nmigrate created/updated to autodate:")
|
|
for collection_name in MIGRATED_COLLECTIONS:
|
|
if get_collection(token, collection_name) is None:
|
|
print(f" {collection_name}: does not exist yet, nothing to migrate")
|
|
continue
|
|
migrate_system_fields(token, collection_name)
|
|
|
|
users = require_collection(token, "users")
|
|
sites = require_collection(token, "sites")
|
|
users_id = users["id"]
|
|
sites_id = sites["id"]
|
|
|
|
threads_id = create_collection(
|
|
token, "ask_threads",
|
|
system_fields() + [
|
|
field("title"),
|
|
relation("owner", users_id, required=True),
|
|
field("archived", "bool"),
|
|
field("last_message_at", "date"),
|
|
],
|
|
[
|
|
"CREATE INDEX `idx_ask_threads_owner_archived_last_message` "
|
|
"ON `ask_threads` (`owner`, `archived`, `last_message_at`)",
|
|
],
|
|
)
|
|
|
|
create_collection(
|
|
token, "ask_messages",
|
|
system_fields() + [
|
|
relation("thread", threads_id, required=True, cascade_delete=True),
|
|
relation("owner", users_id),
|
|
select("kind", ["user", "assistant", "attachment", "image"], required=True),
|
|
field("text", max=100000),
|
|
field("display"),
|
|
field("charts", "json", maxSize=2000000),
|
|
field("action_ids", "json", maxSize=200000),
|
|
field("attachment_ids", "json", maxSize=200000),
|
|
field("tools", "json", maxSize=200000),
|
|
field(
|
|
"file", "file", maxSelect=1, maxSize=26214400,
|
|
mimeTypes=[
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
"application/vnd.ms-excel",
|
|
"text/csv",
|
|
"application/pdf",
|
|
"image/png",
|
|
"image/jpeg",
|
|
"image/webp",
|
|
"text/plain",
|
|
],
|
|
),
|
|
field("parsed", "json", maxSize=5000000),
|
|
field("parse_error"),
|
|
],
|
|
[
|
|
"CREATE INDEX `idx_ask_messages_thread_created` "
|
|
"ON `ask_messages` (`thread`, `created`)",
|
|
],
|
|
)
|
|
|
|
create_collection(
|
|
token, "ask_log",
|
|
system_fields() + [
|
|
select("type", ["change", "memory", "request"], required=True),
|
|
relation("owner", users_id),
|
|
field("shared", "bool"),
|
|
relation("thread", threads_id, required=False, cascade_delete=False),
|
|
select(
|
|
"status",
|
|
["pending", "applied", "rejected", "failed", "undone",
|
|
"open", "planned", "done", "declined"],
|
|
),
|
|
field("text", max=2000),
|
|
field("detail"),
|
|
relation("site", sites_id, required=False),
|
|
field("ops", "json", maxSize=2000000),
|
|
field("before", "json", maxSize=2000000),
|
|
field("after", "json", maxSize=2000000),
|
|
field("error"),
|
|
field("applied_at", "date"),
|
|
field("undone_at", "date"),
|
|
],
|
|
[
|
|
"CREATE INDEX `idx_ask_log_type_status` ON `ask_log` (`type`, `status`)",
|
|
"CREATE INDEX `idx_ask_log_owner_type` ON `ask_log` (`owner`, `type`)",
|
|
"CREATE INDEX `idx_ask_log_thread` ON `ask_log` (`thread`)",
|
|
],
|
|
)
|
|
|
|
if BACKFILL:
|
|
backfill_report(token)
|
|
|
|
print("\nDRY RUN. No schema changes written." if DRY_RUN else "\nSchema applied.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|