Implement identity graph stitching and configurable profile storage

Issue persistent signed anonymous IDs and accept cookie/token-bound events.
Resolve verified email, phone, OAuth, and namespaced third-party IDs into
transactionally merged master identities with historical event aliases.
Protect shared-device account switching, tenant isolation, and event retries.

Store sparse canonical profiles as PostgreSQL JSONB or local SQLite JSON.
Support an optional versioned Cassandra wide-column projection through a
transactional outbox, retry worker, and immediate canonical read fallback.
Document deployment settings and provide opt-in database integration tests.

Validation: 43 local tests passed; 6 external database tests skipped.
Ruff checks, formatting, and offline uv lock verification passed.

tests/test_projection.py

@@ -0,0 +1,146 @@
+import copy
+from dataclasses import replace
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.schema import CreateTable
+
+from adcloud_server.database import Database, outbox, profiles
+from adcloud_server.identity_models import ProfilePatch, StitchRequest
+from adcloud_server.identity_service import IdentityService
+from adcloud_server.projection import ProfileProjector
+from adcloud_server.settings import Settings
+
+
+class RecoverableStore:
+    def __init__(self):
+        self.snapshots = {}
+        self.unavailable = False
+
+    def put(self, tenant, master, revision, document):
+        if self.unavailable:
+            raise OSError("Simulated storage outage")
+        self.snapshots[tenant, master, revision] = copy.deepcopy(document)
+
+    def get(self, tenant, master, revision):
+        if self.unavailable:
+            raise OSError("Simulated storage outage")
+        return self.snapshots.get((tenant, master, revision))
+
+
+@pytest.fixture
+def components(tmp_path):
+    settings = Settings(
+        database_url="sqlite:///" + str(tmp_path / "profiles.db"),
+        identity_secret="x" * 48,
+        profile_backend="cassandra",
+    )
+    db = Database(settings.database_url, settings.tenant_id)
+    db.initialize()
+    store = RecoverableStore()
+    service = IdentityService(settings, db, store)
+    try:
+        yield service, ProfileProjector(db, store), store
+    finally:
+        db.close()
+
+
+def test_wide_table_outage_preserves_immediate_stitched_profile_and_retries(components):
+    service, worker, store = components
+    anon = service.anonymous()
+    result = service.stitch(
+        anon["anonymous_token"],
+        StitchRequest(
+            identifiers=[{"kind": "email", "value": "alice@example.com"}],
+            attributes={"interest": "running"},
+            tags=["runner"],
+        ),
+    )
+    assert service.profile(result["master_id"])["projection_status"] == "pending"
+    store.unavailable = True
+    assert service.profile(result["master_id"])["attributes"] == {"interest": "running"}
+    with pytest.raises(OSError):
+        worker.flush()
+    with service.db.transaction() as conn:
+        pending = (
+            conn.execute(select(outbox).where(outbox.c.delivered == 0)).mappings().all()
+        )
+        assert any(job["revision"] == result["revision"] for job in pending)
+    store.unavailable = False
+    worker.flush()
+    current = service.profile(result["master_id"])
+    assert current["projection_status"] == "current"
+    assert current["tags"] == ["runner"]
+    assert worker.flush() == 0
+
+
+def test_out_of_order_projection_cannot_resurrect_deleted_attributes(components):
+    service, worker, store = components
+    anon = service.anonymous()
+    first = service.patch_profile(
+        anon["master_id"], ProfilePatch(attributes={"remove": "old", "keep": 1})
+    )
+    store.put(
+        service.tenant,
+        anon["master_id"],
+        first["revision"],
+        {"attributes": first["attributes"], "tags": []},
+    )
+    last = service.patch_profile(
+        anon["master_id"], ProfilePatch(attributes={"remove": None})
+    )
+    # A stale snapshot never satisfies the latest revision lookup.
+    assert service.profile(anon["master_id"])["attributes"] == {"keep": 1}
+    assert service.profile(anon["master_id"])["projection_status"] == "pending"
+    worker.flush()
+    assert service.profile(anon["master_id"])["revision"] == last["revision"]
+    assert service.profile(anon["master_id"])["attributes"] == {"keep": 1}
+
+
+def test_relational_to_wide_table_switch_backfills_existing_profiles(components):
+    service, worker, _store = components
+    relational = IdentityService(
+        replace(service.settings, profile_backend="relational"), service.db
+    )
+    anon = relational.anonymous()
+    relational.patch_profile(
+        anon["master_id"], ProfilePatch(attributes={"nested": {"count": 3}})
+    )
+    with service.db.transaction() as conn:
+        assert conn.execute(select(outbox)).first() is None
+    assert worker.backfill() == 1
+    worker.flush()
+    assert service.profile(anon["master_id"])["projection_status"] == "current"
+    assert service.profile(anon["master_id"])["attributes"] == {"nested": {"count": 3}}
+
+
+def test_merge_failure_never_leaves_partial_outbox_records(components, monkeypatch):
+    service, worker, _store = components
+    a = service.anonymous()
+    service.stitch(
+        a["anonymous_token"],
+        StitchRequest(identifiers=[{"kind": "email", "value": "alice@example.com"}]),
+    )
+    b = service.anonymous(rotate=True)
+    worker.flush()
+
+    def fail(*args):
+        raise OSError("Simulated transaction failure")
+
+    monkeypatch.setattr(service, "_queue", fail)
+    with pytest.raises(OSError):
+        service.stitch(
+            b["anonymous_token"],
+            StitchRequest(
+                identifiers=[{"kind": "email", "value": "alice@example.com"}]
+            ),
+        )
+    assert service.profile(b["master_id"])["master_id"] == b["master_id"]
+    assert worker.flush() == 0
+    assert len(service.graph(a["master_id"])["member_ids"]) == 1
+
+
+def test_postgres_document_column_is_jsonb():
+    ddl = str(CreateTable(profiles).compile(dialect=postgresql.dialect()))
+    assert "document JSONB NOT NULL" in ddl