sx-pub Phase 1: DB schema, IPFS wiring, actor + webfinger + collections + status endpoints

Foundation for the sx-pub federated publishing protocol:

- 6 SQLAlchemy models: actor, collections, documents, activities, followers, following
- Conditional DB enablement in sx_docs (DATABASE_URL present → enable DB)
- Python IO helpers: get_or_create_actor (auto-generates RSA keypair),
  list_collections, check_status, seed_default_collections
- 4 defhandler endpoints returning text/sx (no JSON):
  /pub/actor, /pub/webfinger, /pub/collections, /pub/status
- Alembic migration infrastructure for sx service
- Docker compose: DB + pgbouncer + IPFS + env vars

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

sx/sxc/pages/helpers.py

@@ -37,6 +37,10 @@ def _register_sx_helpers() -> None:
         "spec-explorer-data": _spec_explorer_data,
         "spec-explorer-data-by-slug": _spec_explorer_data_by_slug,
         "handler-source": _handler_source,
+        # sx-pub helpers (only functional when DATABASE_URL is set)
+        "pub-actor-data": _pub_actor_data,
+        "pub-collections-data": _pub_collections_data,
+        "pub-status-data": _pub_status_data,
     })
 
 
@@ -1717,3 +1721,22 @@ def _page_helpers_demo_data() -> dict:
     results["attr-keys"] = list(ATTR_DETAILS.keys())
 
     return results
+
+
+# ---------------------------------------------------------------------------
+# sx-pub helpers — thin wrappers for SX access
+# ---------------------------------------------------------------------------
+
+async def _pub_actor_data():
+    from .pub_helpers import get_or_create_actor
+    return await get_or_create_actor()
+
+
+async def _pub_collections_data():
+    from .pub_helpers import list_collections
+    return await list_collections()
+
+
+async def _pub_status_data():
+    from .pub_helpers import check_status
+    return await check_status()

sx/sxc/pages/pub_helpers.py

@@ -0,0 +1,144 @@
+"""sx-pub Python IO helpers — actor management, IPFS status, collections.
+
+These are called from SX defhandlers via (helper "pub-..." args...).
+All DB access uses g.s (per-request async session from register_db).
+"""
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+
+logger = logging.getLogger("sx.pub")
+
+SX_PUB_DOMAIN = os.getenv("SX_PUB_DOMAIN", "pub.sx-web.org")
+
+
+async def get_or_create_actor() -> dict[str, Any]:
+    """Get or create the singleton sx-pub actor. Auto-generates RSA keypair."""
+    from quart import g
+    from sqlalchemy import select
+    from shared.models.sx_pub import SxPubActor
+
+    result = await g.s.execute(
+        select(SxPubActor).where(SxPubActor.preferred_username == "sx")
+    )
+    actor = result.scalar_one_or_none()
+
+    if actor is None:
+        from cryptography.hazmat.primitives.asymmetric import rsa
+        from cryptography.hazmat.primitives import serialization
+
+        private_key = rsa.generate_private_key(
+            public_exponent=65537, key_size=2048,
+        )
+        private_pem = private_key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.NoEncryption(),
+        ).decode("utf-8")
+        public_pem = private_key.public_key().public_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        ).decode("utf-8")
+
+        actor = SxPubActor(
+            preferred_username="sx",
+            display_name="SX Language",
+            summary="Federated SX specification publisher",
+            public_key_pem=public_pem,
+            private_key_pem=private_pem,
+            domain=SX_PUB_DOMAIN,
+        )
+        g.s.add(actor)
+        await g.s.flush()
+        logger.info("Created sx-pub actor id=%d domain=%s", actor.id, SX_PUB_DOMAIN)
+
+        # Seed default collections on first run
+        await seed_default_collections()
+
+    return {
+        "preferred-username": actor.preferred_username,
+        "display-name": actor.display_name or actor.preferred_username,
+        "summary": actor.summary or "",
+        "public-key-pem": actor.public_key_pem,
+        "domain": actor.domain,
+    }
+
+
+async def seed_default_collections() -> None:
+    """Create default collections if they don't exist."""
+    from quart import g
+    from sqlalchemy import select
+    from shared.models.sx_pub import SxPubCollection
+
+    defaults = [
+        ("core-specs", "Core Specifications", "Language spec files — evaluator, parser, primitives, render", 0),
+        ("platforms", "Platforms", "Host platform implementations — JavaScript, Python, OCaml", 1),
+        ("components", "Components", "Reusable UI components published as content-addressed SX", 2),
+        ("libraries", "Libraries", "SX library modules — stdlib, signals, freeze, web forms", 3),
+    ]
+
+    for slug, name, description, order in defaults:
+        exists = await g.s.execute(
+            select(SxPubCollection).where(SxPubCollection.slug == slug)
+        )
+        if exists.scalar_one_or_none() is None:
+            g.s.add(SxPubCollection(
+                slug=slug, name=name, description=description, sort_order=order,
+            ))
+    await g.s.flush()
+    logger.info("Seeded %d default collections", len(defaults))
+
+
+async def list_collections() -> list[dict[str, Any]]:
+    """List all pub collections."""
+    from quart import g
+    from sqlalchemy import select
+    from shared.models.sx_pub import SxPubCollection
+
+    result = await g.s.execute(
+        select(SxPubCollection).order_by(SxPubCollection.sort_order)
+    )
+    return [
+        {
+            "slug": c.slug,
+            "name": c.name,
+            "description": c.description or "",
+        }
+        for c in result.scalars().all()
+    ]
+
+
+async def check_status() -> dict[str, Any]:
+    """Health check — DB, IPFS, actor."""
+    status: dict[str, Any] = {"healthy": "true"}
+
+    # DB
+    try:
+        from quart import g
+        from sqlalchemy import text
+        await g.s.execute(text("SELECT 1"))
+        status["db"] = "connected"
+    except Exception as e:
+        status["db"] = f"error: {e}"
+        status["healthy"] = "false"
+
+    # IPFS
+    try:
+        from shared.utils.ipfs_client import is_available
+        ok = await is_available()
+        status["ipfs"] = "available" if ok else "unavailable"
+    except Exception as e:
+        status["ipfs"] = f"error: {e}"
+
+    # Actor
+    try:
+        actor = await get_or_create_actor()
+        status["actor"] = actor["preferred-username"]
+        status["domain"] = actor["domain"]
+    except Exception as e:
+        status["actor"] = f"error: {e}"
+        status["healthy"] = "false"
+
+    return status