feat: Adjust S2S ingest with single install-source authority

.env.example

@@ -48,6 +48,14 @@ MAIL_FROM="Adex <no-reply@your-domain.com>"
 # openssl rand -hex 32
 CRON_SECRET=""
 
+# Growth ingest secrets (org-level PlatformAuth rows override these fallbacks).
+# INGEST_WEBHOOK_SECRET — HMAC signing key for POST /api/ingest/events.
+# INGEST_ADJUST_SECRET — static token for GET|POST /api/ingest/adjust. MUST be
+#   distinct from INGEST_WEBHOOK_SECRET: the Adjust token travels in cleartext
+#   query strings (lands in access logs) and must not unlock HMAC forgery.
+INGEST_WEBHOOK_SECRET=""
+INGEST_ADJUST_SECRET=""
+
 # Demo seed config (used by `npm run db:seed`).
 #  - SEED_EMAIL: defaults to demo@adexads.com if unset.
 #  - SEED_PASSWORD: if unset, the seed script generates a cryptographically

docs/growth/06-mmp-ingest.md

@@ -0,0 +1,52 @@
+# MMP（Adjust / AppsFlyer）接入新数据链路指南
+
+> 版本 v1 · 2026-07-07 · 结论先行：**MMP 走 S2S callback → ConversionEvent；接之前必须先堵三个洞，否则 GA4 + MMP 双开会让 install 翻倍、CAC 腰斩，且现有测试测不出来。**
+
+## 0. 现状：MMP 只喂旧链路
+
+`src/lib/platforms/adjust.ts` / `appsflyer.ts` 目前只做一件事：`reports/sync` 拉 app 级日聚合（Adjust 维度仅 `day,app`，无渠道拆分）→ 写 `Report(level=account)` → `/dashboard` 展示。**这条旧链路保留不动**——它是"MMP 原生总量视图"，与新链路（ConversionEvent → CohortSnapshot → `/growth` + agent perceive）物理隔离（读写零交集），二者数字对不上是口径差异，不是 bug，不要试图对齐。
+
+真正的冲突全在新链路内部：让 MMP 往 ConversionEvent 写数据时，它会和 GA4 在同一张 CohortSnapshot 里打架。
+
+## 1. 三个必堵的洞（接入前置，缺一不可）
+
+| # | 洞 | 成因 | 后果 | 堵法 |
+| --- | --- | --- | --- | --- |
+| 1 | **双源 install 重复计数** | 幂等键 `@@unique([orgId, source, eventName, userKey, occurredAt])` 把 `source` 算进唯一约束（GA4/Adjust 各落一行），而 `cohorts.ts` 聚合完全不感知 source、只按 userKey 分组累加 | installs 翻倍 → CAC 腰斩 → gate 判定（放量/CAC 上限）资损级误判 | `RawEvent` 加 `source` 字段；cohort 聚合前按 org 配置的 install 权威源单选（见 §2 决策 A） |
+| 2 | **channel 归一化失配** | `channels.ts:resolveChannel` 只认 `adex_*` UTM 和 kol/seo 等裸码；MMP 的 network 名（"Apple Search Ads"、"Meta Installs"）全部兜底成 `organic` | 同一渠道在 CohortSnapshot 裂成两行（如 ASA 裂成 `organic` + `paid_asa`），看板和 gate 各算各的 | 新增 `ADJUST_NETWORK_MAP` 显式映射表 + `resolveAdjustChannel()`；**不要扩 `resolveChannel`**（它服务自有 UTM 主路径） |
+| 3 | **userKey 跨源断裂** | 留存 D1/D7 = "同一 userKey 在 +1/+7 有 GA4 行为事件"；RC 收入也靠 userKey join。MMP 的 adid/idfa 与 GA4 pseudo_id、RC app_user_id 是不同命名空间 | MMP 渠道的 cohort **留存和 LTV 假性归零**；同人被当两个 user 重复计 install | 首选：callback 透传 RC `app_user_id` 统一 userKey（需客户侧配置，Adjust partner parameter 支持）；降级：userKey 加 `adjust:` 前缀、承认跨源不 join（见 §2 决策 B） |
+
+## 2. 两个口径决策（已拍板）
+
+- **决策 A · install 权威源**（已拍板）：接了 Adjust 的 org，install 与渠道归因以 `source='adjust'` 为权威；GA4 只供 first_chat / scene_generated 等 Adjust 不报的漏斗深层事件。实现：`kpi-canon.ts:resolveInstallAuthority({ hasAdjustAuth, adjustInstallCount, ga4InstallCount })`——org 存在 `PlatformAuth(platform='adjust')` → `'adjust'`，否则 `'ga4'`。**防归零保险**：若权威源在计算窗口内 install 数为 0 而另一源 > 0（例如 org 只配了 legacy Report 凭证、没接 `/api/ingest/adjust` callback），本次计算回退另一源，并在返回值 `fallback`/`warning` 字段标注，`growth-sync` cron 会把 warning 打进日志。`cohorts.ts:buildCohortSnapshots` 的 `opts.installAuthority` 据此只过滤 ACQUISITION 类事件（install/signup）的来源；漏斗深层/收入事件不受影响。
+- **决策 B · userKey 统一方式**（已拍板）：userKey 首选 Adjust callback 透传的 RC `app_user_id`（partner parameter，需客户侧在 Adjust 后台配置）；取不到时降级为 `adjust:${adid}` 前缀，明确承认这条 userKey 跨源不可 join——留存/LTV 仍以 GA4/RC 体系计算，Adjust 只贡献 install 计数与渠道校正。实现见 `adjust-ingest.ts:mapAdjustCallback`。
+
+## 3. 接入步骤（Adjust，S2S callback 方案）
+
+**不推荐**用 Report Service API 拉聚合造合成事件——聚合行没有 userKey，进不了 cohort。
+
+1. **堵洞（代码前置）**——已完成
+   - `src/lib/growth/cohorts.ts`：`RawEvent` 加 `source`；`buildCohortSnapshots(events, { installAuthority })` 聚合前按决策 A 单源过滤 ACQUISITION 类事件（install/signup），漏斗深层/收入事件不过滤；`growth-sync` cron 查询同步加 `source` select
+   - `src/lib/growth/channels.ts`：新增 `ADJUST_NETWORK_MAP: Record<string, Channel>` + `resolveAdjustChannel(networkName, campaignName)`。ASA 确定性映射；Meta/TikTok 的 install network 名（`Facebook Installs` / `TikTok Installs` 等）保守映射到 `*_ios`（SKAN，低置信度）——Adjust 的 network_name 分不清 web 与 app-install，`campaign_name` 含 "web" 时才改判 `*_web`；未映射网络名 → `organic`/`inferred`。未改动 `resolveChannel` 一行
+   - `src/lib/growth/adjust-ingest.ts`（新建 + 测试）：`mapAdjustCallback(params, eventTokenMap?) → ConversionEventInput | null`，照 `ga4.ts:mapGa4Event` 的写法；`activity_kind='install'` → `install`；`='event'` 时查 `eventTokenMap`（event token → canonical `EventName`，映射不到 → drop）；`reattribution`/`session` 等 → drop（返回 null，不抛错）
+   - 测试含**双源用例**（`cohorts.test.ts`）：同一真实安装的 GA4 + Adjust 两行（不同 userKey namespace）不设权威时 installs=2（复现洞 1）；设 `installAuthority: 'adjust'` 后 installs=1 且渠道以 Adjust 行为准；另有用例确认 installAuthority 不过滤漏斗深层事件
+2. **鉴权路由**——已完成，选定方案一
+   - `GET|POST /api/ingest/adjust?org=<id>`（`src/app/api/ingest/adjust/route.ts`），静态密钥 + `verifyBearer`（仿 `src/app/api/ingest/revenuecat/route.ts`）。**密钥槽独立**：存 `PlatformAuth(orgId, platform='ingest_adjust').apiKey`，env `INGEST_ADJUST_SECRET` 兜底——不与 `/api/ingest/events` 的 HMAC 密钥共用，因为 Adjust token 走 `?token=` 明文 query 会进访问日志，泄漏不能连带 HMAC 伪造。同一行的 `extra` JSON 配置 **event token 映射**：`{"eventTokenMap":{"<adjust event token>":"trial_start"}}`——不配置时非 install 事件会被安全丢弃（200 ignored），install 不受影响；Adjust 侧密钥支持 `Authorization: Bearer` 或 `?token=` query（callback 是 URL 模板，设不了自定义 header）；单条映射失败跳过（200 ok, ignored），不 5xx；写入沿用 `createMany` + `skipDuplicates` 幂等
+3. **Adjust 后台配置**：Raw Data Exports → Real-time callbacks，install + 关键 event 各配一条，URL 带 placeholder：`{activity_kind}`、`{network_name}`、`{campaign_name}`、`{adid}`、`{idfa}`、`{created_at}`、`{country}`，及透传的 partner parameter（决策 B 的 app_user_id）
+4. **验证清单**：同一 callback 重放两次只落一行（幂等键）；ASA/Meta 的 network 名映射进正确 channel（不落 organic）；GA4+Adjust 双开的 org 在 `/growth` 的 installs 与单开 MMP 一致；`/dashboard` 的 Report.adjust 总量不受影响
+
+## 4. AppsFlyer 附注
+
+同构接入：Push API（S2S 实时回传）对应 Adjust callback，`pid/c` 字段对应 `network_name/campaign_name`（需 `APPSFLYER_NETWORK_MAP`），设备标识 `appsflyer_id` 同样是独立命名空间——三个洞和两个决策**一字不差地适用**。旧链路 `getInstallReport`（`groupings=date,pid`）同样保留不动。AppsFlyer 接入尚未实现，本节仍是规划。
+
+## 5. 涉及文件速查
+
+| 动作 | 文件 | 状态 |
+| --- | --- | --- |
+| 新建 mapper | `src/lib/growth/adjust-ingest.ts`（+ `adjust-ingest.test.ts`） | 已完成 |
+| 新建路由 | `src/app/api/ingest/adjust/route.ts`（GET+POST） | 已完成 |
+| 改：channel 映射 | `src/lib/growth/channels.ts`（`ADJUST_NETWORK_MAP` / `resolveAdjustChannel`，+ 测试） | 已完成 |
+| 改：单源聚合 | `src/lib/growth/cohorts.ts`、`src/app/api/cron/growth-sync/route.ts`（+ 测试） | 已完成 |
+| 改：口径记录 | `src/lib/growth/kpi-canon.ts`（`resolveInstallAuthority`，+ 测试）、本文件 §2 | 已完成 |
+| 不动 | `src/lib/platforms/adjust.ts` / `appsflyer.ts`、`reports/sync`、`/api/ingest/events` 鉴权、`ingest-parse.ts`（`source='adjust'` 已注册） | 按计划不动 |
+| 未实现 | AppsFlyer S2S 接入（§4 仍是规划，非本次范围） | 待办 |

src/app/api/cron/growth-sync/route.ts

@@ -2,7 +2,8 @@ import { NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/cron-auth'
 import { prisma } from '@/lib/prisma'
 import { buildCohortSnapshots, type RawEvent } from '@/lib/growth/cohorts'
-import { EVENTS, type EventName } from '@/lib/growth/events'
+import { EVENTS, SOURCES, type EventName } from '@/lib/growth/events'
+import { resolveInstallAuthority } from '@/lib/growth/kpi-canon'
 
 /**
  * POST /api/cron/growth-sync
@@ -28,12 +29,12 @@ export async function POST(req: NextRequest) {
   const cutoff = new Date(Date.now() - WINDOW_DAYS * 86_400_000)
   const orgs = await prisma.organization.findMany({ select: { id: true } })
 
-  const summary: Array<{ orgId: string; cohorts: number; events: number }> = []
+  const summary: Array<{ orgId: string; cohorts: number; events: number; installAuthority?: string; installAuthorityWarning?: string }> = []
 
   for (const org of orgs) {
     const rows = await prisma.conversionEvent.findMany({
       where: { orgId: org.id, occurredAt: { gte: cutoff } },
-      select: { eventName: true, occurredAt: true, userKey: true, channel: true, revenue: true },
+      select: { eventName: true, occurredAt: true, userKey: true, channel: true, revenue: true, source: true },
     })
 
     const events: RawEvent[] = rows
@@ -44,9 +45,29 @@ export async function POST(req: NextRequest) {
         userKey: r.userKey,
         channel: r.channel,
         revenue: r.revenue,
+        source: r.source,
       }))
 
-    const cohorts = buildCohortSnapshots(events)
+    // Install authority (decision A, docs/growth/06-mmp-ingest.md §2): if the
+    // org has Adjust wired, Adjust's install events are authoritative and the
+    // GA4 install rows for the same window are excluded from cohort placement
+    // (see cohorts.ts). The anti-zeroing guard in resolveInstallAuthority falls
+    // back to GA4 if Adjust is configured but reported 0 installs this window.
+    const hasAdjustAuth = await prisma.platformAuth.findUnique({
+      where: { orgId_platform: { orgId: org.id, platform: 'adjust' } },
+    }).then((a) => !!a?.isActive).catch(() => false)
+    const adjustInstallCount = events.filter((e) => e.eventName === EVENTS.INSTALL && e.source === SOURCES.ADJUST).length
+    const ga4InstallCount = events.filter((e) => e.eventName === EVENTS.INSTALL && e.source === SOURCES.GA4).length
+    const installAuthorityResult = resolveInstallAuthority({
+      hasAdjustAuth,
+      adjustInstallCount,
+      ga4InstallCount,
+    })
+    if (installAuthorityResult.warning) {
+      console.warn(`[growth-sync] org=${org.id} ${installAuthorityResult.warning}`)
+    }
+
+    const cohorts = buildCohortSnapshots(events, { installAuthority: installAuthorityResult.authority })
 
     // Idempotent recompute: clear the window, then insert fresh.
     await prisma.cohortSnapshot.deleteMany({
@@ -72,7 +93,13 @@ export async function POST(req: NextRequest) {
       })
     }
 
-    summary.push({ orgId: org.id, cohorts: cohorts.length, events: events.length })
+    summary.push({
+      orgId: org.id,
+      cohorts: cohorts.length,
+      events: events.length,
+      installAuthority: installAuthorityResult.authority,
+      ...(installAuthorityResult.warning ? { installAuthorityWarning: installAuthorityResult.warning } : {}),
+    })
   }
 
   return NextResponse.json({ ok: true, window: `${WINDOW_DAYS}d`, orgs: summary })

src/app/api/ingest/adjust/route.ts

@@ -0,0 +1,125 @@
+import { NextRequest, NextResponse } from 'next/server'
+import { prisma } from '@/lib/prisma'
+import { verifyBearer, readBearer } from '@/lib/growth/ingest-auth'
+import { mapAdjustCallback, type AdjustCallbackParams, type AdjustEventTokenMap } from '@/lib/growth/adjust-ingest'
+import { EVENTS, type EventName } from '@/lib/growth/events'
+
+const EVENT_NAME_SET = new Set<string>(Object.values(EVENTS))
+
+/**
+ * GET|POST /api/ingest/adjust?org=<orgId>&token=<secret>
+ *
+ * Receives an Adjust real-time S2S callback (docs/growth/06-mmp-ingest.md §3)
+ * and writes a normalized ConversionEvent. Adjust's callback is a URL template
+ * it expands itself — it can't sign a per-request HMAC and typically fires as
+ * a GET, so auth is a static secret compared constant-time, same as
+ * /api/ingest/revenuecat, except the secret may arrive via `?token=` (Adjust
+ * can't set custom headers) in addition to `Authorization: Bearer`.
+ *
+ * Auth: PlatformAuth(orgId, platform='ingest_adjust').apiKey, falling back to
+ * the INGEST_ADJUST_SECRET env var — a slot distinct from the events-route
+ * HMAC secret, because this key travels in cleartext query strings. The same
+ * row's `extra` JSON supplies the event-token map (custom Adjust event tokens
+ * → canonical event names).
+ *
+ * A single callback maps to at most one event; unmapped activity_kind /
+ * event_token combinations are dropped (200 ok, not written) so Adjust never
+ * sees a 4xx/5xx and retries. Idempotent via the ConversionEvent unique key.
+ */
+export async function GET(req: NextRequest) {
+  return handle(req)
+}
+
+export async function POST(req: NextRequest) {
+  return handle(req)
+}
+
+async function handle(req: NextRequest) {
+  const url = new URL(req.url)
+  const orgId = url.searchParams.get('org')
+  if (!orgId) {
+    return NextResponse.json({ error: 'missing ?org' }, { status: 400 })
+  }
+
+  // Distinct secret slot (platform='ingest_adjust', env INGEST_ADJUST_SECRET).
+  // Deliberately NOT the '/api/ingest/events' HMAC secret: this route has to
+  // transmit its key in cleartext via `?token=` (Adjust can't set headers),
+  // which lands in LB/access logs — a leak here must not let anyone forge
+  // HMAC signatures for the events route. The same row's `extra` JSON also
+  // carries the org's Adjust event-token map: {"eventTokenMap":{"abc123":"trial_start"}}.
+  let expected: string | undefined = process.env.INGEST_ADJUST_SECRET
+  let eventTokenMap: AdjustEventTokenMap | undefined
+  try {
+    const auth = await prisma.platformAuth.findUnique({
+      where: { orgId_platform: { orgId, platform: 'ingest_adjust' } },
+    })
+    if (auth?.apiKey) expected = auth.apiKey
+    if (auth?.extra) {
+      try {
+        const extra = JSON.parse(auth.extra) as { eventTokenMap?: Record<string, unknown> }
+        if (extra.eventTokenMap && typeof extra.eventTokenMap === 'object') {
+          // Only keep entries whose value is a canonical EventName — a typo'd
+          // config entry degrades to "token unmapped → dropped", never a bad row.
+          const validated: AdjustEventTokenMap = {}
+          for (const [token, name] of Object.entries(extra.eventTokenMap)) {
+            if (typeof name === 'string' && EVENT_NAME_SET.has(name)) {
+              validated[token] = name as EventName
+            }
+          }
+          eventTokenMap = validated
+        }
+      } catch {
+        // malformed extra JSON — proceed without a map (installs still ingest)
+      }
+    }
+  } catch {
+    // fall through to env fallback
+  }
+
+  const provided = readBearer(req.headers.get('authorization')) ?? url.searchParams.get('token')
+  if (!verifyBearer(provided, expected)) {
+    return NextResponse.json({ error: 'unauthorized' }, { status: 401 })
+  }
+
+  const params: AdjustCallbackParams = Object.fromEntries(url.searchParams.entries())
+  if (req.method === 'POST') {
+    try {
+      const form = await req.formData()
+      for (const [k, v] of form.entries()) {
+        if (typeof v === 'string') params[k] = v
+      }
+    } catch {
+      // no form body — GET-style query params only, that's fine
+    }
+  }
+  // Never persist the auth secret into ConversionEvent.raw — strip AFTER the
+  // form merge, or a form field named `token` would smuggle it back in.
+  delete params.token
+
+  const mapped = mapAdjustCallback(params, eventTokenMap)
+  if (!mapped) {
+    // Not funnel-relevant (reattribution/session, unmapped event token). Ack
+    // so Adjust doesn't retry.
+    return NextResponse.json({ ok: true, ignored: true })
+  }
+
+  const result = await prisma.conversionEvent.createMany({
+    data: [
+      {
+        orgId,
+        source: mapped.source,
+        eventName: mapped.eventName,
+        occurredAt: mapped.occurredAt,
+        userKey: mapped.userKey ?? null,
+        utmCampaign: mapped.utmCampaign ?? null,
+        channel: mapped.channel ?? null,
+        country: mapped.country ?? null,
+        revenue: mapped.revenue ?? 0,
+        raw: JSON.stringify(mapped.raw ?? params),
+      },
+    ],
+    skipDuplicates: true,
+  })
+
+  return NextResponse.json({ ok: true, written: result.count })
+}