Kafka Events Reference

{
  "specversion": "1.0",
  "id": "evt_unique_uuid",
  "source": "eventzr.ads-svc",
  "type": "eventzr.ads.campaign.created.v1",
  "subject": "cmp_xyz789",
  "time": "2026-03-09T10:00:00.000Z",
  "datacontenttype": "application/json",
  "tenantid": "<tenant-uuid>",
  "data": {
    // Domain-specific payload (see examples below)
  }
}

import { Injectable } from '@nestjs/common';
import { KafkaConsumer, OnKafkaEvent } from '@eventzr/kafka';
import { Logger } from '@eventzr/logger';

@Injectable()
export class AdsCampaignConsumer {
  private readonly logger = new Logger(AdsCampaignConsumer.name);

  @OnKafkaEvent('eventzr.ads.campaign.created.v1')
  async handleCampaignCreated(event: CloudEvent<CampaignCreatedPayload>): Promise<void> {
    const { data, tenantid } = event;
    this.logger.log(`Campaign created: ${data.id} for tenant ${tenantid}`);

    // Process the event (e.g., create matching records, send notifications)
    await this.setRLSContext(tenantid);
    await this.campaignRepo.save({
      externalCampaignId: data.id,
      name: data.name,
      tenantId: tenantid,
    });
  }

  @OnKafkaEvent('eventzr.ads.broadcast.delivered.v1')
  async handleBroadcastDelivered(event: CloudEvent<BroadcastDeliveredPayload>): Promise<void> {
    const { data, tenantid } = event;
    this.logger.log(`Broadcast delivered: ${data.broadcastId}, impressions: ${data.impressionsDelivered}`);

    // Update analytics, trigger notifications, etc.
  }

  @OnKafkaEvent('eventzr.ads.campaign.review.completed.v1')
  async handleReviewCompleted(event: CloudEvent<ReviewCompletedPayload>): Promise<void> {
    const { data, tenantid } = event;
    if (data.decision === 'rejected') {
      this.logger.warn(`Creative rejected: ${data.creativeId}, reason: ${data.reason}`);
      // Notify campaign owner via notify-svc
    }
  }
}