microblog.pub/app/actor.py

import typing
from dataclasses import dataclass
from typing import Union
from urllib.parse import urlparse

from sqlalchemy import select
from sqlalchemy.orm import joinedload

from app import activitypub as ap
from app import media
from app.database import AsyncSession

if typing.TYPE_CHECKING:
    from app.models import Actor as ActorModel


def _handle(raw_actor: ap.RawObject) -> str:
    ap_id = ap.get_id(raw_actor["id"])
    domain = urlparse(ap_id)
    if not domain.hostname:
        raise ValueError(f"Invalid actor ID {ap_id}")

    return f'@{raw_actor["preferredUsername"]}@{domain.hostname}'  # type: ignore


class Actor:
    @property
    def ap_actor(self) -> ap.RawObject:
        raise NotImplementedError()

    @property
    def ap_id(self) -> str:
        return ap.get_id(self.ap_actor["id"])

    @property
    def name(self) -> str | None:
        return self.ap_actor.get("name")

    @property
    def summary(self) -> str | None:
        return self.ap_actor.get("summary")

    @property
    def url(self) -> str | None:
        return self.ap_actor.get("url") or self.ap_actor["id"]

    @property
    def preferred_username(self) -> str:
        return self.ap_actor["preferredUsername"]

    @property
    def display_name(self) -> str:
        return self.name or self.preferred_username

    @property
    def handle(self) -> str:
        return _handle(self.ap_actor)

    @property
    def ap_type(self) -> str:
        raise NotImplementedError()

    @property
    def inbox_url(self) -> str:
        return self.ap_actor["inbox"]

    @property
    def shared_inbox_url(self) -> str | None:
        return self.ap_actor.get("endpoints", {}).get("sharedInbox")

    @property
    def icon_url(self) -> str | None:
        return self.ap_actor.get("icon", {}).get("url")

    @property
    def icon_media_type(self) -> str | None:
        return self.ap_actor.get("icon", {}).get("mediaType")

    @property
    def public_key_as_pem(self) -> str:
        return self.ap_actor["publicKey"]["publicKeyPem"]

    @property
    def public_key_id(self) -> str:
        return self.ap_actor["publicKey"]["id"]

    @property
    def proxied_icon_url(self) -> str:
        if self.icon_url:
            return media.proxied_media_url(self.icon_url)
        else:
            return "/static/nopic.png"

    @property
    def resized_icon_url(self) -> str:
        if self.icon_url:
            return media.resized_media_url(self.icon_url, 50)
        else:
            return "/static/nopic.png"

    @property
    def tags(self) -> list[ap.RawObject]:
        return self.ap_actor.get("tag", [])

    @property
    def followers_collection_id(self) -> str:
        return self.ap_actor["followers"]


class RemoteActor(Actor):
    def __init__(self, ap_actor: ap.RawObject) -> None:
        if (ap_type := ap_actor.get("type")) not in ap.ACTOR_TYPES:
            raise ValueError(f"Unexpected actor type: {ap_type}")

        self._ap_actor = ap_actor
        self._ap_type = ap_type

    @property
    def ap_actor(self) -> ap.RawObject:
        return self._ap_actor

    @property
    def ap_type(self) -> str:
        return self._ap_type

    @property
    def is_from_db(self) -> bool:
        return False


LOCAL_ACTOR = RemoteActor(ap_actor=ap.ME)


async def save_actor(db_session: AsyncSession, ap_actor: ap.RawObject) -> "ActorModel":
    from app import models

    if ap_type := ap_actor.get("type") not in ap.ACTOR_TYPES:
        raise ValueError(f"Invalid type {ap_type} for actor {ap_actor}")

    actor = models.Actor(
        ap_id=ap_actor["id"],
        ap_actor=ap_actor,
        ap_type=ap_actor["type"],
        handle=_handle(ap_actor),
    )
    db_session.add(actor)
    await db_session.commit()
    await db_session.refresh(actor)
    return actor


async def fetch_actor(db_session: AsyncSession, actor_id: str) -> "ActorModel":
    from app import models

    existing_actor = (
        await db_session.scalars(
            select(models.Actor).where(models.Actor.ap_id == actor_id)
        )
    ).one_or_none()
    if existing_actor:
        return existing_actor

    ap_actor = await ap.fetch(actor_id)
    return await save_actor(db_session, ap_actor)


@dataclass
class ActorMetadata:
    ap_actor_id: str
    is_following: bool
    is_follower: bool
    is_follow_request_sent: bool
    outbox_follow_ap_id: str | None
    inbox_follow_ap_id: str | None


ActorsMetadata = dict[str, ActorMetadata]


async def get_actors_metadata(
    db_session: AsyncSession,
    actors: list[Union["ActorModel", "RemoteActor"]],
) -> ActorsMetadata:
    from app import models

    ap_actor_ids = [actor.ap_id for actor in actors]
    followers = {
        follower.ap_actor_id: follower.inbox_object.ap_id
        for follower in (
            await db_session.scalars(
                select(models.Follower)
                .where(models.Follower.ap_actor_id.in_(ap_actor_ids))
                .options(joinedload(models.Follower.inbox_object))
            )
        )
        .unique()
        .all()
    }
    following = {
        following.ap_actor_id
        for following in await db_session.execute(
            select(models.Following.ap_actor_id).where(
                models.Following.ap_actor_id.in_(ap_actor_ids)
            )
        )
    }
    sent_follow_requests = {
        follow_req.ap_object["object"]: follow_req.ap_id
        for follow_req in await db_session.execute(
            select(models.OutboxObject.ap_object, models.OutboxObject.ap_id).where(
                models.OutboxObject.ap_type == "Follow",
                models.OutboxObject.undone_by_outbox_object_id.is_(None),
            )
        )
    }
    idx: ActorsMetadata = {}
    for actor in actors:
        if not actor.ap_id:
            raise ValueError("Should never happen")
        idx[actor.ap_id] = ActorMetadata(
            ap_actor_id=actor.ap_id,
            is_following=actor.ap_id in following,
            is_follower=actor.ap_id in followers,
            is_follow_request_sent=actor.ap_id in sent_follow_requests,
            outbox_follow_ap_id=sent_follow_requests.get(actor.ap_id),
            inbox_follow_ap_id=followers.get(actor.ap_id),
        )
    return idx