Event listener

A listener receives every platform event after enrichment and processing and does something of its own with it: writes it to a database, sends it to an external system, computes statistics. Front expects nothing back, so a listener does not affect the response time to the client. This is how chwriter (ClickHouse) and kafkastream (Kafka) are built.

Subscription#

A listener subscribes to the broadcast channel and registers a method with the same name — that is exactly what Front calls with the notification notify('broadcast', 'broadcast', msg):

const channels = [config.rpc.name, BROADCAST];
// ...
async setup() {
  await this.writer.init();                     // first connect to the sink
  this.rpc.register(BROADCAST, this.writer.write);  // then start receiving events
}

Nothing else is needed: a listener does not have to register with Front.

What arrives#

params carries the whole event. Its fields:

{
  "id": "6410748886712844288",
  "time": 1718000000123,
  "key": "in.gen.track.page",
  "service": "track",
  "name": "page",
  "channel": "track",
  "projectId": 1617632140,
  "uid": "6410748886712844300",
  "td": {
    "ip": "93.184.216.34",
    "ua": "Mozilla/5.0 ...",
    "fpid": "1234567890",
    "host": "example.com",
    "ref": "https://google.com/"
  },
  "data": {
    "page": { "url": "https://example.com/", "title": "Example" },
    "sess": { "type": "organic", "num": 3 },
    "mmgeo": { "country_iso": "RU", "city_en": "Moscow" }
  }
}

Field	Description
id	the event's snowflake identifier, assigned by Front
time	the time of receipt, UTC unix milliseconds
key	the routing key in.gen.<service>.<name>
service, name	the source and name of the event, from which the key is built
channel	which channel the data came through
projectId	the project identifier
uid, uid_param	the visitor identifier and its original value
td	transport data: IP, User-Agent, fingerprint, host, referer
data	the event payload; enricher results are appended here too, each under its own service_group
pancake	additional data from Front

The handler returns {}: the response to a notification goes nowhere.

Routing inside a listener#

The decision of what to do with an event is made from key or from the service/name pair. Both approaches, from working services:

chwriter picks a table from the destinations map in its configuration, with the fallbacks <service>/default and other:

clickhouse:
  destinations:
    track/default: events
    track/activity: activity
    track/scroll: activity
    pixel/default: pixels
    other: webhooks

const nameKey = msg.name.toLowerCase().replace(/\s/g, '_');
const table = this.dest[`${msg.service}/${nameKey}`]
  || this.dest[`${msg.service}/default`]
  || this.dest['other'];
if (!table) return {};

kafkastream walks a list of key prefixes. The first match determines the topic and the path to the field that becomes the Kafka message key; null means "skip":

kafkastream:
  mappings:
    - ['in.gen.track.ping', null]
    - ['in.gen.track.', [rstat.web, uid]]
    - ['in.gen.events_mp_ios.', [rstat.mobile, data.properties_s_device_id]]

for (const [prefix, target] of this.map) {
  if (target && key.startsWith(prefix)) {
    [topic, keyPath] = target;
    break;
  }
}

Buffering#

There are many events, and the handler must return control immediately. Neither service writes to its sink on every event; instead it puts the event in a buffer and flushes it on a timer:

• chwriter keeps one buffer per table and sends them to ClickHouse once every uploadInterval seconds. Failed batches are saved to emergency_dir as JSON so that no data is lost.
• kafkastream keeps one array of messages per topic and every 500 ms sends everything in a single producer.sendBatch.

The pattern is the same:

writers: { [topic: string]: Message[] } = {};

async init() {
  await this.producer.connect();
  setInterval(() => this.flush(), 500);
}

write = async (msg: HandyRecord) => {
  this.meter.tick('kafka.write.called');
  // ... topic selection
  (this.writers[topic] ||= []).push({ key: kafkaKey, value: JSON.stringify(rest) });
  return {};
}

flush() {
  const topicMessages = Object.entries(this.writers)
    .filter(([, items]) => items.length > 0)
    .map(([topic, messages]) => { this.writers[topic] = []; return { topic, messages }; });
  if (topicMessages.length) {
    this.producer.sendBatch({ topicMessages }).catch(e => this.log.error(e));
  }
}

A new listener in five steps#

1. Copy kafkastream, replace the name in package.json, config/02_config.yml (rpc.name), and in the Dockerfile (LABEL band.service.title).
2. Describe your configuration section and its type in src/types.ts.
3. Write a sink class with init() and write methods and a buffer with a timer.
4. In AppServer.setup(), await init() and register write on BROADCAST.
5. Run it locally with the REDIS_DSN of your installation (npm run start:dev) and make sure in the logs that the subscription to broadcast succeeded and events are arriving. Next — building the image.

A minimal sink that simply prints event keys:

import { Deps } from '@app/AppServer';
import { Logger, Meter } from '@rockstat/rock-me-ts';

export class Printer {
  log: Logger;
  meter: Meter;
  counter = 0;

  constructor(deps: Deps) {
    this.log = deps.log.for(this);
    this.meter = deps.meter;
  }

  async init() {
    setInterval(() => {
      this.log.info({ eps: this.counter }, 'events per second');
      this.counter = 0;
    }, 1000);
  }

  write = async (msg: { [k: string]: any }) => {
    this.counter++;
    this.meter.tick('printer.event');
    this.log.debug({ key: msg.key, uid: msg.uid }, 'event');
    return {};
  }
}