بدون استاندارد:
S3 event      → { "Records": [...] }           ← AWS-specific
Azure Blob    → [{ "topic": "/subscriptions/..." }] ← Azure-specific
GitHub webhook → { "action": "opened", ... }    ← GitHub-specific
Custom app    → { "eventType": "order", ... }    ← ad-hoc

Consumer باید برای هر source کد جدا بنویسد

{
  "specversion": "1.0",
  "type": "com.example.order.placed",
  "source": "/order-service/production",
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "time": "2026-09-02T10:30:00Z",
  "datacontenttype": "application/json",
  "subject": "order-12345",
  "dataschema": "https://example.com/schemas/order/v1.json",
  "data": {
    "orderId": "12345",
    "customerId": "67890",
    "totalAmount": 99.99,
    "currency": "USD"
  }
}

┌─────────────────────────────────────────┐
│           CloudEvent Envelope            │
│  ┌───────────────────────────────────┐  │
│  │  Context Attributes (metadata)     │  │
│  │  specversion, type, source, id,    │  │
│  │  time, subject, datacontenttype    │  │
│  └───────────────────────────────────┘  │
│  ┌───────────────────────────────────┐  │
│  │  data (business payload)           │  │
│  │  — schema آزاد، per event type     │  │
│  └───────────────────────────────────┘  │
└─────────────────────────────────────────┘

{
  "traceid": "abc-123-def",
  "partitionkey": "customer-67890"
}

com.github.pull_request.opened
com.google.cloud.storage.object.finalize
io.knative.serving.revision.ready
com.example.order.placed
com.example.order.cancelled

/order-service
https://github.com/myorg/myrepo
/k8s/namespaces/default/services/order-svc
urn:uuid:6e8c-1234-5678

POST /events HTTP/1.1
Host: broker.example.com
Content-Type: application/cloudevents+json

{
  "specversion": "1.0",
  "type": "com.example.order.placed",
  "source": "/order-service",
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "data": { "orderId": "12345" }
}

POST /orders HTTP/1.1
Host: broker.example.com
Content-Type: application/json
ce-specversion: 1.0
ce-type: com.example.order.placed
ce-source: /order-service
ce-id: a1b2c3d4-e5f6-7890-abcd-ef1234567890
ce-time: 2026-09-02T10:30:00Z

{"orderId": "12345", "totalAmount": 99.99}

ce_specversion: 1.0
ce_type: com.example.order.placed
ce_source: /order-service
ce_id: a1b2c3d4-e5f6-7890-abcd-ef1234567890

from cloudevents.http import CloudEvent, to_structured
import requests

attributes = {
    "type": "com.example.order.placed",
    "source": "https://order-service.example.com",
}
data = {"orderId": "12345", "amount": 99.99}

event = CloudEvent(attributes, data)
headers, body = to_structured(event)

requests.post(
    "https://broker.example.com/events",
    headers=headers,
    data=body,
)

import (
    cloudevents "github.com/cloudevents/sdk-go/v2"
    "github.com/google/uuid"
)

func sendEvent(ctx context.Context, client cloudevents.Client) error {
    event := cloudevents.NewEvent()
    event.SetSpecVersion(cloudevents.VersionV1)
    event.SetType("com.example.order.placed")
    event.SetSource("/order-service")
    event.SetID(uuid.New().String())
    event.SetData(cloudevents.ApplicationJSON, map[string]interface{}{
        "orderId": "12345",
        "amount":  99.99,
    })
    return client.Send(ctx, event)
}

func receive(ctx context.Context, event cloudevents.Event) {
    fmt.Printf("type=%s source=%s id=%s\n",
        event.Type(), event.Source(), event.ID())

    var order map[string]interface{}
    event.DataAs(&order)
}

-- مثال مفهومی
SELECT * FROM events
WHERE type LIKE 'com.example.order.%'
  AND source = '/order-service'
  AND time > '2026-09-01T00:00:00Z'

Producer → Broker → Trigger (filter by type/attributes) → Subscriber
                │
                └── Channel (Kafka, NATS, InMemory)

apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  name: order-placed-trigger
  namespace: default
spec:
  broker: default
  filter:
    attributes:
      type: com.example.order.placed
  subscriber:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: order-processor

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: order-sensor
spec:
  dependencies:
    - name: order-placed
      eventSourceName: webhook
      eventName: order
      filters:
        data:
          - path: body.type
            type: string
            value:
              - com.example.order.placed
  triggers:
    - template:
        name: run-workflow
        k8s:
          operation: create
          source:
            resource:
              apiVersion: argoproj.io/v1alpha1
              kind: Workflow
              # ...

{
  "specversion": "1.0",
  "type": "io.debezium.postgresql.data.change",
  "source": "/debezium/postgresql/orderdb/orders",
  "id": "001-00000000000000000001-00000000000000000001-0",
  "time": "2026-09-02T10:30:00Z",
  "datacontenttype": "application/json",
  "data": {
    "before": null,
    "after": { "id": 12345, "status": "placed" },
    "op": "c"
  }
}

                    ┌──────────────┐
  S3 ──► CE ───────►│              │
  GitHub ──► CE ───►│ Event Router │──► filter by type ──► Service A
  Debezium ──► CE ─►│  (generic)   │──► filter by source ─► Service B
  Custom ──► CE ───►│              │──► dead letter queue
                    └──────────────┘

Database (Debezium CDC)
    │
    ▼ CloudEvents
Kafka / Knative Broker
    │
    ├── Trigger → Knative Service
    ├── [Argo Events](/blog/argo-project-kubernetes-gitops-cicd/) Sensor → Workflow
    └── [Cilium](/blog/cilium-ebpf-kubernetes-networking/) Hubble (network events)

# Knative — inspect broker events
kubectl get triggers -A
kubectl describe trigger order-placed-trigger

# Argo Events — sensor logs
kubectl logs -n argo-events deploy/sensor-controller -f