a unified schema. 2. Context Enrichment: Joins transaction data with product metadata, customer journey touchpoints, and bundle definitions. 3. Attribution Processor: Applies the selected model (Weighted, Linear, or Custom) to distribute revenue across product dimensions. 4. Ledger Storage: Persists attribution results in a columnar store optimized for analytical queries.

Step-by-Step Implementation

1. Define the Unified Event Schema

All revenue events must conform to a schema that supports multi-product distribution.

import { z } from 'zod';

// Core types for attribution
type ProductId = string;
type TransactionId = string;
type Decimal = string; // Use string for precision

const AttributionEventSchema = z.object({
  transactionId: z.string().uuid(),
  timestamp: z.string().datetime(),
  customerId: z.string().uuid(),
  grossAmount: z.number().min(0),
  currency: z.string().length(3),
  products: z.array(z.object({
    productId: z.string(),
    quantity: z.number(),
    unitPrice: z.number(),
    metadata: z.record(z.unknown()).optional(),
  })),
  // Touchpoints drive weighted attribution
  touchpoints: z.array(z.object({
    productId: z.string(),
    interactionType: z.enum(['view', 'click', 'usage_spike', 'support', 'trial']),
    timestamp: z.string().datetime(),
    weight: z.number().min(0).max(1).optional(), // Pre-assigned weight if known
  })).default([]),
  bundleId: z.string().uuid().optional(),
  source: z.enum(['invoice', 'usage_meter', 'marketplace', 'refund']),
});

type AttributionEvent = z.infer<typeof AttributionEventSchema>;

2. Implement the Rule Engine

A composable rule engine allows business logic to evolve without code deployments. Rules are evaluated in priority order.

interface AttributionRule {
  name: string;
  priority: number;
  matches(event: AttributionEvent): boolean;
  distribute(event: AttributionEvent): Distribution[];
}

interface Distribution {
  productId: ProductId;
  amount: number;
  reason: string;
}

class RuleEngine {
  private rules: AttributionRule[] = [];

  addRule(rule: AttributionRule) {
    this.rules.push(rule);
    this.rules.sort((a, b) => b.priority - a.priority);
  }

  process(event: AttributionEvent): Distribution[] {
    const matchedRule = this.rules.find(r => r.matches(event));
    if (!matchedRule) {
      throw new Error(`No attribution rule matched event ${event.transactionId}`);
    }
    return matchedRule.distribute(event);
  }
}

// Example: Bundle Attribution Rule
const BundleRule: AttributionRule = {
  name: 'Bundle Pro-Rata',
  priority: 100,
  matches: (e) => !!e.bundleId && e.products.length > 1,
  distribute: (e) => {
    const totalUnitValue = e.products.reduce((sum, p) => sum + (p.unitPrice * p.quantity), 0);
    return e.products.map(p => ({
      productId: p.productId,
      amount: (e.grossAmount * p.unitPrice * p.quantity) / totalUnitValue,
      reason: 'Bundle pro-rata split',
    }));
  },
};

// Example: Weighted Touchpoint Rule
const WeightedTouchpointRule: AttributionRule = {
  name: 'Weighted Multi-Touch',
  priority: 50,
  matches: (e) => e.touchpoints.length > 0,
  distribute: (e) => {
    const weights = e.touchpoints.reduce((acc, tp) => {
      acc[tp.productId] = (acc[tp.productId] || 0) + (tp.weight || 1);
      return acc;
    }, {} as Record<ProductId, number>);

    const totalWeight = Object.values(weights).reduce((a, b) => a + b, 0);
    
    return Object.entries(weights).map(([productId, weight]) => ({
      productId,
      amount: (e.grossAmount * weight) / totalWeight,
      reason: 'Weighted touchpoint attribution',
    }));
  },
};

3. Handle Refund Cascades

Refunds must reverse attribution atomically. The engine must support idempotent reversal transactions.

function processRefund(
  refundEvent: Pick<AttributionEvent, 'transactionId' | 'grossAmount' | 'products'>,
  originalAttribution: Distribution[]
): Distribution[] {
  // Verify refund matches original structure
  const originalTotal = originalAttribution.reduce((s, d) => s + d.amount, 0);
  const ratio = refundEvent.grossAmount / originalTotal;

  return originalAttribution.map(dist => ({
    productId: dist.productId,
    amount: -(dist.amount * ratio), // Negative amount indicates reversal
    reason: `Refund cascade from ${refundEvent.transactionId}`,
  }));
}

4. Storage Strategy

Store attribution results in a format optimized for time-series analysis. A schema like product_revenue_ledger allows querying revenue by product, customer, and time bucket.

CREATE TABLE product_revenue_ledger (
    ledger_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    transaction_id UUID NOT NULL,
    product_id UUID NOT NULL,
    customer_id UUID NOT NULL,
    amount DECIMAL(18, 4) NOT NULL,
    currency CHAR(3) NOT NULL,
    attribution_model VARCHAR(50) NOT NULL,
    attribution_reason TEXT,
    event_timestamp TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    UNIQUE(transaction_id, product_id) -- Prevent double counting
);

CREATE INDEX idx_ledger_product_time ON product_revenue_ledger(product_id, event_timestamp);
CREATE INDEX idx_ledger_customer_time ON product_revenue_ledger(customer_id, event_timestamp);

Architecture Decisions

• Stateless Processing: The attribution engine must be stateless regarding the calculation logic. State is derived from the event payload and external context (e.g., bundle definitions fetched via cache). This enables horizontal scaling.
• Decimal Precision: All monetary calculations must use fixed-point arithmetic or string-based decimals to avoid floating-point errors that compound during distribution.
• Idempotency: Every attribution write must be idempotent based on transaction_id and product_id. This prevents duplicate revenue when event streams retry.

Pitfall Guide

1. Double Counting Revenue

Mistake: Aggregating revenue across product tables without deduplication when transactions span multiple products. Fix: Enforce a unique constraint on (transaction_id, product_id) in the ledger. Revenue for a customer is the sum of the ledger, not the sum of product invoices.

2. Ignoring Refund Cascades

Mistake: Recording a refund as a negative transaction on the refunding product only, without reversing attribution on the original products. Fix: Implement a lookup mechanism that retrieves the original attribution distribution and applies a proportional reversal. Refunds must be treated as anti-events that mirror the original distribution.

3. Timezone and Cut-off Errors

Mistake: Attributing revenue to a product based on the transaction time in the user's timezone rather than UTC, causing misalignment in monthly reports. Fix: All attribution timestamps must be normalized to UTC. Business logic for monthly cut-offs should be handled at query time or via a separate reporting layer, not during ingestion.

4. Circular Dependencies in Bundles

Mistake: Defining bundles where Product A includes Product B, and Product B includes Product A, causing infinite loops in attribution logic. Fix: Validate bundle definitions against a DAG (Directed Acyclic Graph) structure during schema validation. Reject circular references at configuration time.

5. Performance Degradation in Real-Time Attribution

Mistake: Running complex graph traversals or Shapley calculations synchronously during the checkout flow. Fix: Decouple attribution from the critical path. Ingest the transaction, acknowledge the user, and process attribution asynchronously via a message queue. Use cached weights for real-time estimates if needed.

6. Lack of Auditability

Mistake: Overwriting attribution results when rules change, destroying historical accuracy. Fix: Use an append-only ledger. If rules change, recalculate attribution for new events only. Maintain a model_version field to track which logic produced each record. Allow backfilling via replay jobs.

7. Hardcoding Business Logic

Mistake: Embedding attribution rules directly in database triggers or application code, making updates risky and slow. Fix: Externalize rules into a configuration store or decision engine. The code should execute rules, not define them. This allows finance and product teams to adjust weights without engineering releases.

Production Bundle

Action Checklist

• Schema Validation: Implement strict Zod/JSON Schema validation for all ingestion events to reject malformed data early.
• Idempotency Keys: Ensure every write to the attribution ledger uses a deterministic composite key to prevent duplicates.
• Decimal Library: Replace native floats with a decimal library (e.g., decimal.js or big.js) for all monetary math.
• Backfill Strategy: Design a replay job that can re-process historical events with new attribution rules for model iteration.
• Variance Alerting: Set up monitoring to alert when sum(attributed_revenue) deviates from billing_system_revenue by >0.01%.
• Refund Handling: Verify refund logic reverses attribution proportionally and updates the ledger atomically.
• Access Control: Implement row-level security or service accounts so product teams can query their attribution without accessing PII.

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Early Stage / < 5 Products	Linear Split	Simplicity outweighs precision. Low engineering overhead.	Minimal
Cross-sell Focus / Bundles	Weighted Rule Engine	Captures product interdependencies. Configurable without code.	Medium (Storage/Compute)
Marketplace / Complex Dependencies	Graph/Shapley Model	Required for non-linear value contribution. High accuracy.	High (Compute/Complexity)
Real-time Dashboard Needs	Approximation Cache	Pre-calculate weights; update asynchronously.	Medium (Cache Infra)
Strict Compliance / Audit	Append-only Ledger + Replay	Immutable history allows full audit trails.	Low (Storage increase)

Configuration Template

Use this YAML structure to define attribution rules in your configuration store.

attribution:
  version: "1.0"
  default_model: "weighted"
  rules:
    - name: "Enterprise Bundle"
      priority: 100
      condition: "event.bundle_id == 'ent-bundle-v2'"
      model: "pro_rata"
      parameters:
        basis: "unit_price"
    
    - name: "Usage-Driven Upsell"
      priority: 50
      condition: "event.touchpoints.exists(tp => tp.type == 'usage_spike')"
      model: "weighted"
      parameters:
        weights:
          view: 0.1
          click: 0.3
          usage_spike: 0.8
          support: 0.2
        normalization: "sum_weights"
    
    - name: "Fallback"
      priority: 0
      condition: "true"
      model: "last_touch"

Quick Start Guide

1. Initialize Ledger: Run the SQL schema creation script. Add the product_revenue_ledger table to your analytics warehouse.
2. Deploy Engine: Containerize the TypeScript attribution engine. Configure it to listen to your billing event stream (e.g., Kafka/RabbitMQ).
3. Load Rules: Import the configuration template via the engine's admin API. Validate rules against a sample event set.
4. Ingest Test Data: Send a mock multi-product transaction. Verify the ledger contains distributed amounts that sum to the gross total.
5. Query Results: Run a query to aggregate revenue by product for the current month. Compare against raw billing totals to confirm reconciliation.

# Example query to verify reconciliation
SELECT 
  SUM(amount) as total_attributed,
  (SELECT SUM(gross_amount) FROM billing_events WHERE date = CURRENT_DATE) as total_billing
FROM product_revenue_ledger
WHERE event_timestamp::date = CURRENT_DATE;