1. Introduction — Why Klaviyo Dominates E-Commerce Marketing

Klaviyo has fundamentally reshaped how e-commerce businesses approach marketing automation. With over 100,000 paying customers across 190 countries, the platform processes more than 8.8 billion customer interactions every single day and has been directly attributed to over $10 billion in revenue for its merchant base. These are not aspirational marketing numbers — they represent real transactional outcomes driven by deeply integrated, data-first marketing systems.

At its core, Klaviyo succeeds because it treats customer data as the primary product. Unlike generic email service providers that bolt on e-commerce features as an afterthought, Klaviyo was purpose-built for platforms like Shopify, BigCommerce, and Magento. It ingests every customer touchpoint — product views, cart additions, purchases, refund requests, loyalty program events — and transforms that raw behavioral data into actionable marketing moments delivered via email, SMS, and push notifications.

The platform's competitive moat lies in three pillars that any system design interview candidate should deeply understand. First, the real-time event ingestion system that synchronizes millions of Shopify webhooks per second without dropping events. Second, the behavioral segmentation engine that can evaluate millions of customers against complex rule sets in near real-time. Third, the predictive analytics layer that uses machine learning to forecast customer lifetime value, churn risk, and next purchase dates — all powered by the first-party data flowing through the system continuously.

Understanding how to build a system of this magnitude requires a senior-plus engineering mindset. You must think about distributed systems, event-driven architectures, eventually consistent data stores, and the unique challenges of sending millions of time-sensitive marketing messages while maintaining deliverability reputations across dozens of email service providers and SMS carriers worldwide.

This article walks through every major subsystem of a Klaviyo-like platform, from Shopify webhook integration through event ingestion, customer profile unification, behavioral segmentation, flow automation, campaign management, predictive analytics, revenue attribution, and multi-channel orchestration. We include concrete data models, API contracts, architecture diagrams, C# implementation code, database schemas, caching strategies, and cost models — everything you need to demonstrate senior-plus depth in a system design interview or production planning session.

2. Functional and Non-Functional Requirements

Functional Requirements

Before designing any system, we must enumerate the capabilities that a Klaviyo-like platform must support. These requirements drive every downstream architectural decision, from data model design to infrastructure sizing.

Core Marketing Capabilities

• Shopify Data Sync: Real-time synchronization of customers, products, orders, and inventory from Shopify stores via webhooks and the Admin REST/GraphQL API. The system must handle stores with millions of products and hundreds of thousands of customers without data loss or significant lag.
• Event Tracking: Ingest behavioral events including product views, add-to-cart, checkout started, order placed, order fulfilled, items returned, email opened, email clicked, SMS received, and coupon used. Events must be attributed to specific customers with accurate timestamps.
• Email Campaigns: Create, schedule, A/B test, and send marketing email campaigns with dynamic content blocks, product recommendation feeds, coupon code generation, and full HTML template support with drag-and-drop editing capabilities.
• SMS Campaigns: Send transactional and marketing SMS messages with opt-in/opt-out management, two-way messaging support, and carrier-level compliance with TCPA regulations across multiple countries.
• Flows (Automation): Build multi-step automation workflows triggered by customer behavior or time delays. Flow steps can include conditional splits, time waits, email sends, SMS sends, webhooks, and profile property updates.
• Segmentation: Define dynamic customer segments based on behavioral criteria (purchased in last 30 days, viewed product X but did not purchase, lifetime value above $500), demographic data, and predictive scores. Segments update in near real-time as new events arrive.
• Revenue Attribution: Track which emails, SMS messages, and flows directly generated revenue. Support last-touch attribution, multi-touch attribution, and UTM-based tracking with configurable attribution windows.
• Product Recommendations: Generate personalized product recommendations using collaborative filtering and content-based algorithms, surfaced in emails and on-site widgets.
• Predictive Analytics: Compute predictive metrics including customer lifetime value (CLV), churn risk score, predicted next order date, average order value prediction, and likelihood to purchase in the next 30 days.
• Consent and Compliance: Manage opt-in/opt-out preferences across channels, enforce CAN-SPAM, GDPR, and TCPA regulations, and maintain auditable consent records for every customer interaction.

Non-Functional Requirements

Out of Scope for This Design

• On-site popup and form builder (would require a JavaScript SDK design)
• Creative asset management and image hosting
• Billing and subscription management for Klaviyo merchants
• Native integrations beyond Shopify such as BigCommerce and Magento

3. Capacity Estimation and Back-of-Envelope Math

Capacity estimation grounds our design in reality. Let us work through the numbers for a platform serving 100,000 merchant stores with a combined customer base of approximately 5 billion unique customer profiles (accounting for overlap across stores, though each store sees its own slice).

Event Ingestion Volume

Consider a mid-size Shopify merchant with 500,000 monthly visitors. A conservative event-per-visitor ratio of 8 events per session (page views, product views, add-to-cart, checkout events) yields roughly 4 million events per month per store. Across 100,000 stores, this is approximately 400 billion events per month, or about 155 million events per second at peak (assuming 20% of traffic occurs during a 4-hour peak window).

In practice, not all stores are mid-size. The distribution is long-tailed — a small percentage of stores generate the majority of events. We must design for a peak ingestion rate of 1 million events per second with a sustained rate of approximately 200,000 events per second.

Storage Estimation

If each event averages 500 bytes (compressed), daily event storage is approximately 2.5 terabytes. Over a year, this accumulates to roughly 900 terabytes of raw event data. With indexing, customer profile data, and product catalogs, total storage requirements approach 1.5 petabytes annually. This demands a horizontally scalable event store with efficient compression and tiered storage — hot data in SSD-backed stores, warm data in object storage, and cold data in archival tiers.

Throughput Calculation

Email sending throughput is the most capacity-constrained subsystem. Sending 2 billion emails per day requires an aggregate throughput of approximately 23,000 emails per second sustained. Given that email delivery has natural burstiness (campaign launches), we must provision for 100,000 emails per second peak throughput. This requires maintaining sending infrastructure across multiple email service providers with IP warm-up pools, reputation management, and bounce/complaint handling.

4. Core Data Model

The data model is the backbone of the entire system. Every subsystem — from event ingestion to segmentation to campaign reporting — operates on these core entities. The design must balance normalization for data integrity with denormalization for query performance at scale.

Entity Relationship Overview

Core Tables

Shops Table

Each merchant store is a top-level tenant. The shop entity stores Shopify credentials, plan information, regional settings, and sending configuration. Multi-tenant isolation begins at this level — all data queries are scoped to a shop identifier.

SQL
CREATE TABLE shops (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    shopify_domain  VARCHAR(255) NOT NULL UNIQUE,
    shopify_access_token VARCHAR(255) NOT NULL,
    shop_name       VARCHAR(255) NOT NULL,
    plan_tier       VARCHAR(50) NOT NULL DEFAULT 'free',
    timezone        VARCHAR(50) NOT NULL DEFAULT 'UTC',
    currency        VARCHAR(3) NOT NULL DEFAULT 'USD',
    email_from      VARCHAR(255),
    sms_sender_id   VARCHAR(50),
    settings        JSONB NOT NULL DEFAULT '{}',
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Customers Table

Customer profiles are the most queried entity in the system. Each customer is scoped to a shop and enriched with computed properties (total lifetime value, last event timestamp, segment membership) that are updated asynchronously as new events flow in.

SQL
CREATE TABLE customers (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    shop_id         UUID NOT NULL REFERENCES shops(id),
    shopify_customer_id BIGINT,
    email           VARCHAR(255),
    phone           VARCHAR(50),
    first_name      VARCHAR(100),
    last_name       VARCHAR(100),
    tags            TEXT[] DEFAULT '{}',
    properties      JSONB NOT NULL DEFAULT '{}',
    email_opt_in    BOOLEAN DEFAULT FALSE,
    sms_opt_in      BOOLEAN DEFAULT FALSE,
    total_revenue   NUMERIC(12,2) DEFAULT 0,
    order_count     INTEGER DEFAULT 0,
    last_order_at   TIMESTAMPTZ,
    last_event_at   TIMESTAMPTZ,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE(shop_id, shopify_customer_id)
);

Events Table

The events table is the highest-volume table in the system. It stores every behavioral event — from page views to purchases — with full attribute payloads. This table is partitioned by time for efficient writes and range queries.

SQL
CREATE TABLE events (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    shop_id         UUID NOT NULL,
    customer_id     UUID,
    event_type      VARCHAR(100) NOT NULL,
    event_value     NUMERIC(12,2),
    properties      JSONB NOT NULL DEFAULT '{}',
    source          VARCHAR(50) NOT NULL,
    timestamp       TIMESTAMPTZ NOT NULL,
    ingested_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
) PARTITION BY RANGE (timestamp);

Flows and Flow States

SQL
CREATE TABLE flows (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    shop_id         UUID NOT NULL REFERENCES shops(id),
    name            VARCHAR(255) NOT NULL,
    status          VARCHAR(20) NOT NULL DEFAULT 'draft',
    trigger_type    VARCHAR(50) NOT NULL,
    trigger_config  JSONB NOT NULL DEFAULT '{}',
    flow_definition JSONB NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE flow_states (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    flow_id         UUID NOT NULL REFERENCES flows(id),
    customer_id     UUID NOT NULL REFERENCES customers(id),
    current_step    INTEGER NOT NULL DEFAULT 0,
    status          VARCHAR(20) NOT NULL DEFAULT 'active',
    entered_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    next_action_at  TIMESTAMPTZ,
    context         JSONB NOT NULL DEFAULT '{}',
    completed_at    TIMESTAMPTZ
);

5. API Design

The API surface of a Klaviyo-like platform serves three distinct consumers: the Shopify integration layer (writing events and reading product catalogs), the merchant dashboard (reading analytics, managing campaigns), and the public API (third-party integrations). All APIs follow RESTful conventions with consistent error handling, rate limiting, and pagination.

Core API Endpoints

Track Event Request and Response

JSON
// POST /api/v1/events/track
{
  "shop_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "customer_email": "jane@example.com",
  "event_type": "AddedToCart",
  "timestamp": "2026-07-01T10:30:00Z",
  "properties": {
    "product_id": "gid://shopify/Product/123456",
    "product_title": "Premium Wireless Headphones",
    "product_price": 149.99,
    "product_image_url": "https://cdn.example.com/product/123456.jpg",
    "quantity": 1,
    "cart_total": 149.99,
    "currency": "USD"
  }
}

// Response 201 Created
{
  "event_id": "evt_9f8e7d6c5b4a3210",
  "status": "accepted",
  "processed_at": "2026-07-01T10:30:00.142Z"
}

Create Flow Request and Response

JSON
// POST /api/v1/flows
{
  "name": "Abandoned Cart Recovery",
  "trigger_type": "event",
  "trigger_config": {
    "event_type": "AddedToCart",
    "filters": {
      "customer.properties.email_opt_in": true
    }
  },
  "flow_definition": {
    "steps": [
      { "type": "wait", "duration": 1, "unit": "hours" },
      { "type": "condition", "rule": "customer.has_purchased_since(event.timestamp)" },
      { "type": "email", "template_id": "tpl_abandoned_cart_1",
        "subject": "You left something behind!", "channel": "email" },
      { "type": "wait", "duration": 24, "unit": "hours" },
      { "type": "email", "template_id": "tpl_abandoned_cart_2",
        "subject": "Your cart is waiting - 10% off inside", "channel": "email" }
    ]
  }
}

Pagination and Filtering

All list endpoints support cursor-based pagination for efficient traversal of large result sets. Filtering uses a consistent syntax across all resources with support for operators like equals, contains, greater_than, less_than, and in for array fields. Sort parameters allow ordering by any indexed field with consistent null handling.

HTTP
GET /api/v1/customers?filter[total_revenue][greater_than]=100
    &filter[email_opt_in][equals]=true
    &sort=-last_order_at
    &page[cursor]=eyJpZCI6MTAwfQ==
    &page[size]=50

6. High-Level Architecture

The architecture of a Klaviyo-like platform is a textbook example of an event-driven system with multiple specialized subsystems communicating through message queues and shared data stores. The design prioritizes eventual consistency for non-critical paths (analytics dashboards) while ensuring strong consistency for critical paths (email sending, compliance).

System Architecture Diagram

Subsystem Responsibilities

7. Shopify Integration Deep Dive

The Shopify integration is the most critical external dependency in the system. Klaviyo's competitive advantage begins here — no other platform syncs Shopify data as comprehensively or as quickly. Building a robust Shopify integration requires understanding the Shopify Admin API, webhook system, and the nuances of OAuth token management for thousands of merchant stores.

Webhook Architecture

Shopify supports webhooks for over 60 resource types. For a marketing platform, the essential webhooks include orders/create, orders/updated, orders/paid, orders/fulfilled, orders/cancelled, customers/create, customers/updated, products/create, products/updated, products/delete, inventory_levels/update, and checkouts/create. Each webhook delivers a JSON payload signed with HMAC-SHA256 for authenticity verification.

Shopify API Rate Limits

Shopify imposes a leaky bucket rate limit of 40 requests per second per store for the Admin API. For stores with large product catalogs (100,000+ SKUs), this means a full catalog sync can take over 40 minutes. The system must implement intelligent sync strategies: prioritize high-sellers, use GraphQL bulk operations for product data, and maintain a persistent cursor for incremental syncs rather than full re-indexes.

OAuth Token Lifecycle

Every connected Shopify store generates an access token that must be securely stored and refreshed. Shopify access tokens do not expire, but they can be revoked by the merchant. The system must monitor token validity by attempting periodic API calls and automatically flagging disconnected stores for re-authorization. Tokens are stored encrypted at rest with shop-specific encryption keys in a dedicated secrets vault, never in the primary application database.

Real-Time Product Catalog Sync

The product catalog drives email content (dynamic product blocks), recommendations, and segmentation (products viewed, inventory-based triggers). The sync service maintains an eventually consistent mirror of each store's product catalog in the local database. Products are synced in near real-time via webhooks and periodically reconciled using bulk GraphQL queries to catch any missed updates. Product images are proxied through a CDN layer for consistent delivery and to avoid hotlinking issues.

8. Event Ingestion Pipeline

The event ingestion pipeline is the front door for all customer behavioral data. It must handle three distinct event sources — Shopify webhooks (orders, customers), the JavaScript tracking SDK (page views, product views), and the public API (custom events) — and funnel them into a unified, deduplicated event stream with sub-second latency.

Ingestion Architecture

Event Schema

Every event flowing through the pipeline conforms to a standardized envelope that includes traceability metadata, making it possible to debug event processing issues end-to-end. The event envelope is serialized using Apache Avro for compact binary representation and schema evolution support.

C#
public class EventEnvelope
{
    public string EventId { get; set; }
    public string ShopId { get; set; }
    public string EventType { get; set; }
    public string? CustomerId { get; set; }
    public string? CustomerEmail { get; set; }
    public string? CustomerPhone { get; set; }
    public DateTime Timestamp { get; set; }
    public DateTime IngestedAt { get; set; }
    public string Source { get; set; }
    public Dictionary<string, object> Properties { get; set; }
    public EventMetadata Metadata { get; set; }
}

public class EventMetadata
{
    public string? IpAddress { get; set; }
    public string? UserAgent { get; set; }
    public string? ReferrerUrl { get; set; }
    public string? UtmSource { get; set; }
    public string? UtmMedium { get; set; }
    public string? UtmCampaign { get; set; }
    public string? ShopDomain { get; set; }
}

Deduplication Strategy

At the scale of billions of events per day, duplicate events are inevitable — Shopify retries webhooks, the JS SDK may fire duplicate pageview events on slow connections, and network partitions can cause exactly-once delivery to be impractical. The deduplication layer uses a sliding window Bloom filter backed by a Redis set for exact-match deduplication within a 5-minute window. The EventId (UUID v4) serves as the deduplication key. False positives from the Bloom filter are handled gracefully — duplicates are logged but silently dropped.

Event Enrichment

Raw events arriving from external sources often lack internal identifiers. The enrichment step resolves customer identifiers (email or phone to internal customer_id), attaches session metadata (UTM parameters, device info), and normalizes event types to a consistent vocabulary. This resolution happens asynchronously using a customer identity resolution service backed by Redis for fast lookups and PostgreSQL for persistent storage.

9. Customer Data Platform

The Customer Data Platform is the heart of the system — it maintains a unified profile for every customer across every touchpoint. When a merchant views a customer profile in the dashboard, the CDP must present a complete picture: demographic information, order history, browsing behavior, email engagement, SMS interactions, segment membership, and predictive scores — all assembled from data scattered across multiple storage systems.

Unified Profile Architecture

Computed Customer Properties

Raw events are transformed into computed properties that power segmentation and personalization. These properties are updated incrementally as new events arrive, not recomputed from scratch on every query. This incremental computation model is critical for performance at scale.

Event Timeline

The event timeline is a chronologically ordered view of every customer interaction, stored as a materialized list within the customer profile. For active customers (events in the last 90 days), the timeline is kept in Redis for sub-millisecond access. For inactive customers, the timeline is reconstructed on-demand from the events table. The timeline powers the customer profile view in the merchant dashboard and provides context for flow decision-making.

Identity Resolution

Customers interact with stores across multiple devices and may provide different email addresses or phone numbers over time. The identity resolution service merges profiles that share common identifiers (email, phone, Shopify customer ID) into a single canonical profile. This merge is irreversible and must handle edge cases like shared email addresses (households) and business vs. personal email addresses. The resolution algorithm uses a weighted graph approach where shared identifiers create edges between profiles, and connected components are merged into single canonical profiles.

10. Behavioral Segmentation Engine

Segmentation is where data becomes actionable. A merchant defines a segment like "Customers who spent over $500 in the last 90 days, viewed the Summer Collection in the last 7 days, and have not placed an order in the last 14 days." The segmentation engine must evaluate this complex rule set against potentially millions of customers and return accurate membership within 30 seconds.

Segment Evaluation Architecture

Segment Rule Types

Incremental Segment Updates

Re-evaluating all segments from scratch on every event is computationally prohibitive. Instead, the system uses an incremental evaluation strategy. When a new event arrives for a customer, the system checks only the segments that could be affected by that event type. For example, a ViewedProduct event might affect "Product Category Browsers" segments but not "High Value Customers" segments. A segment type index maps event types to affected segment definitions, enabling targeted re-evaluation.

C#
public class IncrementalSegmentEvaluator
{
    private readonly Dictionary<string, List<SegmentDefinition>> _eventTypeToSegments;
    private readonly ISegmentQueryExecutor _executor;
    private readonly ISegmentCache _cache;

    public async Task EvaluateOnEvent(string shopId, string eventType, string customerId)
    {
        if (!_eventTypeToSegments.TryGetValue(eventType, out var affectedSegments))
            return;

        foreach (var segment in affectedSegments.Where(s => s.ShopId == shopId))
        {
            bool isMember = await _executor.EvaluateSegmentForCustomer(
                segment.Id, customerId);
            bool wasMember = await _cache.IsMemberAsync(segment.Id, customerId);

            if (isMember && !wasMember)
            {
                await _cache.AddMemberAsync(segment.Id, customerId);
                await NotifySegmentEntry(segment, customerId);
            }
            else if (!isMember && wasMember)
            {
                await _cache.RemoveMemberAsync(segment.Id, customerId);
                await NotifySegmentExit(segment, customerId);
            }
        }
    }
}

Segment Sizing and Preview

Merchants need to see segment sizes before launching campaigns. The segment sizing API returns an approximate count using PostgreSQL statistics-based count estimation for large segments, and exact count for segments under 100,000 members. A background job maintains approximate segment sizes using HyperLogLog counters in Redis, refreshed every 5 minutes for segments with more than 10,000 members.

11. Flows and Automation

Flows are the revenue engine of any e-commerce marketing platform. An abandoned cart flow that recovers just 5% of abandoned carts can generate millions of dollars in incremental revenue for a mid-size merchant. The flow system must be reliable, idempotent, and capable of handling millions of concurrent flow states across thousands of merchants.

Flow Execution Architecture

Flow State Machine

Each customer's journey through a flow is tracked as a state machine. The state includes the current step index, status (active, paused, completed, exited), timing context (when the customer entered, when they reach the next step), and a context object carrying flow-specific variables (e.g., the cart contents that triggered an abandoned cart flow).

Common Flow Patterns

Abandoned Cart Flow

The most common and highest-ROI flow. Triggered when a customer adds items to cart but does not complete checkout within a configurable time window (typically 1 hour). The flow sends a sequence of 2-4 emails with escalating urgency and potential incentives. Key implementation detail: the flow must check at each email step whether the customer has completed a purchase since entering the flow, and exit them if so.

Browse Abandonment Flow

Triggered when a customer views 3+ products in a category without adding anything to cart. Requires higher volume event tracking (product views) and more nuanced personalization (referencing specific products viewed). This flow operates at much higher volume than abandoned cart but with lower conversion rates, making deliverability management critical.

Post-Purchase Flow

Triggered on order confirmation. Sends order confirmation, shipping updates, delivery follow-up, and review request emails. This flow is primarily transactional but includes cross-sell product recommendations in the post-delivery steps. Timing depends on fulfillment status, requiring integration with order fulfillment webhooks.

Winback Flow

Triggered when a customer's last purchase date exceeds a threshold (e.g., 90 days for a store with 60-day average purchase frequency). This flow uses the most aggressive personalization, including predictive churn scores and dynamic discount offers based on the customer's historical purchase patterns.

12. Email Campaign Builder

The email campaign builder is the merchant-facing interface for creating one-time email broadcasts — product launches, sales announcements, newsletters, and seasonal promotions. Unlike flows (which are event-triggered), campaigns are merchant-initiated and sent to a target segment or list at a scheduled time.

Email Rendering Pipeline

Merchant-created email templates contain dynamic content blocks — product recommendation feeds, personalized customer properties, conditional content sections, and coupon codes. The rendering pipeline transforms these templates into final HTML by resolving dynamic variables against customer profiles, fetching product data from the catalog service, and generating unique coupon codes per recipient.

Dynamic Product Feeds

Product recommendation blocks in email templates are resolved at send time, not at template creation time. This ensures recommendations reflect the latest catalog data and the most recent browsing behavior for each recipient. The product feed resolver queries the recommendation service for each customer, fetches current product details (price, availability, image URL) from the catalog cache, and injects the resolved HTML into the template.

A/B Testing Framework

Campaigns support A/B testing across subject lines, send times, content variants, and sender names. The A/B test configuration specifies the variants, the test population size (percentage of recipients used for testing), and the winning metric (open rate, click rate, or revenue per recipient). The system sends the test variants to the test population, waits for a configurable measurement window (typically 2-4 hours), determines the winner based on the specified metric, and automatically sends the winning variant to the remaining recipients.

Email Deliverability Management

Sending millions of emails per day requires careful deliverability management. The system maintains dedicated IP pools per sending domain, warms up new IPs gradually, monitors bounce and complaint rates per IP and domain, automatically pauses sending on IPs that exceed complaint thresholds, and rotates between healthy IPs during campaign sends. Bounce classification (hard bounce vs. soft bounce) determines whether an email address is immediately suppressed or retried with exponential backoff.

13. SMS Marketing

SMS marketing delivers the highest engagement rates of any channel — open rates above 95% and click rates above 20% are common. However, SMS is also the most regulated channel, with strict opt-in requirements, character limits, and carrier-level filtering that can block messages deemed promotional. Building a compliant, effective SMS system requires deep integration with SMS gateways and a robust consent management framework.

SMS Architecture

Opt-In and Compliance

SMS compliance is governed by TCPA in the United States, PECR in the UK, CASL in Canada, and equivalent regulations in other markets. The core requirements are: explicit consent must be collected before sending any marketing SMS, a clear opt-out mechanism must be included in every message, consent records must be retained with timestamps and the specific language the customer agreed to, and transactional messages (order confirmations, shipping updates) do not require prior marketing consent but must include opt-out instructions.

The consent management system maintains a per-customer, per-channel consent record with the following fields: consent status (opted_in, opted_out, never_opted), consent timestamp, consent method (keyword, web form, checkout checkbox), consent language (the exact text shown to the customer), and opt-out timestamp (if applicable). Every SMS send checks this record before proceeding, and any mismatch between consent status and send attempt triggers an immediate alert.

Message Splitting and Character Limits

Standard SMS messages are limited to 160 characters (GSM-7 encoding) or 70 characters (UCS-2 encoding for non-Latin characters). Messages exceeding these limits are split into multiple segments, and carriers charge per segment. The message splitter service detects character encoding, splits at word boundaries to avoid breaking URLs or emojis, and prepends segment indicators (1/3, 2/3, 3/3) when multi-segment messages are detected. Shortened URLs (via a branded short domain) reduce segment count and cost.

Two-Way Messaging

Two-way SMS enables customers to reply to messages with keywords (STOP to opt out, HELP for support, YES to confirm an order). The inbound message handler parses keyword responses, updates consent records for opt-out keywords, routes support keywords to the merchant's support system, and logs all inbound messages to the customer event timeline for context in future marketing decisions.

14. Predictive Analytics

Predictive analytics transforms historical customer data into forward-looking insights that drive smarter marketing decisions. Rather than reacting to what customers have done, predictive models forecast what they are likely to do next, enabling proactive marketing that intercepts customers at the optimal moment in their lifecycle.

Predictive Models

Prediction Pipeline

Predictive models run as batch jobs on a scheduled basis. The pipeline extracts feature vectors from customer profiles and historical events, feeds them through trained ML models (gradient-boosted trees for churn and CLV, time-series forecasting for next purchase date), and writes prediction scores back to the predictive_scores table. Scores are cached in Redis for real-time segment evaluation and dashboard display.

C#
public class ChurnPredictionService
{
    private readonly IFeatureStore _featureStore;
    private readonly IChurnModel _model;
    private readonly IPredictionStore _predictionStore;

    public async Task RunBatchPrediction(string shopId)
    {
        var customers = await _featureStore.GetActiveCustomers(shopId);

        var features = customers.Select(c => new ChurnFeatures
        {
            CustomerId = c.Id,
            DaysSinceLastOrder = c.DaysSinceLastOrder,
            DaysSinceLastEvent = c.DaysSinceLastEvent,
            TotalOrders = c.OrderCount,
            TotalRevenue = c.TotalRevenue,
            AverageOrderValue = c.AverageOrderValue,
            EmailEngagementScore = c.EmailEngagementScore,
            ProductViewCount30d = c.ProductViewsLast30Days,
            AverageDaysBetweenOrders = c.AverageInterPurchaseDays,
            HasPurchaseInLast90Days = c.LastOrderAt > DateTime.UtcNow.AddDays(-90)
        }).ToList();

        var predictions = _model.PredictBatch(features);

        foreach (var prediction in predictions)
        {
            await _predictionStore.UpsertAsync(new PredictionScore
            {
                ShopId = shopId,
                CustomerId = prediction.CustomerId,
                ScoreType = "churn_risk",
                Score = prediction.Probability,
                ComputedAt = DateTime.UtcNow,
                ExpiresAt = DateTime.UtcNow.AddDays(7)
            });
        }
    }
}

Predictive Segments

Predictive scores enable powerful segment definitions that are impossible with behavioral data alone. Merchants can create segments like "High-value customers at risk of churning" (CLV prediction above $500 AND churn risk above 0.7) or "Customers likely to purchase soon who haven't been emailed" (purchase likelihood above 0.6 AND last email sent more than 7 days ago). These predictive segments often deliver 3-5x higher ROI than purely behavioral segments because they target customers at the inflection points in their lifecycle.

15. Revenue Attribution

Revenue attribution answers the fundamental question every merchant asks: "How much money did my marketing generate?" Without accurate attribution, merchants cannot optimize their marketing spend, compare channel effectiveness, or justify ROI to stakeholders. The attribution system must track the complete customer journey from first touch to final purchase and attribute revenue fairly across all contributing marketing touchpoints.

Attribution Models

Attribution Tracking

Every email and SMS message sent through the platform includes tracking pixels and UTM-tagged links that attribute subsequent website activity back to the specific message. The tracking infrastructure includes: an email pixel (1x1 transparent image) hosted on a dedicated tracking domain that fires on email open, click tracking that redirects through a tracking service before forwarding to the final destination, and UTM parameter injection that appends utm_source=klaviyo, utm_medium=email, and utm_campaign={campaign_id} to all links.

Attribution Window

The attribution window defines how long after an email/SMS interaction a subsequent purchase can be attributed to that message. The default window is 5 days for email clicks and 1 day for SMS clicks, but merchants can configure custom windows. The attribution service maintains a lookup table mapping (customer_id, channel, campaign_id, interaction_timestamp) to enable efficient window-based queries when processing orders.

C#
public class RevenueAttributor
{
    private readonly IAttributionStore _store;
    private readonly IOrderEventSource _orders;

    public async Task AttributeOrder(OrderEvent order)
    {
        var window = TimeSpan.FromDays(5);
        var cutoff = order.Timestamp.Subtract(window);

        var interactions = await _store.GetInteractions(
            order.ShopId, order.CustomerId, cutoff, order.Timestamp);

        if (!interactions.Any())
            return;

        var attribution = CalculateAttribution(order, interactions);

        foreach (var attr in attribution)
        {
            await _store.SaveAttribution(new RevenueAttribution
            {
                OrderId = order.Id,
                CampaignId = attr.CampaignId,
                FlowId = attr.FlowId,
                Channel = attr.Channel,
                Revenue = order.TotalRevenue * attr.Weight,
                AttributionModel = "last_touch_email",
                InteractedAt = attr.InteractionTimestamp,
                OrderedAt = order.Timestamp
            });
        }
    }
}

Revenue Dashboard Metrics

The attribution data powers a merchant dashboard showing: total attributed revenue per time period, revenue per email/SMS sent, revenue per flow and campaign, channel comparison (email vs. SMS ROI), and cohort analysis (revenue by customer acquisition month). All dashboard metrics are pre-computed by a background aggregation job and stored in materialized views for sub-second dashboard load times.

16. Product Recommendations

Product recommendations bridge the gap between what a customer has shown interest in and what they are most likely to purchase next. In email marketing, personalized product recommendation blocks consistently outperform generic content by 2-3x in click-through rate. The recommendation engine must operate in real-time (resolving recommendations at email send time) and handle catalogs with millions of products.

Recommendation Strategies

Collaborative Filtering Implementation

Collaborative filtering for email recommendations uses a simplified matrix factorization approach. The system builds a customer-product interaction matrix where rows represent customers, columns represent products, and cell values represent interaction strength (view = 1, add-to-cart = 3, purchase = 5). Singular Value Decomposition (SVD) factorizes this matrix into lower-dimensional representations, and the dot product of a customer's latent vector with each product's latent vector produces a relevance score. The top-N products by score, excluding already-purchased items, become the recommendation set.

Given the scale of millions of customers and products, the full matrix factorization is infeasible to run in real-time. Instead, the system uses a two-phase approach: an offline batch job (run daily) computes the latent factor matrices and stores them, and an online lookup service uses the pre-computed factors to generate personalized recommendations per customer in under 50 milliseconds.

Recommendation Caching

Recommendations change infrequently for most customers (once per day is sufficient), so they are cached aggressively. The cache key is a composite of (customer_id, strategy, catalog_version). When the product catalog changes (new products, price updates, out-of-stock removals), affected catalog versions are detected and the recommendation cache for impacted customers is invalidated. This balance of staleness and freshness ensures recommendations are relevant without excessive recomputation.

17. Deliverability and Compliance

Deliverability is the invisible differentiator between a marketing platform that generates revenue and one that lands in spam folders. A beautifully designed email is worthless if it never reaches the inbox. The deliverability subsystem manages sender reputation, monitors inbox placement rates, enforces compliance regulations, and maintains list hygiene — all operating continuously in the background of every campaign send.

Deliverability Infrastructure

Sender Reputation Management

Each sending domain and IP pool has a reputation score computed from bounce rates, complaint rates, inbox placement test results, and blocklist membership. The reputation manager monitors these signals in real-time and takes automated actions: if bounce rate on an IP exceeds 3%, that IP is temporarily removed from the send pool; if complaint rate exceeds 0.1%, sending on that domain is paused and the merchant is notified; if an IP appears on a major blocklist (Spamhaus, Barracuda), it is immediately quarantined pending investigation.

Compliance Framework

List Hygiene Pipeline

The list hygiene pipeline runs continuously, processing every email address in the system through a multi-stage verification process. Stage one catches syntax errors and disposable email domains. Stage two performs MX record verification to confirm the domain can receive mail. Stage three uses a combination of SMTP probe verification and third-party verification services (ZeroBounce, NeverBounce) to identify risky addresses. Stage four scores each address based on engagement history — addresses with no opens in 6 months are flagged as disengaged and removed from marketing sends to protect domain reputation.

18. Multi-Channel Orchestration

Multi-channel orchestration coordinates messages across email, SMS, and push notifications to deliver cohesive customer experiences without message fatigue. The orchestration engine ensures a customer never receives an email and an SMS about the same promotion on the same day, that high-priority messages (transactional) take precedence over marketing messages, and that channel preferences are respected at every step.

Orchestration Rules

The orchestration layer applies a set of rules before any message is dispatched. These rules operate on a per-customer basis and consider the customer's full message history across all channels within a configurable lookback window (typically 24-48 hours).

• Frequency Capping: Maximum messages per channel per time period (e.g., 2 emails per day, 1 SMS per 3 days)
• Channel Priority: If both email and SMS are queued for the same customer, the higher-priority channel sends first and the other is suppressed or delayed
• Quiet Hours: No messages between 9 PM and 8 AM in the customer's timezone (configurable per channel and regulation)
• Suppression Rules: Recent purchasers (last 24 hours) are suppressed from marketing emails to avoid tone-deaf messaging
• Channel Preference: If a customer has opted out of SMS but is subscribed to email, SMS steps in flows are automatically skipped
• Message Deduplication: If the same content is queued across channels, only one is sent based on channel preference scoring

Orchestration Decision Engine

C#
public class OrchestrationDecider
{
    private readonly ICustomerProfileService _profiles;
    private readonly IMessageHistoryService _history;
    private readonly IComplianceService _compliance;

    public async Task<OrchestrationDecision> ShouldSend(
        string customerId, string channel, string messageId)
    {
        var profile = await _profiles.GetProfile(customerId);
        var recentMessages = await _history.GetRecent(customerId, TimeSpan.FromHours(48));

        if (channel == "sms" && !profile.SmsOptIn)
            return Suppressed("Customer not opted into SMS");

        if (channel == "email" && !profile.EmailOptIn)
            return Suppressed("Customer not opted into email");

        var customerTime = TimeZoneInfo.FindSystemTimeZoneById(profile.Timezone);
        var localTime = TimeZoneInfo.ConvertTimeFromUtc(DateTime.UtcNow, customerTime);
        if (localTime.Hour < 8 || localTime.Hour > 21)
            return Delayed("Quiet hours", GetNextAllowedTime(localTime));

        var channelMessages = recentMessages.Where(m => m.Channel == channel);
        if (channelMessages.Count() >= GetFrequencyCap(channel))
            return Suppressed("Frequency cap exceeded");

        if (recentMessages.Any(m => m.Type == "purchase" &&
            m.Timestamp > DateTime.UtcNow.AddHours(-24)))
            return Suppressed("Recent purchaser suppression");

        return Approved();
    }
}

19. Database Design

The database layer is the most critical infrastructure component — it must handle extreme write volumes from event ingestion, support complex analytical queries for segmentation, and provide fast reads for the merchant dashboard. The design uses a polyglot persistence approach: PostgreSQL for transactional data and complex queries, Redis for caching and real-time state, Elasticsearch for log search and analytics, and S3 for cold storage.

PostgreSQL Schema Strategy

The events table is the largest table in the system, growing by billions of rows per day. To manage this scale, the events table uses PostgreSQL native partitioning by month. Each monthly partition is an independent table with its own indexes, enabling efficient range queries on specific time periods and easy data lifecycle management (dropping old partitions by detaching them).

SQL
CREATE TABLE events (
    id              UUID NOT NULL DEFAULT gen_random_uuid(),
    shop_id         UUID NOT NULL,
    customer_id     UUID,
    event_type      VARCHAR(100) NOT NULL,
    event_value     NUMERIC(12,2),
    properties      JSONB NOT NULL DEFAULT '{}',
    source          VARCHAR(50) NOT NULL,
    timestamp       TIMESTAMPTZ NOT NULL,
    ingested_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
) PARTITION BY RANGE (timestamp);

CREATE TABLE events_2026_07 PARTITION OF events
    FOR VALUES FROM ('2026-07-01') TO ('2026-08-01');
CREATE TABLE events_2026_08 PARTITION OF events
    FOR VALUES FROM ('2026-08-01') TO ('2026-09-01');

CREATE INDEX idx_events_shop_customer_type
    ON events(shop_id, customer_id, event_type, timestamp);
CREATE INDEX idx_events_shop_type_timestamp
    ON events(shop_id, event_type, timestamp);
CREATE INDEX idx_events_customer_timestamp
    ON events(customer_id, timestamp);
CREATE INDEX idx_events_properties ON events USING gin(properties);

Redis Data Structures

Read Replicas and Query Routing

All write operations target the PostgreSQL primary instance. Read operations are routed to read replicas using a connection pool with automatic failover. The routing layer uses query analysis to direct analytical queries (segment evaluation, dashboard aggregations) to read replicas while routing transactional queries (profile updates, event writes) to the primary. This separation prevents analytical workloads from impacting write throughput during peak campaign periods.

20. Caching Strategy

Caching is not optional at this scale — it is a fundamental architectural requirement. Without aggressive caching, every segment evaluation would require full table scans on the events table, every email render would require a database lookup per recipient, and every dashboard page load would trigger dozens of analytical queries. The caching strategy uses a multi-layer approach with different TTLs based on data freshness requirements.

Cache Hierarchy

Cache Invalidation Strategy

Cache invalidation is one of the hardest problems in distributed systems, and it is especially critical for a marketing platform where stale data can mean sending the wrong message to the wrong customer. The system uses event-driven invalidation: when a customer event is processed, a corresponding cache invalidation message is published to a Kafka topic. Cache invalidation consumers update the Redis cache and publish invalidation signals to connected application instances to evict their in-process L1 caches.

Cache Stampede Prevention

When a high-profile customer's cache expires (e.g., a VIP customer during a flash sale), hundreds of concurrent requests may try to rebuild the cache simultaneously. The system uses a distributed lock pattern (Redis SETNX) combined with request coalescing: the first request to detect a cache miss acquires the lock, rebuilds the cache, and writes it. Subsequent requests within the lock window wait on a semaphore and read the newly populated cache when the lock holder completes. This ensures only one database query rebuilds the cache regardless of concurrent request volume.

C#
public class DistributedCacheService<T> where T : class
{
    private readonly IDatabase _redis;
    private readonly ConcurrentDictionary<string, SemaphoreSlim> _locks = new();

    public async Task<T?> GetOrRebuildAsync(
        string key, Func<Task<T?>> rebuildFunc, TimeSpan ttl)
    {
        if (_localCache.TryGetValue(key, out var cached) &&
            cached.ExpiresAt > DateTime.UtcNow)
            return cached.Value;

        var redisValue = await _redis.StringGetAsync(key);
        if (redisValue.HasValue)
            return JsonSerializer.Deserialize<T>(redisValue!);

        var semaphore = _locks.GetOrAdd(key, _ => new SemaphoreSlim(1, 1));
        await semaphore.WaitAsync();
        try
        {
            redisValue = await _redis.StringGetAsync(key);
            if (redisValue.HasValue)
                return JsonSerializer.Deserialize<T>(redisValue!);

            var value = await rebuildFunc();
            if (value != null)
            {
                var serialized = JsonSerializer.Serialize(value);
                await _redis.StringSetAsync(key, serialized, ttl);
                _localCache[key] = new CacheEntry(
                    value, DateTime.UtcNow.AddSeconds(30));
            }
            return value;
        }
        finally
        {
            semaphore.Release();
        }
    }
}

21. Multi-Region Design

A global marketing platform serving merchants in 190 countries must operate across multiple regions to meet data residency requirements (GDPR in Europe, LGPD in Brazil, PIPL in China) and to minimize latency for merchants and their customers. The multi-region architecture must handle cross-region event routing, region-local customer data storage, and global aggregate analytics.

Regional Architecture

Data Residency Strategy

Customer data is stored in the region closest to the merchant's store domain. US merchants have data in US-East, EU merchants in EU-West, and so on. Event ingestion routes events to the correct region based on the merchant's region assignment. Cross-region reads are avoided for customer data — if a merchant queries a customer profile, the request is routed to the region where that customer's data resides. Global analytics (aggregate dashboards, predictive model training) operate on anonymized, aggregated data replicated from all regions to the primary region.

Region Failover

If a region becomes unavailable, traffic is automatically routed to the nearest healthy region with a degraded mode of operation. Customer profile data may be stale (replication lag), but event ingestion continues to the healthy region's Kafka cluster (with events queued for processing when the primary region recovers). This eventually consistent everywhere approach prioritizes data collection over data accuracy during regional outages, ensuring no customer events are lost.

22. Cost Estimation

Understanding the infrastructure cost of a Klaviyo-like platform is essential for system design interviews and production planning. The cost model is driven primarily by event volume, email sending volume, and storage requirements.

Infrastructure Cost Breakdown

Cost Optimization Levers

• Event Sampling: For high-volume stores, sample page view events (1 in 10) while keeping purchase events at 100%. This reduces event volume by 40-60% with minimal impact on segmentation accuracy.
• Tiered Storage: Move events older than 90 days from SSD-backed PostgreSQL to compressed Parquet files in S3, queryable via Athena or Presto for historical analytics.
• Kafka Retention: Set topic retention to 7 days for operational topics and 24 hours for enrichment topics, reducing Kafka storage requirements by 70%.
• Connection Pooling: Use PgBouncer for PostgreSQL connection pooling, reducing connection overhead and allowing the database to handle more concurrent queries with less RAM.
• SMS Batching: Group SMS messages to the same carrier into batches, reducing per-message gateway fees by 10-15%.

23. Interview Q and A

Q1: How do you handle duplicate events from Shopify webhook retries?

Shopify retries webhooks if our endpoint does not respond with a 200 status within 5 seconds. This can cause duplicate events during network partitions or processing delays. We handle this using idempotency keys derived from the Shopify webhook ID (included in the X-Shopify-Webhook-Id header) and a Redis-based deduplication window. Every incoming webhook is checked against a Redis sorted set keyed by shop ID, with the webhook ID as the score and a 5-minute TTL. If the webhook ID already exists in the set, the event is acknowledged but not processed. This approach has zero false negatives (every duplicate is caught) and minimal false positives (the Bloom filter is only used for additional protection against memory pressure in Redis).

Q2: How does the segmentation engine evaluate complex rules at scale?

Segment evaluation uses a hybrid approach: simple property-based segments (e.g., "email opt-in is true") are evaluated using direct PostgreSQL queries with appropriate indexes. Behavioral segments (e.g., "purchased in last 30 days") use time-bounded queries on the events table with the (shop_id, customer_id, event_type, timestamp) composite index. Complex compound segments are decomposed into sub-queries that are executed independently and intersected using PostgreSQL's set operations. For segments that must update in near real-time, we use an incremental evaluation strategy where only customers affected by new events are re-evaluated, rather than re-scoring the entire customer base.

Q3: What happens when a flow email bounces or gets marked as spam?

The email delivery system processes bounce and complaint webhooks from the ESP in near real-time. A hard bounce (invalid address) immediately suppresses the email address from all future sends and exits the customer from any active flows. A soft bounce (mailbox full, temporary failure) retries with exponential backoff up to 3 times before suppressing. A spam complaint (feedback loop) immediately suppresses the address, exits all flows, and logs a compliance event that triggers a merchant notification. The suppression update propagates to the suppression list within 30 seconds, preventing any queued emails from being sent to the problematic address.

Q4: How do you ensure email deliverability during Black Friday when send volumes spike 10x?

Black Friday preparedness begins 3 months before the event. We pre-warm additional IP pools, negotiate increased sending limits with ESPs, and work with merchants to stagger campaign scheduling. During the event, the sending infrastructure applies dynamic throttling — if bounce rates on any IP pool rise above the threshold, send rates on that pool are automatically reduced and traffic is shifted to healthy pools. We also implement a priority queue where transactional emails (order confirmations, shipping updates) always take precedence over marketing campaigns, ensuring the highest-value emails are never delayed.

Q5: How do you handle a customer who opts out of email but remains in a flow that sends emails?

The flow engine checks consent status before every email step. When a customer opts out, the consent management service publishes a CustomerConsentUpdated event to Kafka. The flow engine's consumer processes this event and immediately exits the customer from any active flows that target the opted-out channel. Additionally, the flow state is marked with an exit reason ("channel_opt_out") for merchant reporting. Re-entry into flows is also blocked — if a flow trigger fires for an opted-out customer, the orchestration layer suppresses the entry.

Q6: Describe the trade-offs between real-time vs. batch segment evaluation.

Real-time evaluation (re-evaluating segments as each event arrives) provides the most up-to-date segment membership but is computationally expensive — every event must be checked against all potentially affected segments. Batch evaluation (re-evaluating segments on a schedule, e.g., every 5 minutes) is much more efficient because queries can be optimized across multiple events, but segment membership can be stale by up to the batch interval. Our hybrid approach uses real-time evaluation only for high-priority segments (those with active flow triggers) and batch evaluation for all other segments, balancing freshness with computational cost.

Q7: How do you design the revenue attribution model to handle multi-touch customer journeys?

Multi-touch attribution requires maintaining a complete interaction history for each customer within the attribution window (typically 5-30 days). Every email open, email click, SMS delivery, and SMS click is recorded as an interaction with a timestamp. When an order is placed, the attribution service queries all interactions within the window, applies the merchant's chosen attribution model (last-touch, linear, time-decay), and creates attribution records linking the order to specific campaigns and flows. The attribution window and model are configurable per merchant, and all attribution records are immutable once created to ensure historical accuracy.

Q8: How would you migrate a 50-million-customer store from another platform to your system without data loss?

The migration uses a phased approach. Phase one: import the customer list via CSV upload or API bulk import, mapping fields to our schema. Phase two: historical order data is imported from Shopify (which retains the full order history regardless of the previous marketing platform). Phase three: the JavaScript tracking SDK is deployed on the store, and new behavioral events begin flowing in from that point. Phase four: existing segment definitions are recreated using our segment rule builder, and existing flow automations are mapped to our flow system. During the transition period, the old platform remains active until the new system has sufficient historical data (typically 30 days of behavioral events) to power accurate segmentation and predictive models.

Q9: How do you prevent a rogue merchant from using the platform to send spam?

Spam prevention operates at multiple levels. During onboarding, new accounts are restricted to low send volumes (100 emails per day) with mandatory domain verification and SPF/DKIM/DMARC configuration. As the account builds sending reputation, volume limits are gradually increased. Automated monitoring detects spam indicators: high complaint rates, rapid list growth from unverified sources, unusual sending patterns (thousands of emails to addresses with zero engagement). If spam is detected, the account is automatically suspended pending review. Additionally, all outgoing emails are scanned for common spam indicators (missing physical address, excessive all-caps content, misleading subject lines) before delivery.

Q10: How would you redesign the system if email volume grew 100x?

A 100x growth in email volume (from 2 billion to 200 billion per day) would require fundamental architectural changes. The single-tenant PostgreSQL model for campaign scheduling would need to move to a distributed task queue (like Apache Flink) for campaign orchestration. Email rendering would move to a pre-rendered cache model where templates are rendered once per segment rather than once per recipient. The sending infrastructure would need to expand from 200 dedicated IPs to 2000+ IPs across multiple ESPs, with a custom IP reputation management system. Event processing would move from Kafka to Apache Pulsar or a similar system with built-in tiered storage to handle the increased throughput without proportional storage growth.

24. Full C# Implementation

The following implementation covers the four core services that form the backbone of the marketing platform: event ingestion, customer profile management, segmentation evaluation, and flow orchestration. This code demonstrates production-grade patterns including dependency injection, idempotency, retry logic, and event-driven architecture integration.

Event Ingestion Service

C#
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Npgsql;
using StackExchange.Redis;

public class EventIngestionService
{
    private readonly ILogger<EventIngestionService> _logger;
    private readonly IProducer<string, EventEnvelope> _kafkaProducer;
    private readonly IDatabase _redis;
    private readonly EventIngestionConfig _config;
    private readonly ICustomerResolver _customerResolver;

    public EventIngestionService(
        ILogger<EventIngestionService> logger,
        IProducer<string, EventEnvelope> kafkaProducer,
        IConnectionMultiplexer redis,
        IOptions<EventIngestionConfig> config,
        ICustomerResolver customerResolver)
    {
        _logger = logger;
        _kafkaProducer = kafkaProducer;
        _redis = redis.GetDatabase();
        _config = config.Value;
        _customerResolver = customerResolver;
    }

    public async Task<IngestionResult> IngestEventAsync(TrackEventRequest request)
    {
        var eventId = Guid.NewGuid().ToString();

        if (await IsDuplicateAsync(request.ShopId, eventId))
        {
            _logger.LogDebug("Duplicate event {EventId} for shop {ShopId}",
                eventId, request.ShopId);
            return IngestionResult.Duplicate(eventId);
        }

        var customerId = await _customerResolver.ResolveAsync(
            request.ShopId, request.CustomerEmail, request.CustomerPhone);

        var envelope = new EventEnvelope
        {
            EventId = eventId,
            ShopId = request.ShopId,
            EventType = request.EventType,
            CustomerId = customerId,
            CustomerEmail = request.CustomerEmail,
            CustomerPhone = request.CustomerPhone,
            Timestamp = request.Timestamp ?? DateTime.UtcNow,
            IngestedAt = DateTime.UtcNow,
            Source = request.Source,
            Properties = request.Properties ?? new Dictionary<string, object>(),
            Metadata = request.Metadata ?? new EventMetadata()
        };

        var topic = MapEventToTopic(request.EventType);
        var message = new Message<string, EventEnvelope>
        {
            Key = $"{request.ShopId}:{customerId ?? "anonymous"}",
            Value = envelope
        };

        var deliveryResult = await _kafkaProducer.ProduceAsync(topic, message);

        if (deliveryResult.Status == PersistenceStatus.Persisted)
        {
            _logger.LogInformation(
                "Event {EventId} ingested for shop {ShopId}, type {EventType}",
                eventId, request.ShopId, request.EventType);
            return IngestionResult.Accepted(eventId, deliveryResult.TopicPartitionOffset);
        }

        throw new EventIngestionException(
            $"Failed to persist event {eventId} to Kafka: {deliveryResult.Status}");
    }

    public async Task<BatchIngestionResult> IngestBatchAsync(
        List<TrackEventRequest> requests)
    {
        var tasks = requests.Select(IngestEventAsync);
        var completedResults = await Task.WhenAll(tasks);

        return new BatchIngestionResult
        {
            TotalEvents = requests.Count,
            Accepted = completedResults.Count(r => r.Status == IngestionStatus.Accepted),
            Duplicates = completedResults.Count(r => r.Status == IngestionStatus.Duplicate),
            Failed = completedResults.Count(r => r.Status == IngestionStatus.Failed),
            Results = completedResults.ToList()
        };
    }

    private async Task<bool> IsDuplicateAsync(string shopId, string eventId)
    {
        var key = $"dedup:events:{shopId}";
        var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
        var added = await _redis.SortedSetAddAsync(key, eventId, now);
        if (added)
            await _redis.KeyExpireAsync(key, TimeSpan.FromMinutes(5));
        return !added;
    }

    private string MapEventToTopic(string eventType)
    {
        return eventType switch
        {
            "OrderCreated" or "OrderUpdated" or "OrderCancelled" => "order.events",
            "ProductCreated" or "ProductUpdated" or "ProductDeleted" => "product.events",
            "EmailOpened" or "EmailClicked" or "EmailBounced" => "email.events",
            "SmsDelivered" or "SmsClicked" or "SmsFailed" => "sms.events",
            _ => "customer.events"
        };
    }
}

public class TrackEventRequest
{
    public string ShopId { get; set; } = string.Empty;
    public string EventType { get; set; } = string.Empty;
    public string? CustomerEmail { get; set; }
    public string? CustomerPhone { get; set; }
    public DateTime? Timestamp { get; set; }
    public string Source { get; set; } = "api";
    public Dictionary<string, object> Properties { get; set; } = new();
    public EventMetadata? Metadata { get; set; }
}

Segmentation Engine

C#
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Npgsql;
using StackExchange.Redis;

public class SegmentationEngine
{
    private readonly NpgsqlConnection _connection;
    private readonly IDatabase _redis;
    private readonly ISegmentCache _cache;
    private readonly ILogger<SegmentationEngine> _logger;

    public SegmentationEngine(
        NpgsqlConnection connection,
        IConnectionMultiplexer redis,
        ISegmentCache cache,
        ILogger<SegmentationEngine> logger)
    {
        _connection = connection;
        _redis = redis.GetDatabase();
        _cache = cache;
        _logger = logger;
    }

    public async Task<SegmentEvaluationResult> EvaluateSegmentAsync(
        SegmentDefinition segment, string customerId)
    {
        var cacheKey = $"segment_eval:{segment.Id}:{customerId}";
        var cachedResult = await _redis.StringGetAsync(cacheKey);

        if (cachedResult.HasValue)
        {
            return new SegmentEvaluationResult
            {
                SegmentId = segment.Id,
                CustomerId = customerId,
                IsMember = bool.Parse(cachedResult!),
                EvaluatedAt = DateTime.UtcNow,
                FromCache = true
            };
        }

        bool isMember = await EvaluateSegmentRulesAsync(segment, customerId);

        await _redis.StringSetAsync(cacheKey, isMember.ToString(),
            TimeSpan.FromMinutes(5));

        return new SegmentEvaluationResult
        {
            SegmentId = segment.Id,
            CustomerId = customerId,
            IsMember = isMember,
            EvaluatedAt = DateTime.UtcNow,
            FromCache = false
        };
    }

    private async Task<bool> EvaluateSegmentRulesAsync(
        SegmentDefinition segment, string customerId)
    {
        var sql = BuildSegmentQuery(segment, customerId);
        await using var cmd = new NpgsqlCommand(sql, _connection);
        cmd.Parameters.AddWithValue("@shopId", segment.ShopId);
        cmd.Parameters.AddWithValue("@customerId", customerId);

        var result = await cmd.ExecuteScalarAsync();
        return result != null && (long)result > 0;
    }

    private string BuildSegmentQuery(SegmentDefinition segment, string customerId)
    {
        var conditions = new List<string>();

        foreach (var rule in segment.Rules)
        {
            var condition = rule.RuleType switch
            {
                "property_equals" =>
                    $"c.properties->>'{rule.Property}' = @param_{rule.Property}",
                "property_greater_than" =>
                    $"(c.properties->>'{rule.Property}')::numeric > @param_{rule.Property}",
                "total_revenue_greater_than" =>
                    $"c.total_revenue > {rule.Value}",
                "order_count_greater_than" =>
                    $"c.order_count > {rule.Value}",
                "purchased_in_last_n_days" =>
                    $"c.last_order_at > NOW() - INTERVAL '{rule.Value} days'",
                "no_purchase_in_last_n_days" =>
                    $"(c.last_order_at IS NULL OR c.last_order_at < NOW() - INTERVAL '{rule.Value} days')",
                "has_viewed_product_in_last_n_days" =>
                    $"EXISTS (SELECT 1 FROM events e WHERE e.customer_id = c.id " +
                    $"AND e.event_type = 'ViewedProduct' " +
                    $"AND e.properties->>'product_id' = '{rule.Value}' " +
                    $"AND e.timestamp > NOW() - INTERVAL '{rule.Duration} days')",
                "email_opt_in" => "c.email_opt_in = true",
                "sms_opt_in" => "c.sms_opt_in = true",
                _ => throw new NotSupportedException($"Unknown rule type: {rule.RuleType}")
            };
            conditions.Add(condition);
        }

        var op = segment.ConditionOperator?.ToUpper() == "OR" ? "OR" : "AND";
        var whereClause = string.Join($" {op} ", conditions);

        return $@"
            SELECT COUNT(*) FROM customers c
            WHERE c.shop_id = @shopId AND c.id = @customerId
            AND ({whereClause})";
    }

    public async Task RefreshSegmentMembersAsync(SegmentDefinition segment)
    {
        _logger.LogInformation("Starting refresh for segment {SegmentId}", segment.Id);
        var batchSize = 1000;
        string? cursor = null;
        var totalMembers = 0;

        do
        {
            var (customerIds, nextCursor) = await GetSegmentMembersBatchAsync(
                segment, cursor, batchSize);
            var setKey = $"segment:{segment.Id}:members";

            var batch = _redis.CreateBatch();
            foreach (var cid in customerIds)
                batch.SetAddAsync(setKey, cid);
            batch.Execute();

            totalMembers += customerIds.Count;
            cursor = nextCursor;
        } while (cursor != null);

        await _redis.StringSetAsync(
            $"segment:{segment.Id}:count",
            totalMembers.ToString(),
            TimeSpan.FromMinutes(10));
    }

    private async Task<(List<string> members, string? cursor)> GetSegmentMembersBatchAsync(
        SegmentDefinition segment, string? cursor, int batchSize)
    {
        var whereClause = cursor != null ? $"AND c.id > @cursor" : "";
        var sql = BuildSegmentQuery(segment, null)
            .Replace("AND c.id = @customerId", whereClause) +
            $" ORDER BY c.id LIMIT {batchSize + 1}";

        await using var cmd = new NpgsqlCommand(sql, _connection);
        cmd.Parameters.AddWithValue("@shopId", segment.ShopId);
        if (cursor != null) cmd.Parameters.AddWithValue("@cursor", cursor);

        var members = new List<string>();
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
            members.Add(reader.GetString(0));

        string? nextCursor = members.Count > batchSize ? members.Last() : null;
        if (nextCursor != null) members.RemoveAt(members.Count - 1);
        return (members, nextCursor);
    }
}

Flow Orchestrator

C#
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using Npgsql;
using StackExchange.Redis;

public class FlowOrchestrator
{
    private readonly NpgsqlConnection _connection;
    private readonly IDatabase _redis;
    private readonly IEmailService _emailService;
    private readonly ISmsService _smsService;
    private readonly IOrchestrationDecider _orchestrationDecider;
    private readonly ILogger<FlowOrchestrator> _logger;

    public FlowOrchestrator(
        NpgsqlConnection connection,
        IConnectionMultiplexer redis,
        IEmailService emailService,
        ISmsService smsService,
        IOrchestrationDecider orchestrationDecider,
        ILogger<FlowOrchestrator> logger)
    {
        _connection = connection;
        _redis = redis.GetDatabase();
        _emailService = emailService;
        _smsService = smsService;
        _orchestrationDecider = orchestrationDecider;
        _logger = logger;
    }

    public async Task ProcessFlowTriggerAsync(
        string shopId, string eventType, string customerId)
    {
        var flows = await GetTriggeredFlowsAsync(shopId, eventType);

        foreach (var flow in flows)
        {
            var existingState = await GetFlowStateAsync(flow.Id, customerId);
            if (existingState != null && existingState.Status != FlowStatus.Paused)
                continue;

            var shouldEnter = await EvaluateFlowEntryCriteriaAsync(flow, customerId);
            if (!shouldEnter) continue;

            var state = existingState ?? new FlowState
            {
                FlowId = flow.Id,
                CustomerId = customerId,
                CurrentStep = 0,
                Status = FlowStatus.Active,
                EnteredAt = DateTime.UtcNow,
                Context = new Dictionary<string, object>
                {
                    ["trigger_event"] = eventType,
                    ["triggered_at"] = DateTime.UtcNow
                }
            };

            await ProcessFlowStepAsync(flow, state);
        }
    }

    public async Task ProcessScheduledStepsAsync()
    {
        var dueStates = await GetDueFlowStatesAsync();
        foreach (var state in dueStates)
        {
            var flow = await GetFlowAsync(state.FlowId);
            if (flow == null || flow.Status != "active") continue;
            await ProcessFlowStepAsync(flow, state);
        }
    }

    private async Task ProcessFlowStepAsync(FlowDefinition flow, FlowState state)
    {
        try
        {
            var step = flow.Steps.ElementAtOrDefault(state.CurrentStep);
            if (step == null)
            {
                state.Status = FlowStatus.Completed;
                state.CompletedAt = DateTime.UtcNow;
                await SaveFlowStateAsync(state);
                return;
            }

            switch (step.Type)
            {
                case "email":
                    await ProcessEmailStepAsync(flow, state, step);
                    break;
                case "sms":
                    await ProcessSmsStepAsync(flow, state, step);
                    break;
                case "wait":
                    state.CurrentStep++;
                    state.NextActionAt = DateTime.UtcNow
                        .Add(step.Duration, step.Unit);
                    state.Status = FlowStatus.Waiting;
                    await SaveFlowStateAsync(state);
                    break;
                case "condition":
                    var conditionMet = await EvaluateConditionAsync(
                        step.Rule, state.CustomerId);
                    state.CurrentStep = conditionMet
                        ? state.CurrentStep + 1
                        : state.CurrentStep + 2;
                    await SaveFlowStateAsync(state);
                    await ProcessFlowStepAsync(flow, state);
                    break;
                case "update_profile":
                    await UpdateProfilePropertiesAsync(
                        state.CustomerId, step.Properties);
                    state.CurrentStep++;
                    await SaveFlowStateAsync(state);
                    await ProcessFlowStepAsync(flow, state);
                    break;
                default:
                    state.CurrentStep++;
                    await SaveFlowStateAsync(state);
                    break;
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex,
                "Error processing flow {FlowId} step {Step} for customer {CustomerId}",
                flow.Id, state.CurrentStep, state.CustomerId);
            state.Status = FlowStatus.Paused;
            state.Context["error"] = ex.Message;
            await SaveFlowStateAsync(state);
        }
    }

    private async Task ProcessEmailStepAsync(
        FlowDefinition flow, FlowState state, FlowStep step)
    {
        var decision = await _orchestrationDecider.ShouldSend(
            state.CustomerId, "email", $"{flow.Id}:{step.TemplateId}");

        if (decision.Status == OrchestrationStatus.Suppressed)
        {
            state.CurrentStep++;
            await SkipToNextNonEmailStepAsync(flow, state);
            return;
        }

        if (decision.Status == OrchestrationStatus.Delayed)
        {
            state.NextActionAt = decision.DelayUntil;
            state.Status = FlowStatus.Waiting;
            await SaveFlowStateAsync(state);
            return;
        }

        var emailRequest = new EmailSendRequest
        {
            CustomerId = state.CustomerId,
            TemplateId = step.TemplateId,
            Subject = await ResolveTemplateVariableAsync(step.Subject, state),
            FlowId = flow.Id,
            FlowStep = state.CurrentStep,
            IdempotencyKey = $"{flow.Id}:{state.CustomerId}:{state.CurrentStep}"
        };

        var sendResult = await _emailService.SendAsync(emailRequest);
        if (sendResult.Success)
        {
            state.CurrentStep++;
            state.Status = FlowStatus.Active;
            await SaveFlowStateAsync(state);
        }
    }

    private async Task ProcessSmsStepAsync(
        FlowDefinition flow, FlowState state, FlowStep step)
    {
        var decision = await _orchestrationDecider.ShouldSend(
            state.CustomerId, "sms", $"{flow.Id}:{step.TemplateId}");

        if (decision.Status == OrchestrationStatus.Suppressed)
        {
            state.CurrentStep++;
            await SaveFlowStateAsync(state);
            return;
        }

        var smsRequest = new SmsSendRequest
        {
            CustomerId = state.CustomerId,
            Message = await ResolveTemplateVariableAsync(step.Message, state),
            FlowId = flow.Id,
            FlowStep = state.CurrentStep,
            IdempotencyKey = $"{flow.Id}:{state.CustomerId}:{state.CurrentStep}"
        };

        var sendResult = await _smsService.SendAsync(smsRequest);
        if (sendResult.Success)
        {
            state.CurrentStep++;
            await SaveFlowStateAsync(state);
        }
    }

    public async Task ExitCustomerFromFlowAsync(
        string flowId, string customerId, string reason)
    {
        var state = await GetFlowStateAsync(flowId, customerId);
        if (state == null) return;

        state.Status = FlowStatus.Exited;
        state.CompletedAt = DateTime.UtcNow;
        state.Context["exit_reason"] = reason;
        await SaveFlowStateAsync(state);
    }

    private async Task<List<FlowDefinition>> GetTriggeredFlowsAsync(
        string shopId, string eventType)
    {
        var sql = @"SELECT id, name, status, trigger_type, trigger_config, flow_definition
                    FROM flows
                    WHERE shop_id = @shopId AND status = 'active'
                    AND trigger_type = 'event'
                    AND trigger_config->>'event_type' = @eventType";

        await using var cmd = new NpgsqlCommand(sql, _connection);
        cmd.Parameters.AddWithValue("@shopId", Guid.Parse(shopId));
        cmd.Parameters.AddWithValue("@eventType", eventType);

        var flows = new List<FlowDefinition>();
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            flows.Add(new FlowDefinition
            {
                Id = reader.GetGuid(0).ToString(),
                Name = reader.GetString(1),
                Status = reader.GetString(2),
                TriggerType = reader.GetString(3),
                TriggerConfig = System.Text.Json.JsonSerializer
                    .Deserialize<Dictionary<string, object>>(
                        reader.GetString(4)) ?? new(),
                Steps = System.Text.Json.JsonSerializer
                    .Deserialize<List<FlowStep>>(
                        reader.GetString(5)) ?? new()
            });
        }
        return flows;
    }
}

Revenue Attribution Service

C#
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Npgsql;

public class RevenueAttributor
{
    private readonly NpgsqlConnection _connection;
    private readonly ILogger<RevenueAttributor> _logger;

    public RevenueAttributor(
        NpgsqlConnection connection,
        ILogger<RevenueAttributor> logger)
    {
        _connection = connection;
        _logger = logger;
    }

    public async Task<List<RevenueAttribution>> AttributeOrderAsync(OrderEvent order)
    {
        var interactions = await GetRecentInteractionsAsync(
            order.ShopId, order.CustomerId, order.Timestamp);

        if (!interactions.Any()) return new List<RevenueAttribution>();

        var attributions = CalculateAttributions(order, interactions);

        foreach (var attribution in attributions)
            await InsertAttributionAsync(attribution);

        _logger.LogInformation(
            "Attributed ${Revenue} across {Count} touchpoints for order {OrderId}",
            order.TotalRevenue, attributions.Count, order.Id);

        return attributions;
    }

    private List<RevenueAttribution> CalculateAttributions(
        OrderEvent order, List<MarketingInteraction> interactions)
    {
        var model = order.AttributionModel ?? "last_touch_email";
        return model switch
        {
            "last_touch_email" => CalculateLastTouch(order, interactions, "email"),
            "last_touch_any" => CalculateLastTouchAny(order, interactions),
            "linear" => CalculateLinear(order, interactions),
            "time_decay" => CalculateTimeDecay(order, interactions),
            "position_based" => CalculatePositionBased(order, interactions),
            _ => CalculateLastTouch(order, interactions, "email")
        };
    }

    private List<RevenueAttribution> CalculateLastTouch(
        OrderEvent order, List<MarketingInteraction> interactions, string channel)
    {
        var filtered = interactions
            .Where(i => channel == "any" || i.Channel == channel)
            .OrderByDescending(i => i.Timestamp).ToList();

        if (!filtered.Any()) return new List<RevenueAttribution>();

        var lastTouch = filtered.First();
        return new List<RevenueAttribution>
        {
            new RevenueAttribution
            {
                OrderId = order.Id, ShopId = order.ShopId,
                CustomerId = order.CustomerId,
                CampaignId = lastTouch.CampaignId,
                FlowId = lastTouch.FlowId, Channel = lastTouch.Channel,
                Revenue = order.TotalRevenue, Weight = 1.0m,
                AttributionModel = "last_touch_" + channel,
                InteractedAt = lastTouch.Timestamp, OrderedAt = order.Timestamp
            }
        };
    }

    private List<RevenueAttribution> CalculateLinear(
        OrderEvent order, List<MarketingInteraction> interactions)
    {
        var sorted = interactions.OrderBy(i => i.Timestamp).ToList();
        var weight = 1.0m / sorted.Count;
        return sorted.Select(i => new RevenueAttribution
        {
            OrderId = order.Id, ShopId = order.ShopId,
            CustomerId = order.CustomerId, CampaignId = i.CampaignId,
            FlowId = i.FlowId, Channel = i.Channel,
            Revenue = order.TotalRevenue * weight, Weight = weight,
            AttributionModel = "linear",
            InteractedAt = i.Timestamp, OrderedAt = order.Timestamp
        }).ToList();
    }

    private List<RevenueAttribution> CalculateTimeDecay(
        OrderEvent order, List<MarketingInteraction> interactions)
    {
        var halfLife = TimeSpan.FromDays(3);
        var totalDecay = interactions.Sum(i =>
            Math.Pow(0.5, (order.Timestamp - i.Timestamp).TotalDays / halfLife.TotalDays));

        return interactions.Select(interaction =>
        {
            var decay = Math.Pow(0.5,
                (order.Timestamp - interaction.Timestamp).TotalDays / halfLife.TotalDays);
            var weight = (decimal)(decay / totalDecay);
            return new RevenueAttribution
            {
                OrderId = order.Id, ShopId = order.ShopId,
                CustomerId = order.CustomerId,
                CampaignId = interaction.CampaignId,
                FlowId = interaction.FlowId, Channel = interaction.Channel,
                Revenue = order.TotalRevenue * weight, Weight = weight,
                AttributionModel = "time_decay",
                InteractedAt = interaction.Timestamp, OrderedAt = order.Timestamp
            };
        }).ToList();
    }

    private List<RevenueAttribution> CalculatePositionBased(
        OrderEvent order, List<MarketingInteraction> interactions)
    {
        var sorted = interactions.OrderBy(i => i.Timestamp).ToList();
        if (sorted.Count == 1)
            return CalculateLastTouch(order, interactions, "any");

        var firstWeight = 0.4m;
        var lastWeight = 0.4m;
        var middleWeight = 0.2m / Math.Max(1, sorted.Count - 2);

        return sorted.Select((interaction, index) =>
        {
            decimal weight = index == 0 ? firstWeight
                : index == sorted.Count - 1 ? lastWeight
                : middleWeight;
            return new RevenueAttribution
            {
                OrderId = order.Id, ShopId = order.ShopId,
                CustomerId = order.CustomerId,
                CampaignId = interaction.CampaignId,
                FlowId = interaction.FlowId, Channel = interaction.Channel,
                Revenue = order.TotalRevenue * weight, Weight = weight,
                AttributionModel = "position_based",
                InteractedAt = interaction.Timestamp, OrderedAt = order.Timestamp
            };
        }).ToList();
    }

    private async Task<List<MarketingInteraction>> GetRecentInteractionsAsync(
        string shopId, string customerId, DateTime orderTime)
    {
        var cutoff = orderTime.Subtract(TimeSpan.FromDays(5));
        var sql = @"
            SELECT event_type, properties, timestamp, source FROM events
            WHERE shop_id = @shopId AND customer_id = @customerId
            AND event_type IN ('EmailOpened','EmailClicked','SmsDelivered','SmsClicked')
            AND timestamp BETWEEN @cutoff AND @orderTime ORDER BY timestamp ASC";

        await using var cmd = new NpgsqlCommand(sql, _connection);
        cmd.Parameters.AddWithValue("@shopId", Guid.Parse(shopId));
        cmd.Parameters.AddWithValue("@customerId", Guid.Parse(customerId));
        cmd.Parameters.AddWithValue("@cutoff", cutoff);
        cmd.Parameters.AddWithValue("@orderTime", orderTime);

        var interactions = new List<MarketingInteraction>();
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            var props = System.Text.Json.JsonSerializer
                .Deserialize<Dictionary<string, object>>(reader.GetString(1)) ?? new();
            interactions.Add(new MarketingInteraction
            {
                InteractionType = reader.GetString(0),
                Channel = reader.GetString(0).StartsWith("Email") ? "email" : "sms",
                CampaignId = props.TryGetValue("campaign_id", out var cid)
                    ? cid.ToString() : null,
                FlowId = props.TryGetValue("flow_id", out var fid)
                    ? fid.ToString() : null,
                Timestamp = reader.GetDateTime(2), Source = reader.GetString(3)
            });
        }
        return interactions;
    }

    private async Task InsertAttributionAsync(RevenueAttribution attr)
    {
        var sql = @"
            INSERT INTO revenue_attributions
                (order_id, shop_id, customer_id, campaign_id, flow_id,
                 channel, revenue, weight, attribution_model,
                 interacted_at, ordered_at, created_at)
            VALUES
                (@orderId, @shopId, @customerId, @campaignId, @flowId,
                 @channel, @revenue, @weight, @attributionModel,
                 @interactedAt, @orderedAt, NOW())";

        await using var cmd = new NpgsqlCommand(sql, _connection);
        cmd.Parameters.AddWithValue("@orderId", Guid.Parse(attr.OrderId));
        cmd.Parameters.AddWithValue("@shopId", Guid.Parse(attr.ShopId));
        cmd.Parameters.AddWithValue("@customerId", Guid.Parse(attr.CustomerId));
        cmd.Parameters.AddWithValue("@campaignId",
            (object?)attr.CampaignId ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@flowId",
            (object?)attr.FlowId ?? DBNull.Value);
        cmd.Parameters.AddWithValue("@channel", attr.Channel);
        cmd.Parameters.AddWithValue("@revenue", attr.Revenue);
        cmd.Parameters.AddWithValue("@weight", attr.Weight);
        cmd.Parameters.AddWithValue("@attributionModel", attr.AttributionModel);
        cmd.Parameters.AddWithValue("@interactedAt", attr.InteractedAt);
        cmd.Parameters.AddWithValue("@orderedAt", attr.OrderedAt);
        await cmd.ExecuteNonQueryAsync();
    }
}

25. Conclusion

Designing a Klaviyo-like e-commerce marketing platform is one of the most challenging and rewarding system design exercises because it touches every major area of distributed systems engineering. The event ingestion pipeline must handle millions of events per second with sub-second latency and zero data loss. The customer data platform must unify behavioral data from dozens of sources into coherent profiles updated in real-time. The segmentation engine must evaluate complex behavioral rules against millions of customers in near real-time. The flow orchestrator must manage millions of concurrent automation workflows with idempotent execution. And the entire system must maintain strict compliance with email and SMS regulations across multiple jurisdictions.

The architectural patterns we have explored — event-driven architecture with Kafka, polyglot persistence with PostgreSQL and Redis, incremental computation for segmentation, state machine-based flow execution, and multi-layer caching — form a reusable toolkit that applies far beyond marketing platforms. These patterns appear in fintech transaction processing, healthcare event monitoring, IoT telemetry systems, and any domain where high-volume behavioral data must be transformed into timely, actionable outcomes.

For senior-plus engineers preparing for system design interviews, the key takeaways from this design are: always start with requirements and capacity estimation to ground your design in reality; design the data model before the architecture (the data model reveals the architecture); use event-driven patterns to decouple subsystems and enable independent scaling; implement idempotency everywhere (networks are unreliable, retries are inevitable); and think about compliance and operational concerns (deliverability, consent management, monitoring) from day one rather than treating them as afterthoughts.

The marketing automation space continues to evolve rapidly, with increasing emphasis on AI-driven personalization, real-time interaction marketing, and cross-channel orchestration. The foundational architecture described here provides a solid platform for building these next-generation capabilities while maintaining the reliability and scale that merchants depend on for their revenue.