Event enricher

An enricher receives several event fields from Front, computes additional data from them, and returns it. Front appends the result to the event's data before the event reaches the handler and the listeners. mmgeo is an example: from the IP it determines the country, region, and city, and from the User-Agent, the browser, OS, and device.

Unlike a listener, an enricher sits on the critical path of a request: Front waits for its response, by default no longer than 200 ms. For this reason databases are loaded at startup and results are cached.

Subscription and name#

An enricher listens on the shared enrich channel and its own direct channel. Since there are usually several enrichers of the same type, the name is built from the group and the worker number, and the group is specified separately in rpc.service_group:

rpc:
  service_group: <%= env.SERVICE_GROUP || 'unknown_enrich' %>
  name: <%= env.SERVICE_GROUP || 'unknown_enrich' %>_<%= env.WORKER_ID || '1' %>
  channels:
    - enrich
    - <%= env.SERVICE_GROUP || 'unknown_enrich' %>_<%= env.WORKER_ID || '1' %>

service_group is also the key under which the result lands in the event's data: an enricher of the mmgeo group produces data.mmgeo, and of the rst group — data.rst. chwriter then expands the nested object into prefixed columns: mmgeo_country_iso, rst_asn_org.

Registration#

Front learns about an enricher through the heartbeat and __status (see Message bus and RPC). In its response to __status, the enricher lists which events it is interested in and which fields to send:

async setup() {
  this.rpc.register(ENRICH, this.handle.bind(this));
  this.registerEnricher(ENRICH, ['in.gen.track'], { ip: 'td.ip', ua: 'td.ua' });
  this.rpc.register(METHOD_STATUS, () => this.status());

  const aliver = () => {
    this.meter.tick(`band.${this.name}.alive`);
    this.rpc.notify(SERVICE_DIRECTOR, METHOD_IAMALIVE, { name: this.name });
  };
  setTimeout(aliver, 500);
  setInterval(aliver, 9 * 1000);
}

async status(): Promise<RPCAppStatus> {
  return {
    name: this.name,
    app_state: 'running',
    app_started: Number(this.appStarted),
    app_uptime: Date.now() - Number(this.appStarted),
    register: this.enrichers.map(e => ({
      method: e.method,
      role: 'enricher',
      options: { keys: e.keys, props: e.props }
    }))
  };
}

• keys — routing key prefixes. in.gen.track covers all Web-SDK events: in.gen.track.page, in.gen.track.activity, and the rest.
• props — what to send. The key is the parameter name, the value is the dotted path in the event. Front assembles { ip: msg.td.ip, ua: msg.td.ua } from the event and sends only that.

Handling a request#

Front calls the enrich method with a group request to all enrichers registered for the key at once. The method receives exactly the fields requested in props and returns a flat object with the result, or {} if it has nothing to say:

async handle({ ip, ua }: { ip: string, ua: string }) {
  if (this.service_group === 'mmgeo') {
    if (!ip.includes(':')) {               // the database is for IPv4 only
      try {
        return handleCity(this.mmReader.city(ip));
      } catch (e) {
        this.log.error(e, 'mmgeo ex');
      }
    }
  }
  if (this.service_group === 'uap') {
    const cached = uapCache.get(ua);
    if (cached) return cached;
    const parsed = UAParser(ua);
    const result = {
      browser_name: parsed.browser.name,
      browser_version: parsed.browser.version,
      os_name: parsed.os.name,
      os_version: parsed.os.version,
      device_type: parsed.device.type,
      device_vendor: parsed.device.vendor,
      device_model: parsed.device.model
    };
    uapCache.set(ua, result);
    return result;
  }
  return {};
}

What matters for performance:

• Data in memory. mmgeo reads the GeoLite2 database into a buffer once at startup, and parses the IP reputation file into a Map. No disk or network access in handle.
• Result cache. Parsing a User-Agent is expensive, and there are many identical strings. mmgeo caches results in a TTL cache of 10,000 entries for 30 minutes.
• Staying silent is better than erring. On an exception, respond with {} and write to the log. An error in the response only gets logged by Front, and the value for the group stays empty.

The result in the event#

After enrichment, the event in.gen.track.page reaches listeners with the added blocks:

{
  "key": "in.gen.track.page",
  "td": { "ip": "93.184.216.34", "ua": "Mozilla/5.0 ..." },
  "data": {
    "page": { "url": "https://example.com/" },
    "mmgeo": { "country_en": "Russia", "country_iso": "RU", "region_en": "Moscow", "region_iso": "MOW", "city_en": "Moscow" },
    "uapc": { "browser_family": "Chrome", "browser_version": "124.0.0", "os_family": "Android", "os_version": "14", "device_family": "K", "is_bot": false }
  }
}

For the columns to appear in ClickHouse, add them to the chwriter configuration — a file in config/custom/ with fields of the events table under the group prefix:

clickhouse:
  tables:
    events:
      mmgeo_country_en: String
      mmgeo_country_iso: String
      mmgeo_region_en: String
      mmgeo_city_en: String

A new enricher in five steps#

1. Copy mmgeo, keep a single service_group branch in AppServer and your own logic in handle.
2. In config/02_config.yml, set rpc.service_group, rpc.name, and rpc.channels via SERVICE_GROUP and WORKER_ID.
3. In setup(), register enrich and __status with the needed keys and props, and start the heartbeat.
4. Load the reference data in the constructor, and add a cache for hot values.
5. Run it with SERVICE_GROUP=mygroup and check that a data.mygroup block has appeared in the events at the listeners. Then describe its columns in chwriter.