feat: add script to create space allocations snapshot (#451)

## Description

This PR addresses the issue
storacha/project-tracking#183.

This PR introduces the `billing/scripts/space-allocations-snapshot`
script, which utilizes the `allocation` and `store` tables to calculate
customer storage size and provide an estimate of their overall usage.

Please note that this method is not entirely accurate, as it does not
account for deleted storage.

For detailed instructions on using the script, refer to the
`billing/scripts/space-allocations-snapshot/README.md` file.

billing/data/allocations.js

@@ -0,0 +1,103 @@
+import * as Link from 'multiformats/link'
+import { DecodeFailure, EncodeFailure, Schema } from './lib.js'
+
+/**
+ * @typedef {import('../lib/api').Allocation} Allocation
+ * @typedef {import('../lib/api').AllocationSpaceInsertedAtIndex} AllocationSpaceInsertedAtIndex
+ * @typedef {import('../types').InferStoreRecord<Allocation>} AllocationStoreRecord
+ * @typedef {import('../lib/api').AllocationKey} AllocationKey
+ * @typedef {import('../lib/api').AllocationListKey} AllocationListKey
+ * @typedef {import('../types').InferStoreRecord<AllocationKey>} AllocationKeyStoreRecord
+ * @typedef {{ space: string, insertedAt?: string }} AllocationListStoreRecord
+ * @typedef {import('../types').StoreRecord} StoreRecord
+ */
+
+const schema = Schema.struct({
+  space: Schema.did(),
+  multihash: Schema.text(),
+  cause: Schema.link({ version: 1 }),
+  insertedAt: Schema.date(),
+  size: Schema.bigint().greaterThanEqualTo(0n),
+})
+
+/** @type {import('../lib/api').Validator<Allocation>} */
+export const validate = (input) => schema.read(input)
+
+/** @type {import('../lib/api').Encoder<AllocationKey, AllocationKeyStoreRecord>} */
+export const encodeKey = (input) => ({ ok: { multihash: input.multihash } })
+
+/** @type {import('../lib/api').Encoder<Allocation, AllocationStoreRecord>} */
+export const encode = (input) => {
+  try {
+    return {
+      ok: {
+        space: input.space.toString(),
+        multihash: input.multihash,
+        cause: input.cause.toString(),
+        insertedAt: input.insertedAt.toISOString(),
+        size: input.size.toString(),
+      },
+    }
+  } catch (/** @type {any} */ err) {
+    return {
+      error: new EncodeFailure(`encoding allocation record: ${err.message}`, {
+        cause: err,
+      }),
+    }
+  }
+}
+
+/** @type {import('../lib/api').Decoder<StoreRecord, Allocation>} */
+export const decode = (input) => {
+  try {
+    return {
+      ok: {
+        space: Schema.did().from(input.space),
+        multihash: /** @type {string} */ (input.multihash),
+        cause: Link.parse(/** @type {string} */ (input.cause)),
+        insertedAt: new Date(input.insertedAt),
+        size: BigInt(input.size),
+      },
+    }
+  } catch (/** @type {any} */ err) {
+    return {
+      error: new DecodeFailure(`decoding allocation record: ${err.message}`, {
+        cause: err,
+      }),
+    }
+  }
+}
+
+export const lister = {
+  /** @type {import('../lib/api').Encoder<AllocationListKey, AllocationListStoreRecord>} */
+  encodeKey: (input) => {
+    /** @type  AllocationListStoreRecord */
+    const conditions = { space: input.space.toString() }
+    if (input.insertedAt) {
+      conditions.insertedAt = input.insertedAt.toISOString()
+    }
+    return {
+      ok: {
+        ...conditions,
+      },
+    }
+  },
+  /** @type {import('../lib/api').Decoder<StoreRecord, AllocationSpaceInsertedAtIndex>} */
+  decode: (input) => {
+    try {
+      return {
+        ok: {
+          space: Schema.did().from(input.space),
+          insertedAt: new Date(input.insertedAt),
+          size: BigInt(input.size),
+        },
+      }
+    } catch (/** @type {any} */ err) {
+      return {
+        error: new DecodeFailure(`decoding allocation record: ${err.message}`, {
+          cause: err,
+        }),
+      }
+    }
+  },
+}

billing/data/store.js

@@ -0,0 +1,106 @@
+import * as Link from 'multiformats/link'
+import { DecodeFailure, EncodeFailure, Schema } from './lib.js'
+
+/**
+ * @typedef {import('../lib/api.js').StoreTable} StoreTable
+ * @typedef {import('../lib/api').StoreTableSpaceInsertedAtIndex} StoreTableSpaceInsertedAtIndex
+ * @typedef {import('../types.js').InferStoreRecord<StoreTable>} StoreTableStoreRecord
+ * @typedef {import('../lib/api.js').StoreTableKey} StoreTableKey
+ * @typedef {import('../lib/api.js').StoreTableListKey} StoreTableListKey
+ * @typedef {import('../types.js').InferStoreRecord<StoreTableKey>} StoreTableKeyStoreRecord
+ * @typedef {{ space: string, insertedAt?: string}} StoreTableListStoreRecord
+ * @typedef {import('../types.js').StoreRecord} StoreRecord
+ */
+
+const schema = Schema.struct({
+  space: Schema.did(),
+  link: Schema.link({ version: 1 }),
+  invocation: Schema.link({ version: 1 }),
+  insertedAt: Schema.date(),
+  size: Schema.bigint().greaterThanEqualTo(0n),
+  issuer: Schema.did().optional(),
+})
+
+/** @type {import('../lib/api.js').Validator<StoreTable>} */
+export const validate = (input) => schema.read(input)
+
+/** @type {import('../lib/api.js').Encoder<StoreTableKey, StoreTableKeyStoreRecord>} */
+export const encodeKey = (input) => ({ ok: { link: input.link } })
+
+/** @type {import('../lib/api.js').Encoder<StoreTable, StoreTableStoreRecord>} */
+export const encode = (input) => {
+  try {
+    return {
+      ok: {
+        space: input.space.toString(),
+        link: input.link.toString(),
+        invocation: input.invocation.toString(),
+        insertedAt: input.insertedAt.toISOString(),
+        size: input.size.toString(),
+        issuer: input.issuer?.toString(),
+      },
+    }
+  } catch (/** @type {any} */ err) {
+    return {
+      error: new EncodeFailure(`encoding store record: ${err.message}`, {
+        cause: err,
+      }),
+    }
+  }
+}
+
+/** @type {import('../lib/api.js').Decoder<StoreRecord, StoreTable>} */
+export const decode = (input) => {
+  try {
+    return {
+      ok: {
+        space: Schema.did().from(input.space),
+        link: Link.parse(/** @type {string} */ (input.link)),
+        invocation: Link.parse(/** @type {string} */ (input.invocation)),
+        insertedAt: new Date(input.insertedAt),
+        size: BigInt(input.size),
+        issuer: Schema.did().from(input.issuer),
+      },
+    }
+  } catch (/** @type {any} */ err) {
+    return {
+      error: new DecodeFailure(`decoding store record: ${err.message}`, {
+        cause: err,
+      }),
+    }
+  }
+}
+
+export const lister = {
+  /** @type {import('../lib/api.js').Encoder<StoreTableListKey, StoreTableListStoreRecord>} */
+  encodeKey: (input) => {
+    /** @type  StoreTableListStoreRecord */
+    const conditions = { space: input.space.toString() }
+    if (input.insertedAt) {
+      conditions.insertedAt = input.insertedAt.toISOString()
+    }
+    return {
+      ok: {
+        ...conditions,
+      },
+    }
+  },
+  /** @type {import('../lib/api.js').Decoder<StoreRecord, StoreTableSpaceInsertedAtIndex>} */
+  decode: (input) => {
+    try {
+      return {
+        ok: {
+          space: Schema.did().from(input.space),
+          insertedAt: new Date(input.insertedAt),
+          size: BigInt(input.size),
+        },
+      }
+    } catch (/** @type {any} */ err) {
+      return {
+        error: new DecodeFailure(`decoding allocation record: ${err.message}`, {
+          cause: err,
+        }),
+      }
+    }
+  },
+}

billing/lib/api.ts

@@ -1,4 +1,5 @@
 import { DID, Link, URI, LinkJSON, Result, Capabilities, Unit, Failure, UnknownLink } from '@ucanto/interface'
+import { StoreRecord } from '../types'
 
 // Billing stores /////////////////////////////////////////////////////////////
 
@@ -138,6 +139,65 @@ export type UsageStore = StorePutter<Usage>
  */
 export type EgressTrafficEventStore = StorePutter<EgressTrafficData> & StoreLister<EgressTrafficEventListKey, EgressTrafficData>
 
+export interface Allocation {
+  /** Space DID (did:key:...). */
+  space: ConsumerDID
+  /** Represents a multihash digest which carries information about the hashing algorithm and an actual hash digest. */
+  multihash: string
+  /** UCAN invocation that caused the size change. */
+  cause: Link
+  /** Time the record was added to the database. */
+  insertedAt: Date
+  /** Number of bytes that were added to the space. */
+  size: bigint
+}
+
+export type AllocationSpaceInsertedAtIndex = Omit< Allocation, "multihash" | "cause" >
+export interface AllocationKey { multihash: string }
+export interface AllocationListKey { space: ConsumerDID, insertedAt?: Date }
+
+export type AllocationStore =
+  & StoreGetter<AllocationKey, Allocation>
+  & StoreLister<AllocationListKey, AllocationSpaceInsertedAtIndex>
+  & {
+    listBetween: (space: DID, from: Date, to: Date, options?: Pageable) => Promise<Result<ListSuccess<AllocationSpaceInsertedAtIndex>, EncodeFailure|DecodeFailure|StoreOperationFailure>>
+  }
+
+export interface AllocationSnapshot {
+  [customerDID: CustomerDID] : {
+    spaceAllocations: Array<{[spaceDID: ConsumerDID]: {size: bigint, usage: bigint}}>
+    totalAllocation: bigint,
+    totalUsage: bigint,
+    product: string,
+    provider: ProviderDID
+    recordedAt: Date
+  }
+}
+
+export interface StoreTable {
+  /** Space DID (did:key:...). */
+  space: ConsumerDID
+  link: Link // TODO: should this be CARLink? how to validate using Schema?
+  /** UCAN invocation that caused the size change. */
+  invocation: Link
+  /** Time the record was added to the database. */
+  insertedAt: Date
+  /** Number of bytes that were added to the space. */
+  size: bigint
+  issuer?: DID
+}
+
+export type StoreTableSpaceInsertedAtIndex = Omit< StoreTable, "invocation" | "link" | "issuer" >
+export interface StoreTableKey { link: string }
+export interface StoreTableListKey { space: ConsumerDID, insertedAt?: Date }
+
+export type StoreTableStore = 
+  & StoreGetter<StoreTableKey, StoreTable>
+  & StoreLister<StoreTableListKey, StoreTableSpaceInsertedAtIndex>  
+  & {
+    listBetween: (space: DID, from: Date, to: Date, options?: Pageable) => Promise<Result<ListSuccess<StoreTableSpaceInsertedAtIndex>, EncodeFailure|DecodeFailure|StoreOperationFailure>>
+  }
+
 // Billing queues /////////////////////////////////////////////////////////////
 
 /**
@@ -375,3 +435,9 @@ export interface QueueAdder<T> {
   /** Adds a message to the end of the queue. */
   add: (message: T) => Promise<Result<Unit, EncodeFailure|QueueOperationFailure|Failure>>
 }
+export interface CreateStoreListerContext<K,V> {
+  tableName: string
+  encodeKey: Encoder<K, StoreRecord>
+  decode: Decoder<StoreRecord, V>
+  indexName?: string
+}

billing/lib/space-billing-queue.js

@@ -1,6 +1,5 @@
 import Big from 'big.js'
-
-const GB = 1024 * 1024 * 1024
+import {GB} from './util.js'
 
 /**
  * @param {import('./api').SpaceDiffListKey & { to: Date }} params
@@ -117,3 +116,60 @@ export const storeSpaceUsage = async (instruction, { size, usage }, ctx) => {
 
   return { ok: {} }
 }
+
+/**
+ * Calculates the total allocation for the specified space.  
+ * Additionally, it estimates the usage for the space based on the allocation/store values.  
+ * Note: This value is approximate as it doesn’t account for deleted files.
+ *
+ * @typedef {import('./api').AllocationStore} AllocationStore
+ * @typedef {import('./api').StoreTableStore} StoreTableStore
+ * @typedef {{allocationStore: AllocationStore}} AllocationStoreCtx
+ * @typedef {{storeTableStore: StoreTableStore}} StoreTableStoreCtx
+ * 
+ * @param {"allocationStore" | "storeTableStore"} store
+ * @param {import('./api').SpaceBillingInstruction} instruction
+ * @param { AllocationStoreCtx | StoreTableStoreCtx} ctx
+ * @returns {Promise<import('@ucanto/interface').Result<{ size: bigint , usage: bigint}>>}
+ */
+export const calculateSpaceAllocation = async (store, instruction, ctx) => {
+  console.log(`Calculating total allocation for: ${instruction.space}`)
+  console.log(`Provider: ${instruction.provider}`)
+  console.log(`Customer: ${instruction.customer}`)
+  console.log(`Period: ${instruction.from.toISOString()} - ${instruction.to.toISOString()}`)
+
+  /** @type AllocationStore | StoreTableStore */
+  const ctxStore = store === 'allocationStore' ? 
+  /** @type AllocationStoreCtx */ (ctx).allocationStore :
+  /** @type StoreTableStoreCtx */ (ctx).storeTableStore
+
+  /** @type {string|undefined} */
+  let cursor
+  let size = 0n
+  let usage = 0n
+  while(true){
+    const {ok: allocations, error} = await ctxStore.listBetween(
+      instruction.space, 
+      instruction.from, 
+      instruction.to,
+      {cursor, size: 100}
+    )
+
+    if (error) return { error }
+
+    for (const allocation of allocations.results){
+      size += allocation.size
+      usage += allocation.size * BigInt(instruction.to.getTime() - allocation.insertedAt.getTime())
+    }
+
+    if (!allocations.cursor) break
+    cursor = allocations.cursor
+  }
+
+  console.log(`Total allocation for ${instruction.space}: ${size} bytes`)
+  const duration = instruction.to.getTime() - instruction.from.getTime()
+  const usageGB = new Big(usage.toString()).div(duration).div(GB).toFixed(2)
+  console.log(`Approximate space consumed ${usage} byte/ms (~${usageGB} GiB/month)`)
+
+  return {ok: {size, usage}}
+}

billing/lib/util.js

@@ -1,3 +1,5 @@
+export const GB = 1024 * 1024 * 1024
+
 /** @param {string|number|Date} now */
 export const startOfMonth = (now) => {
   const d = new Date(now)

billing/package.json

@@ -26,6 +26,7 @@
     "@types/big.js": "^6.2.1",
     "aws-lambda": "^1.0.7",
     "c8": "^8.0.1",
+    "csv-parser": "^3.0.0",
     "csv-stringify": "^6.4.6",
     "dotenv": "^16.4.5",
     "entail": "^2.1.1",