rose-ash/federation/sxc/pages/__init__.py

"""Federation defpage setup — registers layouts, page helpers, and loads .sx pages."""
from __future__ import annotations

from typing import Any


def setup_federation_pages() -> None:
    """Register federation-specific layouts, page helpers, and load page definitions."""
    _register_federation_layouts()
    _register_federation_helpers()
    _load_federation_page_files()


def _load_federation_page_files() -> None:
    import os
    from shared.sx.pages import load_page_dir
    load_page_dir(os.path.dirname(__file__), "federation")


# ---------------------------------------------------------------------------
# Layouts
# ---------------------------------------------------------------------------

def _register_federation_layouts() -> None:
    from shared.sx.layouts import register_custom_layout
    register_custom_layout("social", _social_full, _social_oob)


def _social_full(ctx: dict, **kw: Any) -> str:
    from shared.sx.helpers import root_header_sx, header_child_sx, sx_call, SxExpr
    from shared.sx.parser import serialize

    actor = ctx.get("actor")
    actor_data = _serialize_actor(actor) if actor else None
    nav = sx_call("federation-social-nav",
        actor=SxExpr(serialize(actor_data)) if actor_data else None)
    social_hdr = sx_call("federation-social-header", nav=SxExpr(nav))
    root_hdr = root_header_sx(ctx)
    child = header_child_sx(social_hdr)
    return "(<> " + root_hdr + " " + child + ")"


def _social_oob(ctx: dict, **kw: Any) -> str:
    from shared.sx.helpers import root_header_sx, sx_call, SxExpr
    from shared.sx.parser import serialize

    actor = ctx.get("actor")
    actor_data = _serialize_actor(actor) if actor else None
    nav = sx_call("federation-social-nav",
        actor=SxExpr(serialize(actor_data)) if actor_data else None)
    social_hdr = sx_call("federation-social-header", nav=SxExpr(nav))
    child_oob = sx_call("oob-header-sx",
                         parent_id="root-header-child",
                         row=SxExpr(social_hdr))
    root_hdr_oob = root_header_sx(ctx, oob=True)
    return "(<> " + child_oob + " " + root_hdr_oob + ")"


# ---------------------------------------------------------------------------
# Page helpers
# ---------------------------------------------------------------------------

def _register_federation_helpers() -> None:
    from shared.sx.pages import register_page_helpers

    register_page_helpers("federation", {
        "home-timeline-content": _h_home_timeline_content,
        "public-timeline-content": _h_public_timeline_content,
        "compose-content": _h_compose_content,
        "search-content": _h_search_content,
        "following-content": _h_following_content,
        "followers-content": _h_followers_content,
        "actor-timeline-content": _h_actor_timeline_content,
        "notifications-content": _h_notifications_content,
    })


def _serialize_actor(actor) -> dict | None:
    """Serialize an actor profile to a dict for sx defcomps."""
    if not actor:
        return None
    return {
        "id": actor.id,
        "preferred_username": actor.preferred_username,
        "display_name": getattr(actor, "display_name", None),
        "icon_url": getattr(actor, "icon_url", None),
        "summary": getattr(actor, "summary", None),
        "actor_url": getattr(actor, "actor_url", ""),
        "domain": getattr(actor, "domain", ""),
    }


def _serialize_timeline_item(item) -> dict:
    """Serialize a timeline item DTO to a dict for sx defcomps."""
    published = getattr(item, "published", None)
    return {
        "object_id": getattr(item, "object_id", "") or "",
        "author_inbox": getattr(item, "author_inbox", "") or "",
        "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": published.strftime("%b %d, %H:%M") if published else "",
        "before_cursor": published.isoformat() if published else "",
        "url": getattr(item, "url", None),
        "post_type": getattr(item, "post_type", ""),
        "boosted_by": getattr(item, "boosted_by", None),
        "like_count": getattr(item, "like_count", 0) or 0,
        "boost_count": getattr(item, "boost_count", 0) or 0,
        "liked_by_me": getattr(item, "liked_by_me", False),
        "boosted_by_me": getattr(item, "boosted_by_me", False),
    }


def _serialize_remote_actor(a) -> dict:
    """Serialize a remote actor DTO to a dict for sx defcomps."""
    return {
        "id": getattr(a, "id", None),
        "display_name": getattr(a, "display_name", None) or getattr(a, "preferred_username", ""),
        "preferred_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),
    }


def _get_actor():
    """Return current user's actor or None."""
    from quart import g
    return getattr(g, "_social_actor", None)


def _require_actor():
    """Return current user's actor or abort 403."""
    from quart import abort
    actor = _get_actor()
    if not actor:
        abort(403, "You need to choose a federation username first")
    return actor


async def _h_home_timeline_content(**kw):
    from quart import g
    from shared.services.registry import services
    from shared.sx.helpers import sx_call, SxExpr
    from shared.sx.parser import serialize
    actor = _require_actor()
    items = await services.federation.get_home_timeline(g.s, actor.id)
    return sx_call("federation-timeline-content",
        items=SxExpr(serialize([_serialize_timeline_item(i) for i in items])),
        timeline_type="home",
        actor=SxExpr(serialize(_serialize_actor(actor))))


async def _h_public_timeline_content(**kw):
    from quart import g
    from shared.services.registry import services
    from shared.sx.helpers import sx_call, SxExpr
    from shared.sx.parser import serialize
    actor = _get_actor()
    items = await services.federation.get_public_timeline(g.s)
    return sx_call("federation-timeline-content",
        items=SxExpr(serialize([_serialize_timeline_item(i) for i in items])),
        timeline_type="public",
        actor=SxExpr(serialize(_serialize_actor(actor))) if actor else None)


async def _h_compose_content(**kw):
    from quart import request
    from shared.sx.helpers import sx_call
    _require_actor()
    reply_to = request.args.get("reply_to")
    return sx_call("federation-compose-content",
        reply_to=reply_to or None)


async def _h_search_content(**kw):
    from quart import g, request
    from shared.services.registry import services
    from shared.sx.helpers import sx_call, SxExpr
    from shared.sx.parser import serialize
    actor = _get_actor()
    query = request.args.get("q", "").strip()
    actors_list = []
    total = 0
    followed_urls: set[str] = set()
    if query:
        actors_list, total = await services.federation.search_actors(g.s, query)
        if actor:
            following, _ = await services.federation.get_following(
                g.s, actor.preferred_username, page=1, per_page=1000,
            )
            followed_urls = {a.actor_url for a in following}
    return sx_call("federation-search-content",
        query=query,
        actors=SxExpr(serialize([_serialize_remote_actor(a) for a in actors_list])),
        total=total,
        followed_urls=SxExpr(serialize(list(followed_urls))),
        actor=SxExpr(serialize(_serialize_actor(actor))) if actor else None)


async def _h_following_content(**kw):
    from quart import g
    from shared.services.registry import services
    from shared.sx.helpers import sx_call, SxExpr
    from shared.sx.parser import serialize
    actor = _require_actor()
    actors_list, total = await services.federation.get_following(
        g.s, actor.preferred_username,
    )
    return sx_call("federation-following-content",
        actors=SxExpr(serialize([_serialize_remote_actor(a) for a in actors_list])),
        total=total,
        actor=SxExpr(serialize(_serialize_actor(actor))))


async def _h_followers_content(**kw):
    from quart import g
    from shared.services.registry import services
    from shared.sx.helpers import sx_call, SxExpr
    from shared.sx.parser import serialize
    actor = _require_actor()
    actors_list, total = await services.federation.get_followers_paginated(
        g.s, actor.preferred_username,
    )
    following, _ = await services.federation.get_following(
        g.s, actor.preferred_username, page=1, per_page=1000,
    )
    followed_urls = {a.actor_url for a in following}
    return sx_call("federation-followers-content",
        actors=SxExpr(serialize([_serialize_remote_actor(a) for a in actors_list])),
        total=total,
        followed_urls=SxExpr(serialize(list(followed_urls))),
        actor=SxExpr(serialize(_serialize_actor(actor))))


async def _h_actor_timeline_content(id=None, **kw):
    from quart import g, abort
    from shared.services.registry import services
    from shared.sx.helpers import sx_call, SxExpr
    from shared.sx.parser import serialize
    actor = _get_actor()
    actor_id = id
    from shared.models.federation import RemoteActor
    from sqlalchemy import select as sa_select
    remote = (
        await g.s.execute(
            sa_select(RemoteActor).where(RemoteActor.id == actor_id)
        )
    ).scalar_one_or_none()
    if not remote:
        abort(404)
    from shared.services.federation_impl import _remote_actor_to_dto
    remote_dto = _remote_actor_to_dto(remote)
    items = await services.federation.get_actor_timeline(g.s, actor_id)
    is_following = False
    if actor:
        from shared.models.federation import APFollowing
        existing = (
            await g.s.execute(
                sa_select(APFollowing).where(
                    APFollowing.actor_profile_id == actor.id,
                    APFollowing.remote_actor_id == actor_id,
                )
            )
        ).scalar_one_or_none()
        is_following = existing is not None
    return sx_call("federation-actor-timeline-content",
        remote_actor=SxExpr(serialize(_serialize_remote_actor(remote_dto))),
        items=SxExpr(serialize([_serialize_timeline_item(i) for i in items])),
        is_following=is_following,
        actor=SxExpr(serialize(_serialize_actor(actor))) if actor else None)


async def _h_notifications_content(**kw):
    from quart import g
    from shared.services.registry import services
    from shared.sx.helpers import sx_call, SxExpr
    from shared.sx.parser import serialize
    actor = _require_actor()
    items = await services.federation.get_notifications(g.s, actor.id)
    await services.federation.mark_notifications_read(g.s, actor.id)

    notif_dicts = []
    for n in items:
        created = getattr(n, "created_at", None)
        notif_dicts.append({
            "from_actor_name": getattr(n, "from_actor_name", "?"),
            "from_actor_username": getattr(n, "from_actor_username", ""),
            "from_actor_domain": getattr(n, "from_actor_domain", ""),
            "from_actor_icon": getattr(n, "from_actor_icon", None),
            "notification_type": getattr(n, "notification_type", ""),
            "target_content_preview": getattr(n, "target_content_preview", None),
            "created_at_formatted": created.strftime("%b %d, %H:%M") if created else "",
            "read": getattr(n, "read", True),
            "app_domain": getattr(n, "app_domain", ""),
        })
    return sx_call("federation-notifications-content",
        notifications=SxExpr(serialize(notif_dicts)))