""" Federation service s-expression page components. Renders social timeline, compose, search, following/followers, notifications, actor profiles, login, and username selection pages. """ from __future__ import annotations import os from typing import Any from markupsafe import escape from shared.sx.jinja_bridge import load_service_components from shared.sx.helpers import ( sx_call, SxExpr, root_header_sx, full_page_sx, header_child_sx, ) # Load federation-specific .sx components at import time load_service_components(os.path.dirname(os.path.dirname(__file__))) # --------------------------------------------------------------------------- # Social header nav # --------------------------------------------------------------------------- def _social_nav_sx(actor: Any) -> str: """Build the social header nav bar content.""" from quart import url_for, request if not actor: choose_url = url_for("identity.choose_username_form") return sx_call("federation-nav-choose-username", url=choose_url) links = [ ("social.home_timeline", "Timeline"), ("social.public_timeline", "Public"), ("social.compose_form", "Compose"), ("social.following_list", "Following"), ("social.followers_list", "Followers"), ("social.search", "Search"), ] parts = [] for endpoint, label in links: href = url_for(endpoint) bold = " font-bold" if request.path == href else "" parts.append(sx_call( "federation-nav-link", href=href, cls=f"px-2 py-1 rounded hover:bg-stone-200{bold}", label=label, )) # Notifications with live badge notif_url = url_for("social.notifications") notif_count_url = url_for("social.notification_count") notif_bold = " font-bold" if request.path == notif_url else "" parts.append(sx_call( "federation-nav-notification-link", href=notif_url, cls=f"px-2 py-1 rounded hover:bg-stone-200 relative{notif_bold}", count_url=notif_count_url, )) # Profile link profile_url = url_for("activitypub.actor_profile", username=actor.preferred_username) parts.append(sx_call( "federation-nav-link", href=profile_url, cls="px-2 py-1 rounded hover:bg-stone-200", label=f"@{actor.preferred_username}", )) items_sx = "(<> " + " ".join(parts) + ")" return sx_call("federation-nav-bar", items=SxExpr(items_sx)) def _social_header_sx(actor: Any) -> str: """Build the social section header row.""" nav_sx = _social_nav_sx(actor) return sx_call("federation-social-header", nav=SxExpr(nav_sx)) def _social_page(ctx: dict, actor: Any, *, content: str, title: str = "Rose Ash", meta_html: str = "") -> str: """Render a social page with header and content.""" hdr = root_header_sx(ctx) social_hdr = _social_header_sx(actor) child = header_child_sx(social_hdr) header_rows = "(<> " + hdr + " " + child + ")" return full_page_sx(ctx, header_rows=header_rows, content=content, meta_html=meta_html or f'{escape(title)}') # --------------------------------------------------------------------------- # Post card # --------------------------------------------------------------------------- def _interaction_buttons_sx(item: Any, actor: Any) -> str: """Render like/boost/reply buttons for a post.""" from shared.browser.app.csrf import generate_csrf_token from quart import url_for oid = getattr(item, "object_id", "") or "" ainbox = getattr(item, "author_inbox", "") or "" lcount = getattr(item, "like_count", 0) or 0 bcount = getattr(item, "boost_count", 0) or 0 liked = getattr(item, "liked_by_me", False) boosted = getattr(item, "boosted_by_me", False) csrf = generate_csrf_token() safe_id = oid.replace("/", "_").replace(":", "_") target = f"#interactions-{safe_id}" if liked: like_action = url_for("social.unlike") like_cls = "text-red-500 hover:text-red-600" like_icon = "\u2665" else: like_action = url_for("social.like") like_cls = "hover:text-red-500" like_icon = "\u2661" if boosted: boost_action = url_for("social.unboost") boost_cls = "text-green-600 hover:text-green-700" else: boost_action = url_for("social.boost") boost_cls = "hover:text-green-600" reply_url = url_for("social.compose_form", reply_to=oid) if oid else "" reply_sx = sx_call("federation-reply-link", url=reply_url) if reply_url else "" like_form = sx_call( "federation-like-form", action=like_action, target=target, oid=oid, ainbox=ainbox, csrf=csrf, cls=f"flex items-center gap-1 {like_cls}", icon=like_icon, count=str(lcount), ) boost_form = sx_call( "federation-boost-form", action=boost_action, target=target, oid=oid, ainbox=ainbox, csrf=csrf, cls=f"flex items-center gap-1 {boost_cls}", count=str(bcount), ) return sx_call( "federation-interaction-buttons", like=SxExpr(like_form), boost=SxExpr(boost_form), reply=SxExpr(reply_sx) if reply_sx else None, ) def _post_card_sx(item: Any, actor: Any) -> str: """Render a single timeline post card.""" boosted_by = getattr(item, "boosted_by", None) actor_icon = getattr(item, "actor_icon", None) actor_name = getattr(item, "actor_name", "?") actor_username = getattr(item, "actor_username", "") actor_domain = getattr(item, "actor_domain", "") content = getattr(item, "content", "") summary = getattr(item, "summary", None) published = getattr(item, "published", None) url = getattr(item, "url", None) post_type = getattr(item, "post_type", "") boost_sx = sx_call( "federation-boost-label", name=str(escape(boosted_by)), ) if boosted_by else "" if actor_icon: avatar = sx_call("federation-avatar-img", src=actor_icon, cls="w-10 h-10 rounded-full") else: initial = actor_name[0].upper() if actor_name else "?" avatar = sx_call( "federation-avatar-placeholder", cls="w-10 h-10 rounded-full bg-stone-300 flex items-center justify-center text-stone-600 font-bold text-sm", initial=initial, ) domain_str = f"@{escape(actor_domain)}" if actor_domain else "" time_str = published.strftime("%b %d, %H:%M") if published else "" if summary: content_sx = sx_call( "federation-content-cw", summary=str(escape(summary)), content=content, ) else: content_sx = sx_call("federation-content-plain", content=content) original_sx = "" if url and post_type == "remote": original_sx = sx_call("federation-original-link", url=url) interactions_sx = "" if actor: oid = getattr(item, "object_id", "") or "" safe_id = oid.replace("/", "_").replace(":", "_") interactions_sx = sx_call( "federation-interactions-wrap", id=f"interactions-{safe_id}", buttons=SxExpr(_interaction_buttons_sx(item, actor)), ) return sx_call( "federation-post-card", boost=SxExpr(boost_sx) if boost_sx else None, avatar=SxExpr(avatar), actor_name=str(escape(actor_name)), actor_username=str(escape(actor_username)), domain=domain_str, time=time_str, content=SxExpr(content_sx), original=SxExpr(original_sx) if original_sx else None, interactions=SxExpr(interactions_sx) if interactions_sx else None, ) # --------------------------------------------------------------------------- # Timeline items (pagination fragment) # --------------------------------------------------------------------------- def _timeline_items_sx(items: list, timeline_type: str, actor: Any, actor_id: int | None = None) -> str: """Render timeline items with infinite scroll sentinel.""" from quart import url_for parts = [_post_card_sx(item, actor) for item in items] if items: last = items[-1] before = last.published.isoformat() if last.published else "" if timeline_type == "actor" and actor_id is not None: next_url = url_for("social.actor_timeline_page", id=actor_id, before=before) else: next_url = url_for(f"social.{timeline_type}_timeline_page", before=before) parts.append(sx_call("federation-scroll-sentinel", url=next_url)) return "(<> " + " ".join(parts) + ")" if parts else "" # --------------------------------------------------------------------------- # Search results (pagination fragment) # --------------------------------------------------------------------------- def _actor_card_sx(a: Any, actor: Any, followed_urls: set, *, list_type: str = "search") -> str: """Render a single actor card with follow/unfollow button.""" from shared.browser.app.csrf import generate_csrf_token from quart import url_for csrf = generate_csrf_token() display_name = getattr(a, "display_name", None) or getattr(a, "preferred_username", "") username = getattr(a, "preferred_username", "") domain = getattr(a, "domain", "") icon_url = getattr(a, "icon_url", None) actor_url = getattr(a, "actor_url", "") summary = getattr(a, "summary", None) aid = getattr(a, "id", None) safe_id = actor_url.replace("/", "_").replace(":", "_") if icon_url: avatar = sx_call("federation-actor-avatar-img", src=icon_url, cls="w-12 h-12 rounded-full") else: initial = (display_name or username)[0].upper() if (display_name or username) else "?" avatar = sx_call( "federation-actor-avatar-placeholder", cls="w-12 h-12 rounded-full bg-stone-300 flex items-center justify-center text-stone-600 font-bold", initial=initial, ) # Name link if (list_type in ("following", "search")) and aid: name_sx = sx_call( "federation-actor-name-link", href=url_for("social.actor_timeline", id=aid), name=str(escape(display_name)), ) else: name_sx = sx_call( "federation-actor-name-link-external", href=f"https://{domain}/@{username}", name=str(escape(display_name)), ) summary_sx = sx_call("federation-actor-summary", summary=summary) if summary else "" # Follow/unfollow button button_sx = "" if actor: is_followed = actor_url in (followed_urls or set()) if list_type == "following" or is_followed: button_sx = sx_call( "federation-unfollow-button", action=url_for("social.unfollow"), csrf=csrf, actor_url=actor_url, ) else: label = "Follow Back" if list_type == "followers" else "Follow" button_sx = sx_call( "federation-follow-button", action=url_for("social.follow"), csrf=csrf, actor_url=actor_url, label=label, ) return sx_call( "federation-actor-card", cls="bg-white rounded-lg shadow-sm border border-stone-200 p-4 mb-3 flex items-center gap-4", id=f"actor-{safe_id}", avatar=SxExpr(avatar), name=SxExpr(name_sx), username=str(escape(username)), domain=str(escape(domain)), summary=SxExpr(summary_sx) if summary_sx else None, button=SxExpr(button_sx) if button_sx else None, ) def _search_results_sx(actors: list, query: str, page: int, followed_urls: set, actor: Any) -> str: """Render search results with pagination sentinel.""" from quart import url_for parts = [_actor_card_sx(a, actor, followed_urls, list_type="search") for a in actors] if len(actors) >= 20: next_url = url_for("social.search_page", q=query, page=page + 1) parts.append(sx_call("federation-scroll-sentinel", url=next_url)) return "(<> " + " ".join(parts) + ")" if parts else "" def _actor_list_items_sx(actors: list, page: int, list_type: str, followed_urls: set, actor: Any) -> str: """Render actor list items (following/followers) with pagination sentinel.""" from quart import url_for parts = [_actor_card_sx(a, actor, followed_urls, list_type=list_type) for a in actors] if len(actors) >= 20: next_url = url_for(f"social.{list_type}_list_page", page=page + 1) parts.append(sx_call("federation-scroll-sentinel", url=next_url)) return "(<> " + " ".join(parts) + ")" if parts else "" # --------------------------------------------------------------------------- # Notification card # --------------------------------------------------------------------------- def _notification_sx(notif: Any) -> str: """Render a single notification.""" from_name = getattr(notif, "from_actor_name", "?") from_username = getattr(notif, "from_actor_username", "") from_domain = getattr(notif, "from_actor_domain", "") from_icon = getattr(notif, "from_actor_icon", None) ntype = getattr(notif, "notification_type", "") preview = getattr(notif, "target_content_preview", None) created = getattr(notif, "created_at", None) read = getattr(notif, "read", True) app_domain = getattr(notif, "app_domain", "") border = " border-l-4 border-l-stone-400" if not read else "" if from_icon: avatar = sx_call("federation-avatar-img", src=from_icon, cls="w-8 h-8 rounded-full") else: initial = from_name[0].upper() if from_name else "?" avatar = sx_call( "federation-avatar-placeholder", cls="w-8 h-8 rounded-full bg-stone-300 flex items-center justify-center text-stone-600 font-bold text-xs", initial=initial, ) domain_str = f"@{escape(from_domain)}" if from_domain else "" type_map = { "follow": "followed you", "like": "liked your post", "boost": "boosted your post", "mention": "mentioned you", "reply": "replied to your post", } action = type_map.get(ntype, "") if ntype == "follow" and app_domain and app_domain != "federation": action += f" on {escape(app_domain)}" preview_sx = sx_call( "federation-notification-preview", preview=str(escape(preview)), ) if preview else "" time_str = created.strftime("%b %d, %H:%M") if created else "" return sx_call( "federation-notification-card", cls=f"bg-white rounded-lg shadow-sm border border-stone-200 p-4{border}", avatar=SxExpr(avatar), from_name=str(escape(from_name)), from_username=str(escape(from_username)), from_domain=domain_str, action_text=action, preview=SxExpr(preview_sx) if preview_sx else None, time=time_str, ) # --------------------------------------------------------------------------- # Public API: Home page # --------------------------------------------------------------------------- async def render_federation_home(ctx: dict) -> str: """Full page: federation home (minimal).""" hdr = root_header_sx(ctx) return full_page_sx(ctx, header_rows=hdr) # --------------------------------------------------------------------------- # Public API: Login # --------------------------------------------------------------------------- async def render_login_page(ctx: dict) -> str: """Full page: federation login form.""" from shared.browser.app.csrf import generate_csrf_token from quart import url_for error = ctx.get("error", "") email = ctx.get("email", "") action = url_for("auth.start_login") csrf = generate_csrf_token() error_sx = sx_call("federation-error-banner", error=error) if error else "" content = sx_call( "federation-login-form", error=SxExpr(error_sx) if error_sx else None, action=action, csrf=csrf, email=str(escape(email)), ) return _social_page(ctx, None, content=content, title="Login \u2014 Rose Ash") async def render_check_email_page(ctx: dict) -> str: """Full page: check email after magic link sent.""" email = ctx.get("email", "") email_error = ctx.get("email_error") error_sx = sx_call( "federation-check-email-error", error=str(escape(email_error)), ) if email_error else "" content = sx_call( "federation-check-email", email=str(escape(email)), error=SxExpr(error_sx) if error_sx else None, ) return _social_page(ctx, None, content=content, title="Check your email \u2014 Rose Ash") # --------------------------------------------------------------------------- # Public API: Timeline # --------------------------------------------------------------------------- async def render_timeline_page(ctx: dict, items: list, timeline_type: str, actor: Any) -> str: """Full page: timeline (home or public).""" from quart import url_for label = "Home" if timeline_type == "home" else "Public" compose_sx = "" if actor: compose_url = url_for("social.compose_form") compose_sx = sx_call("federation-compose-button", url=compose_url) timeline_sx = _timeline_items_sx(items, timeline_type, actor) content = sx_call( "federation-timeline-page", label=label, compose=SxExpr(compose_sx) if compose_sx else None, timeline=SxExpr(timeline_sx) if timeline_sx else None, ) return _social_page(ctx, actor, content=content, title=f"{label} Timeline \u2014 Rose Ash") async def render_timeline_items(items: list, timeline_type: str, actor: Any, actor_id: int | None = None) -> str: """Pagination fragment: timeline items.""" return _timeline_items_sx(items, timeline_type, actor, actor_id) # --------------------------------------------------------------------------- # Public API: Compose # --------------------------------------------------------------------------- async def render_compose_page(ctx: dict, actor: Any, reply_to: str | None) -> str: """Full page: compose form.""" from shared.browser.app.csrf import generate_csrf_token from quart import url_for csrf = generate_csrf_token() action = url_for("social.compose_submit") reply_sx = "" if reply_to: reply_sx = sx_call( "federation-compose-reply", reply_to=str(escape(reply_to)), ) content = sx_call( "federation-compose-form", action=action, csrf=csrf, reply=SxExpr(reply_sx) if reply_sx else None, ) return _social_page(ctx, actor, content=content, title="Compose \u2014 Rose Ash") # --------------------------------------------------------------------------- # Public API: Search # --------------------------------------------------------------------------- async def render_search_page(ctx: dict, query: str, actors: list, total: int, page: int, followed_urls: set, actor: Any) -> str: """Full page: search.""" from quart import url_for search_url = url_for("social.search") search_page_url = url_for("social.search_page") results_sx = _search_results_sx(actors, query, page, followed_urls, actor) info_sx = "" if query and total: s = "s" if total != 1 else "" info_sx = sx_call( "federation-search-info", cls="text-sm text-stone-500 mb-4", text=f"{total} result{s} for {escape(query)}", ) elif query: info_sx = sx_call( "federation-search-info", cls="text-stone-500 mb-4", text=f"No results found for {escape(query)}", ) content = sx_call( "federation-search-page", search_url=search_url, search_page_url=search_page_url, query=str(escape(query)), info=SxExpr(info_sx) if info_sx else None, results=SxExpr(results_sx) if results_sx else None, ) return _social_page(ctx, actor, content=content, title="Search \u2014 Rose Ash") async def render_search_results(actors: list, query: str, page: int, followed_urls: set, actor: Any) -> str: """Pagination fragment: search results.""" return _search_results_sx(actors, query, page, followed_urls, actor) # --------------------------------------------------------------------------- # Public API: Following / Followers # --------------------------------------------------------------------------- async def render_following_page(ctx: dict, actors: list, total: int, actor: Any) -> str: """Full page: following list.""" items_sx = _actor_list_items_sx(actors, 1, "following", set(), actor) content = sx_call( "federation-actor-list-page", title="Following", count_str=f"({total})", items=SxExpr(items_sx) if items_sx else None, ) return _social_page(ctx, actor, content=content, title="Following \u2014 Rose Ash") async def render_following_items(actors: list, page: int, actor: Any) -> str: """Pagination fragment: following items.""" return _actor_list_items_sx(actors, page, "following", set(), actor) async def render_followers_page(ctx: dict, actors: list, total: int, followed_urls: set, actor: Any) -> str: """Full page: followers list.""" items_sx = _actor_list_items_sx(actors, 1, "followers", followed_urls, actor) content = sx_call( "federation-actor-list-page", title="Followers", count_str=f"({total})", items=SxExpr(items_sx) if items_sx else None, ) return _social_page(ctx, actor, content=content, title="Followers \u2014 Rose Ash") async def render_followers_items(actors: list, page: int, followed_urls: set, actor: Any) -> str: """Pagination fragment: followers items.""" return _actor_list_items_sx(actors, page, "followers", followed_urls, actor) # --------------------------------------------------------------------------- # Public API: Actor timeline # --------------------------------------------------------------------------- async def render_actor_timeline_page(ctx: dict, remote_actor: Any, items: list, is_following: bool, actor: Any) -> str: """Full page: remote actor timeline.""" from shared.browser.app.csrf import generate_csrf_token from quart import url_for csrf = generate_csrf_token() display_name = remote_actor.display_name or remote_actor.preferred_username icon_url = getattr(remote_actor, "icon_url", None) summary = getattr(remote_actor, "summary", None) actor_url = getattr(remote_actor, "actor_url", "") if icon_url: avatar = sx_call("federation-avatar-img", src=icon_url, cls="w-16 h-16 rounded-full") else: initial = display_name[0].upper() if display_name else "?" avatar = sx_call( "federation-avatar-placeholder", cls="w-16 h-16 rounded-full bg-stone-300 flex items-center justify-center text-stone-600 font-bold text-xl", initial=initial, ) summary_sx = sx_call("federation-profile-summary", summary=summary) if summary else "" follow_sx = "" if actor: if is_following: follow_sx = sx_call( "federation-follow-form", action=url_for("social.unfollow"), csrf=csrf, actor_url=actor_url, label="Unfollow", cls="border border-stone-300 rounded px-4 py-2 hover:bg-stone-100", ) else: follow_sx = sx_call( "federation-follow-form", action=url_for("social.follow"), csrf=csrf, actor_url=actor_url, label="Follow", cls="bg-stone-800 text-white rounded px-4 py-2 hover:bg-stone-700", ) timeline_sx = _timeline_items_sx(items, "actor", actor, remote_actor.id) header_sx = sx_call( "federation-actor-profile-header", avatar=SxExpr(avatar), display_name=str(escape(display_name)), username=str(escape(remote_actor.preferred_username)), domain=str(escape(remote_actor.domain)), summary=SxExpr(summary_sx) if summary_sx else None, follow=SxExpr(follow_sx) if follow_sx else None, ) content = sx_call( "federation-actor-timeline-layout", header=SxExpr(header_sx), timeline=SxExpr(timeline_sx) if timeline_sx else None, ) return _social_page(ctx, actor, content=content, title=f"{display_name} \u2014 Rose Ash") async def render_actor_timeline_items(items: list, actor_id: int, actor: Any) -> str: """Pagination fragment: actor timeline items.""" return _timeline_items_sx(items, "actor", actor, actor_id) # --------------------------------------------------------------------------- # Public API: Notifications # --------------------------------------------------------------------------- async def render_notifications_page(ctx: dict, notifications: list, actor: Any) -> str: """Full page: notifications.""" if not notifications: notif_sx = sx_call("federation-notifications-empty") else: items_sx = "(<> " + " ".join(_notification_sx(n) for n in notifications) + ")" notif_sx = sx_call( "federation-notifications-list", items=SxExpr(items_sx), ) content = sx_call("federation-notifications-page", notifs=SxExpr(notif_sx)) return _social_page(ctx, actor, content=content, title="Notifications \u2014 Rose Ash") # --------------------------------------------------------------------------- # Public API: Choose username # --------------------------------------------------------------------------- async def render_choose_username_page(ctx: dict) -> str: """Full page: choose username form.""" from shared.browser.app.csrf import generate_csrf_token from quart import url_for from shared.config import config csrf = generate_csrf_token() error = ctx.get("error", "") username = ctx.get("username", "") ap_domain = config().get("ap_domain", "rose-ash.com") check_url = url_for("identity.check_username") actor = ctx.get("actor") error_sx = sx_call("federation-error-banner", error=error) if error else "" content = sx_call( "federation-choose-username", domain=str(escape(ap_domain)), error=SxExpr(error_sx) if error_sx else None, csrf=csrf, username=str(escape(username)), check_url=check_url, ) return _social_page(ctx, actor, content=content, title="Choose Username \u2014 Rose Ash") # --------------------------------------------------------------------------- # Public API: Actor profile # --------------------------------------------------------------------------- async def render_profile_page(ctx: dict, actor: Any, activities: list, total: int) -> str: """Full page: actor profile.""" from shared.config import config ap_domain = config().get("ap_domain", "rose-ash.com") display_name = actor.display_name or actor.preferred_username summary_sx = sx_call( "federation-profile-summary-text", text=str(escape(actor.summary)), ) if actor.summary else "" activities_sx = "" if activities: parts = [] for a in activities: published = a.published.strftime("%Y-%m-%d %H:%M") if a.published else "" obj_type_sx = sx_call( "federation-activity-obj-type", obj_type=a.object_type, ) if a.object_type else "" parts.append(sx_call( "federation-activity-card", activity_type=a.activity_type, published=published, obj_type=SxExpr(obj_type_sx) if obj_type_sx else None, )) items_sx = "(<> " + " ".join(parts) + ")" activities_sx = sx_call("federation-activities-list", items=SxExpr(items_sx)) else: activities_sx = sx_call("federation-activities-empty") content = sx_call( "federation-profile-page", display_name=str(escape(display_name)), username=str(escape(actor.preferred_username)), domain=str(escape(ap_domain)), summary=SxExpr(summary_sx) if summary_sx else None, activities_heading=f"Activities ({total})", activities=SxExpr(activities_sx), ) return _social_page(ctx, actor, content=content, title=f"@{actor.preferred_username} \u2014 Rose Ash") # --------------------------------------------------------------------------- # Public API: POST handler fragment renderers # --------------------------------------------------------------------------- def render_interaction_buttons(object_id: str, author_inbox: str, like_count: int, boost_count: int, liked_by_me: bool, boosted_by_me: bool, actor: Any) -> str: """Render interaction buttons fragment for HTMX POST response.""" from types import SimpleNamespace item = SimpleNamespace( object_id=object_id, author_inbox=author_inbox, like_count=like_count, boost_count=boost_count, liked_by_me=liked_by_me, boosted_by_me=boosted_by_me, ) return _interaction_buttons_sx(item, actor) def render_actor_card(actor_dto: Any, actor: Any, followed_urls: set, *, list_type: str = "following") -> str: """Render a single actor card fragment for HTMX POST response.""" return _actor_card_sx(actor_dto, actor, followed_urls, list_type=list_type)