Unify domain_events + ap_activities into AP-shaped event bus

All cross-service events now flow through ap_activities with a unified EventProcessor. Internal events use visibility="internal"; federation activities use visibility="public" and get delivered by a wildcard handler. - Add processing columns to APActivity (process_state, actor_uri, etc.) - New emit_activity() / register_activity_handler() API - EventProcessor polls ap_activities instead of domain_events - Rewrite all handlers to accept APActivity - Migrate all 7 emit_event call sites to emit_activity - publish_activity() sets process_state=pending directly (no emit_event bridge) - Migration to drop domain_events table Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Tech debt cleanup: fix session.py, remove stale references, update docs
2026-02-22 16:19:29 +00:00 · 2026-02-22 15:11:31 +00:00 · 2026-02-22 13:41:58 +00:00
30 changed files with 2360 additions and 246 deletions
--- a/README.md
+++ b/README.md
@@ -1,6 +1,6 @@
 # Shared

-Shared infrastructure, models, templates, and configuration used by all four Rose Ash microservices (blog, market, cart, events). Included as a git submodule in each app.
+Shared infrastructure, models, contracts, services, and templates used by all five Rose Ash microservices (blog, market, cart, events, federation). Included as a git submodule in each app.

 ## Structure

@@ -8,53 +8,78 @@ Shared infrastructure, models, templates, and configuration used by all four Ros
 shared/
  db/
    base.py                 # SQLAlchemy declarative Base
-    session.py              # Async session factory (get_session)
-  models/                   # Shared domain models
+    session.py              # Async session factory (get_session, register_db)
+  models/                   # Canonical domain models
    user.py                 # User
    magic_link.py           # MagicLink (auth tokens)
-    domain_event.py         # DomainEvent (transactional outbox)
+    domain_event.py         # DomainEvent (legacy — being removed)
    kv.py                   # KeyValue (key-value store)
-    menu_item.py            # MenuItem
+    menu_item.py            # MenuItem (deprecated — use MenuNode)
+    menu_node.py            # MenuNode (navigation tree)
+    container_relation.py   # ContainerRelation (parent-child content)
    ghost_membership_entities.py  # GhostNewsletter, UserNewsletter
+    federation.py           # ActorProfile, APActivity, APFollower, APFollowing,
+                            # RemoteActor, APRemotePost, APLocalPost,
+                            # APInteraction, APNotification, APAnchor, IPFSPin
+  contracts/
+    dtos.py                 # Frozen dataclasses for cross-domain data transfer
+    protocols.py            # Service protocols (Blog, Calendar, Market, Cart, Federation)
+    widgets.py              # Widget types (NavWidget, CardWidget, AccountPageWidget)
+  services/
+    registry.py             # Typed singleton: services.blog, .calendar, .market, .cart, .federation
+    blog_impl.py            # SqlBlogService
+    calendar_impl.py        # SqlCalendarService
+    market_impl.py          # SqlMarketService
+    cart_impl.py            # SqlCartService
+    federation_impl.py      # SqlFederationService
+    federation_publish.py   # try_publish() — inline AP publication helper
+    stubs.py                # No-op stubs for absent domains
+    navigation.py           # get_navigation_tree()
+    relationships.py        # attach_child, get_children, detach_child
+    widget_registry.py      # Widget registry singleton
+    widgets/                # Per-domain widget registration
  infrastructure/
    factory.py              # create_base_app() — Quart app factory
    cart_identity.py        # current_cart_identity() (user_id or session_id)
    cart_loader.py          # Cart data loader for context processors
    context.py              # Jinja2 context processors
-    internal_api.py         # Inter-app HTTP client (get/post via httpx)
    jinja_setup.py          # Jinja2 template environment setup
    urls.py                 # URL helpers (coop_url, market_url, etc.)
    user_loader.py          # Load current user from session
    http_utils.py           # HTTP utility functions
  events/
-    bus.py                  # emit_event(), register_handler()
-    processor.py            # EventProcessor (polls domain_events, runs handlers)
-  browser/app/
-    csrf.py                 # CSRF protection
-    errors.py               # Error handlers
-    middleware.py            # Request/response middleware
-    redis_cacher.py         # Tag-based Redis page caching
-    authz.py                # Authorization helpers
-    filters/                # Jinja2 template filters (currency, truncate, etc.)
-    utils/                  # HTMX helpers, UTC time, parsing
-    payments/sumup.py       # SumUp checkout API integration
-  browser/templates/        # ~300 Jinja2 templates shared across all apps
-  config.py                 # YAML config loader
+    bus.py                  # emit_activity(), register_activity_handler()
+    processor.py            # EventProcessor (polls ap_activities, runs handlers)
+    handlers/               # Shared activity handlers
+      container_handlers.py #   Navigation rebuild on attach/detach
+      login_handlers.py     #   Cart/entry adoption on login
+      order_handlers.py     #   Order lifecycle events
+      ap_delivery_handler.py #  AP activity delivery to follower inboxes (wildcard)
+  utils/
+    __init__.py
+    calendar_helpers.py     # Calendar period/entry utilities
+    http_signatures.py      # RSA keypair generation, HTTP signature signing/verification
+    ipfs_client.py          # Async IPFS client (add_bytes, add_json, pin_cid)
+    anchoring.py            # Merkle trees + OpenTimestamps Bitcoin anchoring
+    webfinger.py            # WebFinger actor resolution
+  browser/
+    app/                    # Middleware, CSRF, errors, Redis caching, authz, filters
+    templates/              # ~300 Jinja2 templates shared across all apps
  containers.py             # ContainerType, container_filter, content_filter helpers
+  config.py                 # YAML config loader
  log_config/setup.py       # Logging configuration (JSON formatter)
-  utils.py                  # host_url and other shared utilities
  static/                   # Shared static assets (CSS, JS, images, FontAwesome)
  editor/                   # Koenig (Ghost) rich text editor build
-  alembic/                  # Database migrations (25 versions)
-    env.py                  # Imports models from all apps (with try/except guards)
-    versions/               # Migration files — single head: j0h8e4f6g7
+  alembic/                  # Database migrations
 ```

 ## Key Patterns

- **App factory:** All apps call `create_base_app()` which sets up DB sessions, CSRF, error handling, event processing, logging, and the glue handler registry.
- **Event bus:** `emit_event()` writes to `domain_events` table in the caller's transaction. `EventProcessor` polls and dispatches to registered handlers.
- **Inter-app HTTP:** `internal_api.get/post("cart", "/internal/cart/summary")` for cross-app reads. URLs resolved from `app-config.yaml`.
+- **App factory:** All apps call `create_base_app()` which sets up DB sessions, CSRF, error handling, event processing, logging, widget registration, and domain service wiring.
+- **Service contracts:** Cross-domain communication via typed Protocols + frozen DTO dataclasses. Apps call `services.calendar.method()`, never import models from other domains.
+- **Service registry:** Typed singleton (`services.blog`, `.calendar`, `.market`, `.cart`, `.federation`). Apps wire their own domain + stubs for others via `register_domain_services()`.
+- **Activity bus:** `emit_activity()` writes to `ap_activities` table in the caller's transaction. `EventProcessor` polls pending activities and dispatches to registered handlers. Internal events use `visibility="internal"`; federation activities use `visibility="public"` and are delivered to follower inboxes by the wildcard delivery handler.
+- **Widget registry:** Domain services register widgets (nav, card, account); templates consume via `widgets.container_nav`, `widgets.container_cards`.
 - **Cart identity:** `current_cart_identity()` returns `{"user_id": int|None, "session_id": str|None}` from the request session.

 ## Alembic Migrations
@@ -62,8 +87,5 @@ shared/
 All apps share one PostgreSQL database. Migrations are managed here and run from the blog app's entrypoint (other apps skip migrations on startup).

 ```bash
-# From any app directory (shared/ must be on sys.path)
 alembic -c shared/alembic.ini upgrade head
 ```
-
-Current head: `j0h8e4f6g7` (drop cross-domain FK constraints).
--- a/init.py
+++ b/init.py
@@ -1 +1 @@
-# shared package — extracted from blog/shared_lib/
+# shared package — infrastructure, models, contracts, and services
--- a/alembic/env.py
+++ b/alembic/env.py
@@ -19,7 +19,7 @@ from shared.db.base import Base

 # Import ALL models so Base.metadata sees every table
 import shared.models  # noqa: F401  User, KV, MagicLink, MenuItem, Ghost*
-for _mod in ("blog.models", "market.models", "cart.models", "events.models", "federation.models", "glue.models"):
+for _mod in ("blog.models", "market.models", "cart.models", "events.models", "federation.models"):
    try:
        __import__(_mod)
    except ImportError:
--- a/alembic/versions/l2j0g6h8i9_add_fediverse_tables.py
+++ b/alembic/versions/l2j0g6h8i9_add_fediverse_tables.py
@@ -0,0 +1,138 @@
+"""add fediverse social tables
+
+Revision ID: l2j0g6h8i9
+Revises: k1i9f5g7h8
+Create Date: 2026-02-22
+
+Creates:
+- ap_remote_actors    — cached profiles of remote actors
+- ap_following        — outbound follows (local → remote)
+- ap_remote_posts     — ingested posts from remote actors
+- ap_local_posts      — native posts composed in federation UI
+- ap_interactions     — likes and boosts
+- ap_notifications    — follow/like/boost/mention/reply notifications
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import JSONB
+
+revision = "l2j0g6h8i9"
+down_revision = "k1i9f5g7h8"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # -- ap_remote_actors --
+    op.create_table(
+        "ap_remote_actors",
+        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
+        sa.Column("actor_url", sa.String(512), unique=True, nullable=False),
+        sa.Column("inbox_url", sa.String(512), nullable=False),
+        sa.Column("shared_inbox_url", sa.String(512), nullable=True),
+        sa.Column("preferred_username", sa.String(255), nullable=False),
+        sa.Column("display_name", sa.String(255), nullable=True),
+        sa.Column("summary", sa.Text, nullable=True),
+        sa.Column("icon_url", sa.String(512), nullable=True),
+        sa.Column("public_key_pem", sa.Text, nullable=True),
+        sa.Column("domain", sa.String(255), nullable=False),
+        sa.Column("fetched_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+    )
+    op.create_index("ix_ap_remote_actor_url", "ap_remote_actors", ["actor_url"], unique=True)
+    op.create_index("ix_ap_remote_actor_domain", "ap_remote_actors", ["domain"])
+
+    # -- ap_following --
+    op.create_table(
+        "ap_following",
+        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
+        sa.Column("actor_profile_id", sa.Integer, sa.ForeignKey("ap_actor_profiles.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("remote_actor_id", sa.Integer, sa.ForeignKey("ap_remote_actors.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("state", sa.String(20), nullable=False, server_default="pending"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("accepted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.UniqueConstraint("actor_profile_id", "remote_actor_id", name="uq_following"),
+    )
+    op.create_index("ix_ap_following_actor", "ap_following", ["actor_profile_id"])
+    op.create_index("ix_ap_following_remote", "ap_following", ["remote_actor_id"])
+
+    # -- ap_remote_posts --
+    op.create_table(
+        "ap_remote_posts",
+        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
+        sa.Column("remote_actor_id", sa.Integer, sa.ForeignKey("ap_remote_actors.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("activity_id", sa.String(512), unique=True, nullable=False),
+        sa.Column("object_id", sa.String(512), unique=True, nullable=False),
+        sa.Column("object_type", sa.String(64), nullable=False, server_default="Note"),
+        sa.Column("content", sa.Text, nullable=True),
+        sa.Column("summary", sa.Text, nullable=True),
+        sa.Column("url", sa.String(512), nullable=True),
+        sa.Column("attachment_data", JSONB, nullable=True),
+        sa.Column("tag_data", JSONB, nullable=True),
+        sa.Column("in_reply_to", sa.String(512), nullable=True),
+        sa.Column("conversation", sa.String(512), nullable=True),
+        sa.Column("published", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("fetched_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+    )
+    op.create_index("ix_ap_remote_post_actor", "ap_remote_posts", ["remote_actor_id"])
+    op.create_index("ix_ap_remote_post_published", "ap_remote_posts", ["published"])
+    op.create_index("ix_ap_remote_post_object", "ap_remote_posts", ["object_id"], unique=True)
+
+    # -- ap_local_posts --
+    op.create_table(
+        "ap_local_posts",
+        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
+        sa.Column("actor_profile_id", sa.Integer, sa.ForeignKey("ap_actor_profiles.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("content", sa.Text, nullable=False),
+        sa.Column("visibility", sa.String(20), nullable=False, server_default="public"),
+        sa.Column("in_reply_to", sa.String(512), nullable=True),
+        sa.Column("published", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+    )
+    op.create_index("ix_ap_local_post_actor", "ap_local_posts", ["actor_profile_id"])
+    op.create_index("ix_ap_local_post_published", "ap_local_posts", ["published"])
+
+    # -- ap_interactions --
+    op.create_table(
+        "ap_interactions",
+        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
+        sa.Column("actor_profile_id", sa.Integer, sa.ForeignKey("ap_actor_profiles.id", ondelete="CASCADE"), nullable=True),
+        sa.Column("remote_actor_id", sa.Integer, sa.ForeignKey("ap_remote_actors.id", ondelete="CASCADE"), nullable=True),
+        sa.Column("post_type", sa.String(20), nullable=False),
+        sa.Column("post_id", sa.Integer, nullable=False),
+        sa.Column("interaction_type", sa.String(20), nullable=False),
+        sa.Column("activity_id", sa.String(512), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+    )
+    op.create_index("ix_ap_interaction_post", "ap_interactions", ["post_type", "post_id"])
+    op.create_index("ix_ap_interaction_actor", "ap_interactions", ["actor_profile_id"])
+    op.create_index("ix_ap_interaction_remote", "ap_interactions", ["remote_actor_id"])
+
+    # -- ap_notifications --
+    op.create_table(
+        "ap_notifications",
+        sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
+        sa.Column("actor_profile_id", sa.Integer, sa.ForeignKey("ap_actor_profiles.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("notification_type", sa.String(20), nullable=False),
+        sa.Column("from_remote_actor_id", sa.Integer, sa.ForeignKey("ap_remote_actors.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("from_actor_profile_id", sa.Integer, sa.ForeignKey("ap_actor_profiles.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("target_activity_id", sa.Integer, sa.ForeignKey("ap_activities.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("target_remote_post_id", sa.Integer, sa.ForeignKey("ap_remote_posts.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("read", sa.Boolean, nullable=False, server_default="false"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+    )
+    op.create_index("ix_ap_notification_actor", "ap_notifications", ["actor_profile_id"])
+    op.create_index("ix_ap_notification_read", "ap_notifications", ["actor_profile_id", "read"])
+    op.create_index("ix_ap_notification_created", "ap_notifications", ["created_at"])
+
+
+def downgrade() -> None:
+    op.drop_table("ap_notifications")
+    op.drop_table("ap_interactions")
+    op.drop_table("ap_local_posts")
+    op.drop_table("ap_remote_posts")
+    op.drop_table("ap_following")
+    op.drop_table("ap_remote_actors")
--- a/alembic/versions/m3k1h7i9j0_add_activity_bus_columns.py
+++ b/alembic/versions/m3k1h7i9j0_add_activity_bus_columns.py
@@ -0,0 +1,113 @@
+"""add unified event bus columns to ap_activities
+
+Revision ID: m3k1h7i9j0
+Revises: l2j0g6h8i9
+Create Date: 2026-02-22
+
+Adds processing and visibility columns so ap_activities can serve as the
+unified event bus for both internal domain events and federation delivery.
+"""
+
+revision = "m3k1h7i9j0"
+down_revision = "l2j0g6h8i9"
+branch_labels = None
+depends_on = None
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade() -> None:
+    # Add new columns with defaults so existing rows stay valid
+    op.add_column(
+        "ap_activities",
+        sa.Column("actor_uri", sa.String(512), nullable=True),
+    )
+    op.add_column(
+        "ap_activities",
+        sa.Column(
+            "visibility", sa.String(20),
+            nullable=False, server_default="public",
+        ),
+    )
+    op.add_column(
+        "ap_activities",
+        sa.Column(
+            "process_state", sa.String(20),
+            nullable=False, server_default="completed",
+        ),
+    )
+    op.add_column(
+        "ap_activities",
+        sa.Column(
+            "process_attempts", sa.Integer(),
+            nullable=False, server_default="0",
+        ),
+    )
+    op.add_column(
+        "ap_activities",
+        sa.Column(
+            "process_max_attempts", sa.Integer(),
+            nullable=False, server_default="5",
+        ),
+    )
+    op.add_column(
+        "ap_activities",
+        sa.Column("process_error", sa.Text(), nullable=True),
+    )
+    op.add_column(
+        "ap_activities",
+        sa.Column(
+            "processed_at", sa.DateTime(timezone=True), nullable=True,
+        ),
+    )
+
+    # Backfill actor_uri from the related actor_profile
+    op.execute(
+        """
+        UPDATE ap_activities a
+        SET actor_uri = CONCAT(
+            'https://',
+            COALESCE(current_setting('app.ap_domain', true), 'rose-ash.com'),
+            '/users/',
+            p.preferred_username
+        )
+        FROM ap_actor_profiles p
+        WHERE a.actor_profile_id = p.id
+          AND a.actor_uri IS NULL
+        """
+    )
+
+    # Make actor_profile_id nullable (internal events have no actor profile)
+    op.alter_column(
+        "ap_activities", "actor_profile_id",
+        existing_type=sa.Integer(),
+        nullable=True,
+    )
+
+    # Index for processor polling
+    op.create_index(
+        "ix_ap_activity_process", "ap_activities", ["process_state"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_ap_activity_process", table_name="ap_activities")
+
+    # Restore actor_profile_id NOT NULL (remove any rows without it first)
+    op.execute(
+        "DELETE FROM ap_activities WHERE actor_profile_id IS NULL"
+    )
+    op.alter_column(
+        "ap_activities", "actor_profile_id",
+        existing_type=sa.Integer(),
+        nullable=False,
+    )
+
+    op.drop_column("ap_activities", "processed_at")
+    op.drop_column("ap_activities", "process_error")
+    op.drop_column("ap_activities", "process_max_attempts")
+    op.drop_column("ap_activities", "process_attempts")
+    op.drop_column("ap_activities", "process_state")
+    op.drop_column("ap_activities", "visibility")
+    op.drop_column("ap_activities", "actor_uri")
--- a/alembic/versions/n4l2i8j0k1_drop_domain_events_table.py
+++ b/alembic/versions/n4l2i8j0k1_drop_domain_events_table.py
@@ -0,0 +1,46 @@
+"""drop domain_events table
+
+Revision ID: n4l2i8j0k1
+Revises: m3k1h7i9j0
+Create Date: 2026-02-22
+
+The domain_events table is no longer used — all events now flow through
+ap_activities with the unified activity bus.
+"""
+
+revision = "n4l2i8j0k1"
+down_revision = "m3k1h7i9j0"
+branch_labels = None
+depends_on = None
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import JSONB
+
+
+def upgrade() -> None:
+    op.drop_index("ix_domain_events_state", table_name="domain_events")
+    op.drop_index("ix_domain_events_event_type", table_name="domain_events")
+    op.drop_table("domain_events")
+
+
+def downgrade() -> None:
+    op.create_table(
+        "domain_events",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column("event_type", sa.String(128), nullable=False),
+        sa.Column("aggregate_type", sa.String(64), nullable=False),
+        sa.Column("aggregate_id", sa.Integer(), nullable=False),
+        sa.Column("payload", JSONB(), nullable=True),
+        sa.Column("state", sa.String(20), nullable=False, server_default="pending"),
+        sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("max_attempts", sa.Integer(), nullable=False, server_default="5"),
+        sa.Column("last_error", sa.Text(), nullable=True),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True),
+            nullable=False, server_default=sa.func.now(),
+        ),
+        sa.Column("processed_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index("ix_domain_events_event_type", "domain_events", ["event_type"])
+    op.create_index("ix_domain_events_state", "domain_events", ["state"])
--- a/browser/templates/_types/day/_nav.html
+++ b/browser/templates/_types/day/_nav.html
@@ -25,6 +25,15 @@
  {% endcall %}
 </div>

+{# Container nav widgets (market links, etc.) #}
+{% if container_nav_widgets %}
+  {% for wdata in container_nav_widgets %}
+    {% with ctx=wdata.ctx %}
+      {% include wdata.widget.template with context %}
+    {% endwith %}
+  {% endfor %}
+{% endif %}
+
 {# Admin link #}
 {% if g.rights.admin %}
    {% from 'macros/admin_nav.html' import admin_nav_item %}
--- a/browser/templates/_types/entry/_nav.html
+++ b/browser/templates/_types/entry/_nav.html
@@ -22,6 +22,14 @@
  {% endcall %}
 </div>

+{% if container_nav_widgets %}
+  {% for wdata in container_nav_widgets %}
+    {% with ctx=wdata.ctx %}
+      {% include wdata.widget.template with context %}
+    {% endwith %}
+  {% endfor %}
+{% endif %}
+
 {# Admin link #}
 {% if g.rights.admin %}

--- a/contracts/dtos.py
+++ b/contracts/dtos.py
@@ -187,3 +187,68 @@ class APAnchorDTO:
    ots_proof_cid: str | None = None
    confirmed_at: datetime | None = None
    bitcoin_txid: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class RemoteActorDTO:
+    id: int
+    actor_url: str
+    inbox_url: str
+    preferred_username: str
+    domain: str
+    display_name: str | None = None
+    summary: str | None = None
+    icon_url: str | None = None
+    shared_inbox_url: str | None = None
+    public_key_pem: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class RemotePostDTO:
+    id: int
+    remote_actor_id: int
+    object_id: str
+    content: str
+    summary: str | None = None
+    url: str | None = None
+    attachments: list[dict] = field(default_factory=list)
+    tags: list[dict] = field(default_factory=list)
+    published: datetime | None = None
+    actor: RemoteActorDTO | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class TimelineItemDTO:
+    id: str  # composite key for cursor pagination
+    post_type: str  # "local" | "remote" | "boost"
+    content: str  # HTML
+    published: datetime
+    actor_name: str
+    actor_username: str
+    object_id: str | None = None
+    summary: str | None = None
+    url: str | None = None
+    attachments: list[dict] = field(default_factory=list)
+    tags: list[dict] = field(default_factory=list)
+    actor_domain: str | None = None  # None = local
+    actor_icon: str | None = None
+    actor_url: str | None = None
+    boosted_by: str | None = None
+    like_count: int = 0
+    boost_count: int = 0
+    liked_by_me: bool = False
+    boosted_by_me: bool = False
+    author_inbox: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class NotificationDTO:
+    id: int
+    notification_type: str  # follow/like/boost/mention/reply
+    from_actor_name: str
+    from_actor_username: str
+    created_at: datetime
+    read: bool
+    from_actor_domain: str | None = None
+    from_actor_icon: str | None = None
+    target_content_preview: str | None = None
--- a/contracts/protocols.py
+++ b/contracts/protocols.py
@@ -22,6 +22,10 @@ from .dtos import (
    ActorProfileDTO,
    APActivityDTO,
    APFollowerDTO,
+    RemoteActorDTO,
+    RemotePostDTO,
+    TimelineItemDTO,
+    NotificationDTO,
 )


@@ -217,6 +221,11 @@ class FederationService(Protocol):
        self, session: AsyncSession, username: str,
    ) -> list[APFollowerDTO]: ...

+    async def get_followers_paginated(
+        self, session: AsyncSession, username: str,
+        page: int = 1, per_page: int = 20,
+    ) -> tuple[list[RemoteActorDTO], int]: ...
+
    async def add_follower(
        self, session: AsyncSession, username: str,
        follower_acct: str, follower_inbox: str, follower_actor_url: str,
@@ -227,5 +236,108 @@ class FederationService(Protocol):
        self, session: AsyncSession, username: str, follower_acct: str,
    ) -> bool: ...

+    # -- Remote actors --------------------------------------------------------
+    async def get_or_fetch_remote_actor(
+        self, session: AsyncSession, actor_url: str,
+    ) -> RemoteActorDTO | None: ...
+
+    async def search_remote_actor(
+        self, session: AsyncSession, acct: str,
+    ) -> RemoteActorDTO | None: ...
+
+    # -- Following (outbound) -------------------------------------------------
+    async def send_follow(
+        self, session: AsyncSession, local_username: str, remote_actor_url: str,
+    ) -> None: ...
+
+    async def get_following(
+        self, session: AsyncSession, username: str,
+        page: int = 1, per_page: int = 20,
+    ) -> tuple[list[RemoteActorDTO], int]: ...
+
+    async def accept_follow_response(
+        self, session: AsyncSession, local_username: str, remote_actor_url: str,
+    ) -> None: ...
+
+    async def unfollow(
+        self, session: AsyncSession, local_username: str, remote_actor_url: str,
+    ) -> None: ...
+
+    # -- Remote posts ---------------------------------------------------------
+    async def ingest_remote_post(
+        self, session: AsyncSession, remote_actor_id: int,
+        activity_json: dict, object_json: dict,
+    ) -> None: ...
+
+    async def delete_remote_post(
+        self, session: AsyncSession, object_id: str,
+    ) -> None: ...
+
+    async def get_remote_post(
+        self, session: AsyncSession, object_id: str,
+    ) -> RemotePostDTO | None: ...
+
+    # -- Timelines ------------------------------------------------------------
+    async def get_home_timeline(
+        self, session: AsyncSession, actor_profile_id: int,
+        before: datetime | None = None, limit: int = 20,
+    ) -> list[TimelineItemDTO]: ...
+
+    async def get_public_timeline(
+        self, session: AsyncSession,
+        before: datetime | None = None, limit: int = 20,
+    ) -> list[TimelineItemDTO]: ...
+
+    async def get_actor_timeline(
+        self, session: AsyncSession, remote_actor_id: int,
+        before: datetime | None = None, limit: int = 20,
+    ) -> list[TimelineItemDTO]: ...
+
+    # -- Local posts ----------------------------------------------------------
+    async def create_local_post(
+        self, session: AsyncSession, actor_profile_id: int,
+        content: str, visibility: str = "public",
+        in_reply_to: str | None = None,
+    ) -> int: ...
+
+    async def delete_local_post(
+        self, session: AsyncSession, actor_profile_id: int, post_id: int,
+    ) -> None: ...
+
+    # -- Interactions ---------------------------------------------------------
+    async def like_post(
+        self, session: AsyncSession, actor_profile_id: int,
+        object_id: str, author_inbox: str,
+    ) -> None: ...
+
+    async def unlike_post(
+        self, session: AsyncSession, actor_profile_id: int,
+        object_id: str, author_inbox: str,
+    ) -> None: ...
+
+    async def boost_post(
+        self, session: AsyncSession, actor_profile_id: int,
+        object_id: str, author_inbox: str,
+    ) -> None: ...
+
+    async def unboost_post(
+        self, session: AsyncSession, actor_profile_id: int,
+        object_id: str, author_inbox: str,
+    ) -> None: ...
+
+    # -- Notifications --------------------------------------------------------
+    async def get_notifications(
+        self, session: AsyncSession, actor_profile_id: int,
+        before: datetime | None = None, limit: int = 20,
+    ) -> list[NotificationDTO]: ...
+
+    async def unread_notification_count(
+        self, session: AsyncSession, actor_profile_id: int,
+    ) -> int: ...
+
+    async def mark_notifications_read(
+        self, session: AsyncSession, actor_profile_id: int,
+    ) -> None: ...
+
    # -- Stats ----------------------------------------------------------------
    async def get_stats(self, session: AsyncSession) -> dict: ...
--- a/db/session.py
+++ b/db/session.py
@@ -15,7 +15,7 @@ _engine = create_async_engine(
    future=True,
    echo=False,
    pool_pre_ping=True,
-    pool_size=-1  # ned to look at this!!!
+    pool_size=0,  # 0 = unlimited (NullPool equivalent for asyncpg)
 )

 _Session = async_sessionmaker(
@@ -34,43 +34,42 @@ async def get_session():
        await sess.close()


-
 def register_db(app: Quart):

-  @app.before_request
-  async def open_session():
-      g.s = _Session()
-      g.tx = await g.s.begin()
-      g.had_error = False
+    @app.before_request
+    async def open_session():
+        g.s = _Session()
+        g.tx = await g.s.begin()
+        g.had_error = False

-  @app.after_request
-  async def maybe_commit(response):
-      # Runs BEFORE bytes are sent.
-      if not g.had_error and 200 <= response.status_code < 400:
-          try:
-              if hasattr(g, "tx"):
-                  await g.tx.commit()
-          except Exception as e:
-              print(f'commit failed {e}')
-              if hasattr(g, "tx"):
-                  await g.tx.rollback()
-              from quart import make_response
-              return await make_response("Commit failed", 500)
-      return response
+    @app.after_request
+    async def maybe_commit(response):
+        # Runs BEFORE bytes are sent.
+        if not g.had_error and 200 <= response.status_code < 400:
+            try:
+                if hasattr(g, "tx"):
+                    await g.tx.commit()
+            except Exception as e:
+                print(f'commit failed {e}')
+                if hasattr(g, "tx"):
+                    await g.tx.rollback()
+                from quart import make_response
+                return await make_response("Commit failed", 500)
+        return response

-  @app.teardown_request
-  async def finish(exc):
-      try:
-          # If an exception occurred OR we didn't commit (still in txn), roll back.
-          if hasattr(g, "s"):
-              if exc is not None or g.s.in_transaction():
-                  if hasattr(g, "tx"):
-                      await g.tx.rollback()
-      finally:
-          if hasattr(g, "s"):
-              await g.s.close()
+    @app.teardown_request
+    async def finish(exc):
+        try:
+            # If an exception occurred OR we didn't commit (still in txn), roll back.
+            if hasattr(g, "s"):
+                if exc is not None or g.s.in_transaction():
+                    if hasattr(g, "tx"):
+                        await g.tx.rollback()
+        finally:
+            if hasattr(g, "s"):
+                await g.s.close()

-  @app.errorhandler(Exception)
-  async def mark_error(e):
-      g.had_error = True
-      raise
+    @app.errorhandler(Exception)
+    async def mark_error(e):
+        g.had_error = True
+        raise
--- a/events/init.py
+++ b/events/init.py
@@ -1,4 +1,9 @@
-from .bus import emit_event, register_handler
+from .bus import emit_activity, register_activity_handler, get_activity_handlers
 from .processor import EventProcessor

-__all__ = ["emit_event", "register_handler", "EventProcessor"]
+__all__ = [
+    "emit_activity",
+    "register_activity_handler",
+    "get_activity_handlers",
+    "EventProcessor",
+]
--- a/events/bus.py
+++ b/events/bus.py
@@ -1,56 +1,109 @@
 """
-Transactional outbox event bus.
+Unified activity bus.

-emit_event() writes to the domain_events table within the caller's existing
-DB transaction — atomic with whatever domain change triggered the event.
+emit_activity() writes an APActivity row with process_state='pending' within
+the caller's existing DB transaction — atomic with the domain change.

-register_handler() registers async handler functions that the EventProcessor
-will call when processing events of a given type.
+register_activity_handler() registers async handler functions that the
+EventProcessor dispatches when processing pending activities.
 """
 from __future__ import annotations

+import uuid
 from collections import defaultdict
-from typing import Any, Awaitable, Callable, Dict, List
+from typing import Awaitable, Callable, Dict, List, Tuple

 from sqlalchemy.ext.asyncio import AsyncSession

-from shared.models.domain_event import DomainEvent
+from shared.models.federation import APActivity

-# handler signature: async def handler(event: DomainEvent, session: AsyncSession) -> None
-HandlerFn = Callable[[DomainEvent, AsyncSession], Awaitable[None]]
+# ---------------------------------------------------------------------------
+# Activity-handler registry
+# ---------------------------------------------------------------------------
+# Handler signature: async def handler(activity: APActivity, session: AsyncSession) -> None
+ActivityHandlerFn = Callable[[APActivity, AsyncSession], Awaitable[None]]

-_handlers: Dict[str, List[HandlerFn]] = defaultdict(list)
+# Keyed by (activity_type, object_type). object_type="*" is wildcard.
+_activity_handlers: Dict[Tuple[str, str], List[ActivityHandlerFn]] = defaultdict(list)


-async def emit_event(
+def register_activity_handler(
+    activity_type: str,
+    fn: ActivityHandlerFn,
+    *,
+    object_type: str | None = None,
+) -> None:
+    """Register an async handler for an activity type + optional object type.
+
+    Use ``activity_type="*"`` as a wildcard that fires for every activity
+    (e.g. federation delivery handler).
+    """
+    key = (activity_type, object_type or "*")
+    _activity_handlers[key].append(fn)
+
+
+def get_activity_handlers(
+    activity_type: str,
+    object_type: str | None = None,
+) -> List[ActivityHandlerFn]:
+    """Return all matching handlers for an activity.
+
+    Matches in order:
+    1. Exact (activity_type, object_type)
+    2. (activity_type, "*")  — type-level wildcard
+    3. ("*", "*")            — global wildcard (e.g. delivery)
+    """
+    handlers: List[ActivityHandlerFn] = []
+    ot = object_type or "*"
+
+    # Exact match
+    if ot != "*":
+        handlers.extend(_activity_handlers.get((activity_type, ot), []))
+    # Type-level wildcard
+    handlers.extend(_activity_handlers.get((activity_type, "*"), []))
+    # Global wildcard
+    if activity_type != "*":
+        handlers.extend(_activity_handlers.get(("*", "*"), []))
+
+    return handlers
+
+
+# ---------------------------------------------------------------------------
+# emit_activity — the primary way to emit events
+# ---------------------------------------------------------------------------
+async def emit_activity(
    session: AsyncSession,
-    event_type: str,
-    aggregate_type: str,
-    aggregate_id: int,
-    payload: Dict[str, Any] | None = None,
-) -> DomainEvent:
+    *,
+    activity_type: str,
+    actor_uri: str,
+    object_type: str,
+    object_data: dict | None = None,
+    source_type: str | None = None,
+    source_id: int | None = None,
+    visibility: str = "internal",
+    actor_profile_id: int | None = None,
+) -> APActivity:
    """
-    Write a domain event to the outbox table in the current transaction.
+    Write an AP-shaped activity to ap_activities with process_state='pending'.

-    Call this inside your service function, using the same session that
-    performs the domain change. The event and the change commit together.
+    Called inside a service function using the same session that performs the
+    domain change.  The activity and the change commit together.
    """
-    event = DomainEvent(
-        event_type=event_type,
-        aggregate_type=aggregate_type,
-        aggregate_id=aggregate_id,
-        payload=payload or {},
+    activity_uri = f"internal:{uuid.uuid4()}" if visibility == "internal" else f"urn:uuid:{uuid.uuid4()}"
+
+    activity = APActivity(
+        activity_id=activity_uri,
+        activity_type=activity_type,
+        actor_profile_id=actor_profile_id,
+        actor_uri=actor_uri,
+        object_type=object_type,
+        object_data=object_data or {},
+        is_local=True,
+        source_type=source_type,
+        source_id=source_id,
+        visibility=visibility,
+        process_state="pending",
    )
-    session.add(event)
-    await session.flush()  # assign event.id
-    return event
-
-
-def register_handler(event_type: str, fn: HandlerFn) -> None:
-    """Register an async handler for a given event type."""
-    _handlers[event_type].append(fn)
-
-
-def get_handlers(event_type: str) -> List[HandlerFn]:
-    """Return all registered handlers for an event type."""
-    return _handlers.get(event_type, [])
+    session.add(activity)
+    await session.flush()
+    return activity
--- a/events/handlers/init.py
+++ b/events/handlers/init.py
@@ -1,4 +1,4 @@
-"""Shared event handlers (replaces glue.setup.register_glue_handlers)."""
+"""Shared event handlers."""


 def register_shared_handlers():
@@ -6,5 +6,4 @@ def register_shared_handlers():
    import shared.events.handlers.container_handlers  # noqa: F401
    import shared.events.handlers.login_handlers  # noqa: F401
    import shared.events.handlers.order_handlers  # noqa: F401
-    # federation_handlers removed — publication is now inline at write sites
    import shared.events.handlers.ap_delivery_handler  # noqa: F401
--- a/events/handlers/ap_delivery_handler.py
+++ b/events/handlers/ap_delivery_handler.py
@@ -1,7 +1,7 @@
 """Deliver AP activities to remote followers.

-On ``federation.activity_created`` → load activity + actor + followers →
-sign with HTTP Signatures → POST to each follower inbox.
+Registered as a wildcard handler — fires for every activity.  Skips
+non-public activities and those without an actor profile.
 """
 from __future__ import annotations

@@ -11,7 +11,7 @@ import httpx
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession

-from shared.events.bus import register_handler, DomainEvent
+from shared.events.bus import register_activity_handler
 from shared.models.federation import ActorProfile, APActivity, APFollower
 from shared.services.registry import services

@@ -33,12 +33,10 @@ def _build_activity_json(activity: APActivity, actor: ActorProfile, domain: str)
    object_id = activity.activity_id + "/object"

    if activity.activity_type == "Delete":
-        # Delete: object is a Tombstone with just id + type
        obj.setdefault("id", object_id)
        obj.setdefault("type", "Tombstone")
    else:
-        # Create/Update: full object with attribution
-        obj["id"] = object_id
+        obj.setdefault("id", object_id)
        obj.setdefault("type", activity.object_type)
        obj.setdefault("attributedTo", actor_url)
        obj.setdefault("published", activity.published.isoformat() if activity.published else None)
@@ -104,30 +102,20 @@ async def _deliver_to_inbox(
        return False


-async def on_activity_created(event: DomainEvent, session: AsyncSession) -> None:
-    """Deliver a newly created activity to all followers."""
+async def on_any_activity(activity: APActivity, session: AsyncSession) -> None:
+    """Deliver a public activity to all followers of its actor."""
    import os

+    # Only deliver public activities that have an actor profile
+    if activity.visibility != "public":
+        return
+    if activity.actor_profile_id is None:
+        return
    if not services.has("federation"):
        return

-    payload = event.payload
-    activity_id_uri = payload.get("activity_id")
-    if not activity_id_uri:
-        return
-
    domain = os.getenv("AP_DOMAIN", "rose-ash.com")

-    # Load the activity
-    activity = (
-        await session.execute(
-            select(APActivity).where(APActivity.activity_id == activity_id_uri)
-        )
-    ).scalar_one_or_none()
-    if not activity:
-        log.warning("Activity not found: %s", activity_id_uri)
-        return
-
    # Load actor with private key
    actor = (
        await session.execute(
@@ -135,7 +123,7 @@ async def on_activity_created(event: DomainEvent, session: AsyncSession) -> None
        )
    ).scalar_one_or_none()
    if not actor or not actor.private_key_pem:
-        log.warning("Actor not found or missing key for activity %s", activity_id_uri)
+        log.warning("Actor not found or missing key for activity %s", activity.activity_id)
        return

    # Load followers
@@ -146,14 +134,13 @@ async def on_activity_created(event: DomainEvent, session: AsyncSession) -> None
    ).scalars().all()

    if not followers:
-        log.debug("No followers to deliver to for %s", activity_id_uri)
+        log.debug("No followers to deliver to for %s", activity.activity_id)
        return

    # Build activity JSON
    activity_json = _build_activity_json(activity, actor, domain)

-    # Deliver to each follower inbox
-    # Deduplicate inboxes (multiple followers might share a shared inbox)
+    # Deduplicate inboxes
    inboxes = {f.follower_inbox for f in followers if f.follower_inbox}

    log.info(
@@ -166,4 +153,5 @@ async def on_activity_created(event: DomainEvent, session: AsyncSession) -> None
            await _deliver_to_inbox(client, inbox_url, activity_json, actor, domain)


-register_handler("federation.activity_created", on_activity_created)
+# Wildcard: fires for every activity
+register_activity_handler("*", on_any_activity)
--- a/events/handlers/container_handlers.py
+++ b/events/handlers/container_handlers.py
@@ -2,18 +2,18 @@ from __future__ import annotations

 from sqlalchemy.ext.asyncio import AsyncSession

-from shared.events import register_handler
-from shared.models.domain_event import DomainEvent
+from shared.events import register_activity_handler
+from shared.models.federation import APActivity
 from shared.services.navigation import rebuild_navigation


-async def on_child_attached(event: DomainEvent, session: AsyncSession) -> None:
+async def on_child_attached(activity: APActivity, session: AsyncSession) -> None:
    await rebuild_navigation(session)


-async def on_child_detached(event: DomainEvent, session: AsyncSession) -> None:
+async def on_child_detached(activity: APActivity, session: AsyncSession) -> None:
    await rebuild_navigation(session)


-register_handler("container.child_attached", on_child_attached)
-register_handler("container.child_detached", on_child_detached)
+register_activity_handler("Add", on_child_attached, object_type="rose:ContainerRelation")
+register_activity_handler("Remove", on_child_detached, object_type="rose:ContainerRelation")
--- a/events/handlers/federation_handlers.py
+++ b/events/handlers/federation_handlers.py
@@ -1,8 +0,0 @@
-"""Federation event handlers — REMOVED.
-
-Federation publication is now inline at the write site (ghost_sync, entries,
-market routes) via shared.services.federation_publish.try_publish().
-
-AP delivery (federation.activity_created → inbox POST) remains async via
-ap_delivery_handler.
-"""
--- a/events/handlers/login_handlers.py
+++ b/events/handlers/login_handlers.py
@@ -2,24 +2,22 @@ from __future__ import annotations

 from sqlalchemy.ext.asyncio import AsyncSession

-from shared.events import register_handler
-from shared.models.domain_event import DomainEvent
+from shared.events import register_activity_handler
+from shared.models.federation import APActivity
 from shared.services.registry import services


-async def on_user_logged_in(event: DomainEvent, session: AsyncSession) -> None:
-    payload = event.payload
-    user_id = payload["user_id"]
-    session_id = payload["session_id"]
+async def on_user_logged_in(activity: APActivity, session: AsyncSession) -> None:
+    data = activity.object_data
+    user_id = data["user_id"]
+    session_id = data["session_id"]

-    # Adopt cart items (if cart service is registered)
    if services.has("cart"):
        await services.cart.adopt_cart_for_user(session, user_id, session_id)

-    # Adopt calendar entries and tickets (if calendar service is registered)
    if services.has("calendar"):
        await services.calendar.adopt_entries_for_user(session, user_id, session_id)
        await services.calendar.adopt_tickets_for_user(session, user_id, session_id)


-register_handler("user.logged_in", on_user_logged_in)
+register_activity_handler("rose:Login", on_user_logged_in)
--- a/events/handlers/order_handlers.py
+++ b/events/handlers/order_handlers.py
@@ -4,19 +4,19 @@ import logging

 from sqlalchemy.ext.asyncio import AsyncSession

-from shared.events import register_handler
-from shared.models.domain_event import DomainEvent
+from shared.events import register_activity_handler
+from shared.models.federation import APActivity

 log = logging.getLogger(__name__)


-async def on_order_created(event: DomainEvent, session: AsyncSession) -> None:
-    log.info("order.created: order_id=%s", event.payload.get("order_id"))
+async def on_order_created(activity: APActivity, session: AsyncSession) -> None:
+    log.info("order.created: order_id=%s", activity.object_data.get("order_id"))


-async def on_order_paid(event: DomainEvent, session: AsyncSession) -> None:
-    log.info("order.paid: order_id=%s", event.payload.get("order_id"))
+async def on_order_paid(activity: APActivity, session: AsyncSession) -> None:
+    log.info("order.paid: order_id=%s", activity.object_data.get("order_id"))


-register_handler("order.created", on_order_created)
-register_handler("order.paid", on_order_paid)
+register_activity_handler("Create", on_order_created, object_type="rose:Order")
+register_activity_handler("rose:OrderPaid", on_order_paid)
--- a/events/processor.py
+++ b/events/processor.py
@@ -1,6 +1,6 @@
 """
-Event processor — polls the domain_events outbox table and dispatches
-to registered handlers.
+Event processor — polls the ap_activities table and dispatches to registered
+activity handlers.

 Runs as an asyncio background task within each app process.
 Uses SELECT ... FOR UPDATE SKIP LOCKED for safe concurrent processing.
@@ -11,16 +11,16 @@ import asyncio
 import traceback
 from datetime import datetime, timezone

-from sqlalchemy import select, update
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession

 from shared.db.session import get_session
-from shared.models.domain_event import DomainEvent
-from .bus import get_handlers
+from shared.models.federation import APActivity
+from .bus import get_activity_handlers


 class EventProcessor:
-    """Background event processor that polls the outbox table."""
+    """Background event processor that polls the ap_activities table."""

    def __init__(
        self,
@@ -64,54 +64,52 @@ class EventProcessor:
                await asyncio.sleep(self._poll_interval)

    async def _process_batch(self) -> int:
-        """Fetch and process a batch of pending events. Returns count processed."""
+        """Fetch and process a batch of pending activities. Returns count processed."""
        processed = 0
        async with get_session() as session:
-            # FOR UPDATE SKIP LOCKED: safe for concurrent processors
            stmt = (
-                select(DomainEvent)
+                select(APActivity)
                .where(
-                    DomainEvent.state == "pending",
-                    DomainEvent.attempts < DomainEvent.max_attempts,
+                    APActivity.process_state == "pending",
+                    APActivity.process_attempts < APActivity.process_max_attempts,
                )
-                .order_by(DomainEvent.created_at)
+                .order_by(APActivity.created_at)
                .limit(self._batch_size)
                .with_for_update(skip_locked=True)
            )
            result = await session.execute(stmt)
-            events = result.scalars().all()
+            activities = result.scalars().all()

-            for event in events:
-                await self._process_one(session, event)
+            for activity in activities:
+                await self._process_one(session, activity)
                processed += 1

            await session.commit()
        return processed

-    async def _process_one(self, session: AsyncSession, event: DomainEvent) -> None:
-        """Run all handlers for a single event."""
-        handlers = get_handlers(event.event_type)
+    async def _process_one(self, session: AsyncSession, activity: APActivity) -> None:
+        """Run all handlers for a single activity."""
+        handlers = get_activity_handlers(activity.activity_type, activity.object_type)
        now = datetime.now(timezone.utc)

-        event.state = "processing"
-        event.attempts += 1
+        activity.process_state = "processing"
+        activity.process_attempts += 1
        await session.flush()

        if not handlers:
-            # No handlers registered — mark completed (nothing to do)
-            event.state = "completed"
-            event.processed_at = now
+            activity.process_state = "completed"
+            activity.processed_at = now
            return

        try:
            for handler in handlers:
-                await handler(event, session)
-            event.state = "completed"
-            event.processed_at = now
+                await handler(activity, session)
+            activity.process_state = "completed"
+            activity.processed_at = now
        except Exception as exc:
-            event.last_error = f"{exc.__class__.__name__}: {exc}"
-            if event.attempts >= event.max_attempts:
-                event.state = "failed"
-                event.processed_at = now
+            activity.process_error = f"{exc.__class__.__name__}: {exc}"
+            if activity.process_attempts >= activity.process_max_attempts:
+                activity.process_state = "failed"
+                activity.processed_at = now
            else:
-                event.state = "pending"  # retry
+                activity.process_state = "pending"  # retry
--- a/models/init.py
+++ b/models/init.py
@@ -8,8 +8,6 @@ from .ghost_membership_entities import (
    GhostNewsletter, UserNewsletter,
    GhostTier, GhostSubscription,
 )
-from .domain_event import DomainEvent
-
 from .ghost_content import Tag, Post, Author, PostAuthor, PostTag, PostLike
 from .page_config import PageConfig
 from .order import Order, OrderItem
@@ -29,4 +27,5 @@ from .container_relation import ContainerRelation
 from .menu_node import MenuNode
 from .federation import (
    ActorProfile, APActivity, APFollower, APInboxItem, APAnchor, IPFSPin,
+    RemoteActor, APFollowing, APRemotePost, APLocalPost, APInteraction, APNotification,
 )
--- a/models/federation.py
+++ b/models/federation.py
@@ -50,14 +50,19 @@ class ActorProfile(Base):


 class APActivity(Base):
-    """An ActivityPub activity (local or remote)."""
+    """An ActivityPub activity (local or remote).
+
+    Also serves as the unified event bus: internal domain events and public
+    federation activities both live here, distinguished by ``visibility``.
+    The ``EventProcessor`` polls rows with ``process_state='pending'``.
+    """
    __tablename__ = "ap_activities"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    activity_id: Mapped[str] = mapped_column(String(512), unique=True, nullable=False)
    activity_type: Mapped[str] = mapped_column(String(64), nullable=False)
-    actor_profile_id: Mapped[int] = mapped_column(
-        Integer, ForeignKey("ap_actor_profiles.id", ondelete="CASCADE"), nullable=False,
+    actor_profile_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("ap_actor_profiles.id", ondelete="CASCADE"), nullable=True,
    )
    object_type: Mapped[str | None] = mapped_column(String(64), nullable=True)
    object_data: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
@@ -83,6 +88,27 @@ class APActivity(Base):
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )

+    # --- Unified event-bus columns ---
+    actor_uri: Mapped[str | None] = mapped_column(
+        String(512), nullable=True,
+    )
+    visibility: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="public", server_default="public",
+    )
+    process_state: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="completed", server_default="completed",
+    )
+    process_attempts: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default="0",
+    )
+    process_max_attempts: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=5, server_default="5",
+    )
+    process_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    processed_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+    )
+
    # Relationships
    actor_profile = relationship("ActorProfile", back_populates="activities")

@@ -90,6 +116,7 @@ class APActivity(Base):
        Index("ix_ap_activity_actor", "actor_profile_id"),
        Index("ix_ap_activity_source", "source_type", "source_id"),
        Index("ix_ap_activity_published", "published"),
+        Index("ix_ap_activity_process", "process_state"),
    )

    def __repr__(self) -> str:
@@ -193,3 +220,207 @@ class IPFSPin(Base):

    def __repr__(self) -> str:
        return f"<IPFSPin {self.id} {self.ipfs_cid[:16]}...>"
+
+
+class RemoteActor(Base):
+    """Cached profile of a remote actor we interact with."""
+    __tablename__ = "ap_remote_actors"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    actor_url: Mapped[str] = mapped_column(String(512), unique=True, nullable=False)
+    inbox_url: Mapped[str] = mapped_column(String(512), nullable=False)
+    shared_inbox_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    preferred_username: Mapped[str] = mapped_column(String(255), nullable=False)
+    display_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    summary: Mapped[str | None] = mapped_column(Text, nullable=True)
+    icon_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    public_key_pem: Mapped[str | None] = mapped_column(Text, nullable=True)
+    domain: Mapped[str] = mapped_column(String(255), nullable=False)
+    fetched_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+
+    __table_args__ = (
+        Index("ix_ap_remote_actor_url", "actor_url", unique=True),
+        Index("ix_ap_remote_actor_domain", "domain"),
+    )
+
+    def __repr__(self) -> str:
+        return f"<RemoteActor {self.id} {self.preferred_username}@{self.domain}>"
+
+
+class APFollowing(Base):
+    """Outbound follow: local actor → remote actor."""
+    __tablename__ = "ap_following"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    actor_profile_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("ap_actor_profiles.id", ondelete="CASCADE"), nullable=False,
+    )
+    remote_actor_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("ap_remote_actors.id", ondelete="CASCADE"), nullable=False,
+    )
+    state: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="pending", server_default="pending",
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+    accepted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    # Relationships
+    actor_profile = relationship("ActorProfile")
+    remote_actor = relationship("RemoteActor")
+
+    __table_args__ = (
+        UniqueConstraint("actor_profile_id", "remote_actor_id", name="uq_following"),
+        Index("ix_ap_following_actor", "actor_profile_id"),
+        Index("ix_ap_following_remote", "remote_actor_id"),
+    )
+
+    def __repr__(self) -> str:
+        return f"<APFollowing {self.id} [{self.state}]>"
+
+
+class APRemotePost(Base):
+    """A federated post ingested from a remote actor."""
+    __tablename__ = "ap_remote_posts"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    remote_actor_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("ap_remote_actors.id", ondelete="CASCADE"), nullable=False,
+    )
+    activity_id: Mapped[str] = mapped_column(String(512), unique=True, nullable=False)
+    object_id: Mapped[str] = mapped_column(String(512), unique=True, nullable=False)
+    object_type: Mapped[str] = mapped_column(String(64), nullable=False, default="Note")
+    content: Mapped[str | None] = mapped_column(Text, nullable=True)
+    summary: Mapped[str | None] = mapped_column(Text, nullable=True)
+    url: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    attachment_data: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
+    tag_data: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
+    in_reply_to: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    conversation: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    published: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    fetched_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+
+    # Relationships
+    remote_actor = relationship("RemoteActor")
+
+    __table_args__ = (
+        Index("ix_ap_remote_post_actor", "remote_actor_id"),
+        Index("ix_ap_remote_post_published", "published"),
+        Index("ix_ap_remote_post_object", "object_id", unique=True),
+    )
+
+    def __repr__(self) -> str:
+        return f"<APRemotePost {self.id} {self.object_type}>"
+
+
+class APLocalPost(Base):
+    """A native post composed in the federation UI."""
+    __tablename__ = "ap_local_posts"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    actor_profile_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("ap_actor_profiles.id", ondelete="CASCADE"), nullable=False,
+    )
+    content: Mapped[str] = mapped_column(Text, nullable=False)
+    visibility: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="public", server_default="public",
+    )
+    in_reply_to: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    published: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(),
+    )
+
+    # Relationships
+    actor_profile = relationship("ActorProfile")
+
+    __table_args__ = (
+        Index("ix_ap_local_post_actor", "actor_profile_id"),
+        Index("ix_ap_local_post_published", "published"),
+    )
+
+    def __repr__(self) -> str:
+        return f"<APLocalPost {self.id}>"
+
+
+class APInteraction(Base):
+    """Like or boost (local or remote)."""
+    __tablename__ = "ap_interactions"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    actor_profile_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("ap_actor_profiles.id", ondelete="CASCADE"), nullable=True,
+    )
+    remote_actor_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("ap_remote_actors.id", ondelete="CASCADE"), nullable=True,
+    )
+    post_type: Mapped[str] = mapped_column(String(20), nullable=False)  # local/remote
+    post_id: Mapped[int] = mapped_column(Integer, nullable=False)
+    interaction_type: Mapped[str] = mapped_column(String(20), nullable=False)  # like/boost
+    activity_id: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+
+    __table_args__ = (
+        Index("ix_ap_interaction_post", "post_type", "post_id"),
+        Index("ix_ap_interaction_actor", "actor_profile_id"),
+        Index("ix_ap_interaction_remote", "remote_actor_id"),
+    )
+
+    def __repr__(self) -> str:
+        return f"<APInteraction {self.id} {self.interaction_type}>"
+
+
+class APNotification(Base):
+    """Notification for a local actor."""
+    __tablename__ = "ap_notifications"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    actor_profile_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("ap_actor_profiles.id", ondelete="CASCADE"), nullable=False,
+    )
+    notification_type: Mapped[str] = mapped_column(String(20), nullable=False)
+    from_remote_actor_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("ap_remote_actors.id", ondelete="SET NULL"), nullable=True,
+    )
+    from_actor_profile_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("ap_actor_profiles.id", ondelete="SET NULL"), nullable=True,
+    )
+    target_activity_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("ap_activities.id", ondelete="SET NULL"), nullable=True,
+    )
+    target_remote_post_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("ap_remote_posts.id", ondelete="SET NULL"), nullable=True,
+    )
+    read: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="false")
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+
+    # Relationships
+    actor_profile = relationship("ActorProfile", foreign_keys=[actor_profile_id])
+    from_remote_actor = relationship("RemoteActor")
+    from_actor_profile = relationship("ActorProfile", foreign_keys=[from_actor_profile_id])
+
+    __table_args__ = (
+        Index("ix_ap_notification_actor", "actor_profile_id"),
+        Index("ix_ap_notification_read", "actor_profile_id", "read"),
+        Index("ix_ap_notification_created", "created_at"),
+    )
--- a/models/menu_item.py
+++ b/models/menu_item.py
@@ -6,7 +6,7 @@ from shared.db.base import Base


 class MenuItem(Base):
-    """Deprecated — kept so the table isn't dropped. Use glue.models.MenuNode."""
+    """Deprecated — kept so the table isn't dropped. Use shared.models.menu_node.MenuNode."""
    __tablename__ = "menu_items"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
--- a/services/calendar_impl.py
+++ b/services/calendar_impl.py
@@ -371,7 +371,7 @@ class SqlCalendarService:
            entries_by_post.setdefault(post_id, []).append(_entry_to_dto(entry))
        return entries_by_post

-    # -- writes (absorb glue lifecycle) ---------------------------------------
+    # -- writes ---------------------------------------------------------------

    async def adopt_entries_for_user(
        self, session: AsyncSession, user_id: int, session_id: str,
--- a/services/federation_impl.py
+++ b/services/federation_impl.py
--- a/services/federation_publish.py
+++ b/services/federation_publish.py
@@ -1,13 +1,14 @@
 """Inline federation publication — called at write time, not via async handler.

-Replaces the old pattern where emit_event("post.published") → async handler →
-publish_activity().  Now the originating service calls try_publish() directly,
-which creates the APActivity in the same DB transaction.  AP delivery
-(federation.activity_created → inbox POST) stays async.
+The originating service calls try_publish() directly, which creates the
+APActivity (with process_state='pending') in the same DB transaction.
+The EventProcessor picks it up and the delivery wildcard handler POSTs
+to follower inboxes.
 """
 from __future__ import annotations

 import logging
+import os

 from sqlalchemy.ext.asyncio import AsyncSession

@@ -48,10 +49,20 @@ async def try_publish(
    if existing:
        if activity_type == "Create" and existing.activity_type != "Delete":
            return  # already published (allow re-Create after Delete/unpublish)
+        if activity_type == "Update" and existing.activity_type == "Update":
+            return  # already updated (Ghost fires duplicate webhooks)
        if activity_type == "Delete" and existing.activity_type == "Delete":
            return  # already deleted
-    elif activity_type == "Delete":
-        return  # never published, nothing to delete
+    elif activity_type in ("Delete", "Update"):
+        return  # never published, nothing to delete/update
+
+    # Stable object ID: same source always gets the same object id so
+    # Mastodon treats Create/Update/Delete as the same post.
+    domain = os.getenv("AP_DOMAIN", "rose-ash.com")
+    object_data["id"] = (
+        f"https://{domain}/users/{actor.preferred_username}"
+        f"/objects/{source_type.lower()}/{source_id}"
+    )

    try:
        await services.federation.publish_activity(
--- a/services/relationships.py
+++ b/services/relationships.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 from sqlalchemy import select, func
 from sqlalchemy.ext.asyncio import AsyncSession

-from shared.events import emit_event
+from shared.events import emit_activity
 from shared.models.container_relation import ContainerRelation


@@ -40,17 +40,19 @@ async def attach_child(
            if label is not None:
                existing.label = label
            await session.flush()
-            await emit_event(
+            await emit_activity(
                session,
-                event_type="container.child_attached",
-                aggregate_type="container_relation",
-                aggregate_id=existing.id,
-                payload={
+                activity_type="Add",
+                actor_uri="internal:system",
+                object_type="rose:ContainerRelation",
+                object_data={
                    "parent_type": parent_type,
                    "parent_id": parent_id,
                    "child_type": child_type,
                    "child_id": child_id,
                },
+                source_type="container_relation",
+                source_id=existing.id,
            )
            return existing
        # Already attached and active — no-op
@@ -77,17 +79,19 @@ async def attach_child(
    session.add(rel)
    await session.flush()

-    await emit_event(
+    await emit_activity(
        session,
-        event_type="container.child_attached",
-        aggregate_type="container_relation",
-        aggregate_id=rel.id,
-        payload={
+        activity_type="Add",
+        actor_uri="internal:system",
+        object_type="rose:ContainerRelation",
+        object_data={
            "parent_type": parent_type,
            "parent_id": parent_id,
            "child_type": child_type,
            "child_id": child_id,
        },
+        source_type="container_relation",
+        source_id=rel.id,
    )

    return rel
@@ -139,17 +143,19 @@ async def detach_child(
    rel.deleted_at = func.now()
    await session.flush()

-    await emit_event(
+    await emit_activity(
        session,
-        event_type="container.child_detached",
-        aggregate_type="container_relation",
-        aggregate_id=rel.id,
-        payload={
+        activity_type="Remove",
+        actor_uri="internal:system",
+        object_type="rose:ContainerRelation",
+        object_data={
            "parent_type": parent_type,
            "parent_id": parent_id,
            "child_type": child_type,
            "child_id": child_id,
        },
+        source_type="container_relation",
+        source_id=rel.id,
    )

    return True
--- a/services/stubs.py
+++ b/services/stubs.py
@@ -227,5 +227,71 @@ class StubFederationService:
    async def remove_follower(self, session, username, follower_acct):
        return False

+    async def get_or_fetch_remote_actor(self, session, actor_url):
+        return None
+
+    async def search_remote_actor(self, session, acct):
+        return None
+
+    async def send_follow(self, session, local_username, remote_actor_url):
+        raise RuntimeError("FederationService not available")
+
+    async def get_following(self, session, username, page=1, per_page=20):
+        return [], 0
+
+    async def get_followers_paginated(self, session, username, page=1, per_page=20):
+        return [], 0
+
+    async def accept_follow_response(self, session, local_username, remote_actor_url):
+        pass
+
+    async def unfollow(self, session, local_username, remote_actor_url):
+        pass
+
+    async def ingest_remote_post(self, session, remote_actor_id, activity_json, object_json):
+        pass
+
+    async def delete_remote_post(self, session, object_id):
+        pass
+
+    async def get_remote_post(self, session, object_id):
+        return None
+
+    async def get_home_timeline(self, session, actor_profile_id, before=None, limit=20):
+        return []
+
+    async def get_public_timeline(self, session, before=None, limit=20):
+        return []
+
+    async def get_actor_timeline(self, session, remote_actor_id, before=None, limit=20):
+        return []
+
+    async def create_local_post(self, session, actor_profile_id, content, visibility="public", in_reply_to=None):
+        raise RuntimeError("FederationService not available")
+
+    async def delete_local_post(self, session, actor_profile_id, post_id):
+        raise RuntimeError("FederationService not available")
+
+    async def like_post(self, session, actor_profile_id, object_id, author_inbox):
+        pass
+
+    async def unlike_post(self, session, actor_profile_id, object_id, author_inbox):
+        pass
+
+    async def boost_post(self, session, actor_profile_id, object_id, author_inbox):
+        pass
+
+    async def unboost_post(self, session, actor_profile_id, object_id, author_inbox):
+        pass
+
+    async def get_notifications(self, session, actor_profile_id, before=None, limit=20):
+        return []
+
+    async def unread_notification_count(self, session, actor_profile_id):
+        return 0
+
+    async def mark_notifications_read(self, session, actor_profile_id):
+        pass
+
    async def get_stats(self, session):
        return {"actors": 0, "activities": 0, "followers": 0}
--- a/utils/anchoring.py
+++ b/utils/anchoring.py
@@ -145,7 +145,7 @@ async def upgrade_ots_proof(proof_bytes: bytes) -> tuple[bytes, bool]:
    """
    # OpenTimestamps upgrade is done via the `ots` CLI or their calendar API.
    # For now, return the proof as-is with is_confirmed=False.
-    # TODO: Implement calendar-based upgrade polling.
+    # Calendar-based upgrade polling not yet implemented.
    return proof_bytes, False


--- a/utils/webfinger.py
+++ b/utils/webfinger.py
@@ -0,0 +1,68 @@
+"""WebFinger client for resolving remote AP actor profiles."""
+from __future__ import annotations
+
+import logging
+
+import httpx
+
+log = logging.getLogger(__name__)
+
+AP_CONTENT_TYPE = "application/activity+json"
+
+
+async def resolve_actor(acct: str) -> dict | None:
+    """Resolve user@domain to actor JSON via WebFinger + actor fetch.
+
+    Args:
+        acct: Handle in the form ``user@domain`` (no leading ``@``).
+
+    Returns:
+        Actor JSON-LD dict, or None if resolution fails.
+    """
+    acct = acct.lstrip("@")
+    if "@" not in acct:
+        return None
+
+    _, domain = acct.rsplit("@", 1)
+    webfinger_url = f"https://{domain}/.well-known/webfinger"
+
+    try:
+        async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client:
+            # Step 1: WebFinger lookup
+            resp = await client.get(
+                webfinger_url,
+                params={"resource": f"acct:{acct}"},
+                headers={"Accept": "application/jrd+json, application/json"},
+            )
+            if resp.status_code != 200:
+                log.debug("WebFinger %s returned %d", webfinger_url, resp.status_code)
+                return None
+
+            data = resp.json()
+            # Find self link with AP content type
+            actor_url = None
+            for link in data.get("links", []):
+                if link.get("rel") == "self" and link.get("type") in (
+                    AP_CONTENT_TYPE,
+                    "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
+                ):
+                    actor_url = link.get("href")
+                    break
+
+            if not actor_url:
+                log.debug("No AP self link in WebFinger response for %s", acct)
+                return None
+
+            # Step 2: Fetch actor JSON
+            resp = await client.get(
+                actor_url,
+                headers={"Accept": AP_CONTENT_TYPE},
+            )
+            if resp.status_code == 200:
+                return resp.json()
+
+            log.debug("Actor fetch %s returned %d", actor_url, resp.status_code)
+    except Exception:
+        log.exception("WebFinger resolution failed for %s", acct)
+
+    return None