core: add AtUri datatype

core/filters.py

···
-from core.models import Record
+from core import lexicon
+from core.models import AtUri, Record
 
 
 def filter_moderated(
···
     return [
         r
         for r in records
-        if r.uri.split("/")[2] not in banned_dids and r.uri not in hidden_posts
+        if AtUri.parse(r.uri).did not in banned_dids and r.uri not in hidden_posts
     ]

core/models.py

···
 from dataclasses import dataclass
 
 
+class AtUri:
+    """Parsed AT URI with did, collection, and rkey fields."""
+
+    __slots__ = ("did", "collection", "rkey")
+
+    def __init__(self, did: str, collection: str, rkey: str):
+        self.did = did
+        self.collection = collection
+        self.rkey = rkey
+
+    @classmethod
+    def parse(cls, uri: str) -> "AtUri":
+        parts = uri.split("/")
+        return cls(parts[2], parts[3], parts[4])
+
+    def __str__(self) -> str:
+        return f"at://{self.did}/{self.collection}/{self.rkey}"
+
+    def __eq__(self, other):
+        if isinstance(other, AtUri):
+            return str(self) == str(other)
+        if isinstance(other, str):
+            return str(self) == other
+        return NotImplemented
+
+    def __hash__(self):
+        return hash(str(self))
+
+
 # errors
 
 

core/records.py

···
 from core import lexicon
 from core.constellation import get_replies, get_threads
 from core.filters import filter_moderated
-from core.models import BBS, Board, Reply, Thread
+from core.models import AtUri, BBS, Board, Reply, Thread
 from core.slingshot import get_records_batch, resolve_identities_batch
 from core.util import now_iso
 
···
     cursor: str | None = None,
 ) -> tuple[list[Thread], str | None]:
     """Fetch and hydrate threads for a board."""
-    board_uri = f"at://{bbs.identity.did}/{lexicon.BOARD}/{board.slug}"
+    board_uri = str(AtUri(bbs.identity.did, lexicon.BOARD, board.slug))
     backlinks = await get_threads(client, board_uri, cursor=cursor)
     records = await get_records_batch(client, backlinks.records)
     records = filter_moderated(records, bbs.site.banned_dids, bbs.site.hidden_posts)
 
-    dids = [r.uri.split("/")[2] for r in records]
+    parsed = {r.uri: AtUri.parse(r.uri) for r in records}
+    dids = [p.did for p in parsed.values()]
     authors = await resolve_identities_batch(client, dids)
 
     threads = [
···
             title=r.value["title"],
             body=r.value["body"],
             created_at=r.value["createdAt"],
-            author=authors[r.uri.split("/")[2]],
+            author=authors[parsed[r.uri].did],
             updated_at=r.value.get("updatedAt"),
             attachments=r.value.get("attachments"),
         )
         for r in records
-        if r.uri.split("/")[2] in authors
+        if parsed[r.uri].did in authors
     ]
     threads.sort(key=lambda t: t.created_at, reverse=True)
     return threads, backlinks.cursor
···
     records = await get_records_batch(client, backlinks.records)
     records = filter_moderated(records, bbs.site.banned_dids, bbs.site.hidden_posts)
 
-    dids = [r.uri.split("/")[2] for r in records]
+    parsed = {r.uri: AtUri.parse(r.uri) for r in records}
+    dids = [p.did for p in parsed.values()]
     authors = await resolve_identities_batch(client, dids)
 
     replies = [
···
             subject_uri=r.value["subject"],
             body=r.value["body"],
             created_at=r.value["createdAt"],
-            author=authors[r.uri.split("/")[2]],
+            author=authors[parsed[r.uri].did],
             updated_at=r.value.get("updatedAt"),
             attachments=r.value.get("attachments"),
             quote=r.value.get("quote"),
         )
         for r in records
-        if r.uri.split("/")[2] in authors
+        if parsed[r.uri].did in authors
     ]
     replies.sort(key=lambda t: t.created_at)
     return replies, backlinks.cursor
···
         thread_uri = tr["uri"]
         thread_title = tr["value"].get("title", "")
         board_uri = tr["value"].get("board", "")
-        bbs_did = board_uri.split("/")[2] if len(board_uri.split("/")) > 2 else did
+        bbs_did = AtUri.parse(board_uri).did if board_uri else did
         try:
             bbs_authors = await resolve_identities_batch(client, [bbs_did])
             bbs_handle = bbs_authors[bbs_did].handle if bbs_did in bbs_authors else ""
···
         try:
             backlinks = await get_replies(client, thread_uri, limit=50)
             records = await get_records_batch(client, backlinks.records)
-            records = [r for r in records if r.uri.split("/")[2] != did]
+            parsed = {r.uri: AtUri.parse(r.uri) for r in records}
+            records = [r for r in records if parsed[r.uri].did != did]
             if not records:
                 continue
 
-            dids = [r.uri.split("/")[2] for r in records]
+            dids = [parsed[r.uri].did for r in records]
             authors = await resolve_identities_batch(client, dids)
 
             for r in records:
-                author_did = r.uri.split("/")[2]
+                author_did = parsed[r.uri].did
                 if author_did not in authors:
                     continue
                 all_items.append(
···
                 continue
 
             records = await get_records_batch(client, backlinks.records)
-            records = [r for r in records if r.uri.split("/")[2] != did]
+            parsed = {r.uri: AtUri.parse(r.uri) for r in records}
+            records = [r for r in records if parsed[r.uri].did != did]
             if not records:
                 continue
 
-            dids = [r.uri.split("/")[2] for r in records]
+            dids = [parsed[r.uri].did for r in records]
             authors = await resolve_identities_batch(client, dids)
 
             for r in records:
-                author_did = r.uri.split("/")[2]
+                author_did = parsed[r.uri].did
                 if author_did not in authors:
                     continue
                 all_items.append(

core/resolver.py

···
 import httpx
 
 from core.models import (
+    AtUri,
     BBS,
     Board,
     News,
···
         raise NetworkError("Could not reach the network.")
 
     sv = site_record.value
-    site_uri = f"at://{identity.did}/{lexicon.SITE}/self"
+    site_uri = str(AtUri(identity.did, lexicon.SITE, "self"))
 
     # Fetch boards and news backlinks concurrently
     board_slugs = sv["boards"]
···
         news_records = await get_records_batch(client, sysop_news)
     news = [
         News(
-            tid=r.uri.split("/")[-1],
+            tid=AtUri.parse(r.uri).rkey,
             site_uri=r.value["site"],
             title=r.value["title"],
             body=r.value["body"],

tui/screens/activity.py

···
         from core.resolver import resolve_bbs
         from core import lexicon
         from core.slingshot import get_record, resolve_identity
-        from core.models import Thread
+        from core.models import AtUri, Thread
 
-        parts = item["thread_uri"].split("/")
-        thread_did = parts[2]
-        thread_tid = parts[-1]
+        thread = AtUri.parse(item["thread_uri"])
+        thread_did = thread.did
+        thread_tid = thread.rkey
         handle = item.get("bbs_handle") or self.app.user_session.get("handle", "")
 
         client = self.app.http_client

tui/screens/compose.py

···
 from pathlib import Path
 
 from core import lexicon
+from core.models import AtUri
 from core.records import create_thread_record, create_reply_record, upload_blob
 from tui.util import require_session
 
···
             self.notify("Title and body cannot be empty.", severity="error")
             return
 
-        board_uri = f"at://{self.bbs.identity.did}/{lexicon.BOARD}/{self.board.slug}"
+        board_uri = str(AtUri(self.bbs.identity.did, lexicon.BOARD, self.board.slug))
 
         # Handle file attachment
         attachments = []

tui/screens/thread.py

···
 from textual.widgets import Button, Footer
 
 from core import lexicon
-from core.models import BBS, Thread
+from core.models import AtUri, BBS, Thread
 from tui.fetchers import delete_record, fetch_replies
 from tui.util import require_session
 from tui.widgets.post import Post
···
         self._replies_map: dict[str, object] = {}
 
     def compose(self) -> ComposeResult:
-        board_slug = self.thread.board_uri.split("/")[-1]
+        board_slug = AtUri.parse(self.thread.board_uri).rkey
         board_name = next(
             (b.name for b in self.bbs.site.boards if b.slug == board_slug), board_slug
         )

tui/widgets/post.py

···
 from textual.widget import Widget
 from textual.widgets import Static
 
+from core.models import AtUri
 from tui.util import format_datetime
 
 
···
     @property
     def rkey(self) -> str | None:
         if self.record_uri:
-            return self.record_uri.split("/")[-1]
+            return AtUri.parse(self.record_uri).rkey
         return None
 
     def compose(self) -> ComposeResult:

web/routes.py

···
 from quart import Blueprint, current_app, render_template, request
 
 from core.models import (
+    AtUri,
     BBSNotFoundError,
     NetworkError,
     NoBBSError,
···
             {
                 "uri": t.uri,
                 "did": t.author.did,
-                "rkey": t.uri.split("/")[-1],
+                "rkey": AtUri.parse(t.uri).rkey,
                 "handle": t.author.handle,
                 "title": t.title,
                 "body": t.body,
···
         attachments=thread_record.value.get("attachments"),
     )
 
-    board_slug = thread_obj.board_uri.split("/")[-1]
+    board_slug = AtUri.parse(thread_obj.board_uri).rkey
     current_board = next((b for b in bbs.site.boards if b.slug == board_slug), None)
 
     return await render_template(
···
         return {"replies": [], "cursor": None}
 
     # Build a minimal Thread object for hydrate_replies
-    thread_uri = f"at://{did}/{lexicon.THREAD}/{tid}"
+    thread_uri = str(AtUri(did, lexicon.THREAD, tid))
     from core.models import MiniDoc
 
     dummy_thread = Thread(
···
             {
                 "uri": r.uri,
                 "did": r.author.did,
-                "rkey": r.uri.split("/")[-1],
+                "rkey": AtUri.parse(r.uri).rkey,
                 "handle": r.author.handle,
                 "pds_url": r.author.pds or "",
                 "body": r.body,

web/routes_sysop.py

···
 from quart import Blueprint, current_app, redirect, render_template, request
 
 from core import lexicon
+from core.models import AtUri
 from core.util import now_iso
 from web.helpers import get_user
 from web.routes_write import _authed_pds_post, authed_delete_record
···
     # Delete news records (via Constellation backlinks)
     from core.constellation import get_news
 
-    site_uri = f"at://{user['did']}/{lexicon.SITE}/self"
+    site_uri = str(AtUri(user["did"], lexicon.SITE, "self"))
     try:
         backlinks = await get_news(client, site_uri)
         for ref in backlinks.records:
···
         hidden_posts = []
         if bbs.site.hidden_posts:
             hidden_dids = list(
-                {
-                    uri.split("/")[2]
-                    for uri in bbs.site.hidden_posts
-                    if len(uri.split("/")) > 2
-                }
+                {AtUri.parse(uri).did for uri in bbs.site.hidden_posts}
             )
             hidden_authors = await resolve_identities_batch(client, hidden_dids)
 
             for uri in bbs.site.hidden_posts:
-                parts = uri.split("/")
-                did = parts[2] if len(parts) > 2 else "?"
+                did = AtUri.parse(uri).did
                 handle = hidden_authors[did].handle if did in hidden_authors else did
 
                 try:
···
     if not title or not body:
         return redirect(f"/bbs/{handle}")
 
-    site_uri = f"at://{user['did']}/{lexicon.SITE}/self"
+    site_uri = str(AtUri(user["did"], lexicon.SITE, "self"))
     now = now_iso()
 
     await _authed_pds_post(

web/routes_write.py

···
 from quart import Blueprint, current_app, redirect, request
 
 from core import lexicon
+from core.models import AtUri
 from core.records import upload_blob
 from core.util import now_iso
 from web.helpers import get_user, session_updater
···
     if bbs.site.is_banned(user["did"]):
         return redirect(f"/bbs/{handle}/board/{slug}")
 
-    board_uri = f"at://{bbs.identity.did}/{lexicon.BOARD}/{slug}"
+    board_uri = str(AtUri(bbs.identity.did, lexicon.BOARD, slug))
 
     # Handle file attachments
     attachments = []
···
     if not body:
         return redirect(f"/bbs/{handle}/thread/{did}/{tid}")
 
-    thread_uri = f"at://{did}/{lexicon.THREAD}/{tid}"
+    thread_uri = str(AtUri(did, lexicon.THREAD, tid))
 
     # Handle file attachments
     client = current_app.http_client