feat: billing dry run (#342)

A script to do dry run of the billing pipeline. It reads production
data, but does not write to any production tables.

billing/.gitignore

@@ -0,0 +1,3 @@
+.env.local
+*.csv
+*.json

billing/data/consumer.js

@@ -52,7 +52,7 @@ export const decode = input => {
         consumer: Schema.did().from(input.consumer),
         provider: Schema.did({ method: 'web' }).from(input.provider),
         subscription: /** @type {string} */ (input.subscription),
-        cause: Link.parse(/** @type {string} */ (input.cause)),
+        cause: input.cause ? Link.parse(/** @type {string} */ (input.cause)) : undefined,
         insertedAt: new Date(input.insertedAt),
         updatedAt: input.updatedAt ? new Date(input.updatedAt) : undefined
       }

billing/package.json

@@ -24,7 +24,10 @@
     "@types/big.js": "^6.2.1",
     "aws-lambda": "^1.0.7",
     "c8": "^8.0.1",
+    "csv-stringify": "^6.4.6",
+    "dotenv": "^16.4.5",
     "entail": "^2.1.1",
+    "p-all": "^5.0.0",
     "testcontainers": "^10.7.1"
   },
   "engines": {

billing/scripts/.env.template

@@ -0,0 +1,7 @@
+AWS_REGION=
+
+CUSTOMER_TABLE_NAME=
+SUBSCRIPTION_TABLE_NAME=
+CONSUMER_TABLE_NAME=
+SPACE_DIFF_TABLE_NAME=
+SPACE_SNAPSHOT_TABLE_NAME=

billing/scripts/README.md

@@ -0,0 +1,26 @@
+# dry-run.js
+
+## Usage
+
+Create your own `.env.local` file and fill in the relevant details:
+
+```sh
+cp billing/scripts/.env.template billing/scripts/.env.local
+```
+
+Run the billing pipeline:
+
+```sh
+cd billing/scripts
+node dry-run.js
+```
+
+This will output a CSV file (`summary-[from]-[to].csv`) with ordered per customer information about what they will be charged.
+
+Much more info is collected, and is output to a JSON file `usage-[from]-[to].json` if you want to do some more spelunking.
+
+Note: this is only as up to date as the `productInfo` found in `helpers.js` which is (at time of writing) set as:
+
+* `did:web:starter.web3.storage` cost: $0, overage: 0.15 / GB, included: 5 * GB
+* `did:web:lite.web3.storage` cost: $10, overage: 0.05 / GB, included: 100 * GB,
+* `did:web:business.web3.storage` cost: $100, overage: 0.03 / GB, included: 2 * TB

billing/scripts/dry-run.js

@@ -0,0 +1,145 @@
+/**
+ * Perform a dry run of the billing pipeline, printing out a report of the
+ * usage per customer/space.
+ */
+import dotenv from 'dotenv'
+import { DynamoDBClient } from '@aws-sdk/client-dynamodb'
+import * as CSV from 'csv-stringify/sync'
+import fs from 'node:fs'
+import all from 'p-all'
+import { startOfLastMonth, startOfMonth } from '../lib/util.js'
+import { calculateCost, createMemoryQueue, createMemoryStore } from './helpers.js'
+import { EndOfQueue } from '../test/helpers/queue.js'
+import { expect, mustGetEnv } from '../functions/lib.js'
+import * as BillingCron from '../lib/billing-cron.js'
+import * as CustomerBillingQueue from '../lib/customer-billing-queue.js'
+import * as SpaceBillingQueue from '../lib/space-billing-queue.js'
+import { createCustomerStore } from '../tables/customer.js'
+import { createSubscriptionStore } from '../tables/subscription.js'
+import { createConsumerStore } from '../tables/consumer.js'
+import { createSpaceDiffStore } from '../tables/space-diff.js'
+import { createSpaceSnapshotStore } from '../tables/space-snapshot.js'
+import * as Usage from '../data/usage.js'
+
+dotenv.config({ path: '.env.local' })
+
+const concurrency = 5
+
+const CUSTOMER_TABLE_NAME = mustGetEnv('CUSTOMER_TABLE_NAME')
+const SUBSCRIPTION_TABLE_NAME = mustGetEnv('SUBSCRIPTION_TABLE_NAME')
+const CONSUMER_TABLE_NAME = mustGetEnv('CONSUMER_TABLE_NAME')
+const SPACE_DIFF_TABLE_NAME = mustGetEnv('SPACE_DIFF_TABLE_NAME')
+const SPACE_SNAPSHOT_TABLE_NAME = mustGetEnv('SPACE_SNAPSHOT_TABLE_NAME')
+
+const dynamo = new DynamoDBClient()
+
+const customerStore = createCustomerStore(dynamo, { tableName: CUSTOMER_TABLE_NAME })
+const subscriptionStore = createSubscriptionStore(dynamo, { tableName: SUBSCRIPTION_TABLE_NAME })
+const consumerStore = createConsumerStore(dynamo, { tableName: CONSUMER_TABLE_NAME })
+const spaceDiffStore = createSpaceDiffStore(dynamo, { tableName: SPACE_DIFF_TABLE_NAME })
+const readableSpaceSnapshotStore = createSpaceSnapshotStore(dynamo, { tableName: SPACE_SNAPSHOT_TABLE_NAME })
+
+/** @type {import('../lib/api.js').StorePutter<import('../lib/api.js').SpaceSnapshot> & import('../lib/api.js').StoreLister<any, import('../lib/api.js').SpaceSnapshot> & import('../lib/api.js').StoreGetter<any, import('../lib/api.js').SpaceSnapshot>} */
+const writableSpaceSnapshotStore = createMemoryStore()
+/** @type {import('../lib/api.js').StorePutter<import('../lib/api.js').Usage> & import('../lib/api.js').StoreLister<any, import('../lib/api.js').Usage> & import('../lib/api.js').StoreGetter<any, import('../lib/api.js').Usage>} */
+const usageStore = createMemoryStore()
+
+/** @type {import('../lib/api.js').QueueAdder<import('../lib/api.js').CustomerBillingInstruction> & import('../test/lib/api.js').QueueRemover<import('../lib/api.js').CustomerBillingInstruction>} */
+const customerBillingQueue = createMemoryQueue()
+/** @type {import('../lib/api.js').QueueAdder<import('../lib/api.js').SpaceBillingInstruction> & import('../test/lib/api.js').QueueRemover<import('../lib/api.js').SpaceBillingInstruction>} */
+const spaceBillingQueue = createMemoryQueue()
+
+const now = new Date()
+const from = startOfLastMonth(now)
+const to = startOfMonth(now)
+
+console.log(`Running billing for period: ${from.toISOString()} - ${to.toISOString()}`)
+expect(await BillingCron.enqueueCustomerBillingInstructions({ from, to }, {
+  customerStore,
+  customerBillingQueue
+}))
+
+const customerBillingInstructions = []
+while (true) {
+  const removeResult = await customerBillingQueue.remove()
+  if (removeResult.error) {
+    if (removeResult.error.name === EndOfQueue.name) break
+    throw removeResult.error
+  }
+  customerBillingInstructions.push(removeResult.ok)
+}
+
+await all(customerBillingInstructions.map(instruction => async () => {
+  expect(await CustomerBillingQueue.enqueueSpaceBillingInstructions(instruction, {
+    subscriptionStore,
+    consumerStore,
+    spaceBillingQueue
+  }))
+
+  const spaceBillingInstructions = []
+  while (true) {
+    const removeResult = await spaceBillingQueue.remove()
+    if (removeResult.error) {
+      if (removeResult.error.name === EndOfQueue.name) break
+      throw removeResult.error
+    }
+    spaceBillingInstructions.push(removeResult.ok)
+  }
+
+  await all(spaceBillingInstructions.map(instruction => async () => {
+    const usage = expect(await SpaceBillingQueue.calculatePeriodUsage(instruction, {
+      spaceDiffStore,
+      spaceSnapshotStore: readableSpaceSnapshotStore
+    }))
+
+    expect(await SpaceBillingQueue.storeSpaceUsage(instruction, usage, {
+      spaceSnapshotStore: writableSpaceSnapshotStore,
+      usageStore
+    }))
+  }), { concurrency })
+}), { concurrency })
+
+console.log(`✅ Billing run completed successfully`)
+
+const { results } = expect(await usageStore.list({}))
+
+await fs.promises.writeFile(
+  `./usage-${from.toISOString()}-${to.toISOString()}.json`,
+  JSON.stringify(results.map(r => Usage.encode(r).ok))
+)
+
+/** @type {Map<string, import('../lib/api.js').Usage[]>} */
+const usageByCustomer = new Map()
+for (const usage of results) {
+  let customerUsages = usageByCustomer.get(usage.customer)
+  if (!customerUsages) {
+    customerUsages = []
+    usageByCustomer.set(usage.customer, customerUsages)
+  }
+  customerUsages.push(usage)
+}
+
+/** @type {Array<[string, string, number]>} */
+const data = []
+const duration = to.getTime() - from.getTime()
+
+for (const [customer, usages] of usageByCustomer.entries()) {
+  let product
+  let totalUsage = 0n
+  for (const u of usages) {
+    product = product ?? u.product
+    totalUsage += u.usage
+  }
+  if (!product) throw new Error('missing product')
+  try {
+    data.push([customer, product, calculateCost(product, totalUsage, duration)])
+  } catch (err) {
+    console.warn(`failed to calculate cost for: ${customer}`, err)
+  }
+}
+data.sort((a, b) => b[2] - a[2])
+
+await fs.promises.writeFile(
+  `./summary-${from.toISOString()}-${to.toISOString()}.csv`,
+  CSV.stringify(data, { header: true, columns: ['Customer', 'Product', 'Total'] })
+)

billing/scripts/helpers.js

@@ -0,0 +1,66 @@
+import Big from 'big.js'
+import { StoreOperationFailure } from '../tables/lib.js'
+import { EndOfQueue } from '../test/helpers/queue.js'
+
+/**
+ * @template T
+ * @returns {import('../lib/api.js').QueueAdder<T> & import('../test/lib/api.js').QueueRemover<T>}
+ */
+export const createMemoryQueue = () => {
+  /** @type {T[]} */
+  const items = []
+  return {
+    add: async (message) => {
+      items.push(message)
+      return { ok: {} }
+    },
+    remove: async () => {
+      const item = items.shift()
+      return item ? { ok: item } : { error: new EndOfQueue() }
+    }
+  }
+}
+
+/**
+ * @template T
+ * @returns {import('../lib/api.js').StorePutter<T> & import('../lib/api.js').StoreLister<any, T> & import('../lib/api.js').StoreGetter<any, T>}
+ */
+export const createMemoryStore = () => {
+  /** @type {T[]} */
+  const items = []
+  return {
+    put: async (item) => {
+      items.push(item)
+      return { ok: {} }
+    },
+    get: async () => ({ error: new StoreOperationFailure('not implemented') }),
+    list: async () => ({ ok: { results: items } })
+  }
+}
+
+const GB = 1024 * 1024 * 1024
+const TB = 1024 * GB
+
+/** @type {Record<string, { cost: number, overage: number, included: number }>} */
+const productInfo = {
+  'did:web:starter.web3.storage': { cost: 0, overage: 0.15 / GB, included: 5 * GB },
+  'did:web:lite.web3.storage': { cost: 10, overage: 0.05 / GB, included: 100 * GB },
+  'did:web:business.web3.storage': { cost: 100, overage: 0.03 / GB, included: 2 * TB }
+}
+
+/**
+ * @param {string} product
+ * @param {bigint} usage Usage in bytes/ms
+ * @param {number} duration Duration in ms
+ */
+export const calculateCost = (product, usage, duration) => {
+  const info = productInfo[product]
+  if (!info) throw new Error(`missing product info: ${product}`)
+
+  let overageBytes = new Big(usage.toString()).div(duration).minus(info.included)
+  if (overageBytes.lt(0)) {
+    overageBytes = new Big(0)
+  }
+  const overageCost = overageBytes.mul(info.overage).toNumber()
+  return info.cost + overageCost
+}