foxhole/app/boxes.py

#!/usr/bin/env python3
from app import models
from app.database import AsyncSession
from app.activitypub import ME
from loguru import logger
from uuid import uuid4
from app.config import MANUALLY_APPROVES_FOLLOWERS
from app.config import BASE_URL

import app.activitypub as ap
import uuid


def allocate_outbox_id() -> str:
    return str(uuid.uuid4())


def build_object_id(id) -> str:
    return f"{BASE_URL}/tail/{id}"

async def save_incoming(
        db_session: AsyncSession,
        payload: dict,
) -> models.IncomingActivity | None:
    ap_id: str
    if "@context" not in payload:
        logger.warning(f"invalid object: {payload}")
        return None

    if "id" in payload:
        ap_id = payload["id"]
    else:
        ap_id = str(uuid4())

    incoming_activity = models.IncomingActivity(
        ap_id=ap_id,
        ap_object=payload,
    )
    await process_incoming(db_session, payload)
    if db_session.add(incoming_activity):
        return incoming_activity

    await db_session.commit()
    await db_session.refresh(incoming_activity)
    return incoming_activity


async def process_incoming(
        db_session: AsyncSession,
        ap_object: dict,
) -> bool:

    if "Follow" == ap_object["type"]:
        await _handle_follow(db_session, ap_object)
        return True

    return False


async def _handle_follow(
        db_session: AsyncSession,
        ap_object: dict,
) -> bool:
    if ME["id"] != ap_object["object"]:
        # await db_session.delete(ap_object)
        logger.warning("no match follow object!" + ap_object["object"])
        return False

    if MANUALLY_APPROVES_FOLLOWERS:
        # TODO
        return False

    await _send_accept(db_session, ap_object)
    return True


async def _send_accept(
        db_session: AsyncSession,
        ap_object: dict,
) -> None :

    reply_id = allocate_outbox_id()

    url = ap_object["actor"] + "/inbox"
    out = {
        "@context": ap.AS_CTX,
        "id": build_object_id(reply_id),
        "type": "Accept",
        "actor": ME["id"],
        "object": ap_object["id"],
    }
    #TODO outcoming
    await ap.post(url, out)