diff --git a/.env.example b/.env.example index b2a0068..0f2ed65 100644 --- a/.env.example +++ b/.env.example @@ -18,3 +18,6 @@ CASSANDRA_USERNAME= CASSANDRA_PASSWORD= CASSANDRA_TLS=true CASSANDRA_CA_FILE= + +# Require application-scoped credentials and registered event versions (default true). +EVENT_REGISTRY_REQUIRED=true diff --git a/README.md b/README.md index 7cd56c2..a7d81e7 100644 --- a/README.md +++ b/README.md @@ -82,3 +82,13 @@ Dockerfile 提供 test、runtime 两个目标；Gitea Actions 在 main 推送后 详细配置、runner 要求、公网端口、持久化和回滚见 [Docker 部署说明](docs/docker-deployment.md)。 测试部署使用 `compose.yml` 编排 PostgreSQL、Cassandra、API 与画像 worker；CI 运行全部数据库集成测试，公网 API 默认 http://47.76.92.159:18080/docs。详见 Docker 部署文档。 + + +## 事件字典与标准特征 + +已录入 33 个标准事件，涵盖访问与内容、电商交易、注册留资、广告和推送，支持客户事件别名、参数版本、草稿试算、不可变规则发布、停用和回滚。 +默认接收端必须携带应用接入凭据及已发布事件版本；未注册或参数不符的事件不入库。 +新增 `adcloud-events` worker 将事件按冻结的规则版本转成可去重的标准特征贡献；画像提供时间窗口和上限，身份缝合后自动汇总。 +本地可运行 `uv run adcloud-events`，Cassandra 模式还需 `uv run adcloud-profiles`。 +完整目录、API、配置和兼容说明见 [事件注册表](docs/events.md)。 +开发者控制台已加入事件字典页面；客户账号登录和角色授权尚未接入，目前使用应用范围管理凭据。 diff --git a/compose.yml b/compose.yml index df44e8d..0c67f9e 100644 --- a/compose.yml +++ b/compose.yml @@ -67,6 +67,12 @@ services: command: [adcloud-profiles] healthcheck: disable: true + event-worker: + <<: *app + restart: unless-stopped + command: [adcloud-events] + healthcheck: + disable: true integration: image: ${TEST_IMAGE:?Set TEST_IMAGE} profiles: [ci] @@ -85,6 +91,7 @@ services: depends_on: api: {condition: service_healthy} worker: {condition: service_started} + event-worker: {condition: service_started} healthcheck: disable: true diff --git a/docs/docker-deployment.md b/docs/docker-deployment.md index f156c4e..0ed940f 100644 --- a/docs/docker-deployment.md +++ b/docs/docker-deployment.md @@ -8,8 +8,8 @@ 2. 使用 Gitea 自动签发的短期作业 token 下载精确提交的源码，不将认证信息放入 URL、命令行或日志。 3. 用 uv 和 uv.lock 构建 test 镜像，执行 Ruff 与 pytest。外部数据库测试未配置环境时跳过。 4. 构建不含开发依赖的非 root runtime 镜像，包含 PostgreSQL / Cassandra Python 驱动，允许以后配置存储；不运行数据库服务。 -5. 通过独立 Compose 项目启动临时 PostgreSQL / Cassandra，执行全部 54 个测试（包含 6 个真实数据库测试）；结束后清理该 CI 项目的容器和数据卷。 -6. 用 Compose 部署持久化 PostgreSQL / Cassandra、API 和画像 worker，检查 JSONB 存储、画像投影与宿主机 HTTP 端口。 +5. 通过独立 Compose 项目启动临时 PostgreSQL / Cassandra，执行全部数据库与注册表测试（包含 6 个真实数据库测试）；结束后清理该 CI 项目的容器和数据卷。 +6. 用 Compose 部署持久化 PostgreSQL / Cassandra、API、特征 worker 和画像 worker，检查 JSONB 存储、画像投影与宿主机 HTTP 端口。 7. 如提供公网 URL，再探测公网 `/health`。后续可从外部执行匿名和埋点测试。 Docker 镜像直接保存在该 daemon 上，本版不依赖镜像仓库登录。 @@ -68,9 +68,9 @@ docker build --target runtime --tag adcloud-server:test . DEPLOY_IMAGE=adcloud-server:test bash scripts/compose_stack.sh deploy ``` -当前公网环境使用 PostgreSQL 16 保存身份图谱和 JSONB 权威画像，Cassandra 4.1 保存画像投影，worker 持续消费 outbox。数据库不发布宿主机端口，只能通过项目 Docker 内网连接。 +当前公网环境使用 PostgreSQL 16 保存身份图谱和 JSONB 权威画像，Cassandra 4.1 保存画像投影，adcloud-events 消费事件规则任务，adcloud-profiles 消费画像 outbox。数据库不发布宿主机端口，只能通过项目 Docker 内网连接。 Cassandra 在此合成测试环境内关闭 TLS / 认证，单节点 RF=1；生产环境需另外配置认证、TLS、备份和多节点副本。 -数据库和应用合计内存限制约 3.5 GiB，服务器应留出至少 4 GiB 可用内存，首次启动 Cassandra 可能需要几分钟。 +数据库和应用合计内存限制约 4 GiB，服务器应留出至少 4.5 GiB 可用内存（CI 临时数据库会额外占用内存），首次启动 Cassandra 可能需要几分钟。 `compose.yml` 使用健康检查和一次性 CQL 初始化任务控制启动顺序。 临时 CI 使用 `adcloud-ci--` 独立项目；PostgreSQL 密码随机生成。Cassandra 测试同时连接 CI PostgreSQL，验证双存储组合。失败时也会清理该项目。 @@ -109,3 +109,5 @@ uv run python scripts/smoke_api.py --base-url http://SERVER_IP:18080 --public-on 不要将 SERVICE_TOKEN 放入浏览器或 SDK。需要管理接口测试时，在服务器的受控环境读取部署配置，或配置你自己的受保护测试凭据。 参考：[Docker 端口映射](https://docs.docker.com/engine/network/port-publishing/)、[runner v0.6.1 配置](https://gitea.com/gitea/act_runner/src/tag/v0.6.1/internal/pkg/config/config.example.yaml)、[uv Docker 集成](https://docs.astral.sh/uv/guides/integration/docker/)。 + +注册表默认强制应用凭据和已发布事件版本；smoke 会创建合成应用、注册 add_cart、重试同一 event_id 并验证数码意向 +1，只输出验证结论。公网 public-only 检查健康、匿名 Cookie 和未注册事件拒绝，不携带管理凭据。 diff --git a/docs/events.md b/docs/events.md new file mode 100644 index 0000000..877db5b --- /dev/null +++ b/docs/events.md @@ -0,0 +1,122 @@ +# 事件注册表与标准特征 + +## 身份与凭据 + +本版沿用现有部署 TENANT_ID：一个部署上下文对应一个租户，该租户内按 app_id 隔离注册表和接入凭据。尚未实现客户账号登录、多租户请求路由或客户角色权限；不能把不同客户的应用都放进 default 租户并声称彼此身份图谱隔离。 +平台受控后端使用 SERVICE_TOKEN 调用 POST /v1/registry/apps 创建应用，返回 collection_key、server_key、management_key（仅此时返回原值）；数据库只保存带租户密钥的 HMAC。 +collection_key 是 SDK 使用的受限写入凭据，不是用户认证或广告反作弊凭据。server_key 只在客户可信后端保存，用于上报服务端专属事件。management_key 用于本版控制台注册管理，页面内存保存，刷新清空；不使用 SERVICE_TOKEN，不存 localStorage。正式管理操作需 HTTPS，并最终接入客户登录。 +事件请求头 X-Adcloud-Key 必须匹配 app_id；管理请求头 X-Adcloud-Management-Key 必须属于同一应用。 +匿名 token 与 Master ID 沿用身份图谱；事件本身不验证邮箱、不执行登录态身份缝合。 + +## 公共请求字段 + +必选：app_id、name、event_version、event_id（UUID）、occurred_at（带时区，不能为未来或超过 90 天的历史事件）、properties。 +可选：session_id、context、anonymous_token。浏览器可使用 Cookie，SDK 也可显式发送此前签发的匿名 token。 +context 为有 32 KiB 上限的可选 JSON 上下文；properties 为有 32 KiB 上限的注册参数对象。业务身份/密码/设备推送 token 不应作为埋点字段；query、URL 等可能包含隐私数据，应由客户在上报前清理。 +匿名签发 POST /v1/identity/anonymous 也接受可选 anonymous_token，用于 SDK 在 Cookie 不可用时恢复此前签发的匿名身份。 + +## 标准事件目录 + +下表参数都位于 properties。默认拒绝未注册参数、空字符串、类型不符和超出范围的值。quantity 为正整数；金额为最小货币单位非负整数，并要求三位大写 currency。 +items 为 1–100 个商品对象，每项必选 product_id / quantity，可选 category_id / unit_price。 +content.engage.action: like / save / share / comment；不采集评论正文。 +content.progress 是累计进度信号，本版没有将其自动累加成时长特征。 +ad.response.result: filled / no_fill / error；filled 需 delivery_id。ad.error 需 request_id 或 delivery_id。 +ad.impression / ad.click 是客户端信号，本版不实现广告归属验证、有效曝光认定、反作弊或计费。 +delivery_id 是单次广告投递 ID；message_id 必须是单个接收方的一次消息投递 ID，群发活动使用 campaign_id。push.open 按 message_id 统计首次打开，非每次重复打开。 +push.delivery.result.status: accepted / delivered / failed，只有供应商确实提供的状态才上报。 + +| 标准事件 | 分类 / 含义 | 必选参数 | 可选参数 | 来源 | 业务去重键 | +|---|---|---|---|---|---| +| `page.view` | 访问与内容 / 页面访问 | page_id | referrer, title, url | SDK / 服务端 | event_id | +| `content.view` | 访问与内容 / 内容查看 | content_id, content_type | author_id, category_id | SDK / 服务端 | event_id | +| `content.engage` | 访问与内容 / 内容互动 | content_id, action | category_id, content_type | SDK / 服务端 | event_id | +| `content.progress` | 访问与内容 / 内容消费进度 | content_id, consumed_ms | duration_ms, progress_percent | SDK / 服务端 | event_id | +| `search.submit` | 访问与内容 / 发起搜索 | query | category_id, result_count | SDK / 服务端 | event_id | +| `navigation.click` | 访问与内容 / 导航点击 | target_id | destination, placement_id | SDK / 服务端 | event_id | +| `commerce.product.view` | 电商与交易 / 商品查看 | product_id | category_id, currency, price | SDK / 服务端 | event_id | +| `commerce.cart.add` | 电商与交易 / 加入购物车 | product_id, quantity | cart_id, category_id, currency, price | SDK / 服务端 | event_id | +| `commerce.cart.remove` | 电商与交易 / 移出购物车 | product_id, quantity | cart_id, category_id | SDK / 服务端 | event_id | +| `commerce.checkout.start` | 电商与交易 / 开始结算 | checkout_id, items | amount, coupon_id, currency | SDK / 服务端 | event_id | +| `commerce.order.create` | 电商与交易 / 创建订单 | order_id, items, amount, currency | checkout_id, coupon_id | 服务端 | order_id | +| `commerce.payment.success` | 电商与交易 / 支付成功 | order_id, payment_id, amount, currency | items, payment_method | 服务端 | payment_id | +| `commerce.order.cancel` | 电商与交易 / 订单取消 | order_id | reason_code | 服务端 | order_id | +| `commerce.refund.success` | 电商与交易 / 退款成功 | order_id, refund_id, amount, currency | items, reason_code | 服务端 | refund_id | +| `account.signup.success` | 注册与线索 / 注册成功 | registration_id, method | campaign_id | 服务端 | registration_id | +| `account.login.success` | 注册与线索 / 登录成功 | method | login_id | 服务端 | event_id | +| `lead.form.start` | 注册与线索 / 开始填写表单 | form_id | campaign_id, placement_id | SDK / 服务端 | event_id | +| `lead.submit.success` | 注册与线索 / 留资成功 | lead_id, form_id | campaign_id, lead_type | 服务端 | lead_id | +| `conversion.complete` | 注册与线索 / 业务转化完成 | conversion_id, conversion_type | currency, related_object_id, value | 服务端 | conversion_id | +| `ad.request` | 广告 / 请求广告 | request_id, placement_id | format, page_id | SDK / 服务端 | request_id | +| `ad.response` | 广告 / 广告返回 | request_id, result | delivery_id, latency_ms, reason_code | 服务端 | request_id | +| `ad.render` | 广告 / 广告渲染 | delivery_id | render_ms | SDK / 服务端 | delivery_id | +| `ad.impression` | 广告 / 广告展示信号 | delivery_id, impression_id | visible_ms, visible_ratio | SDK / 服务端 | impression_id | +| `ad.click` | 广告 / 广告点击信号 | delivery_id, click_id | destination_id, impression_id | SDK / 服务端 | click_id | +| `ad.video.progress` | 广告 / 广告视频进度 | delivery_id, milestone | played_ms | SDK / 服务端 | delivery_id, milestone | +| `ad.dismiss` | 广告 / 关闭广告 | delivery_id, action | reason_code | SDK / 服务端 | event_id | +| `ad.error` | 广告 / 广告错误 | stage, error_code | delivery_id, request_id | SDK / 服务端 | event_id | +| `push.permission.result` | 推送 / 推送权限结果 | status, channel | prompt_id | SDK / 服务端 | event_id | +| `push.send` | 推送 / 发送推送 | message_id, channel | campaign_id | 服务端 | message_id | +| `push.delivery.result` | 推送 / 推送投递结果 | message_id, status | failure_reason, provider_code | 服务端 | message_id, status | +| `push.open` | 推送 / 打开推送 | message_id | campaign_id | SDK / 服务端 | message_id | +| `push.click` | 推送 / 点击推送入口 | message_id, action_id | destination_id | SDK / 服务端 | message_id, action_id | +| `push.unsubscribe` | 推送 / 取消推送订阅 | channel | reason_code, subscription_id | SDK / 服务端 | event_id | + +## 注册、发布、试算 + +公共 GET /v1/registry/catalog 返回事件字段类型、必选字段、枚举、范围，以及标准特征字典。 +应用管理接口路径前缀 /v1/registry/apps/{app_id}： + +- GET /events：查看应用所有事件参数版本和草稿。 +- PUT /events/{name}/versions/{version}：新增或保存草稿。 +- POST 同一路径 /preview：传入 properties，显示命中规则和特征增量，无事件或特征写入。 +- POST /publish：产生不可变 rule_revision 并激活。 +- POST /disable：拒绝新事件；已经接受的任务继续执行，原 event_id 的幂等重试仍返回原回执。 +- GET /publications：列出不可变历史发布配置。 +- POST /rollback：传入 revision，激活该历史规则；不重算此前接受的事件。 + +配置示例： + +```json +{ + "standard_event": "commerce.cart.add", + "description": "用户将数码商品加入购物车", + "extra_fields": {}, + "required_extra": [], + "rules": [{ + "conditions": [{"field": "properties.category_id", "operator": "eq", "value": "digital"}], + "effects": [{"feature": "intent.purchase.digital", "operation": "increment", "value": 1}] + }] +} +``` + +name 可以是客户自己的 add_cart；参数仍使用标准字段名。本版不做参数字段重命名或商品目录补全。 +规则条件引用可选字段时，该字段会成为该事件定义的必选字段；上例因此必须提供 category_id。 +扩展字段支持 string / integer / number / boolean；不能覆盖标准字段。发布过的参数结构不可变，改结构需要新 event_version；只改规则增量或描述可发布新 rule_revision。 +规则内条件全部满足才执行；多条规则独立匹配。支持 eq / in / contains / gt / gte / lt / lte，contains 只支持字符串。 +仅支持正数 increment，增量精度最多 6 位小数，单个增量 ≤100。规则只能引用平台已有特征；不运行任意代码、脚本或正则表达式。 +最多 20 条规则、每条 20 个条件和 20 个增量、30 个扩展字段，配置正文最大 64 KiB。 + +## 持久化与计算 + +PostgreSQL 新增 event_apps、event_definitions、event_rule_publications、event_receipts、event_feature_contributions。元数据与冻结的请求/规则快照使用 JSONB，本地使用 SQLite JSON；已有表不改列，初始化时增加新表。 +POST /v1/events 在同一事务中保存原始事件和规则任务；adcloud-events worker 在同一事务中写特征贡献、更新画像并入画像 outbox、标记任务完成。失败回滚，任务仍可重试。 +同一租户下 event_id 不能用于不同事件；有业务去重键的事件还会按应用和标准事件业务键去重，避免同一支付、广告点击或推送投递重复计算。 +贡献绑定原始身份节点。读取时按 Master ID 对应的所有图谱成员汇总；缝合前后不复制贡献，不重复加分。GET /v1/users/{master_id} 提供 features；GET /v1/users/{master_id}/features 提供最近 100 个贡献及其 rule_revision，使用现有可信服务鉴权。 +标准特征值按事件发生时间计算窗口和上限，而非简单覆盖 attributes；不能通过画像 PATCH 修改标准特征。 +adcloud-profiles 将含特征快照的画像同步 Cassandra。画像读取的 features 按关系库贡献实时计算，避免过期分数被 Cassandra 缓存延长；Cassandra 画像仍保持其版本回退机制。 +没有自动打标签、历史事件重放、商品类目查询、通用累积观看时长或客户自建标准特征功能。购买次数等特征按客户映射的业务事件计数，映射时应明确最终成交的业务含义。 + +| 标准特征 | 展示名称 | 窗口 | 上限 | +|---|---|---|---| +| `intent.purchase.digital` | 数码购买意向 | 7 天 | 100 | +| `interest.category.digital` | 数码兴趣 | 30 天 | 100 | +| `engagement.ad.click` | 广告点击活跃度 | 7 天 | 1000 | +| `engagement.push.open` | 推送打开活跃度 | 30 天 | 1000 | +| `commerce.purchase.count` | 购买次数 | 30 天 | 10000 | +| `conversion.lead.count` | 留资次数 | 30 天 | 10000 | + +## 兼容与部署 + +EVENT_REGISTRY_REQUIRED 默认 true。旧版未带应用字段的埋点将返回 422，应先创建应用、注册并发布事件、更新 SDK。仅本地旧代码联调可设置 false；一旦带 app_id 仍必须通过注册表校验。平台专属结果事件不会因为兼容开关而允许 collection_key 上报。 +Compose 已增加常驻 event-worker；部署 smoke 创建合成应用、注册加购、发送与重试事件，并验证 +1、身份缝合及 Cassandra 当前版本。临时 CI 对 SQLite 与真实 PostgreSQL 都测试注册、参数、去重、规则版本、worker 回滚和窗口。 diff --git a/docs/identity.md b/docs/identity.md index 6977779..9bd6c65 100644 --- a/docs/identity.md +++ b/docs/identity.md @@ -83,14 +83,15 @@ await fetch("https://YOUR_API_HOST/v1/identity/anonymous", { }); ``` -后端第一次收到匿名请求就签发 ID；第三方 React SDK 的自动初始化集成尚未修改。 +后端第一次收到匿名请求就签发 ID；React SDK 的 AdCloudProvider 在挂载时初始化，track 也会等待初始化完成。 浏览器请求需使用 `credentials: "include"`。匿名写接口检查 Origin；CORS 使用明确允许来源并启用凭据。 Cookie 默认 Secure、HttpOnly、SameSite=Lax，有效期默认一年；本地 HTTP 开发明确设置 `COOKIE_SECURE=false`。 跨站点 Cookie 如需 SameSite=None，必须同时启用 Secure。浏览器第三方 Cookie 策略可能阻止 Cookie，应优先将 API 放在第一方域名下。 不设置父域 Cookie，不通过第三方设备指纹跨站点识别用户。 -事件可提供 UUID `event_id` 进行幂等重试。相同 ID 的不同主体或内容返回 409；省略 event_id 时每次生成新事件。 -`occurred_at` 如果提供，必须含时区且不能晚于服务端当前时间。 +默认事件接收要求注册并发布对应 app_id / name / event_version，携带应用 collection 或 server 凭据，以及 UUID event_id 和带时区的 occurred_at。 +相同 ID 的不同主体或内容返回 409；未注册、未发布、参数不符或来源不合法会明确拒绝。详见 [事件字典与规则](events.md)。 +普通画像属性继续按目标账号优先合并；标准特征由独立贡献记录按图谱成员汇总，因此身份缝合不覆盖数值、不重复加分。 Stitch 示例（只在 OTP/OAuth 验证后由可信后端执行）： diff --git a/pyproject.toml b/pyproject.toml index 9e8b76f..ac6b9cc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,7 @@ postgres = ["psycopg[binary]>=3.2,<4"] cassandra = ["cassandra-driver>=3.29,<4"] [project.scripts] +adcloud-events = "adcloud_server.eventing.worker:main" adcloud-profiles = "adcloud_server.projection:main" [dependency-groups] diff --git a/scripts/compose_stack.sh b/scripts/compose_stack.sh index a884c14..8fe554e 100755 --- a/scripts/compose_stack.sh +++ b/scripts/compose_stack.sh @@ -30,7 +30,7 @@ cleanup() { local result=$? trap - EXIT if ! $successful && $legacy_stopped; then - docker compose -f compose.yml stop api worker >/dev/null || true + docker compose -f compose.yml stop api worker event-worker >/dev/null || true docker start adcloud-server-test >/dev/null || true echo 'Compose deployment failed; legacy SQLite service restored where possible.' fi @@ -64,7 +64,7 @@ if docker container inspect adcloud-server-test >/dev/null 2>&1; then legacy_stopped=true fi docker compose -f compose.yml up -d --wait --wait-timeout 180 api -docker compose -f compose.yml up -d worker +docker compose -f compose.yml up -d worker event-worker docker compose -f compose.yml --profile checks run --rm smoke # Verify the published host port too; config stays read-only in this probe. docker run --rm --network host --volume "$CONFIG_VOLUME:/config:ro" --entrypoint python "$DEPLOY_IMAGE" /app/scripts/smoke_api.py --base-url "http://127.0.0.1:$DEPLOY_PORT" --require-projection diff --git a/scripts/demo_identity.py b/scripts/demo_identity.py index a97825a..6cc0259 100644 --- a/scripts/demo_identity.py +++ b/scripts/demo_identity.py @@ -4,6 +4,7 @@ import argparse import http.cookiejar import json import urllib.request +from datetime import UTC, datetime from uuid import uuid4 from adcloud_server.settings import Settings @@ -19,10 +20,11 @@ opener = urllib.request.build_opener( ) -def api(path, body=None, trusted=False, method=None): +def api(path, body=None, trusted=False, method=None, extra_headers=None): headers = {"Content-Type": "application/json"} if trusted: headers["Authorization"] = "Bearer " + settings.service_token + headers.update(extra_headers or {}) request = urllib.request.Request( args.base_url.rstrip("/") + path, data=None if body is None else json.dumps(body).encode(), @@ -34,7 +36,31 @@ def api(path, body=None, trusted=False, method=None): anon = api("/v1/identity/anonymous?rotate=true", {}) -api("/v1/events", {"name": "product.view", "properties": {"sku": "demo-running-shoe"}}) +app_id = "demo-" + uuid4().hex +application = api( + "/v1/registry/apps", {"app_id": app_id, "name": "Identity demo"}, trusted=True +) +registration = "/v1/registry/apps/" + app_id + "/events/product.view/versions/1" +management = {"X-Adcloud-Management-Key": application["management_key"]} +api( + registration, + {"standard_event": "commerce.product.view", "description": "Demo product view"}, + method="PUT", + extra_headers=management, +) +api(registration + "/publish", {}, extra_headers=management) +api( + "/v1/events", + { + "name": "product.view", + "app_id": app_id, + "event_version": 1, + "event_id": str(uuid4()), + "occurred_at": datetime.now(UTC).isoformat(), + "properties": {"product_id": "demo-running-shoe"}, + }, + extra_headers={"X-Adcloud-Key": application["collection_key"]}, +) api( "/v1/users/" + anon["master_id"], {"attributes": {"interest": "running", "unused": None}, "tags": ["runner"]}, diff --git a/scripts/smoke_api.py b/scripts/smoke_api.py index 8188220..92e45e9 100644 --- a/scripts/smoke_api.py +++ b/scripts/smoke_api.py @@ -7,6 +7,7 @@ import os import time import urllib.error import urllib.request +from datetime import UTC, datetime from pathlib import Path from uuid import uuid4 @@ -17,10 +18,13 @@ def check(base_url: str, service_token: str, require_projection=False): jar = http.cookiejar.CookieJar() opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) - def api(path, body=None, *, trusted=False, method=None, expected=200): + def api( + path, body=None, *, trusted=False, method=None, expected=200, extra_headers=None + ): headers = {"Content-Type": "application/json"} if trusted: headers["Authorization"] = "Bearer " + service_token + headers.update(extra_headers or {}) request = urllib.request.Request( base_url.rstrip("/") + path, data=None if body is None else json.dumps(body).encode(), @@ -46,14 +50,54 @@ def check(base_url: str, service_token: str, require_projection=False): assert any(cookie.name == "adcloud_anonymous" for cookie in jar), ( "Missing anonymous cookie" ) - event = api( - "/v1/events", - { - "event_id": str(uuid4()), - "name": "deployment.smoke", - "properties": {"source": "gitea-actions"}, - }, + if not service_token: + api("/v1/events", {"name": "unregistered"}, expected=422) + print( + "Public HTTP smoke passed: routes, anonymous cookie, unregistered event rejection" + ) + return + app_id = "smoke-" + uuid4().hex + application = api( + "/v1/registry/apps", + {"app_id": app_id, "name": "Deployment smoke"}, + trusted=True, + expected=201, ) + management = {"X-Adcloud-Management-Key": application["management_key"]} + registration = "/v1/registry/apps/" + app_id + "/events/add_cart/versions/1" + config = { + "standard_event": "commerce.cart.add", + "description": "Synthetic digital cart", + "rules": [ + { + "conditions": [ + { + "field": "properties.category_id", + "operator": "eq", + "value": "digital", + } + ], + "effects": [{"feature": "intent.purchase.digital", "value": 1}], + } + ], + } + api(registration, config, method="PUT", extra_headers=management) + api(registration + "/publish", {}, extra_headers=management) + body = { + "event_id": str(uuid4()), + "app_id": app_id, + "event_version": 1, + "occurred_at": datetime.now(UTC).isoformat(), + "name": "add_cart", + "properties": { + "product_id": "synthetic-camera", + "quantity": 1, + "category_id": "digital", + }, + } + headers = {"X-Adcloud-Key": application["collection_key"]} + event = api("/v1/events", body, extra_headers=headers) + assert api("/v1/events", body, extra_headers=headers)["duplicate"] is True assert event["anonymous_id"] == anon["anonymous_id"], "Anonymous cookie not reused" if service_token: api("/v1/users/" + anon["master_id"], expected=401) @@ -86,13 +130,17 @@ def check(base_url: str, service_token: str, require_projection=False): assert storage["profile_backend"] == "cassandra" for _ in range(60): profile = api("/v1/users/" + linked["master_id"], trusted=True) - if profile["projection_status"] == "current": + if ( + profile["projection_status"] == "current" + and profile["features"].get("intent.purchase.digital") == 1 + ): break time.sleep(1) assert profile["projection_status"] == "current", ( "Worker projection not current" ) assert profile["attributes"]["smoke"] is True + assert profile["features"] == {"intent.purchase.digital": 1} print( "HTTP smoke passed: routes, anonymous cookie, events" + (", auth, stitching, profile and history" if service_token else "") diff --git a/src/adcloud_server/api/v1/events.py b/src/adcloud_server/api/v1/events.py index a7312d1..c4e7079 100644 --- a/src/adcloud_server/api/v1/events.py +++ b/src/adcloud_server/api/v1/events.py @@ -1,6 +1,7 @@ +from typing import Annotated from uuid import UUID -from fastapi import APIRouter, Depends, Query, Request, Response +from fastapi import APIRouter, Depends, Header, Query, Request, Response from adcloud_server.dependencies import ( IdentityDependency, @@ -9,6 +10,7 @@ from adcloud_server.dependencies import ( set_anonymous_cookie, trusted_service, ) +from adcloud_server.eventing.service import EventRegistry from adcloud_server.identity_models import ( EventRequest, ) @@ -32,9 +34,14 @@ def record_event( request: Request, response: Response, service: IdentityDependency, + collection_key: Annotated[str | None, Header(alias="X-Adcloud-Key")] = None, ): """Receive a cookie/token-bound event; missing IDs get a new anonymous ID.""" - result = service.record_event(anonymous_token(request, body.anonymous_token), body) + token = anonymous_token(request, body.anonymous_token) + if body.app_id or service.settings.event_registry_required: + result = EventRegistry(service).receive(token, body, collection_key) + else: + result = service.record_event(token, body) set_anonymous_cookie(response, service, result) return result diff --git a/src/adcloud_server/api/v1/identity.py b/src/adcloud_server/api/v1/identity.py index 18024dc..b043708 100644 --- a/src/adcloud_server/api/v1/identity.py +++ b/src/adcloud_server/api/v1/identity.py @@ -1,6 +1,7 @@ +from typing import Annotated from uuid import UUID -from fastapi import APIRouter, Depends, Query, Request, Response +from fastapi import APIRouter, Body, Depends, Query, Request, Response from adcloud_server.dependencies import ( IdentityDependency, @@ -10,6 +11,7 @@ from adcloud_server.dependencies import ( trusted_service, ) from adcloud_server.identity_models import ( + AnonymousRequest, ResolveRequest, StitchRequest, ) @@ -41,9 +43,12 @@ def anonymous( response: Response, service: IdentityDependency, rotate: bool = Query(False), + body: Annotated[AnonymousRequest | None, Body()] = None, ): """Create/reuse a signed anonymous ID; rotate on logout/account switching.""" - result = service.anonymous(anonymous_token(request), rotate=rotate) + result = service.anonymous( + anonymous_token(request, body.anonymous_token if body else None), rotate=rotate + ) set_anonymous_cookie(response, service, result) return result diff --git a/src/adcloud_server/api/v1/registry.py b/src/adcloud_server/api/v1/registry.py new file mode 100644 index 0000000..1eaaf0a --- /dev/null +++ b/src/adcloud_server/api/v1/registry.py @@ -0,0 +1,147 @@ +import re +from typing import Annotated + +from fastapi import APIRouter, Depends, Header, HTTPException, Path, Response +from sqlalchemy import select, update + +from adcloud_server.dependencies import IdentityDependency, trusted_service +from adcloud_server.eventing.catalog import EVENTS, FEATURES +from adcloud_server.eventing.models import AppCreate, EventDefinition, Preview, Rollback +from adcloud_server.eventing.rules import evaluate, validate_properties +from adcloud_server.eventing.service import EventRegistry +from adcloud_server.eventing.tables import definitions, publications + +router = APIRouter() +Key = Annotated[str | None, Header(alias="X-Adcloud-Management-Key")] +Version = Annotated[int, Path(ge=1, le=10000)] + + +def registry(identity: IdentityDependency, response: Response): + response.headers["Cache-Control"] = "no-store" + return EventRegistry(identity) + + +Registry = Annotated[EventRegistry, Depends(registry)] + + +@router.get("/catalog") +def catalog(): + return {"events": EVENTS, "features": FEATURES} + + +@router.post("/apps", dependencies=[Depends(trusted_service)], status_code=201) +def create_app(body: AppCreate, service: Registry): + return service.create_app(body) + + +@router.get("/apps/{app_id}") +def app(app_id: str, service: Registry, key: Key = None): + with service.db.transaction() as conn: + application = service.authorize(conn, app_id, key, ["management"]) + return {"app_id": app_id, "name": application["name"]} + + +@router.get("/apps/{app_id}/events") +def list_definitions(app_id: str, service: Registry, key: Key = None): + with service.db.transaction() as conn: + service.authorize(conn, app_id, key, ["management"]) + rows = conn.execute( + select(definitions) + .where(*service.scope(definitions, app_id)) + .order_by(definitions.c.name, definitions.c.version) + ).mappings() + return {"items": [dict(row) for row in rows]} + + +@router.put("/apps/{app_id}/events/{name}/versions/{version}") +def save( + app_id: str, + name: str, + version: Version, + body: EventDefinition, + service: Registry, + key: Key = None, +): + if not re.fullmatch(r"[A-Za-z0-9_.:-]{1,128}", name): + raise HTTPException(422, "Invalid event name") + return service.save(app_id, name, version, body, key) + + +@router.post("/apps/{app_id}/events/{name}/versions/{version}/publish") +def publish( + app_id: str, name: str, version: Version, service: Registry, key: Key = None +): + return service.publish(app_id, name, version, key) + + +@router.post("/apps/{app_id}/events/{name}/versions/{version}/preview") +def preview( + app_id: str, + name: str, + version: Version, + body: Preview, + service: Registry, + key: Key = None, +): + with service.db.transaction() as conn: + service.authorize(conn, app_id, key, ["management"]) + config = service.definition(conn, app_id, name, version)["draft"] + validate_properties(config, body.properties) + return evaluate(config, body.properties) + + +@router.post("/apps/{app_id}/events/{name}/versions/{version}/disable") +def disable( + app_id: str, name: str, version: Version, service: Registry, key: Key = None +): + with service.db.transaction() as conn: + service.authorize(conn, app_id, key, ["management"]) + service.definition(conn, app_id, name, version) + conn.execute( + update(definitions) + .where(*service.where(definitions, app_id, name, version)) + .values(status="disabled") + ) + return {"status": "disabled"} + + +@router.get("/apps/{app_id}/events/{name}/versions/{version}/publications") +def list_publications( + app_id: str, name: str, version: Version, service: Registry, key: Key = None +): + with service.db.transaction() as conn: + service.authorize(conn, app_id, key, ["management"]) + rows = conn.execute( + select(publications) + .where(*service.where(publications, app_id, name, version)) + .order_by(publications.c.revision) + ).mappings() + return {"items": [dict(row) for row in rows]} + + +@router.post("/apps/{app_id}/events/{name}/versions/{version}/rollback") +def rollback( + app_id: str, + name: str, + version: Version, + body: Rollback, + service: Registry, + key: Key = None, +): + with service.db.transaction() as conn: + service.authorize(conn, app_id, key, ["management"]) + service.definition(conn, app_id, name, version) + config = conn.execute( + select(publications.c.config).where( + *service.where(publications, app_id, name, version), + publications.c.revision == body.revision, + ) + ).scalar_one_or_none() + if config is None: + raise HTTPException(404, "Publication revision not found") + conn.execute( + update(definitions) + .where(*service.where(definitions, app_id, name, version)) + .values(status="published", revision=body.revision) + ) + return {"rule_revision": body.revision, "config": config} diff --git a/src/adcloud_server/api/v1/router.py b/src/adcloud_server/api/v1/router.py index 1ed7a1c..55ec5aa 100644 --- a/src/adcloud_server/api/v1/router.py +++ b/src/adcloud_server/api/v1/router.py @@ -1,9 +1,11 @@ from fastapi import APIRouter -from adcloud_server.api.v1 import ads, events, identity, users +from adcloud_server.api.v1 import ads, events, identity, registry, users router = APIRouter(prefix="/v1") router.include_router(events.router, prefix="/events", tags=["events"]) router.include_router(users.router, prefix="/users", tags=["users"]) router.include_router(ads.router, prefix="/ads", tags=["ads"]) router.include_router(identity.router, prefix="/identity", tags=["identity"]) + +router.include_router(registry.router, prefix="/registry", tags=["registry"]) diff --git a/src/adcloud_server/api/v1/users.py b/src/adcloud_server/api/v1/users.py index dbfa050..69f3101 100644 --- a/src/adcloud_server/api/v1/users.py +++ b/src/adcloud_server/api/v1/users.py @@ -36,3 +36,44 @@ def update_profile( ): response.headers["Cache-Control"] = "no-store" return service.patch_profile(str(master_id), body) + + +@router.get("/{master_id}/features", dependencies=[Depends(trusted_service)]) +def features(master_id: UUID, response: Response, service: IdentityDependency): + from sqlalchemy import select + + from adcloud_server.eventing.tables import contributions, receipts + from adcloud_server.eventing.worker import feature_values + + response.headers["Cache-Control"] = "no-store" + with service.db.transaction() as conn: + root = service._root(conn, str(master_id)) + values = feature_values(conn, service.tenant, root) + rows = conn.execute( + select( + contributions.c.event_id, + contributions.c.feature, + contributions.c.delta, + contributions.c.occurred_at, + receipts.c.app_id, + receipts.c.rule_revision, + ) + .select_from( + contributions.join( + receipts, + (contributions.c.tenant_id == receipts.c.tenant_id) + & (contributions.c.event_id == receipts.c.event_id), + ) + ) + .where( + contributions.c.tenant_id == service.tenant, + contributions.c.subject_id.in_(service._members(root)), + ) + .order_by(contributions.c.occurred_at.desc()) + .limit(100) + ).mappings() + return { + "master_id": root, + "features": values, + "recent_contributions": [dict(row) for row in rows], + } diff --git a/src/adcloud_server/database.py b/src/adcloud_server/database.py index e8f5c37..6ba5751 100644 --- a/src/adcloud_server/database.py +++ b/src/adcloud_server/database.py @@ -135,6 +135,10 @@ class Database: from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.dialects.sqlite import insert as sqlite_insert + from adcloud_server.eventing import ( + tables as _event_tables, # noqa: F401 -- Register additive tables. + ) + statement = ( (sqlite_insert if self.sqlite else pg_insert)(tenants) .values(id=self.tenant_id) diff --git a/src/adcloud_server/eventing/__init__.py b/src/adcloud_server/eventing/__init__.py new file mode 100644 index 0000000..0bf3db1 --- /dev/null +++ b/src/adcloud_server/eventing/__init__.py @@ -0,0 +1 @@ +"""Versioned event definitions and deterministic feature rules.""" diff --git a/src/adcloud_server/eventing/catalog.py b/src/adcloud_server/eventing/catalog.py new file mode 100644 index 0000000..84871ee --- /dev/null +++ b/src/adcloud_server/eventing/catalog.py @@ -0,0 +1,275 @@ +"""Platform-owned vocabulary. Customers register aliases of these events.""" + + +def field(kind="string", **kwargs): + return {"type": kind, **kwargs} + + +S = field() +ID = field(max_length=128) +INTEGER = field("integer", minimum=0) +MONEY = field("integer", minimum=0) +CURRENCY = field(max_length=3, min_length=3) +ITEMS = field( + "array", + items=field( + "object", + fields={ + "product_id": ID, + "quantity": field("integer", minimum=1), + "category_id": ID, + "unit_price": MONEY, + }, + required=["product_id", "quantity"], + ), + min_items=1, +) + + +def event(category, title, required, optional=None, *, source="client", dedup=None): + return { + "category": category, + "title": title, + "source": source, + "fields": {**required, **(optional or {})}, + "required": list(required), + "business_keys": dedup or [], + } + + +EVENTS = { + "page.view": event( + "访问与内容", "页面访问", {"page_id": ID}, {"url": S, "title": S, "referrer": S} + ), + "content.view": event( + "访问与内容", + "内容查看", + {"content_id": ID, "content_type": S}, + {"category_id": ID, "author_id": ID}, + ), + "content.engage": event( + "访问与内容", + "内容互动", + {"content_id": ID, "action": field(enum=["like", "save", "share", "comment"])}, + {"content_type": S, "category_id": ID}, + ), + "content.progress": event( + "访问与内容", + "内容消费进度", + {"content_id": ID, "consumed_ms": INTEGER}, + { + "duration_ms": INTEGER, + "progress_percent": field("number", minimum=0, maximum=100), + }, + ), + "search.submit": event( + "访问与内容", + "发起搜索", + {"query": field(max_length=512)}, + {"category_id": ID, "result_count": INTEGER}, + ), + "navigation.click": event( + "访问与内容", + "导航点击", + {"target_id": ID}, + {"placement_id": ID, "destination": S}, + ), + "commerce.product.view": event( + "电商与交易", + "商品查看", + {"product_id": ID}, + {"category_id": ID, "price": MONEY, "currency": CURRENCY}, + ), + "commerce.cart.add": event( + "电商与交易", + "加入购物车", + {"product_id": ID, "quantity": field("integer", minimum=1)}, + {"category_id": ID, "price": MONEY, "currency": CURRENCY, "cart_id": ID}, + ), + "commerce.cart.remove": event( + "电商与交易", + "移出购物车", + {"product_id": ID, "quantity": field("integer", minimum=1)}, + {"category_id": ID, "cart_id": ID}, + ), + "commerce.checkout.start": event( + "电商与交易", + "开始结算", + {"checkout_id": ID, "items": ITEMS}, + {"amount": MONEY, "currency": CURRENCY, "coupon_id": ID}, + ), + "commerce.order.create": event( + "电商与交易", + "创建订单", + {"order_id": ID, "items": ITEMS, "amount": MONEY, "currency": CURRENCY}, + {"checkout_id": ID, "coupon_id": ID}, + source="server", + dedup=["order_id"], + ), + "commerce.payment.success": event( + "电商与交易", + "支付成功", + {"order_id": ID, "payment_id": ID, "amount": MONEY, "currency": CURRENCY}, + {"items": ITEMS, "payment_method": S}, + source="server", + dedup=["payment_id"], + ), + "commerce.order.cancel": event( + "电商与交易", + "订单取消", + {"order_id": ID}, + {"reason_code": S}, + source="server", + dedup=["order_id"], + ), + "commerce.refund.success": event( + "电商与交易", + "退款成功", + {"order_id": ID, "refund_id": ID, "amount": MONEY, "currency": CURRENCY}, + {"items": ITEMS, "reason_code": S}, + source="server", + dedup=["refund_id"], + ), + "account.signup.success": event( + "注册与线索", + "注册成功", + {"registration_id": ID, "method": S}, + {"campaign_id": ID}, + source="server", + dedup=["registration_id"], + ), + "account.login.success": event( + "注册与线索", "登录成功", {"method": S}, {"login_id": ID}, source="server" + ), + "lead.form.start": event( + "注册与线索", + "开始填写表单", + {"form_id": ID}, + {"campaign_id": ID, "placement_id": ID}, + ), + "lead.submit.success": event( + "注册与线索", + "留资成功", + {"lead_id": ID, "form_id": ID}, + {"lead_type": S, "campaign_id": ID}, + source="server", + dedup=["lead_id"], + ), + "conversion.complete": event( + "注册与线索", + "业务转化完成", + {"conversion_id": ID, "conversion_type": S}, + {"value": MONEY, "currency": CURRENCY, "related_object_id": ID}, + source="server", + dedup=["conversion_id"], + ), + "ad.request": event( + "广告", + "请求广告", + {"request_id": ID, "placement_id": ID}, + {"format": S, "page_id": ID}, + dedup=["request_id"], + ), + "ad.response": event( + "广告", + "广告返回", + {"request_id": ID, "result": field(enum=["filled", "no_fill", "error"])}, + {"delivery_id": ID, "latency_ms": INTEGER, "reason_code": S}, + source="server", + dedup=["request_id"], + ), + "ad.render": event( + "广告", + "广告渲染", + {"delivery_id": ID}, + {"render_ms": INTEGER}, + dedup=["delivery_id"], + ), + "ad.impression": event( + "广告", + "广告展示信号", + {"delivery_id": ID, "impression_id": ID}, + {"visible_ratio": field("number", minimum=0, maximum=1), "visible_ms": INTEGER}, + dedup=["impression_id"], + ), + "ad.click": event( + "广告", + "广告点击信号", + {"delivery_id": ID, "click_id": ID}, + {"impression_id": ID, "destination_id": ID}, + dedup=["click_id"], + ), + "ad.video.progress": event( + "广告", + "广告视频进度", + { + "delivery_id": ID, + "milestone": field(enum=["start", "25", "50", "75", "complete"]), + }, + {"played_ms": INTEGER}, + dedup=["delivery_id", "milestone"], + ), + "ad.dismiss": event( + "广告", + "关闭广告", + {"delivery_id": ID, "action": field(enum=["close", "hide"])}, + {"reason_code": S}, + ), + "ad.error": event( + "广告", + "广告错误", + {"stage": field(enum=["request", "render", "play"]), "error_code": S}, + {"request_id": ID, "delivery_id": ID}, + ), + "push.permission.result": event( + "推送", + "推送权限结果", + {"status": field(enum=["granted", "denied", "default"]), "channel": S}, + {"prompt_id": ID}, + ), + "push.send": event( + "推送", + "发送推送", + {"message_id": ID, "channel": S}, + {"campaign_id": ID}, + source="server", + dedup=["message_id"], + ), + "push.delivery.result": event( + "推送", + "推送投递结果", + {"message_id": ID, "status": field(enum=["accepted", "delivered", "failed"])}, + {"provider_code": S, "failure_reason": S}, + source="server", + dedup=["message_id", "status"], + ), + "push.open": event( + "推送", + "打开推送", + {"message_id": ID}, + {"campaign_id": ID}, + dedup=["message_id"], + ), + "push.click": event( + "推送", + "点击推送入口", + {"message_id": ID, "action_id": ID}, + {"destination_id": ID}, + dedup=["message_id", "action_id"], + ), + "push.unsubscribe": event( + "推送", + "取消推送订阅", + {"channel": S}, + {"subscription_id": ID, "reason_code": S}, + ), +} + +FEATURES = { + "intent.purchase.digital": {"title": "数码购买意向", "window_days": 7, "cap": 100}, + "interest.category.digital": {"title": "数码兴趣", "window_days": 30, "cap": 100}, + "engagement.ad.click": {"title": "广告点击活跃度", "window_days": 7, "cap": 1000}, + "engagement.push.open": {"title": "推送打开活跃度", "window_days": 30, "cap": 1000}, + "commerce.purchase.count": {"title": "购买次数", "window_days": 30, "cap": 10000}, + "conversion.lead.count": {"title": "留资次数", "window_days": 30, "cap": 10000}, +} diff --git a/src/adcloud_server/eventing/models.py b/src/adcloud_server/eventing/models.py new file mode 100644 index 0000000..ca70d0a --- /dev/null +++ b/src/adcloud_server/eventing/models.py @@ -0,0 +1,47 @@ +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, JsonValue + + +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", allow_inf_nan=False) + + +class AppCreate(StrictModel): + app_id: str = Field(pattern=r"^[a-zA-Z0-9_-]{1,64}$") + name: str = Field(min_length=1, max_length=128) + + +class Condition(StrictModel): + field: str = Field(pattern=r"^properties\.[A-Za-z0-9_]{1,64}$") + operator: Literal["eq", "in", "contains", "gt", "gte", "lt", "lte"] + value: JsonValue + + +class Effect(StrictModel): + feature: str = Field(max_length=128) + operation: Literal["increment"] = "increment" + value: float = Field(gt=0, le=100, multiple_of=0.000001) + + +class Rule(StrictModel): + conditions: list[Condition] = Field(default_factory=list, max_length=20) + effects: list[Effect] = Field(min_length=1, max_length=20) + + +class EventDefinition(StrictModel): + standard_event: str = Field(max_length=128) + description: str = Field(min_length=1, max_length=1000) + extra_fields: dict[str, Literal["string", "integer", "number", "boolean"]] = Field( + default_factory=dict, max_length=30 + ) + required_extra: list[str] = Field(default_factory=list, max_length=30) + rules: list[Rule] = Field(default_factory=list, max_length=20) + + +class Preview(StrictModel): + properties: dict[str, JsonValue] + + +class Rollback(StrictModel): + revision: int = Field(ge=1) diff --git a/src/adcloud_server/eventing/rules.py b/src/adcloud_server/eventing/rules.py new file mode 100644 index 0000000..32b9a84 --- /dev/null +++ b/src/adcloud_server/eventing/rules.py @@ -0,0 +1,188 @@ +import json +import math +import operator as comparisons +import re + +from fastapi import HTTPException + +from adcloud_server.eventing.catalog import EVENTS, FEATURES + + +def invalid(message): + raise HTTPException(422, message) + + +def schema(config): + standard = EVENTS.get(config["standard_event"]) + if standard is None: + invalid("Unknown standard event") + return { + **standard["fields"], + **{key: {"type": kind} for key, kind in config["extra_fields"].items()}, + }, standard["required"] + config["required_extra"] + config.get( + "required_fields", [] + ) + + +def validate_value(value, spec, path): + kind = spec["type"] + valid = { + "string": isinstance(value, str) and bool(value.strip()), + "integer": isinstance(value, int) and not isinstance(value, bool), + "number": isinstance(value, (int, float)) + and not isinstance(value, bool) + and (isinstance(value, int) or math.isfinite(value)), + "boolean": isinstance(value, bool), + "array": isinstance(value, list), + "object": isinstance(value, dict), + }[kind] + if not valid: + invalid(f"{path} requires {kind}") + if kind == "string" and not spec.get("min_length", 1) <= len(value) <= spec.get( + "max_length", 2048 + ): + invalid(f"{path} has invalid length") + if kind in ("integer", "number") and ( + value < spec.get("minimum", -1e15) or value > spec.get("maximum", 1e15) + ): + invalid(f"{path} is outside its allowed range") + if "enum" in spec and value not in spec["enum"]: + invalid(f"{path} is not an allowed value") + if kind == "array": + if not spec.get("min_items", 0) <= len(value) <= 100: + invalid(f"{path} has invalid item count") + for index, item in enumerate(value): + validate_value(item, spec["items"], f"{path}[{index}]") + if kind == "object": + validate_object(value, spec["fields"], spec["required"], path) + + +def validate_object(properties, fields, required, path="properties"): + missing = sorted(set(required) - properties.keys()) + unknown = sorted(properties.keys() - fields.keys()) + if missing: + invalid(f"{path} missing required fields: {', '.join(missing)}") + if unknown: + invalid(f"{path} contains unregistered fields: {', '.join(unknown)}") + for key, value in properties.items(): + validate_value(value, fields[key], path + "." + key) + + +def validate_properties(config, properties): + fields, required = schema(config) + validate_object(properties, fields, required) + monetary = {"price", "amount", "value"} & properties.keys() + if monetary and "currency" not in properties: + invalid("Monetary values require currency") + if "currency" in properties and not re.fullmatch( + r"[A-Z]{3}", properties["currency"] + ): + invalid("currency must be a three-letter uppercase code") + if ( + config["standard_event"] == "ad.response" + and properties["result"] == "filled" + and not properties.get("delivery_id") + ): + invalid("A filled ad response requires delivery_id") + if config["standard_event"] == "ad.error" and not ( + properties.get("request_id") or properties.get("delivery_id") + ): + invalid("ad.error requires request_id or delivery_id") + if ( + config["standard_event"] == "content.progress" + and "duration_ms" in properties + and properties["consumed_ms"] > properties["duration_ms"] + ): + invalid("consumed_ms cannot exceed duration_ms") + + +def validate_definition(config): + if len(json.dumps(config, ensure_ascii=False).encode()) > 65536: + invalid("Rule configuration exceeds 64 KiB") + fields, required = schema(config) + extras = config["extra_fields"] + forbidden = { + "email", + "phone", + "password", + "oauth_token", + "push_token", + "service_token", + } + if any( + not re.fullmatch(r"[A-Za-z0-9_]{1,64}", key) + or key in forbidden + or key in EVENTS[config["standard_event"]]["fields"] + for key in extras + ): + invalid( + "Extra fields must be new valid names and exclude identity/credential fields" + ) + if set(config["required_extra"]) - extras.keys(): + invalid("required_extra must refer to declared extra fields") + # A referenced optional field becomes required: score rules never silently + # depend on missing product category or other metadata. + for rule in config["rules"]: + for condition in rule["conditions"]: + key = condition["field"].removeprefix("properties.") + if key not in fields: + invalid("Rule condition refers to an unregistered field") + if key not in required and key not in config.get("required_fields", []): + if key in extras: + if key not in config["required_extra"]: + config["required_extra"].append(key) + else: + config.setdefault("required_fields", []).append(key) + kind = fields[key]["type"] + operator, value = condition["operator"], condition["value"] + if operator in ("gt", "gte", "lt", "lte") and ( + kind not in ("integer", "number") + or not isinstance(value, (int, float)) + or isinstance(value, bool) + ): + invalid("Numeric comparisons require numeric fields and values") + if operator == "in": + if not isinstance(value, list) or not 1 <= len(value) <= 100: + invalid("in requires a nonempty bounded list") + for item in value: + validate_value(item, fields[key], "condition.value") + elif operator == "contains": + if kind != "string" or not isinstance(value, str) or not value: + invalid("contains requires a nonempty string") + elif operator not in ("gt", "gte", "lt", "lte"): + validate_value(value, fields[key], "condition.value") + for effect in rule["effects"]: + if effect["feature"] not in FEATURES: + invalid("Unknown standard feature") + return config + + +def evaluate(config, properties): + matches, effects = [], [] + for index, rule in enumerate(config["rules"]): + matched = True + for condition in rule["conditions"]: + value = properties.get(condition["field"].removeprefix("properties.")) + expected, operator = condition["value"], condition["operator"] + if value is None: + matched = False + break + if operator == "in": + passed = value in expected + elif operator == "contains": + passed = expected in value + else: + passed = { + "eq": comparisons.eq, + "gt": comparisons.gt, + "gte": comparisons.ge, + "lt": comparisons.lt, + "lte": comparisons.le, + }[operator](value, expected) + if not passed: + matched = False + break + if matched: + matches.append(index) + effects.extend(rule["effects"]) + return {"matched_rules": matches, "effects": effects} diff --git a/src/adcloud_server/eventing/service.py b/src/adcloud_server/eventing/service.py new file mode 100644 index 0000000..68855d6 --- /dev/null +++ b/src/adcloud_server/eventing/service.py @@ -0,0 +1,340 @@ +import hashlib +import hmac +import json +import secrets +from datetime import UTC, datetime, timedelta + +from fastapi import HTTPException +from sqlalchemy import insert, select, update + +from adcloud_server.eventing.catalog import EVENTS +from adcloud_server.eventing.rules import ( + schema, + validate_definition, + validate_properties, +) +from adcloud_server.eventing.tables import apps, definitions, publications, receipts + + +class EventRegistry: + def __init__(self, identity): + self.identity = identity + self.db = identity.db + self.tenant = identity.tenant + + def scope(self, table, app_id=None): + terms = [table.c.tenant_id == self.tenant] + if app_id is not None: + terms.append( + table.c.app_id == app_id + if "app_id" in table.c + else table.c.id == app_id + ) + return terms + + def digest(self, value): + return hmac.new( + self.identity.settings.identity_secret.encode(), + value.encode(), + hashlib.sha256, + ).hexdigest() + + def app(self, conn, app_id): + row = ( + conn.execute(select(apps).where(*self.scope(apps, app_id))) + .mappings() + .one_or_none() + ) + if row is None: + raise HTTPException(404, "Application not found") + return dict(row) + + def authorize(self, conn, app_id, key, roles): + application = self.app(conn, app_id) + digest = self.digest(key or "") + if not any( + hmac.compare_digest(digest, application["keys"][role]) for role in roles + ): + raise HTTPException(401, "Application-scoped credential required") + return application + + def create_app(self, body): + keys = { + role: secrets.token_urlsafe(48) + for role in ("collection", "server", "management") + } + with self.db.transaction() as conn: + if conn.execute( + select(apps.c.id).where(*self.scope(apps, body.app_id)) + ).first(): + raise HTTPException(409, "Application already exists") + conn.execute( + insert(apps).values( + tenant_id=self.tenant, + id=body.app_id, + name=body.name, + keys={role: self.digest(value) for role, value in keys.items()}, + ) + ) + return { + "app_id": body.app_id, + "name": body.name, + **{role + "_key": value for role, value in keys.items()}, + } + + def where(self, table, app_id, name, version): + return [ + *self.scope(table, app_id), + table.c.name == name, + table.c.version == version, + ] + + def definition(self, conn, app_id, name, version): + row = ( + conn.execute( + select(definitions).where( + *self.where(definitions, app_id, name, version) + ) + ) + .mappings() + .one_or_none() + ) + if row is None: + raise HTTPException(404, "Event version is not registered") + return dict(row) + + def save(self, app_id, name, version, body, key): + config = validate_definition(body.model_dump()) + with self.db.transaction() as conn: + self.authorize(conn, app_id, key, ["management"]) + row = conn.execute( + select(definitions).where( + *self.where(definitions, app_id, name, version) + ) + ).first() + if row: + published = conn.execute( + select(publications.c.config) + .where(*self.where(publications, app_id, name, version)) + .order_by(publications.c.revision) + .limit(1) + ).scalar_one_or_none() + if published is not None and ( + published["standard_event"] != config["standard_event"] + or schema(published)[0] != schema(config)[0] + or set(schema(published)[1]) != set(schema(config)[1]) + ): + raise HTTPException( + 409, + "Published parameter schemas are immutable; create a new event_version", + ) + conn.execute( + update(definitions) + .where(*self.where(definitions, app_id, name, version)) + .values(draft=config) + ) + else: + conn.execute( + insert(definitions).values( + tenant_id=self.tenant, + app_id=app_id, + name=name, + version=version, + draft=config, + status="draft", + revision=0, + ) + ) + return self.definition(conn, app_id, name, version) + + def publish(self, app_id, name, version, key): + with self.db.transaction() as conn: + self.authorize(conn, app_id, key, ["management"]) + row = self.definition(conn, app_id, name, version) + config = validate_definition(row["draft"]) + previous = conn.execute( + select(publications.c.revision) + .where(*self.where(publications, app_id, name, version)) + .order_by(publications.c.revision.desc()) + .limit(1) + ).scalar_one_or_none() + revision = (previous or 0) + 1 + conn.execute( + insert(publications).values( + tenant_id=self.tenant, + app_id=app_id, + name=name, + version=version, + revision=revision, + config=config, + ) + ) + conn.execute( + update(definitions) + .where(*self.where(definitions, app_id, name, version)) + .values(status="published", revision=revision) + ) + return { + "event_version": version, + "rule_revision": revision, + "config": config, + } + + def receive(self, token, request, key): + if ( + not request.app_id + or not request.event_version + or not request.event_id + or not request.occurred_at + ): + raise HTTPException( + 422, + "Registered events require app_id, event_version, event_id and occurred_at", + ) + if request.occurred_at < datetime.now(UTC) - timedelta(days=90): + raise HTTPException(422, "Events older than 90 days are not accepted") + with self.db.transaction() as conn: + self.authorize(conn, request.app_id, key, ["collection", "server"]) + payload = request.model_dump(mode="json", exclude={"anonymous_token"}) + payload["occurred_at"] = request.occurred_at.astimezone(UTC).isoformat() + previous = ( + conn.execute( + select(receipts).where( + receipts.c.tenant_id == self.tenant, + receipts.c.event_id == str(request.event_id), + ) + ) + .mappings() + .one_or_none() + ) + if previous: + if ( + previous["app_id"] != request.app_id + or previous["payload"] != payload + ): + raise HTTPException( + 409, "Event ID already used by a different registered event" + ) + if EVENTS[previous["config"]["standard_event"]]["source"] == "server": + self.authorize(conn, request.app_id, key, ["server"]) + subject = self.identity._anonymous( + conn, token, strict=bool(request.anonymous_token) + ) + if subject != previous["subject_id"]: + raise HTTPException( + 409, "Duplicate event belongs to a different anonymous identity" + ) + return { + **self.identity._result(conn, subject), + "event_id": previous["event_id"], + "duplicate": True, + "rule_revision": previous["rule_revision"], + "standard_event": previous["config"]["standard_event"], + } + row = self.definition( + conn, request.app_id, request.name, request.event_version + ) + if row["status"] != "published": + raise HTTPException( + 409, "Event version is not published or is disabled" + ) + config = conn.execute( + select(publications.c.config).where( + *self.where( + publications, + request.app_id, + request.name, + request.event_version, + ), + publications.c.revision == row["revision"], + ) + ).scalar_one() + standard = EVENTS[config["standard_event"]] + if standard["source"] == "server": + self.authorize(conn, request.app_id, key, ["server"]) + validate_properties(config, request.properties) + # Freeze the complete request and publication before background work. + payload = request.model_dump(mode="json", exclude={"anonymous_token"}) + payload["occurred_at"] = request.occurred_at.astimezone(UTC).isoformat() + business_key = None + if standard["business_keys"]: + business_key = self.digest( + json.dumps( + [ + config["standard_event"], + *[request.properties[k] for k in standard["business_keys"]], + ], + separators=(",", ":"), + ) + ) + existing = ( + conn.execute( + select(receipts).where( + receipts.c.tenant_id == self.tenant, + receipts.c.event_id == str(request.event_id), + ) + ) + .mappings() + .one_or_none() + ) + if existing: + if ( + existing["app_id"] != request.app_id + or existing["payload"] != payload + ): + raise HTTPException( + 409, "Event ID already used by a different registered event" + ) + elif business_key: + existing = ( + conn.execute( + select(receipts).where( + *self.scope(receipts, request.app_id), + receipts.c.business_key == business_key, + ) + ) + .mappings() + .one_or_none() + ) + if existing and ( + existing["payload"]["properties"] != request.properties + or existing["payload"]["name"] != request.name + ): + raise HTTPException( + 409, "Business key already used with different event data" + ) + if existing: + subject = self.identity._anonymous( + conn, token, strict=bool(request.anonymous_token) + ) + if subject != existing["subject_id"]: + raise HTTPException( + 409, "Duplicate event belongs to a different anonymous identity" + ) + result = self.identity._result(conn, subject) + return { + **result, + "event_id": existing["event_id"], + "duplicate": True, + "rule_revision": existing["rule_revision"], + "standard_event": existing["config"]["standard_event"], + } + result = self.identity.record_event(token, request, connection=conn) + conn.execute( + insert(receipts).values( + tenant_id=self.tenant, + event_id=result["event_id"], + app_id=request.app_id, + subject_id=result["anonymous_id"], + payload=payload, + config=config, + rule_revision=row["revision"], + business_key=business_key, + processed=0, + ) + ) + return { + **result, + "rule_revision": row["revision"], + "standard_event": config["standard_event"], + } diff --git a/src/adcloud_server/eventing/tables.py b/src/adcloud_server/eventing/tables.py new file mode 100644 index 0000000..d120ae3 --- /dev/null +++ b/src/adcloud_server/eventing/tables.py @@ -0,0 +1,65 @@ +from sqlalchemy import Column, Index, Integer, Numeric, String, Table, UniqueConstraint + +from adcloud_server.database import document_type, metadata + +apps = Table( + "event_apps", + metadata, + Column("tenant_id", String(64), primary_key=True), + Column("id", String(64), primary_key=True), + Column("name", String(128), nullable=False), + Column("keys", document_type, nullable=False), +) +definitions = Table( + "event_definitions", + metadata, + Column("tenant_id", String(64), primary_key=True), + Column("app_id", String(64), primary_key=True), + Column("name", String(128), primary_key=True), + Column("version", Integer, primary_key=True), + Column("draft", document_type, nullable=False), + Column("status", String(16), nullable=False), + Column("revision", Integer, nullable=False), +) +publications = Table( + "event_rule_publications", + metadata, + Column("tenant_id", String(64), primary_key=True), + Column("app_id", String(64), primary_key=True), + Column("name", String(128), primary_key=True), + Column("version", Integer, primary_key=True), + Column("revision", Integer, primary_key=True), + Column("config", document_type, nullable=False), +) +receipts = Table( + "event_receipts", + metadata, + Column("tenant_id", String(64), primary_key=True), + Column("event_id", String(36), primary_key=True), + Column("app_id", String(64), nullable=False), + Column("subject_id", String(36), nullable=False), + Column("payload", document_type, nullable=False), + Column("config", document_type, nullable=False), + Column("rule_revision", Integer, nullable=False), + Column("business_key", String(64)), + Column("processed", Integer, nullable=False, default=0), + UniqueConstraint("tenant_id", "app_id", "business_key"), +) +Index("ix_event_receipts_pending", receipts.c.tenant_id, receipts.c.processed) +contributions = Table( + "event_feature_contributions", + metadata, + Column("tenant_id", String(64), primary_key=True), + Column("event_id", String(36), primary_key=True), + Column("effect_index", Integer, primary_key=True), + Column("subject_id", String(36), nullable=False), + Column("feature", String(128), nullable=False), + Column("delta", Numeric(20, 6), nullable=False), + Column("occurred_at", String(40), nullable=False), +) +Index( + "ix_feature_subject", + contributions.c.tenant_id, + contributions.c.subject_id, + contributions.c.occurred_at, +) diff --git a/src/adcloud_server/eventing/worker.py b/src/adcloud_server/eventing/worker.py new file mode 100644 index 0000000..efd1edd --- /dev/null +++ b/src/adcloud_server/eventing/worker.py @@ -0,0 +1,125 @@ +"""Transactional feature ledger; immutable event IDs make retries idempotent.""" + +import argparse +import time +from datetime import UTC, datetime, timedelta +from decimal import Decimal + +from sqlalchemy import func, insert, select, update + +from adcloud_server.database import Database, aliases +from adcloud_server.eventing.catalog import FEATURES +from adcloud_server.eventing.rules import evaluate +from adcloud_server.eventing.tables import contributions, receipts +from adcloud_server.identity_service import IdentityService +from adcloud_server.settings import Settings + + +def feature_values(conn, tenant, root, at=None): + at = at or datetime.now(UTC) + members = select(aliases.c.id).where( + aliases.c.tenant_id == tenant, aliases.c.canonical_id == root + ) + totals = {} + for key, definition in FEATURES.items(): + total = conn.execute( + select(func.sum(contributions.c.delta)).where( + contributions.c.tenant_id == tenant, + contributions.c.subject_id.in_(members), + contributions.c.feature == key, + contributions.c.occurred_at + >= (at - timedelta(days=definition["window_days"])).isoformat(), + contributions.c.occurred_at <= at.isoformat(), + ) + ).scalar_one() + if total: + totals[key] = float(min(total, definition["cap"])) + return totals + + +class FeatureWorker: + def __init__(self, identity): + self.identity = identity + + def flush(self, limit=100): + service = self.identity + processed = 0 + for _ in range(limit): + with service.db.transaction() as conn: + row = ( + conn.execute( + select(receipts) + .where( + receipts.c.tenant_id == service.tenant, + receipts.c.processed == 0, + ) + .order_by(receipts.c.event_id) + .limit(1) + ) + .mappings() + .one_or_none() + ) + if row is None: + break + effects = evaluate(row["config"], row["payload"]["properties"])[ + "effects" + ] + for index, effect in enumerate(effects): + conn.execute( + insert(contributions).values( + tenant_id=service.tenant, + event_id=row["event_id"], + effect_index=index, + subject_id=row["subject_id"], + feature=effect["feature"], + delta=Decimal(str(effect["value"])), + occurred_at=row["payload"]["occurred_at"], + ) + ) + if effects: + root = service._root(conn, row["subject_id"]) + service._save_profile( + conn, root, service._profile(conn, root)["document"] + ) + conn.execute( + update(receipts) + .where( + receipts.c.tenant_id == service.tenant, + receipts.c.event_id == row["event_id"], + ) + .values(processed=1) + ) + processed += 1 + return processed + + +def main(): + parser = argparse.ArgumentParser( + description="Apply registered events to standard features" + ) + parser.add_argument("--once", action="store_true") + args = parser.parse_args() + settings = Settings.from_env() + if not settings.identity_secret: + parser.error("Set IDENTITY_SECRET") + db = Database(settings.database_url, settings.tenant_id) + try: + db.initialize() + worker = FeatureWorker(IdentityService(settings, db)) + while True: + try: + count = worker.flush() + if args.once: + print("Processed feature events:", count) + return + except Exception as error: # noqa: BLE001 -- Keep failed jobs for retry without logging payloads. + if args.once: + raise SystemExit( + "Feature processing failed: " + type(error).__name__ + ) from None + print("Feature retry required:", type(error).__name__, flush=True) + time.sleep(1) + except KeyboardInterrupt: + return + finally: + db.close() diff --git a/src/adcloud_server/identity_models.py b/src/adcloud_server/identity_models.py index 1c30b1b..7300cd5 100644 --- a/src/adcloud_server/identity_models.py +++ b/src/adcloud_server/identity_models.py @@ -71,12 +71,16 @@ class ResolveRequest(BaseModel): class EventRequest(BaseModel): model_config = ConfigDict(extra="forbid", allow_inf_nan=False) event_id: UUID | None = None + app_id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9_-]{1,64}$") + event_version: int | None = Field(default=None, ge=1, le=10000) + session_id: str | None = Field(default=None, max_length=128) + context: dict[str, JsonValue] = Field(default_factory=dict) name: str = Field(min_length=1, max_length=128, pattern=r"^[A-Za-z0-9_.:-]+$") properties: dict[str, JsonValue] = Field(default_factory=dict) occurred_at: datetime | None = None anonymous_token: str | None = Field(default=None, max_length=2048) - @field_validator("properties") + @field_validator("properties", "context") @classmethod def properties_valid(cls, value): if len(json.dumps(value, ensure_ascii=False, allow_nan=False).encode()) > 32768: @@ -91,3 +95,8 @@ class EventRequest(BaseModel): "occurred_at requires a timezone and cannot be in the future" ) return value + + +class AnonymousRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + anonymous_token: str | None = Field(default=None, max_length=2048) diff --git a/src/adcloud_server/identity_responses.py b/src/adcloud_server/identity_responses.py index b833f19..cd08a7f 100644 --- a/src/adcloud_server/identity_responses.py +++ b/src/adcloud_server/identity_responses.py @@ -2,7 +2,7 @@ from datetime import datetime from typing import Literal from uuid import UUID -from pydantic import BaseModel, JsonValue +from pydantic import BaseModel, Field, JsonValue class AnonymousResponse(BaseModel): @@ -31,6 +31,7 @@ class ProfileDocument(BaseModel): class ProfileResponse(ProfileDocument): + features: dict[str, float] = Field(default_factory=dict) event_count: int profile_backend: Literal["relational", "cassandra"] projection_status: Literal["not_enabled", "current", "pending", "unavailable"] @@ -59,6 +60,8 @@ class GraphResponse(BaseModel): class EventResponse(AnonymousResponse): + rule_revision: int | None = None + standard_event: str | None = None event_id: UUID duplicate: bool diff --git a/src/adcloud_server/identity_service.py b/src/adcloud_server/identity_service.py index 1b75a53..e39ad47 100644 --- a/src/adcloud_server/identity_service.py +++ b/src/adcloud_server/identity_service.py @@ -1,6 +1,7 @@ import hashlib import hmac import json +from contextlib import nullcontext from datetime import UTC, datetime from uuid import uuid4 @@ -220,6 +221,9 @@ class IdentityService: } def _save_profile(self, conn, root, document): + from adcloud_server.eventing.worker import feature_values + + document = {**document, "features": feature_values(conn, self.tenant, root)} try: validate_attributes(document["attributes"]) if len(document["tags"]) > 500: @@ -365,6 +369,9 @@ class IdentityService: root = self._root(conn, master_id) row = self._profile(conn, root) document = row["document"] + from adcloud_server.eventing.worker import feature_values + + features = feature_values(conn, self.tenant, root) count = conn.execute( select(func.count()) .select_from(events) @@ -390,6 +397,7 @@ class IdentityService: "event_count": count, "profile_backend": self.settings.profile_backend, "projection_status": projection_status, + "features": features, } def graph(self, master_id): @@ -433,8 +441,10 @@ class IdentityService: "edges": [dict(item) for item in connections], } - def record_event(self, token, request): - with self.db.transaction() as conn: + def record_event(self, token, request, connection=None): + with ( + nullcontext(connection) if connection is not None else self.db.transaction() + ) as conn: subject = self._anonymous(conn, token, strict=bool(request.anonymous_token)) event_id = str(request.event_id or uuid4()) existing = ( diff --git a/src/adcloud_server/main.py b/src/adcloud_server/main.py index efb27c2..eb41d24 100644 --- a/src/adcloud_server/main.py +++ b/src/adcloud_server/main.py @@ -47,7 +47,13 @@ def create_app(settings: Settings | None = None) -> FastAPI: allow_origins=list(settings.cors_origins), allow_credentials=True, allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"], - allow_headers=["Authorization", "Content-Type", "X-Request-ID"], + allow_headers=[ + "Authorization", + "Content-Type", + "X-Request-ID", + "X-Adcloud-Key", + "X-Adcloud-Management-Key", + ], ) application.include_router(v1_router) diff --git a/src/adcloud_server/profile_store.py b/src/adcloud_server/profile_store.py index 07af5ee..e745129 100644 --- a/src/adcloud_server/profile_store.py +++ b/src/adcloud_server/profile_store.py @@ -74,6 +74,8 @@ class CassandraProfileStore: for key, value in document["attributes"].items() } fields["tags"] = json.dumps(document["tags"], ensure_ascii=False) + if "features" in document: + fields["features"] = json.dumps(document["features"], allow_nan=False) fields["_complete"] = str(revision) for field, value in fields.items(): batch.add(self._write, (tenant, master_id, revision, field, value)) @@ -94,6 +96,11 @@ class CassandraProfileStore: if key.startswith("a:") }, "tags": json.loads(fields["tags"]), + **( + {"features": json.loads(fields["features"])} + if "features" in fields + else {} + ), } def close(self): diff --git a/src/adcloud_server/settings.py b/src/adcloud_server/settings.py index 60a4d2e..e4868cf 100644 --- a/src/adcloud_server/settings.py +++ b/src/adcloud_server/settings.py @@ -7,6 +7,7 @@ from typing import Literal class Settings: database_url: str = field(default="sqlite:///.data/adcloud.db", repr=False) profile_backend: Literal["relational", "cassandra"] = "relational" + event_registry_required: bool = True tenant_id: str = "default" identity_secret: str = field(default="", repr=False) service_token: str = field(default="", repr=False) @@ -61,6 +62,7 @@ class Settings: return cls( database_url=os.getenv("DATABASE_URL", "sqlite:///.data/adcloud.db"), profile_backend=os.getenv("PROFILE_BACKEND", "relational"), + event_registry_required=flag("EVENT_REGISTRY_REQUIRED", True), tenant_id=os.getenv("TENANT_ID", "default"), identity_secret=os.getenv("IDENTITY_SECRET", ""), service_token=os.getenv("SERVICE_TOKEN", ""), diff --git a/tests/test_event_registry.py b/tests/test_event_registry.py new file mode 100644 index 0000000..d2e3334 --- /dev/null +++ b/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 + ) diff --git a/tests/test_identity.py b/tests/test_identity.py index 29d61bd..a923f0b 100644 --- a/tests/test_identity.py +++ b/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,