Exporting Paystack Data to a Data Warehouse

Your payments ledger in PostgreSQL works fine for operational queries: verifying a transaction, checking an order status, running daily reconciliation. But as you grow, some queries become problematic:

• Heavy aggregation queries (monthly revenue by product by channel) slow down your production database.
• Analysts want to join payment data with marketing data, user behavior data, and product data. These live in different systems.
• Historical analysis over years of data grinds your production database to a halt.
• You want to run machine learning models on payment patterns (fraud detection, churn prediction).

A data warehouse is a separate database optimized for analytical queries. You copy your payment data there and run heavy queries without affecting production performance. Common choices include Google BigQuery, Amazon Redshift, Snowflake, or even a separate PostgreSQL instance dedicated to analytics.

You do not need a data warehouse until you are processing hundreds of transactions per day or running complex analytical queries. Start with direct ledger queries (see the financial reports guide) and move to a warehouse when those queries start impacting production performance.

ETL stands for Extract, Transform, Load. The pipeline has three stages:

Extract: Pull new records from your payments ledger since the last run.

Transform: Flatten nested data, add derived columns, normalize formats.

Load: Insert the transformed records into the warehouse.

// etl/pipeline.js - The main pipeline
async function runPipeline() {
  var startTime = Date.now();
  console.log('ETL pipeline starting');

  // Get the last successful extraction timestamp
  var lastRun = await getLastRunTimestamp();
  var cutoff = new Date().toISOString();

  console.log('Extracting records since ' + lastRun);

  // Extract
  var rawRecords = await extract(lastRun, cutoff);
  console.log('Extracted ' + rawRecords.length + ' records');

  if (rawRecords.length === 0) {
    console.log('No new records. Pipeline complete.');
    await updateLastRunTimestamp(cutoff);
    return;
  }

  // Transform
  var transformed = transform(rawRecords);
  console.log('Transformed ' + transformed.length + ' records');

  // Load
  await load(transformed);
  console.log('Loaded ' + transformed.length + ' records into warehouse');

  // Update high-water mark
  await updateLastRunTimestamp(cutoff);

  var duration = Date.now() - startTime;
  console.log('ETL pipeline complete in ' + duration + 'ms');
}

async function getLastRunTimestamp() {
  var result = await db.query(
    'SELECT last_value FROM etl_state WHERE pipeline_name = $1',
    ['payment_ledger']
  );

  if (result.rows.length === 0) {
    return '2026-01-01T00:00:00Z';  // First run: start from beginning
  }

  return result.rows[0].last_value;
}

async function updateLastRunTimestamp(timestamp) {
  await db.query(
    'INSERT INTO etl_state (pipeline_name, last_value, updated_at) '
    + 'VALUES ($1, $2, NOW()) '
    + 'ON CONFLICT (pipeline_name) '
    + 'DO UPDATE SET last_value = $2, updated_at = NOW()',
    ['payment_ledger', timestamp]
  );
}

// Extract new records from the payments ledger
async function extract(since, until) {
  var result = await db.query(
    'SELECT id, reference, event_type, direction, '
    + 'gross_amount, fee_amount, net_amount, currency, '
    + 'channel, paystack_id, customer_email, order_id, '
    + 'settlement_id, metadata, created_at, settled_at '
    + 'FROM payment_ledger '
    + 'WHERE created_at > $1 AND created_at <= $2 '
    + 'ORDER BY created_at',
    [since, until]
  );

  return result.rows;
}

Key extraction decisions:

• Extract from the ledger, not the Paystack API. The ledger is your source of truth. It has already been reconciled. API extraction introduces rate limits, pagination complexity, and potential inconsistencies.
• Use incremental extraction. Only pull records created after the last successful run. This avoids re-processing old data and keeps the pipeline fast. The etl_state table tracks the high-water mark.
• Include the metadata column. The JSONB metadata contains the full Paystack response. In the warehouse, you can extract specific fields for analysis (card type, bank name, IP address) without going back to the API.

// Transform extracted records for the warehouse
function transform(records) {
  return records.map(function(record) {
    var metadata = typeof record.metadata === 'string'
      ? JSON.parse(record.metadata) : record.metadata;

    var createdAt = new Date(record.created_at);

    return {
      // Core fields (pass through)
      ledger_id: record.id,
      reference: record.reference,
      event_type: record.event_type,
      direction: record.direction,
      gross_amount: record.gross_amount,
      fee_amount: record.fee_amount,
      net_amount: record.net_amount,
      currency: record.currency,
      channel: record.channel,
      customer_email: record.customer_email,
      order_id: record.order_id,
      created_at: record.created_at,
      settled_at: record.settled_at,

      // Derived fields (for easier analytics)
      created_date: createdAt.toISOString().split('T')[0],
      created_hour: createdAt.getUTCHours(),
      created_day_of_week: createdAt.getUTCDay(),
      is_weekend: createdAt.getUTCDay() === 0 || createdAt.getUTCDay() === 6,

      // Gross amount in major currency unit for readability
      gross_amount_major: record.gross_amount / 100,
      fee_amount_major: record.fee_amount / 100,
      net_amount_major: record.net_amount / 100,

      // Effective fee rate
      fee_rate: record.gross_amount > 0
        ? Math.round(record.fee_amount * 10000 / record.gross_amount) / 100
        : 0,

      // Flattened metadata fields
      card_type: metadata && metadata.authorization
        ? metadata.authorization.card_type : null,
      card_bank: metadata && metadata.authorization
        ? metadata.authorization.bank : null,
      card_last4: metadata && metadata.authorization
        ? metadata.authorization.last4 : null,
      customer_id: metadata && metadata.customer
        ? metadata.customer.id : null,
      ip_address: metadata ? metadata.ip_address : null,

      // Settlement delay in days (null if not yet settled)
      settlement_delay_days: record.settled_at
        ? Math.round((new Date(record.settled_at) - createdAt) / 86400000)
        : null,
    };
  });
}

The transformation adds value by making common analytical questions answerable with simple queries. "What is our revenue by day of week?" becomes a GROUP BY on created_day_of_week instead of extracting the day from a timestamp. "What is our average fee rate by channel?" is already calculated as fee_rate.

The load step depends on your warehouse. Here is a generic pattern that works for most SQL-based warehouses:

// Load transformed records into the warehouse
// (Example: loading into a separate PostgreSQL analytics database)
async function load(records) {
  if (records.length === 0) return;

  var warehouseDb = require('./warehouse-db');

  // Use a staging table for safe loading
  await warehouseDb.query('CREATE TEMP TABLE IF NOT EXISTS staging_ledger (LIKE payment_facts INCLUDING ALL)');
  await warehouseDb.query('TRUNCATE staging_ledger');

  // Batch insert into staging
  var batchSize = 500;
  for (var i = 0; i < records.length; i += batchSize) {
    var batch = records.slice(i, i + batchSize);

    var values = [];
    var placeholders = [];
    var paramIndex = 1;

    for (var j = 0; j < batch.length; j++) {
      var r = batch[j];
      var rowPlaceholders = [];

      var fields = [
        r.ledger_id, r.reference, r.event_type, r.direction,
        r.gross_amount, r.fee_amount, r.net_amount, r.currency,
        r.channel, r.customer_email, r.order_id,
        r.created_at, r.settled_at, r.created_date,
        r.created_hour, r.created_day_of_week, r.is_weekend,
        r.gross_amount_major, r.fee_rate, r.card_type,
        r.card_bank, r.settlement_delay_days,
      ];

      for (var k = 0; k < fields.length; k++) {
        rowPlaceholders.push('$' + paramIndex);
        values.push(fields[k]);
        paramIndex++;
      }

      placeholders.push('(' + rowPlaceholders.join(', ') + ')');
    }

    await warehouseDb.query(
      'INSERT INTO staging_ledger '
      + '(ledger_id, reference, event_type, direction, '
      + 'gross_amount, fee_amount, net_amount, currency, '
      + 'channel, customer_email, order_id, '
      + 'created_at, settled_at, created_date, '
      + 'created_hour, created_day_of_week, is_weekend, '
      + 'gross_amount_major, fee_rate, card_type, '
      + 'card_bank, settlement_delay_days) '
      + 'VALUES ' + placeholders.join(', '),
      values
    );
  }

  // Merge from staging into fact table (upsert)
  await warehouseDb.query(
    'INSERT INTO payment_facts SELECT * FROM staging_ledger '
    + 'ON CONFLICT (ledger_id) DO NOTHING'
  );

  await warehouseDb.query('DROP TABLE IF EXISTS staging_ledger');
}

The staging table pattern is a safety measure. If the load fails halfway through, the fact table is not corrupted. The staging table is temporary and will be cleaned up on the next run.

-- Fact table in the data warehouse
CREATE TABLE payment_facts (
  ledger_id             BIGINT PRIMARY KEY,
  reference             VARCHAR(100) NOT NULL,
  event_type            VARCHAR(50) NOT NULL,
  direction             VARCHAR(10),
  gross_amount          BIGINT,
  fee_amount            BIGINT,
  net_amount            BIGINT,
  currency              VARCHAR(3),
  channel               VARCHAR(30),
  customer_email        VARCHAR(255),
  order_id              BIGINT,
  created_at            TIMESTAMPTZ,
  settled_at            TIMESTAMPTZ,

  -- Derived / flattened fields
  created_date          DATE,
  created_hour          SMALLINT,
  created_day_of_week   SMALLINT,
  is_weekend            BOOLEAN,
  gross_amount_major    NUMERIC(20, 2),
  fee_rate              NUMERIC(5, 2),
  card_type             VARCHAR(20),
  card_bank             VARCHAR(100),
  settlement_delay_days INT
);

CREATE INDEX idx_facts_date ON payment_facts(created_date);
CREATE INDEX idx_facts_currency ON payment_facts(currency);
CREATE INDEX idx_facts_channel ON payment_facts(channel);
CREATE INDEX idx_facts_event ON payment_facts(event_type);

The fact table mirrors the transformed data structure. Indexes are on the columns you will filter and group by most often. Unlike the production ledger, the warehouse schema is optimized for read performance, not write safety.

// Schedule the ETL pipeline
var cron = require('node-cron');

// Run daily at 7 AM, after reconciliation (which runs at 6 AM)
cron.schedule('0 7 * * *', async function() {
  try {
    await runPipeline();
  } catch (err) {
    console.error('ETL pipeline failed: ' + err.message);
    await sendAlert('ETL Pipeline Failed', { error: err.message });
  }
});

Monitor the pipeline with a simple health check:

• Track the last successful run timestamp. Alert if it has not run in 48 hours.
• Track the number of records processed per run. A sudden drop might indicate an extraction problem.
• Compare the row count in the warehouse against the row count in the ledger. They should match (within the extraction window).

The pipeline should run after your daily reconciliation job completes. This ensures the data in the warehouse has been reconciled and is trustworthy for reporting.

A full ETL pipeline is not always necessary. Consider these simpler alternatives:

Read replica: Set up a PostgreSQL read replica of your production database. Run analytical queries against the replica instead of the primary. This adds no new infrastructure beyond the replica and requires no ETL code. Most cloud PostgreSQL providers support replicas natively.

Materialized views: Create materialized views on your production database for common aggregation queries. Refresh them on a schedule. This avoids a separate warehouse while still giving fast analytical queries.

-- Materialized view for daily revenue
CREATE MATERIALIZED VIEW daily_revenue AS
SELECT
  DATE(created_at) as date,
  currency,
  channel,
  COUNT(*) as count,
  SUM(gross_amount) as gross,
  SUM(fee_amount) as fees,
  SUM(net_amount) as net
FROM payment_ledger
WHERE event_type = 'charge.success'
GROUP BY DATE(created_at), currency, channel;

-- Refresh daily
REFRESH MATERIALIZED VIEW daily_revenue;

CSV exports to Google Sheets: For very small operations, a daily CSV export of ledger data loaded into Google Sheets might be enough. Your accountant can build pivot tables and charts directly. This does not scale, but it works for businesses processing fewer than 100 transactions per day.

Start with the simplest approach that meets your needs. Graduate to a full data warehouse when the simpler approaches hit their limits.