---
name: eventhub-callback
description: Build, debug, and extend EventHub webhook integrations for the CITV EventHub callback processing center. Use whenever the user mentions EventHub, eh.citv.cc, webhook callback forwarding, live room events, VOD events, content moderation callbacks, risk identification events, or any of the supported event sources (Tencent Cloud Live, Tencent Cloud VOD, Tencent CMS, Tencent TCMS, Volcengine Live, Volcengine SaaS Live, Volcengine Risk Identification, SaaS Live, VZan Live, CITV compliance/open API/analytics). Also trigger when working on parsers, dispatchers, or handlers under `internal/dispatcher/` in the TaskHub Go backend, when the user wants to add a new event source, when debugging signature verification, or when reviewing the `eventhub` extension object structure.
---

# EventHub Callback Integration

EventHub is a unified webhook callback processing center that receives and normalizes events from many Chinese cloud platforms (Tencent Cloud, Volcengine, VZan, CITV-internal) so the Checker pipeline can run content compliance checks on them.

The TaskHub Go backend is the consumer side: it parses these events off Kafka (`internal/dispatcher/**`) and reacts. This skill covers the protocol, the event sources, and the patterns to use when adding or modifying an integration.

## Event Flow

Always keep this four-stage model in mind when reasoning about a callback:

```
Source platform  ->  Customer's receiver service  ->  EventHub  ->  Checker
(Tencent/Volc/...)    (your backend)               (eh.citv.cc)   (compliance)
```

Two implications that come up constantly:

- The customer's service is what receives the cloud-platform's raw callback first, then forwards a *modified* payload to EventHub. EventHub does not see the raw platform payload directly except via the forwarded root object.
- EventHub does not own the sign/signature. The customer recomputes it using the AppSecret that EventHub issued for that `appid`, and writes the resulting `sign` into the root object before forwarding.

## Endpoint Shape

EventHub uses one HTTPS endpoint per platform, with a path parameter for the customer-assigned `appid`:

| Platform | Path |
|---|---|
| Tencent Cloud Live | `https://eh.citv.cc/event/tx/live/{appid}` |
| Tencent Cloud VOD | `https://eh.citv.cc/event/tx/vod/{appid}` |
| Tencent CMS / TCMS | `https://eh.citv.cc/event/tx/cms/{appid}` (and `/tcms`) |
| Volcengine Live | `https://eh.citv.cc/event/volc/live/{appid}` |
| Volcengine SaaS Live | `https://eh.citv.cc/event/volc/live-saas/{appid}` |
| Volcengine Risk | `https://eh.citv.cc/event/volc/risk/{appid}` |
| VZan Live | `https://eh.citv.cc/event/vzan/live/{appid}` |
| CITV compliance / open / analytics | `https://eh.citv.cc/event/citv/{path}/{appid}` |

All endpoints accept `POST application/json`. EventHub resolves `{appid}` to an application config (the customer's `AppSecret` lives there) and uses that secret to verify the `sign` field in the root object.

## Payload Structure

The body is **the raw cloud-platform callback object as the root, with an `eventhub` field added on top**. Do not wrap or re-shape the root — keep every field the platform sent, and only attach `eventhub` as an extra sibling.

```json
{
  "...every original field from the cloud platform...": "...",
  "eventhub": {
    "title": "Live room title",
    "type": 1,
    "ccid": "merchant_123_room_456",
    "domains": ["shop.citv.cn", "live.shop.citv.cn"],
    "pageUrl": "https://shop.citv.cn/live/123",
    "auditUrl": "https://internal.citv.cn/audit/123",
    "auditors": ["ai"],
    "startTime": 1773194400000,
    "endTime":   1773196602000,
    "views": 26,
    "providers": [
      { "type": "live", "serviceId": "sPq1qe1riZ", "contentId": "merchant_123_room_456_h5_mix" }
    ],
    "vodToLive": "",
    "vodSourceCcid": ""
  }
}
```

### `eventhub` field reference

| Field | Type | Required | Notes |
|---|---|---|---|
| `title` | string | yes | Display name — live room title, VOD title, etc. |
| `type` | number | yes | Content type: `1` = live, `2` = VOD, `3` = pseudo-live (VodToLive) |
| `ccid` | string | yes | Customer's content id; used to join with the customer's own data |
| `domains` | string[] | yes | Fully qualified hostnames that should be allowed to serve this content |
| `pageUrl` | string | yes | End-user-facing URL for the live room or VOD page |
| `auditUrl` | string | yes | Internal URL the auditor opens when reviewing |
| `auditors` | string[] | yes | Auditor ids; `["ai"]` means AI-only review |
| `startTime` | number | yes | 13-digit ms timestamp; when the content started |
| `endTime` | number | yes | 13-digit ms timestamp; when the content ended |
| `views` | number | optional | Real-time viewer count snapshot |
| `providers` | object[] | yes | One entry per cloud platform the content lives on |
| `providers[].type` | string | yes | `"live"` or `"media"` |
| `providers[].serviceId` | string | yes | App/console id at the cloud provider |
| `providers[].contentId` | string | yes | Id of the content at the cloud provider |
| `vodToLive` | string | optional | When `type=3`, the VOD content id being looped |
| `vodSourceCcid` | string | optional | When `type=3`, the original VOD `ccid` |

## Supported Event Sources

When the user mentions one of these, jump straight to its page in the docs (or its parser under `internal/dispatcher/`):

- **Tencent Cloud Live** — push start, stream disconnect, screenshot, recording ready, etc. Parser: `internal/dispatcher/live_room/tencent_live/`.
- **Tencent Cloud VOD** — `NewFileUpload`, `ProcedureStateChanged`, `ReviewAudioVideoComplete`. Parser: `internal/dispatcher/vod/tencent_vod/`.
- **Tencent Cloud CMS** — content security risk verdicts (`PASS` / `REVIEW` / `BLOCK`). Parser: `internal/dispatcher/live_risk/tencent_cms/`.
- **Tencent Cloud TCMS** — Tianyu content moderation verdicts. Parser: `internal/dispatcher/live_risk/tencent_tcms/`.
- **Volcengine Live** — Volc live room events. (V1)
- **Volcengine SaaS Live** — Volc SaaS live room events. Parser: `internal/dispatcher/live_room/volc_saas_live/`.
- **Volcengine Risk Identification** — risk verdicts. Parser: `internal/dispatcher/live_risk/volcengine_live_risk/`.
- **SaaS Live** — generic SaaS live platforms.
- **VZan Live** — Weizan live room events. Parser: `internal/dispatcher/live_room/vzan_live/`.
- **CITV compliance callback** — internal business callbacks (v1 and v2).
- **CITV open API** — internal open API callbacks.
- **CITV Live Analytics** — page-view / analytics events, typically the `/analytics/page-views` ingestion endpoint, behind `PublicCORS` allowing `https://*.citv.cn` origins.

Routing in the Go backend is keyed by `dispatcher.Event.SourceType` and dispatched in `internal/transport/kafka/processor/processor.go`. Each source has its own package under `internal/dispatcher/<area>/<source>/` with `types.go`, `parser.go`, and usually `parser_test.go`.

## Code Patterns in This Repo

When adding or modifying an EventHub integration in TaskHub, follow these patterns — they are already established in every existing dispatcher.

### 1. Define the source-specific event type

Mirror the cloud platform's JSON exactly, and embed the `EventHub` extension struct on top so it deserializes from the `eventhub` field. See `internal/dispatcher/live_room/tencent_live/types.go:6` (`TencentLiveEvent`) for the canonical example. The `EventHub` struct is repeated in each source package (deliberate — each source owns its own copy and can add fields without cross-package coupling).

Common event-type constants per source:

- Tencent Live: `1` = stream start, `2` = stream stop, `0` = stop (legacy), `100` = recording generated.
- Volcengine Risk: `PASS` / `REVIEW` / `BLOCK`.
- Tencent VOD: `NewFileUpload`, `ProcedureStateChanged`, `ReviewAudioVideoComplete`.

### 2. Custom `UnmarshalJSON` when the upstream has aliases

Cloud platforms drift between camelCase and snake_case, and Kafka messages can hold either form depending on the producer. Use the `aux struct` + `*alias` trick so both forms decode into the same Go field. Example: `TencentLiveEvent.UnmarshalJSON` accepts both `video_url` and `videoUrl`. Prefer the snake_case (legacy) form when both are present.

### 3. Parser returns a normalized internal event

The `parser.go` for each source converts the source-specific event into a normalized internal event the rest of the pipeline consumes. Do not let raw source payloads leak past the dispatcher boundary.

### 4. Tests live next to the parser

Every parser has a `parser_test.go` with table-driven cases. When adding a new event type, add a test in the same file. Tests are fast and need no external services — `go test -v ./internal/dispatcher/<area>/<source>` runs them.

### 5. New source = new package

To add a new platform: create `internal/dispatcher/<area>/<new_source>/` with `types.go`, `parser.go`, `parser_test.go`, then register the `SourceType` constant and wire the dispatcher in `internal/transport/kafka/processor/processor.go` (see the `SourceType` table in the project `CLAUDE.md`).

## Common Tasks

### Add a new event source
1. Create `internal/dispatcher/<area>/<new_source>/{types.go,parser.go,parser_test.go}`.
2. Define the source-specific event type and a local `EventHub` struct.
3. Implement `parser.go` returning the normalized internal event.
4. Add a Kafka route by `SourceType` in `internal/transport/kafka/processor/processor.go`.
5. Add a fixture-based test in `parser_test.go`.

### Debug a missing / unparseable callback
1. Confirm the `eventhub` extension object is present and well-formed (every required field above).
2. Confirm `sign` is the customer's recomputed value using their `AppSecret` (not the cloud platform's original).
3. Confirm the path uses the customer's `appid` and that the platform segment is correct.
4. Inspect the Kafka topic — `KAFKA_TOPIC_LIVE` (default `live_room_events`) for live/risk, `KAFKA_TOPIC_MERCHANT` (default `eventhub-merchant`) for merchant events.
5. If the dispatcher logged "no source processor matches", the `SourceType` was not registered.

### Forward to EventHub from a customer service
- Receive the cloud-platform callback at the URL configured in the cloud console.
- Keep the root object intact; do not move fields into `eventhub`.
- Append the `eventhub` object with the required fields populated from your own business data (`ccid`, `pageUrl`, `auditUrl`, `auditors`, `domains`, `providers`, timestamps).
- Recompute the root object's `sign` using the EventHub-issued `AppSecret` for that `appid` (the cloud platform's original signature is not what EventHub verifies against).
- `POST` the merged JSON to `https://eh.citv.cc/event/<platform>/<kind>/{appid}` with `Content-Type: application/json`.

## What NOT to Do

- Do not wrap the cloud-platform root in another object — EventHub expects the platform's own JSON as the root, with `eventhub` attached.
- Do not strip fields the platform sent. Keep the full original payload; future event types may rely on fields not yet used.
- Do not share one `EventHub` struct across dispatcher packages. Each source owns its own copy.
- Do not bypass `parser.go` and pass source-specific types further down the pipeline.
- Do not invent new `type` values for the `eventhub.type` field — only `1` (live), `2` (VOD), `3` (pseudo-live) are defined.