paskia/db/operations.py

@@ -23,6 +23,7 @@ from paskia.db.structs import (
     Org,
     OriginEntry,
     Permission,
+    RemoteConfig,
     ResetToken,
     Role,
     Session,
@@ -728,9 +729,10 @@ def update_domain(
     *,
     rp_name: str | None,
     origins: dict[str, bool | OriginEntry],
+    remote: RemoteConfig | None = None,
     ctx: SessionContext | None = None,
 ) -> None:
-    """Replace a domain's rp_name and origins table (wholesale).
+    """Replace a domain's rp_name, origins table and remote (wholesale).
 
     The rp-id itself is immutable: credentials are stamped with it, so
     changing it would orphan them — delete and recreate the domain instead.
@@ -742,6 +744,7 @@ def update_domain(
     with _transaction("admin:update_domain", ctx):
         domain.rp_name = rp_name
         domain.origins = origins
+        domain.remote = remote
 
 
 def delete_domain(rp_id: str, *, ctx: SessionContext | None = None) -> None:

paskia/fastapi/admin/domains.py

@@ -12,7 +12,7 @@ immediately.
 from fastapi import Body, FastAPI, Request
 
 from paskia import db, domains
-from paskia.db.structs import Config, DomainConfig, OriginEntry
+from paskia.db.structs import Config, DomainConfig, OriginEntry, RemoteConfig
 from paskia.fastapi import authz
 from paskia.fastapi.admin.errors import install_error_handlers
 from paskia.fastapi.response import MsgspecResponse
@@ -27,6 +27,14 @@ install_error_handlers(app)
 
 
 def _domain_to_api(domain: domains.Domain) -> ApiDomain:
+    remote = domain.config.remote
+    if remote is not None:
+        # The sync token is a bearer secret: never echoed back
+        remote = RemoteConfig(
+            url=remote.url,
+            cache_ttl=remote.cache_ttl,
+            refresh_interval=remote.refresh_interval,
+        )
     return ApiDomain(
         rp_id=domain.rp_id,
         rp_name=domain.rp_name,
@@ -34,6 +42,26 @@ def _domain_to_api(domain: domains.Domain) -> ApiDomain:
         site_url=domain.site_url,
         auth_site_url=domain.auth_site_url,
         auth_host=domain.own_auth_host,
+        remote=remote,
+    )
+
+
+def _normalize_remote(value, existing: RemoteConfig | None = None) -> RemoteConfig | None:
+    """Parse a remote object from the admin UI (raises on malformed).
+
+    An absent/empty token keeps the previously stored one — the token is
+    write-only over the API.
+    """
+    if value is None:
+        return None
+    if not isinstance(value, dict) or not isinstance(value.get("url"), str):
+        raise ValueError("remote must be an object with a url")
+    token = str(value.get("token") or "") or (existing.token if existing else "")
+    return RemoteConfig(
+        url=value["url"].rstrip("/"),
+        token=token,
+        cache_ttl=int(value.get("cache_ttl") or 60),
+        refresh_interval=int(value.get("refresh_interval") or 300),
     )
 
 
@@ -123,6 +151,7 @@ async def admin_create_domain(
     new = DomainConfig(
         rp_name=(payload.get("rp_name") or "").strip() or None,
         origins=_normalize_origins_map(payload.get("origins")),
+        remote=_normalize_remote(payload.get("remote")),
     )
 
     config = db.data().config
@@ -156,9 +185,15 @@ async def admin_update_domain(
     if rp_id not in config.domains:
         raise ValueError(f"Domain {rp_id} not found")
 
+    current_remote = config.domains[rp_id].remote
     updated = DomainConfig(
         rp_name=(payload.get("rp_name") or "").strip() or None,
         origins=_normalize_origins_map(payload.get("origins")),
+        remote=(
+            _normalize_remote(payload["remote"], existing=current_remote)
+            if "remote" in payload
+            else current_remote
+        ),
     )
     would_be = Config(
         domains={k: updated if k == rp_id else v for k, v in config.domains.items()},
@@ -171,6 +206,7 @@ async def admin_update_domain(
         rp_id,
         rp_name=updated.rp_name,
         origins=updated.origins,
+        remote=updated.remote,
         ctx=ctx,
     )
     _rebuild_registry()

paskia/syncfeed.py

@@ -55,7 +55,10 @@ class SyncFeed:
                 self.subscribers.discard(queue)
 
     def replay_since(self, seq: int) -> list[dict] | None:
-        """Events after seq, or None when the ring no longer reaches back."""
+        """Events after seq, or None when the ring no longer reaches back
+        (or the claimed seq is ahead of us, which cannot be reconciled)."""
+        if seq > self.seq:
+            return None
         if not self.events:
             return [] if seq == self.seq else None
         oldest = self.events[0]["seq"]

paskia/util/apistructs.py

@@ -14,7 +14,15 @@ import msgspec
 from uarite import uaparse
 
 from paskia import db
-from paskia.db.structs import Credential, Org, OriginEntry, Permission, Role, User
+from paskia.db.structs import (
+    Credential,
+    Org,
+    OriginEntry,
+    Permission,
+    RemoteConfig,
+    Role,
+    User,
+)
 
 # -------------------------------------------------------------------------
 # API structs - inherit from db structs, add uuid for serialization
@@ -194,6 +202,7 @@ class ApiDomain(msgspec.Struct):
     site_url: str
     auth_site_url: str
     auth_host: str | None
+    remote: RemoteConfig | None = None
 
 
 class ApiTokenInfo(msgspec.Struct, omit_defaults=True):

tests/test_remote.py

@@ -0,0 +1,198 @@
+"""Tests for remote (satellite) domains: config, replica application, feed."""
+
+import asyncio
+import os
+import secrets
+from datetime import UTC, datetime
+from uuid import UUID
+
+import pytest
+
+import paskia.db.operations as ops_db
+from paskia import domains, satellite, syncfeed
+from paskia.db.structs import (
+    Config,
+    Credential,
+    DomainConfig,
+    OriginEntry,
+    RemoteConfig,
+    Session,
+    User,
+)
+from paskia.db.structs import DB
+from paskia.util.crypto import hash_secret
+
+from .conftest import TEST_RP_ID
+
+REMOTE_URL = "http://remote.test"
+
+
+def _remote_domain_config(**kw) -> Config:
+    return Config(
+        domains={
+            TEST_RP_ID: DomainConfig(origins={f"**.{TEST_RP_ID}": True}),
+            "example.com": DomainConfig(
+                origins={
+                    "**.example.com": True,
+                    "auth.example.com": OriginEntry(auth_host=True),
+                },
+                remote=RemoteConfig(url=REMOTE_URL, token="t", **kw),
+            ),
+        }
+    )
+
+
+def test_remote_domain_valid():
+    domains.validate_config(_remote_domain_config())
+
+
+def test_remote_domain_requires_auth_host():
+    config = _remote_domain_config()
+    config.domains["example.com"].origins = {"**.example.com": True}
+    with pytest.raises(ValueError, match="auth host"):
+        domains.validate_config(config)
+
+
+def test_remote_domain_requires_http_url():
+    config = _remote_domain_config()
+    config.domains["example.com"].remote.url = "ftp://x"
+    with pytest.raises(ValueError, match="http"):
+        domains.validate_config(config)
+
+
+def test_sanitize_preserves_remote():
+    config, warnings = domains.sanitize_config(_remote_domain_config())
+    assert not warnings
+    assert config.domains["example.com"].remote.url == REMOTE_URL
+
+
+def test_apply_upsert_and_delete():
+    replica = DB()
+    user = User.create(display_name="U", role=UUID(int=1))
+    user.uuid = UUID(int=2)
+    satellite._apply(
+        replica, "users", str(user.uuid), "upsert", _builtins(user)
+    )
+    assert replica.users[user.uuid].display_name == "U"
+    satellite._apply(replica, "users", str(user.uuid), "delete", None)
+    assert not replica.users
+
+
+def _builtins(obj):
+    import msgspec
+
+    return msgspec.to_builtins(obj)
+
+
+def test_apply_session_roundtrip():
+    """Sessions keep their string key and datetime/UUID fields."""
+    replica = DB()
+    session = Session.create(
+        user=UUID(int=1),
+        credential=UUID(int=2),
+        key=hash_secret("cookie", "sekret"),
+        host="app2.example.com",
+        ip="127.0.0.1",
+        user_agent="ua",
+        validated=datetime.now(UTC),
+        rp_id="example.com",
+    )
+    satellite._apply(replica, "sessions", session.key, "upsert", _builtins(session))
+    stored = replica.sessions[session.key]
+    assert stored.host == "app2.example.com"
+    assert stored.validated == session.validated
+    assert stored.user_uuid == UUID(int=1)
+
+
+def test_apply_credential_bytes_roundtrip():
+    """credential_id/public_key are bytes over the wire (base64 in JSON)."""
+    replica = DB()
+    cred = Credential.create(
+        credential_id=os.urandom(32),
+        user=UUID(int=1),
+        aaguid=UUID(int=0),
+        public_key=os.urandom(64),
+        sign_count=3,
+        rp_id="example.com",
+    )
+    cred.uuid = UUID(int=9)
+    # Simulate the full wire path: builtins -> JSON -> builtins
+    import msgspec
+
+    wire = msgspec.json.decode(msgspec.json.encode(_builtins(cred)))
+    satellite._apply(replica, "credentials", str(cred.uuid), "upsert", wire)
+    stored = replica.credentials[cred.uuid]
+    assert stored.credential_id == cred.credential_id
+    assert stored.public_key == cred.public_key
+    assert stored.sign_count == 3
+
+
+def test_feed_emit_and_replay():
+    feed = syncfeed.SyncFeed()
+    user = User.create(display_name="A", role=UUID(int=1))
+    feed.emit("users", "k1", user)
+    feed.emit("users", "k1", None)
+    assert feed.seq == 2
+    assert feed.replay_since(0)[0]["op"] == "upsert"
+    assert feed.replay_since(1)[0]["op"] == "delete"
+    assert feed.replay_since(2) == []
+    assert feed.replay_since(99) is None
+
+
+def test_feed_ring_overflow_replay_none():
+    feed = syncfeed.SyncFeed()
+    feed.events = __import__("collections").deque(maxlen=3)
+    for i in range(5):
+        feed.emit("users", f"k{i}", None)
+    assert feed.replay_since(0) is None  # fell off the ring
+    assert [e["seq"] for e in feed.replay_since(4)] == [5]
+    assert feed.replay_since(5) == []
+
+
+@pytest.mark.asyncio
+async def test_operations_emit_events(test_db):
+    """Writes through db.operations land on the sync feed."""
+    syncfeed.feed.events.clear()
+    syncfeed.feed.seq = 0
+    user = next(iter(test_db.users.values()))
+    ops_db.update_user_display_name(user.uuid, "Renamed")
+    tables = {e["table"] for e in syncfeed.feed.events}
+    assert "users" in tables
+    key = syncfeed.feed.events[-1]["key"]
+    assert syncfeed.feed.events[-1]["fields"]["display_name"] == "Renamed"
+    assert key == str(user.uuid)
+
+
+@pytest.mark.asyncio
+async def test_replica_refresh_and_evict():
+    replica = satellite.RemoteReplica(RemoteConfig(url=REMOTE_URL, token="t"))
+    token = secrets.token_urlsafe(12)
+    session = Session.create(
+        user=UUID(int=1),
+        credential=UUID(int=2),
+        key=hash_secret("cookie", token),
+        host="app2.example.com",
+        ip="1.1.1.1",
+        user_agent="ua",
+        validated=datetime(2020, 1, 1, tzinfo=UTC),
+    )
+    replica.db.sessions[session.key] = session
+
+    now = datetime.now(UTC)
+    replica.refresh_session(session.key, now, "2.2.2.2", "new-ua")
+    assert replica.db.sessions[session.key].validated == now
+    queued = replica._pending_refresh[session.key]
+    assert queued["type"] == "session_refresh"
+    assert queued["ip"] == "2.2.2.2"
+
+    replica.evict_session(token)
+    assert not replica.db.sessions
+
+
+def test_availability_gate():
+    replica = satellite.RemoteReplica(
+        RemoteConfig(url=REMOTE_URL, token="t", cache_ttl=60)
+    )
+    assert not replica.available()  # never synced
+    replica.last_contact = __import__("time").monotonic()
+    assert replica.available()