feat: implement Pages as Spaces Phase 1

- Add PageConfig model with feature flags (calendar, market)
- Auto-create PageConfig on Ghost page sync
- Add create_page() for Ghost /pages/ API endpoint
- Add /new-page/ route for creating pages
- Add ?type=pages blog filter with Posts|Pages tab toggle
- Add list_pages() to DBClient with PageConfig eager loading
- Add PUT /<slug>/admin/features/ route for feature toggles
- Add feature badges (calendar, market) on page cards
- Add features panel to page admin dashboard
- Update shared_lib submodule with PageConfig model

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

bp/blog/ghost/ghost_posts.py

@@ -73,6 +73,35 @@ async def create_post(
     return resp.json()["posts"][0]
 
 
+async def create_page(
+    title: str,
+    lexical_json: str,
+    status: str = "draft",
+    feature_image: str | None = None,
+    custom_excerpt: str | None = None,
+    feature_image_caption: str | None = None,
+) -> dict:
+    """Create a new page in Ghost (via /pages/ endpoint). Returns the created page dict."""
+    page_body: dict = {
+        "title": title,
+        "lexical": lexical_json,
+        "mobiledoc": None,
+        "status": status,
+    }
+    if feature_image:
+        page_body["feature_image"] = feature_image
+    if custom_excerpt:
+        page_body["custom_excerpt"] = custom_excerpt
+    if feature_image_caption is not None:
+        page_body["feature_image_caption"] = feature_image_caption
+    payload = {"pages": [page_body]}
+    url = f"{GHOST_ADMIN_API_URL}/pages/"
+    async with httpx.AsyncClient(timeout=30) as client:
+        resp = await client.post(url, json=payload, headers=_auth_header())
+        _check(resp)
+    return resp.json()["pages"][0]
+
+
 async def update_post(
     ghost_id: str,
     lexical_json: str,

bp/blog/ghost/ghost_sync.py

@@ -13,6 +13,7 @@ from sqlalchemy.orm.attributes import flag_modified  # for non-Mutable JSON colu
 from models.ghost_content import (
     Post, Author, Tag, PostAuthor, PostTag
 )
+from models.page_config import PageConfig
 
 # User-centric membership models
 from models import User
@@ -238,6 +239,15 @@ async def _upsert_post(sess: AsyncSession, gp: Dict[str, Any], author_map: Dict[
         tt = tag_map[t["id"]]
         sess.add(PostTag(post_id=obj.id, tag_id=tt.id, sort_order=idx))
 
+    # Auto-create PageConfig for pages
+    if obj.is_page:
+        existing_pc = (await sess.execute(
+            select(PageConfig).where(PageConfig.post_id == obj.id)
+        )).scalar_one_or_none()
+        if existing_pc is None:
+            sess.add(PageConfig(post_id=obj.id, features={}))
+            await sess.flush()
+
     return obj
 
 async def _ghost_find_member_by_email(email: str) -> Optional[dict]:

bp/blog/ghost_db.py

@@ -6,6 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload, joinedload
 
 from models.ghost_content import Post, Author, Tag, PostTag
+from models.page_config import PageConfig
 from models.tag_group import TagGroup, TagGroupTag
 
 
@@ -235,6 +236,78 @@ class DBClient:
 
         return posts, pagination
 
+    async def list_pages(
+        self,
+        limit: int = 10,
+        page: int = 1,
+        search: Optional[str] = None,
+    ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
+        """
+        List published pages (is_page=True) with their PageConfig eagerly loaded.
+        Returns (pages, pagination).
+        """
+        base_filters = [
+            Post.deleted_at.is_(None),
+            Post.status == "published",
+            Post.is_page.is_(True),
+        ]
+
+        q = select(Post).where(*base_filters)
+
+        if search:
+            term = f"%{search.strip().lower()}%"
+            q = q.where(
+                or_(
+                    func.lower(func.coalesce(Post.title, "")).like(term),
+                    func.lower(func.coalesce(Post.excerpt, "")).like(term),
+                    func.lower(func.coalesce(Post.plaintext, "")).like(term),
+                )
+            )
+
+        q = q.order_by(desc(Post.published_at))
+
+        if page < 1:
+            page = 1
+        offset_val = (page - 1) * limit
+
+        q_no_limit = q.with_only_columns(Post.id).order_by(None)
+        count_q = select(func.count()).select_from(q_no_limit.subquery())
+        total = int((await self.sess.execute(count_q)).scalar() or 0)
+
+        q = (
+            q.options(
+                joinedload(Post.primary_author),
+                joinedload(Post.primary_tag),
+                selectinload(Post.authors),
+                selectinload(Post.tags),
+                joinedload(Post.page_config),
+            )
+            .limit(limit)
+            .offset(offset_val)
+        )
+
+        rows: List[Post] = list((await self.sess.execute(q)).scalars())
+
+        def _page_to_public(p: Post) -> Dict[str, Any]:
+            d = _post_to_public(p)
+            pc = p.page_config
+            d["features"] = pc.features if pc else {}
+            return d
+
+        pages_list = [_page_to_public(p) for p in rows]
+
+        pages_total = (total + limit - 1) // limit if limit else 1
+        pagination = {
+            "page": page,
+            "limit": limit,
+            "pages": pages_total,
+            "total": total,
+            "next": page + 1 if page < pages_total else None,
+            "prev": page - 1 if page > 1 else None,
+        }
+
+        return pages_list, pagination
+
     async def posts_by_slug(
         self,
         slug: str,

bp/blog/routes.py

@@ -18,6 +18,7 @@ from .ghost_db import DBClient  # adjust import path
 from db.session import get_session
 from .filters.qs import makeqs_factory, decode
 from .services.posts_data import posts_data
+from .services.pages_data import pages_data
 
 from suma_browser.app.redis_cacher import cache_page, invalidate_tag_cache
 from suma_browser.app.utils.htmx import is_htmx_request
@@ -81,7 +82,24 @@ def register(url_prefix, title):
     async def home():
 
       q = decode()
+      content_type = request.args.get("type", "posts")
 
+      if content_type == "pages":
+          data = await pages_data(g.s, q.page, q.search)
+          context = {
+              **data,
+              "content_type": "pages",
+              "search": q.search,
+          }
+          if not is_htmx_request():
+              html = await render_template("_types/blog/index.html", **context)
+          elif q.page > 1:
+              html = await render_template("_types/blog/_page_cards.html", **context)
+          else:
+              html = await render_template("_types/blog/_oob_elements.html", **context)
+          return await make_response(html)
+
+      # Default: posts listing
       # Drafts filter requires login; ignore if not logged in
       show_drafts = bool(q.drafts and g.user)
       is_admin = bool((g.get("rights") or {}).get("admin"))
@@ -99,6 +117,7 @@ def register(url_prefix, title):
 
       context = {
           **data,
+          "content_type": "posts",
           "selected_tags": q.selected_tags,
           "selected_authors": q.selected_authors,
           "selected_groups": q.selected_groups,
@@ -196,6 +215,81 @@ def register(url_prefix, title):
       return redirect(host_url(url_for("blog.post.admin.edit", slug=ghost_post["slug"])))
 
 
+    @blogs_bp.get("/new-page/")
+    @require_admin
+    async def new_page():
+      if not is_htmx_request():
+        html = await render_template("_types/blog_new/index.html", is_page=True)
+      else:
+        html = await render_template("_types/blog_new/_oob_elements.html", is_page=True)
+      return await make_response(html)
+
+    @blogs_bp.post("/new-page/")
+    @require_admin
+    async def new_page_save():
+      from .ghost.ghost_posts import create_page
+      from .ghost.lexical_validator import validate_lexical
+      from .ghost.ghost_sync import sync_single_page
+
+      form = await request.form
+      title = form.get("title", "").strip() or "Untitled"
+      lexical_raw = form.get("lexical", "")
+      status = form.get("status", "draft")
+      feature_image = form.get("feature_image", "").strip()
+      custom_excerpt = form.get("custom_excerpt", "").strip()
+      feature_image_caption = form.get("feature_image_caption", "").strip()
+
+      # Validate
+      try:
+        lexical_doc = json.loads(lexical_raw)
+      except (json.JSONDecodeError, TypeError):
+        html = await render_template(
+            "_types/blog_new/index.html",
+            save_error="Invalid JSON in editor content.",
+            is_page=True,
+        )
+        return await make_response(html, 400)
+
+      ok, reason = validate_lexical(lexical_doc)
+      if not ok:
+        html = await render_template(
+            "_types/blog_new/index.html",
+            save_error=reason,
+            is_page=True,
+        )
+        return await make_response(html, 400)
+
+      # Create in Ghost (as page)
+      ghost_page = await create_page(
+          title=title,
+          lexical_json=lexical_raw,
+          status=status,
+          feature_image=feature_image or None,
+          custom_excerpt=custom_excerpt or None,
+          feature_image_caption=feature_image_caption or None,
+      )
+
+      # Sync to local DB (uses pages endpoint)
+      await sync_single_page(g.s, ghost_page["id"])
+      await g.s.flush()
+
+      # Set user_id on the newly created page
+      from models.ghost_content import Post
+      from sqlalchemy import select
+      local_post = (await g.s.execute(
+          select(Post).where(Post.ghost_id == ghost_page["id"])
+      )).scalar_one_or_none()
+      if local_post and local_post.user_id is None:
+          local_post.user_id = g.user.id
+          await g.s.flush()
+
+      # Clear blog listing cache
+      await invalidate_tag_cache("blog")
+
+      # Redirect to the page admin
+      return redirect(host_url(url_for("blog.post.admin.edit", slug=ghost_page["slug"])))
+
+
     @blogs_bp.get("/drafts/")
     async def drafts():
       return redirect(host_url(url_for("blog.home")) + "?drafts=1")

bp/blog/services/pages_data.py

@@ -0,0 +1,18 @@
+from ..ghost_db import DBClient
+
+
+async def pages_data(session, page, search):
+    client = DBClient(session)
+
+    pages, pagination = await client.list_pages(
+        limit=10,
+        page=page,
+        search=search,
+    )
+
+    return {
+        "pages": pages,
+        "page": pagination.get("page", page),
+        "total_pages": pagination.get("pages", 1),
+        "search": search,
+    }