Metering Usage

/// Simple implementation of bulk usage reporting

import fetch from 'node-fetch';
import { config } from 'dotenv';
config();

// Define global variables to store state
let queue = []; // Queue structure: [{customer_id, feature_slug, timestamp, units}, ...]
let last_sent = new Date(); // Timestamp of the last sent data

// Record usage events and update the queue
const recordUsage = (customer_id, feature_slug, units) => {
  // Create an event with the given data and current timestamp
  const event = {
    customer_id,
    feature_slug,
    timestamp: new Date().toISOString(),
    units,
  };

  // Add the event to the queue
  queue.push(event);

  // Send usage data every minute or 10.000 entries
  if (new Date() - last_sent >= 60*1000 || queue.length > 10000) {
    submitRecords();
  }
};

// Submit usage records to the API
const submitRecords = async () => {
  // Move the current queue to a new variable and reset the global queue and last_sent
  const q = queue;
  queue = [];
  last_sent = new Date();

  // Send the events in the queue to the API
  try {
    const response = await fetch('https://api.wingback.com/v1/c/usage', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'wb-key': process.env.WB_API_KEY,
      },
      body: JSON.stringify(q),
    });

    // Handle the API response
    if (response.status === 200) {
      const feedback = await response.json();
      if (feedback.length > 0) {
        // Handle non-empty feedback array
      }
    } else if (response.status >= 400 && response.status <= 499) {
      // Invalid format or API key invalid?
      const errorText = await response.text();
      console.log('error', errorText);
    } else if (response.status >= 500) {
      // Implement retry logic with exponential backoff
    }
  } catch ((err) => {
    // Probably a connection error, add retry logic
  })
};

// Periodic timer to check if we need to submit events
setInterval(() => {
  if (new Date() - last_sent >= 60 * 1000 || queue.length > 10000) {
    submitRecords();
  }
}, 30*1000);

// Usage Example:
recordUsage('Cust_98672780-0099-4ec3-b13e-0716d1d38d02', 'FEATURE_A', 10)

// Naive implementation of usage aggregation for high-frequency events using atomic counters

import fetch from 'node-fetch';
import { config } from 'dotenv';
config();

// Define global variables to store state
// IMPORTANT: JavaScript Map() does not support atomics, so initializing its elements as in this example is not thread safe
let queue = new Map(); // Queue structure: {customer_id: {period_timestamp: {feature_slug: Uint32Array[1]}}}
let last_sent = new Date(); // Timestamp of the last sent data

/// Record usage events and update the queue
const recordUsage = (customer_id, feature_slug, units) => {
  // Align the timestamp to the start of 5-minute intervals
  const date = new Date();
  date.setMinutes(Math.floor(date.getMinutes() / 5) * 5);
  date.setSeconds(0);
  date.setMilliseconds(0);
  const period_timestamp = date.toISOString();

  // Update queue structure with the given usage data
  if (!queue.has(customer_id)) {
    queue.set(customer_id, new Map());
  }
  const customer = queue.get(customer_id);

  if (!customer.has(period_timestamp)) {
    customer.set(period_timestamp, new Map());
  }
  const period = customer.get(period_timestamp);

  if (!period.has(feature_slug)) {
    const buffer = new SharedArrayBuffer(4);
    period.set(feature_slug, new Uint32Array(buffer));
  }
  const slug = period.get(feature_slug);

  // Atomically add the usage units
  Atomics.add(slug, 0, units);

  // Send usage data every minute
  if (new Date() - last_sent >= 60 * 1000) {
    submitRecords();
  }
};

// Submit usage records to the API
const submitRecords = async () => {
  if (queue.length == 0) return;

  // Move the current queue to a new variable and reset the global queue and last_sent
  const q = queue;
  queue = new Map();
  last_sent = new Date();

  // Flatten the queue data into an array of events
  const flattened_events = []; // Format: [{customer_id, feature_slug, timestamp, units}, ...]

  for (const [customer_id, customer] of q.entries()) {
    for (const [period_timestamp, period] of customer.entries()) {
      for (const [feature_slug, slug] of period.entries()) {
        flattened_events.push({
          customer_id,
          feature_slug,
          timestamp: period_timestamp,
          units: slug[0],
        });
      }
    }
  }

  if (flattened_events.length == 0) return;

  // Send the flattened events to the API
  try {
    const response = await fetch('https://api.test.wingback.com/v1/c/usage/bulk', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'wb-key': process.env.WB_API_KEY,
      },
      body: JSON.stringify(flattened_events),
    });

    // Handle the API response
    if (response.status === 200) {
      const feedback = await response.json();
      if (feedback.length > 0) {
        // Handle non-empty feedback array
      }
    } else if (response.status >= 400 && response.status <= 499) {
      // Invalid format or API key invalid?
      const errorText = await response.text();
      console.log('error', errorText);
    } else if (response.status >= 500) {
      // Implement retry logic with exponential backoff
    }
  } catch ((err) => {
    // Probably a connection error, add retry logic
  })
};

// Periodic timer to check if we need to submit events
setInterval(() => {
  if (new Date() - last_sent >= 60 * 1000) {
    submitRecords();
  }
}, 30*1000);

// Usage Example:
recordUsage('Cust_98672780-0099-4ec3-b13e-0716d1d38d02', 'FEATURE_A', 10)