Add versioned event registry and transactional feature rules

tests/test_event_registry.py

@@ -0,0 +1,479 @@
+from datetime import UTC, datetime, timedelta
+from uuid import uuid4
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from adcloud_server.eventing.catalog import EVENTS
+from adcloud_server.eventing.tables import contributions, receipts
+from adcloud_server.eventing.worker import FeatureWorker, feature_values
+from adcloud_server.main import create_app
+from adcloud_server.settings import Settings
+
+AUTH = {"Authorization": "Bearer " + "service-" + "x" * 48}
+CONFIG = {
+    "standard_event": "commerce.cart.add",
+    "description": "加购数码商品",
+    "rules": [
+        {
+            "conditions": [
+                {
+                    "field": "properties.category_id",
+                    "operator": "eq",
+                    "value": "digital",
+                }
+            ],
+            "effects": [{"feature": "intent.purchase.digital", "value": 1}],
+        }
+    ],
+}
+
+
+@pytest.fixture(params=["sqlite", "postgres"])
+def client(request, tmp_path):
+    import os
+
+    url = "sqlite:///" + str(tmp_path / "events.db")
+    if request.param == "postgres":
+        url = os.getenv("ADCLOUD_TEST_POSTGRES_URL")
+        if not url:
+            pytest.skip("No PostgreSQL integration URL configured")
+    settings = Settings(
+        database_url=url,
+        tenant_id="registry-" + uuid4().hex,
+        identity_secret="identity-" + "x" * 48,
+        service_token="service-" + "x" * 48,
+        cookie_secure=False,
+    )
+    with TestClient(create_app(settings)) as test_client:
+        yield test_client
+
+
+def app(client, app_id="shop"):
+    response = client.post(
+        "/v1/registry/apps", headers=AUTH, json={"app_id": app_id, "name": "测试商店"}
+    )
+    assert response.status_code == 201, response.text
+    return response.json()
+
+
+def path(application, name="add_cart", version=1):
+    return f"/v1/registry/apps/{application['app_id']}/events/{name}/versions/{version}"
+
+
+def manage(application):
+    return {"X-Adcloud-Management-Key": application["management_key"]}
+
+
+def register(client, application, config=None, name="add_cart"):
+    response = client.put(
+        path(application, name), json=config or CONFIG, headers=manage(application)
+    )
+    assert response.status_code == 200, response.text
+    response = client.post(
+        path(application, name) + "/publish", headers=manage(application)
+    )
+    assert response.status_code == 200, response.text
+    return response.json()
+
+
+def event(**kwargs):
+    return {
+        "app_id": "shop",
+        "name": "add_cart",
+        "event_version": 1,
+        "event_id": str(uuid4()),
+        "occurred_at": datetime.now(UTC).isoformat(),
+        "properties": {"product_id": "camera", "quantity": 1, "category_id": "digital"},
+        **kwargs,
+    }
+
+
+def send(client, application, body=None, server=False):
+    return client.post(
+        "/v1/events",
+        headers={
+            "X-Adcloud-Key": application["server_key" if server else "collection_key"]
+        },
+        json=body or event(),
+    )
+
+
+def worker(client):
+    return FeatureWorker(client.app.state.identity_service)
+
+
+def test_registration_preview_requires_category_and_scope(client):
+    application = app(client)
+    other = app(client, "other")
+    assert (
+        client.put(path(application), json=CONFIG, headers=manage(other)).status_code
+        == 401
+    )
+    saved = client.put(
+        path(application), json=CONFIG, headers=manage(application)
+    ).json()
+    assert saved["draft"]["required_fields"] == ["category_id"]
+    assert send(client, application).status_code == 409
+    preview = client.post(
+        path(application) + "/preview",
+        headers=manage(application),
+        json={"properties": event()["properties"]},
+    )
+    assert preview.json()["effects"][0]["value"] == 1
+    bad = event()
+    bad["properties"].pop("category_id")
+    assert (
+        client.post(
+            path(application) + "/preview",
+            headers=manage(application),
+            json={"properties": bad["properties"]},
+        ).status_code
+        == 422
+    )
+    register(client, application)
+    assert send(client, application, bad).status_code == 422
+    assert send(client, application, event(name="unregistered")).status_code == 404
+    assert client.post("/v1/events", json={"name": "page.view"}).status_code == 422
+
+
+def test_receipt_and_worker_are_idempotent_and_exclude_tokens(client):
+    application = app(client)
+    register(client, application)
+    body = event()
+    first = send(client, application, body)
+    assert first.status_code == 200, first.text
+    assert send(client, application, body).json()["duplicate"] is True
+    assert worker(client).flush() == 1
+    assert worker(client).flush() == 0
+    profile = client.get("/v1/users/" + first.json()["master_id"], headers=AUTH).json()
+    assert profile["features"] == {"intent.purchase.digital": 1}
+    with client.app.state.identity_service.db.transaction() as conn:
+        rows = (
+            conn.execute(
+                select(receipts).where(
+                    receipts.c.tenant_id == client.app.state.settings.tenant_id
+                )
+            )
+            .mappings()
+            .all()
+        )
+        assert len(rows) == 1 and "anonymous_token" not in rows[0]["payload"]
+
+
+def test_server_only_events_and_business_deduplication(client):
+    application = app(client)
+    register(
+        client,
+        application,
+        {
+            "standard_event": "commerce.payment.success",
+            "description": "支付成功",
+            "rules": [
+                {"effects": [{"feature": "commerce.purchase.count", "value": 1}]}
+            ],
+        },
+        "paid",
+    )
+    body = event(
+        name="paid",
+        properties={
+            "order_id": "o1",
+            "payment_id": "p1",
+            "amount": 100,
+            "currency": "CNY",
+        },
+    )
+    assert send(client, application, body).status_code == 401
+    assert send(client, application, body, server=True).status_code == 200
+    retry = {**body, "event_id": str(uuid4())}
+    assert send(client, application, retry, server=True).json()["duplicate"] is True
+    assert worker(client).flush() == 1
+    retry["properties"] = {**body["properties"], "amount": 200}
+    assert send(client, application, retry, server=True).status_code == 409
+
+
+def test_frozen_rule_revisions_and_rollback(client):
+    application = app(client)
+    register(client, application)
+    first = send(client, application).json()
+    updated = {
+        **CONFIG,
+        "rules": [
+            {
+                "conditions": CONFIG["rules"][0]["conditions"],
+                "effects": [{"feature": "intent.purchase.digital", "value": 2}],
+            }
+        ],
+    }
+    assert (
+        client.put(
+            path(application), json=updated, headers=manage(application)
+        ).status_code
+        == 200
+    )
+    assert (
+        client.post(path(application) + "/publish", headers=manage(application)).json()[
+            "rule_revision"
+        ]
+        == 2
+    )
+    second = send(client, application).json()
+    worker(client).flush()
+    assert (
+        client.get("/v1/users/" + first["master_id"], headers=AUTH).json()["features"][
+            "intent.purchase.digital"
+        ]
+        == 3
+    )
+    assert first["rule_revision"] == 1 and second["rule_revision"] == 2
+    assert (
+        client.post(
+            path(application) + "/rollback",
+            json={"revision": 1},
+            headers=manage(application),
+        ).status_code
+        == 200
+    )
+    assert send(client, application).json()["rule_revision"] == 1
+    assert (
+        client.post(
+            path(application) + "/disable", headers=manage(application)
+        ).status_code
+        == 200
+    )
+    assert send(client, application).status_code == 409
+
+
+def test_stitch_sums_features_across_anonymous_nodes(client):
+    application = app(client)
+    register(client, application)
+    first = send(client, application).json()
+    worker(client).flush()
+    client.post(
+        "/v1/identity/stitch",
+        headers=AUTH,
+        json={"identifiers": [{"kind": "email", "value": "alice@example.com"}]},
+    )
+    client.post("/v1/identity/anonymous?rotate=true")
+    second = send(client, application).json()
+    worker(client).flush()
+    assert first["anonymous_id"] != second["anonymous_id"]
+    merged = client.post(
+        "/v1/identity/stitch",
+        headers=AUTH,
+        json={"identifiers": [{"kind": "email", "value": "alice@example.com"}]},
+    ).json()
+    assert client.get("/v1/users/" + merged["master_id"], headers=AUTH).json()[
+        "features"
+    ] == {"intent.purchase.digital": 2}
+    assert worker(client).flush() == 0
+
+
+def test_transaction_failure_keeps_task_without_partial_effects(client, monkeypatch):
+    application = app(client)
+    register(client, application)
+    send(client, application)
+    identity = client.app.state.identity_service
+    original = identity._save_profile
+
+    def fail(*args):
+        raise RuntimeError("simulated failure")
+
+    monkeypatch.setattr(identity, "_save_profile", fail)
+    with pytest.raises(RuntimeError):
+        worker(client).flush()
+    with identity.db.transaction() as conn:
+        assert (
+            conn.execute(
+                select(contributions).where(
+                    contributions.c.tenant_id == identity.tenant
+                )
+            ).first()
+            is None
+        )
+        assert (
+            conn.execute(
+                select(receipts.c.processed).where(
+                    receipts.c.tenant_id == identity.tenant
+                )
+            ).scalar_one()
+            == 0
+        )
+    monkeypatch.setattr(identity, "_save_profile", original)
+    assert worker(client).flush() == 1
+
+
+def test_feature_window_expiry_and_cap(client):
+    application = app(client)
+    register(
+        client,
+        application,
+        {
+            **CONFIG,
+            "rules": [
+                {"effects": [{"feature": "intent.purchase.digital", "value": 100}]}
+            ],
+        },
+    )
+    result = send(client, application).json()
+    send(client, application)
+    worker(client).flush()
+    identity = client.app.state.identity_service
+    with identity.db.transaction() as conn:
+        assert feature_values(conn, identity.tenant, result["master_id"]) == {
+            "intent.purchase.digital": 100
+        }
+        assert (
+            feature_values(
+                conn,
+                identity.tenant,
+                result["master_id"],
+                datetime.now(UTC) + timedelta(days=8),
+            )
+            == {}
+        )
+
+
+@pytest.mark.parametrize("standard", list(EVENTS))
+def test_catalog_required_fields_are_enforced(standard):
+    from fastapi import HTTPException
+
+    from adcloud_server.eventing.rules import validate_properties
+
+    with pytest.raises(HTTPException) as error:
+        validate_properties(
+            {"standard_event": standard, "extra_fields": {}, "required_extra": []}, {}
+        )
+    assert error.value.status_code == 422
+
+
+@pytest.mark.parametrize("standard", list(EVENTS))
+def test_catalog_accepts_valid_required_and_optional_parameters(standard):
+    from adcloud_server.eventing.rules import validate_properties
+
+    def sample(spec, name):
+        if name == "currency":
+            return "CNY"
+        if "enum" in spec:
+            return spec["enum"][0]
+        if spec["type"] == "string":
+            return "example"
+        if spec["type"] in ("integer", "number"):
+            return spec.get("minimum", 0)
+        if spec["type"] == "array":
+            return [sample(spec["items"], "item")]
+        if spec["type"] == "object":
+            return {key: sample(value, key) for key, value in spec["fields"].items()}
+        raise AssertionError("Unsupported catalogue field")
+
+    properties = {
+        key: sample(spec, key) for key, spec in EVENTS[standard]["fields"].items()
+    }
+    validate_properties(
+        {"standard_event": standard, "extra_fields": {}, "required_extra": []},
+        properties,
+    )
+
+
+@pytest.mark.parametrize(
+    "operator,value,expected",
+    [
+        ("gte", 1, 1),
+        ("gt", 0, 1),
+        ("lte", 1, 1),
+        ("lt", 2, 1),
+        ("eq", 2, 0),
+        ("in", [1, 2], 1),
+    ],
+)
+def test_condition_operators(operator, value, expected):
+    from adcloud_server.eventing.rules import evaluate
+
+    config = {
+        "rules": [
+            {
+                "conditions": [
+                    {
+                        "field": "properties.quantity",
+                        "operator": operator,
+                        "value": value,
+                    }
+                ],
+                "effects": [{"feature": "intent.purchase.digital", "value": 1}],
+            }
+        ]
+    }
+    assert len(evaluate(config, {"quantity": 1})["effects"]) == expected
+
+
+def test_schema_changes_require_new_version_and_retries_use_original_publication(
+    client,
+):
+    application = app(client)
+    register(client, application)
+    body = event()
+    assert send(client, application, body).status_code == 200
+    changed = {**CONFIG, "extra_fields": {"variant_id": "string"}}
+    assert (
+        client.put(
+            path(application), headers=manage(application), json=changed
+        ).status_code
+        == 409
+    )
+    assert (
+        client.put(
+            path(application, version=2), headers=manage(application), json=changed
+        ).status_code
+        == 200
+    )
+    client.post(path(application) + "/disable", headers=manage(application))
+    assert send(client, application, body).json()["duplicate"] is True
+    bad = event(
+        properties={"product_id": "camera", "quantity": True, "category_id": "digital"}
+    )
+    client.post(
+        path(application) + "/rollback",
+        headers=manage(application),
+        json={"revision": 1},
+    )
+    assert send(client, application, bad).status_code == 422
+
+
+def test_invalid_mapping_and_unregistered_extensions_are_rejected(client):
+    application = app(client)
+    for config in (
+        {**CONFIG, "extra_fields": {"email": "string"}},
+        {**CONFIG, "rules": [{"effects": [{"feature": "unknown", "value": 1}]}]},
+        {
+            **CONFIG,
+            "rules": [
+                {
+                    "conditions": [
+                        {"field": "properties.missing", "operator": "eq", "value": "x"}
+                    ],
+                    "effects": CONFIG["rules"][0]["effects"],
+                }
+            ],
+        },
+    ):
+        assert (
+            client.put(
+                path(application), headers=manage(application), json=config
+            ).status_code
+            == 422
+        )
+    register(client, application)
+    body = event()
+    body["properties"]["unregistered"] = "x"
+    assert send(client, application, body).status_code == 422
+    with client.app.state.identity_service.db.transaction() as conn:
+        assert (
+            conn.execute(
+                select(receipts).where(
+                    receipts.c.tenant_id == client.app.state.settings.tenant_id
+                )
+            ).first()
+            is None
+        )

tests/test_identity.py

@@ -23,6 +23,7 @@ def settings(tmp_path):
         identity_secret=SECRET,
         service_token=SERVICE_TOKEN,
         cookie_secure=False,
+        event_registry_required=False,
     )
 
 
@@ -336,6 +337,7 @@ def test_tenants_cannot_resolve_or_reuse_each_others_ids(settings):
         identity_secret=SECRET,
         service_token=SERVICE_TOKEN,
         cookie_secure=False,
+        event_registry_required=False,
     )
     with (
         TestClient(create_app(settings)) as first,