Cleanup and improved webmentions support

app/webmentions.py

@@ -1,8 +1,3 @@
-from dataclasses import asdict
-from dataclasses import dataclass
-from typing import Any
-from typing import Optional
-
 from bs4 import BeautifulSoup  # type: ignore
 from fastapi import APIRouter
 from fastapi import Depends
@@ -10,54 +5,19 @@ from fastapi import HTTPException
 from fastapi import Request
 from fastapi.responses import JSONResponse
 from loguru import logger
+from sqlalchemy import select
 
+from app import models
 from app.boxes import get_outbox_object_by_ap_id
 from app.database import AsyncSession
 from app.database import get_db_session
-from app.database import now
 from app.utils import microformats
 from app.utils.url import check_url
 from app.utils.url import is_url_valid
-from app.utils.url import make_abs
 
 router = APIRouter()
 
 
-@dataclass
-class Webmention:
-    actor_icon_url: str
-    actor_name: str
-    url: str
-    received_at: str
-
-    @classmethod
-    def from_microformats(
-        cls, items: list[dict[str, Any]], url: str
-    ) -> Optional["Webmention"]:
-        for item in items:
-            if item["type"][0] == "h-card":
-                return cls(
-                    actor_icon_url=make_abs(
-                        item["properties"]["photo"][0], url
-                    ),  # type: ignore
-                    actor_name=item["properties"]["name"][0],
-                    url=url,
-                    received_at=now().isoformat(),
-                )
-            if item["type"][0] == "h-entry":
-                author = item["properties"]["author"][0]
-                return cls(
-                    actor_icon_url=make_abs(
-                        author["properties"]["photo"][0], url
-                    ),  # type: ignore
-                    actor_name=author["properties"]["name"][0],
-                    url=url,
-                    received_at=now().isoformat(),
-                )
-
-        return None
-
-
 def is_source_containing_target(source_html: str, target_url: str) -> bool:
     soup = BeautifulSoup(source_html, "html5lib")
     for link in soup.find_all("a"):
@@ -92,40 +52,64 @@ async def webmention_endpoint(
 
     logger.info(f"Received webmention {source=} {target=}")
 
+    existing_webmention_in_db = (
+        await db_session.execute(
+            select(models.Webmention).where(
+                models.Webmention.source == source,
+                models.Webmention.target == target,
+            )
+        )
+    ).scalar_one_or_none()
+    if existing_webmention_in_db:
+        logger.info("Found existing Webmention, will try to update or delete")
+
     mentioned_object = await get_outbox_object_by_ap_id(db_session, target)
     if not mentioned_object:
         logger.info(f"Invalid target {target=}")
+
+        if existing_webmention_in_db:
+            logger.info("Deleting existing Webmention")
+            existing_webmention_in_db.is_deleted = True
+            await db_session.commit()
         raise HTTPException(status_code=400, detail="Invalid target")
 
     maybe_data_and_html = await microformats.fetch_and_parse(source)
     if not maybe_data_and_html:
         logger.info("failed to fetch source")
+
+        if existing_webmention_in_db:
+            logger.info("Deleting existing Webmention")
+            mentioned_object.webmentions_count = mentioned_object.webmentions_count - 1
+            existing_webmention_in_db.is_deleted = True
+            await db_session.commit()
         raise HTTPException(status_code=400, detail="failed to fetch source")
 
     data, html = maybe_data_and_html
 
     if not is_source_containing_target(html, target):
         logger.warning("target not found in source")
+
+        if existing_webmention_in_db:
+            logger.info("Deleting existing Webmention")
+            mentioned_object.webmentions_count = mentioned_object.webmentions_count - 1
+            existing_webmention_in_db.is_deleted = True
+            await db_session.commit()
+
         raise HTTPException(status_code=400, detail="target not found in source")
 
-    try:
-        webmention = Webmention.from_microformats(data["items"], source)
-        if not webmention:
-            raise ValueError("Failed to fetch target data")
-    except Exception:
-        logger.warning("Failed build Webmention for {source=} with {data=}")
-        return JSONResponse(content={}, status_code=200)
-
-    logger.info(f"{webmention=}")
-
-    if mentioned_object.webmentions is None:
-        mentioned_object.webmentions = [asdict(webmention)]
+    if existing_webmention_in_db:
+        existing_webmention_in_db.is_deleted = False
+        existing_webmention_in_db.source_microformats = data
     else:
-        mentioned_object.webmentions = [asdict(webmention)] + [
-            wm  # type: ignore
-            for wm in mentioned_object.webmentions  # type: ignore
-            if wm["url"] != source  # type: ignore
-        ]
+        new_webmention = models.Webmention(
+            source=source,
+            target=target,
+            source_microformats=data,
+            outbox_object_id=mentioned_object.id,
+        )
+        db_session.add(new_webmention)
+
+        mentioned_object.webmentions_count = mentioned_object.webmentions_count + 1
 
     await db_session.commit()