Meiro Pipes — Architecture

Architecture Overview

Pipes is the core processing engine behind Meiro's customer data infrastructure. A single backend handles the full event lifecycle — collection, schema validation, identity resolution, routing, warehouse modeling, and real-time observability. Everything is API-driven, with a built-in JavaScript sandbox for custom transforms and delivery logic.

Data Processing Pipeline

Every event that enters Pipes passes through a seven-stage pipeline. Stages 1–3 are synchronous (in the request path). Stages 4–7 are asynchronous (background workers).

Sources

Sources are the entry points for event data. Each source has a slug, a JavaScript transform function, and bound secrets. The source's transform normalizes raw incoming payloads into arrays of structured events before they enter the processing pipeline.

Webhook Sources

Accepts HTTP POST at /collect/:sourceSlug. Authentication is handled upstream or via IP filtering at the infrastructure layer. The transform function receives the raw request body and headers and must return an array of event objects.

{ event_type: string, event_time: ISO8601 string, event_payload: object }

Web SDK (mpt.js)

Served from your Pipes instance at /mpt.js with a 5-minute cache. The SDK manages four cookies: bp_device (2yr), bp_session (30min), bp_session_count (2yr), bp_first_visit (2yr).

Auto-tracked events (no config required): The SDK ships with over 40 built-in event types — including page views, sessions, clicks, form interactions, video engagement, search, file downloads, and more. All tracked automatically out of the box.

Custom event tracking is defined in SDK config served from /sdk-config/:sourceSlug, cached via ETags and localStorage. Config exposes: sdk.emit(), sdk.on.page(), sdk.on.click(), sdk.on.submit(), sdk.dom.text(), sdk.on.dataLayer().

Collect Endpoint Limits

Scheduled Imports

Run a JavaScript fetch function on a cron schedule. The function can call external APIs, read from object storage, or query any HTTP endpoint, then return arrays of events. Imports can also be triggered manually on demand.

Event Types & Validation

Event Types define the accepted vocabulary for a source — names like purchase, page_view, or subscription_changed. Each has three components:

Identity Resolution

Identifier Types are the keys that connect events, warehouse rows, and profiles. Each type has three properties controlling merge behavior:

Resolution Algorithm

email:      maxIdentifiers=2,  priority=300  // stable, rare; >2 is usually pathological
device_id:  maxIdentifiers=3,  priority=200  // phone + laptop + tablet
cookie:     maxIdentifiers=20, priority=100  // transient, many per real user

Pipes & Event Destinations

A Pipe connects one Source to one Event Destination via an optional JavaScript transform function. The transform receives the event batch and returns a filtered or reshaped array. An Event Destination receives the transformed batch in a custom send() function.

Pipe Transform Function

// Pipe transform — filter to purchase events only
function transform(payload) {
  return payload.events.filter(e => e.event_type === 'purchase');
}

Event Destination send() Function

// Event Destination — POST to a webhook
async function send(payload) {
  const url = env.secrets.WEBHOOK_URL;
  for (const e of payload.events) {
    const res = await fetch(url, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(e)
    });
    if (!res.ok) throw new Error(`Delivery failed: ${res.status}`);
  }
}

Warehouse Modeling

Warehouse Models map tables in Snowflake, BigQuery, or Databricks to Identifier Types, making historical and batch data available to the same identity graph that processes real-time events.

Setup Sequence

1. Create or confirm Identifier Types — identifier columns in the warehouse table must map to an existing type.
2. Create the warehouse connection: provide host, credentials, catalog, and schema.
3. Validate the connection before modeling.
4. Create a Model: specify table name, schema, and identifierRules mapping column names to Identifier Types.
5. Validate by checking whether known test profiles resolve against DWH-backed attributes.

Piper — AI Agent

Every Pipes instance ships with Piper, a context-aware AI agent with full read access to the instance configuration — sources, event types, transforms, destinations, pipes, secrets bindings, queue depths, error logs, and delivery history.

Piper is accessible via the in-product chat and external AI agents that install the CDI skill.

Operations & Observability

Pipes exposes four queues with inspectable depth and worker health: identity resolution, event delivery, profile refresh, and profile delivery. The dashboard provides aggregate counts of events received, stored, and routed.

Troubleshooting Sequence

Follow this order before debugging downstream integrations:

1. Is the source receiving traffic? Check dashboard event volume.
2. Is the payload matching the correct Event Type? Check identifier extraction on representative events.
3. Is the Pipe enabled? Is the Event Destination enabled?
4. Is the transform logic dropping events unexpectedly? Test the transform function directly.
5. What does delivery history show for the affected Pipe and Destination?
6. Is there queue backpressure? Check /api/health/queues.

Version History & Soft Delete

Every entity in Pipes — sources, event types, pipes, destinations, models, identifier types — has version history and soft delete. Accidentally deleted or changed configuration can be restored or reverted without data loss.

CLI

Pipes ships with a CLI that mirrors the full product API — making it fully CI/CD-compatible. A quick taste:

mpcli status
mpcli api GET /api/health/queues

Platform Specifications