chore: address review comments

stacks/ucan-invocation-stack.js

@@ -28,7 +28,7 @@ export function UcanInvocationStack({ stack, app }) {
 
   // Get eventBus reference
   const { eventBus } = use(BusStack)
-  const { w3MetricsTable } = use(UploadDbStack)
+  const { adminMetricsTable } = use(UploadDbStack)
 
   const ucanBucket = new Bucket(stack, 'ucan-store', {
     cors: true,
@@ -52,14 +52,14 @@ export function UcanInvocationStack({ stack, app }) {
     }
   })
 
-  const metricsAccumulatedSizeDLQ = new Queue(stack, 'metrics-accumulated-size-dlq')
-  const metricsAccumulatedSizeConsumer = new Function(stack, 'metrics-accumulated-size-consumer', {
+  const metricsSizeTotalDLQ = new Queue(stack, 'metrics-size-total-dlq')
+  const metricsSizeTotalConsumer = new Function(stack, 'metrics-size-total-consumer', {
     environment: {
-      TABLE_NAME: w3MetricsTable.tableName
+      TABLE_NAME: adminMetricsTable.tableName
     },
-    permissions: [w3MetricsTable],
-    handler: 'functions/metrics-accumulated-size.consumer',
-    deadLetterQueue: metricsAccumulatedSizeDLQ.cdk.queue,
+    permissions: [adminMetricsTable],
+    handler: 'functions/metrics-size-total.consumer',
+    deadLetterQueue: metricsSizeTotalDLQ.cdk.queue,
   })
 
   // create a kinesis stream
@@ -71,8 +71,8 @@ export function UcanInvocationStack({ stack, app }) {
     },
     consumers: {
       // consumer1: 'functions/consumer1.handler'
-      metricsAccumulatedSizeConsumer: {
-        function: metricsAccumulatedSizeConsumer,
+      metricsSizeTotalConsumer: {
+        function: metricsSizeTotalConsumer,
         // TODO: Set kinesis filters when supported by SST
         // https://github.com/serverless-stack/sst/issues/1407
         cdk: {

stacks/upload-db-stack.js

@@ -2,7 +2,7 @@ import { Table } from '@serverless-stack/resources'
 
 import { storeTableProps, uploadTableProps } from '../upload-api/tables/index.js'
 import {
-  w3MetricsTableProps
+  adminMetricsTableProps
 } from '../ucan-invocation/tables/index.js'
 import { setupSentry } from './config.js'
 
@@ -29,11 +29,11 @@ export function UploadDbStack({ stack, app }) {
   /**
    * This table tracks w3 wider metrics.
    */
-  const w3MetricsTable = new Table(stack, 'w3-metrics', w3MetricsTableProps)
+  const adminMetricsTable = new Table(stack, 'metrics', adminMetricsTableProps)
 
   return {
     storeTable,
     uploadTable,
-    w3MetricsTable
+    adminMetricsTable
   }
 }

test/helpers/context.js

@@ -14,7 +14,7 @@ dotenv.config({
  *
  * @typedef {object} Context
  * @property {string} apiEndpoint
- * @property {Dynamo} w3MetricsDynamo
+ * @property {Dynamo} metricsDynamo
  *
  * @typedef {import("ava").TestFn<Awaited<Context>>} TestContextFn
  */

test/upload-api.test.js → test/integration.test.js

@@ -3,7 +3,7 @@ import git from 'git-rev-sync'
 import pWaitFor from 'p-wait-for'
 import { HeadObjectCommand } from '@aws-sdk/client-s3'
 
-import { W3_METRICS_NAMES } from '../ucan-invocation/constants.js'
+import { METRICS_NAMES } from '../ucan-invocation/constants.js'
 import { test } from './helpers/context.js'
 import {
   stage,
@@ -21,17 +21,17 @@ import { getAllTableRows } from './helpers/table.js'
 test.before(t => {
   t.context = {
     apiEndpoint: getApiEndpoint(),
-    w3MetricsDynamo: getDynamoDb('w3-metrics')
+    metricsDynamo: getDynamoDb('metrics')
   }
 })
 
-test('GET /', async t => {
+test('upload-api GET /', async t => {
   const apiEndpoint = getApiEndpoint()
   const response = await fetch(apiEndpoint)
   t.is(response.status, 200)
 })
 
-test('GET /version', async t => {
+test('upload-api /version', async t => {
   const apiEndpoint = getApiEndpoint()
 
   const response = await fetch(`${apiEndpoint}/version`)
@@ -42,12 +42,13 @@ test('GET /version', async t => {
   t.is(body.commit, git.long('.'))
 })
 
-test('POST / client can upload a file and list it', async t => {
+// Integration test for all flow from uploading a file to Kinesis events consumers and replicator
+test('w3infra integration flow', async t => {
   const client = await getClient(t.context.apiEndpoint)
 
   // Get metrics before upload
-  const beforeOperationW3Metrics = await getW3Metrics(t)
-  const beforeW3AccumulatedSize = beforeOperationW3Metrics.find(row => row.name === W3_METRICS_NAMES.STORE_ADD_ACCUM_SIZE)
+  const beforeOperationMetrics = await getMetrics(t)
+  const beforeAccumulatedSize = beforeOperationMetrics.find(row => row.name === METRICS_NAMES.STORE_ADD_ACCUM_SIZE)
 
   const s3Client = getAwsBucketClient()
   const r2Client = getCloudflareBucketClient()
@@ -73,6 +74,8 @@ test('POST / client can upload a file and list it', async t => {
   )
   t.is(carparkRequest.$metadata.httpStatusCode, 200)
 
+  const carSize = carparkRequest.ContentLength
+
   // Check dudewhere
   const dudewhereRequest = await r2Client.send(
     new HeadObjectCommand({
@@ -153,22 +156,27 @@ test('POST / client can upload a file and list it', async t => {
   })
 
   // Check metrics were updated
-  beforeW3AccumulatedSize && await pWaitFor(async () => {
-    const afterOperationW3Metrics = await getW3Metrics(t)
-    const afterW3AccumulatedSize = afterOperationW3Metrics.find(row => row.name === W3_METRICS_NAMES.STORE_ADD_ACCUM_SIZE)
+  beforeAccumulatedSize && await pWaitFor(async () => {
+    const afterOperationMetrics = await getMetrics(t)
+    const afterAccumulatedSize = afterOperationMetrics.find(row => row.name === METRICS_NAMES.STORE_ADD_ACCUM_SIZE)
+
+    // If staging accept more broad condition given multiple parallel tests can happen there
+    if (stage === 'staging') {
+      return afterAccumulatedSize?.value > beforeAccumulatedSize.value
+    }
 
-    return afterW3AccumulatedSize?.value > beforeW3AccumulatedSize.value
+    return afterAccumulatedSize?.value === beforeAccumulatedSize.value + carSize
   })
 })
 
 /**
  * @param {import("ava").ExecutionContext<import("./helpers/context.js").Context>} t
  */
-async function getW3Metrics (t) {
-  const w3Metrics = await getAllTableRows(
-    t.context.w3MetricsDynamo.client,
-    t.context.w3MetricsDynamo.tableName
+async function getMetrics (t) {
+  const metrics = await getAllTableRows(
+    t.context.metricsDynamo.client,
+    t.context.metricsDynamo.tableName
   )
 
-  return w3Metrics
+  return metrics
 }

ucan-invocation/constants.js

@@ -2,6 +2,6 @@
 export const STORE_ADD = 'store/add'
 
 // Metrics
-export const W3_METRICS_NAMES = {
-  STORE_ADD_ACCUM_SIZE: `${STORE_ADD}-accumulated-size`
+export const METRICS_NAMES = {
+  STORE_ADD_SIZE_TOTAL: `${STORE_ADD}-size-total`
 }

ucan-invocation/functions/metrics-accumulated-size.js → ucan-invocation/functions/metrics-size-total.js

@@ -1,6 +1,6 @@
 import * as Sentry from '@sentry/serverless'
 
-import { createW3MetricsTable } from '../tables/w3-metrics.js'
+import { createMetricsTable } from '../tables/metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
 import { STORE_ADD } from '../constants.js'
 
@@ -24,23 +24,23 @@ async function handler(event) {
     DYNAMO_DB_ENDPOINT: dbEndpoint,
   } = process.env
 
-  await updateAccumulatedSize(ucanInvocations, {
-    w3MetricsTable: createW3MetricsTable(AWS_REGION, tableName, {
+  await updateSizeTotal(ucanInvocations, {
+    metricsTable: createMetricsTable(AWS_REGION, tableName, {
       endpoint: dbEndpoint
     })
   })
 }
 
 /**
  * @param {import('../types').UcanInvocation[]} ucanInvocations
- * @param {import('../types').W3AccumulatedSizeCtx} ctx
+ * @param {import('../types').TotalSizeCtx} ctx
  */
-export async function updateAccumulatedSize (ucanInvocations, ctx) {
+export async function updateSizeTotal (ucanInvocations, ctx) {
   const invocationsWithStoreAdd = ucanInvocations.filter(
     inv => inv.value.att.find(a => a.can === STORE_ADD)
   ).flatMap(inv => inv.value.att)
 
-  await ctx.w3MetricsTable.incrementAccumulatedSize(invocationsWithStoreAdd)
+  await ctx.metricsTable.incrementStoreAddSizeTotal(invocationsWithStoreAdd)
 }
 
 export const consumer = Sentry.AWSLambda.wrapHandler(handler)

ucan-invocation/tables/index.js

@@ -1,7 +1,7 @@
 /** @typedef {import('@serverless-stack/resources').TableProps} TableProps */
 
 /** @type TableProps */
-export const w3MetricsTableProps = {
+export const adminMetricsTableProps = {
   fields: {
     name: 'string',        // `total-size`
     value: 'number',         // `101`

ucan-invocation/tables/w3-metrics.js → ucan-invocation/tables/metrics.js

@@ -4,7 +4,7 @@ import {
 } from '@aws-sdk/client-dynamodb'
 import { marshall } from '@aws-sdk/util-dynamodb'
 
-import { W3_METRICS_NAMES } from '../constants.js'
+import { METRICS_NAMES } from '../constants.js'
 
 /**
  * @typedef {import('@ucanto/interface').Ability} Ability
@@ -22,21 +22,21 @@ import { W3_METRICS_NAMES } from '../constants.js'
  * @param {string} tableName
  * @param {object} [options]
  * @param {string} [options.endpoint]
- * @returns {import('../types').W3MetricsTable}
+ * @returns {import('../types').MetricsTable}
  */
-export function createW3MetricsTable (region, tableName, options = {}) {
+export function createMetricsTable (region, tableName, options = {}) {
   const dynamoDb = new DynamoDBClient({
     region,
     endpoint: options.endpoint
   })
 
   return {
     /**
-     * Increment accumulated value from new given operations.
+     * Increment total value from new given operations.
      *
      * @param {Capabilities} operationsInv
      */
-    incrementAccumulatedSize: async (operationsInv) => {
+    incrementStoreAddSizeTotal: async (operationsInv) => {
       // @ts-expect-error
       const invTotalSize = operationsInv.reduce((acc, c) => acc + c.nb?.size, 0)
 
@@ -48,7 +48,7 @@ export function createW3MetricsTable (region, tableName, options = {}) {
             ':value': { N: String(invTotalSize) },
           },
         Key: marshall({
-          name: W3_METRICS_NAMES.STORE_ADD_ACCUM_SIZE
+          name: METRICS_NAMES.STORE_ADD_SIZE_TOTAL
         })
       })
 

ucan-invocation/test/functions/metrics-accumulated-size.test.js → ucan-invocation/test/functions/metrics-size-total.test.js

@@ -6,14 +6,14 @@ import { marshall, unmarshall } from '@aws-sdk/util-dynamodb'
 import * as Signer from '@ucanto/principal/ed25519'
 import * as StoreCapabilities from '@web3-storage/capabilities/store'
 
-import { w3MetricsTableProps } from '../../tables/index.js'
+import { adminMetricsTableProps } from '../../tables/index.js'
 import { createDynamodDb, dynamoDBTableConfig } from '../helpers/resources.js'
 import { createSpace } from '../helpers/ucanto.js'
 import { randomCAR } from '../helpers/random.js'
 
-import { updateAccumulatedSize } from '../../functions/metrics-accumulated-size.js'
-import { createW3MetricsTable } from '../../tables/w3-metrics.js'
-import { W3_METRICS_NAMES } from '../../constants.js'
+import { updateSizeTotal } from '../../functions/metrics-size-total.js'
+import { createMetricsTable } from '../../tables/metrics.js'
+import { METRICS_NAMES } from '../../constants.js'
 
 const REGION = 'us-west-2'
 
@@ -35,7 +35,7 @@ test('handles a batch of single invocation with store/add', async t => {
   const { spaceDid } = await createSpace(alice)
   const car = await randomCAR(128)
 
-  const w3MetricsTable = createW3MetricsTable(REGION, tableName, {
+  const metricsTable = createMetricsTable(REGION, tableName, {
     endpoint: t.context.dbEndpoint
   })
 
@@ -58,13 +58,13 @@ test('handles a batch of single invocation with store/add', async t => {
   }]
 
   // @ts-expect-error
-  await updateAccumulatedSize(invocations, {
-    w3MetricsTable
+  await updateSizeTotal(invocations, {
+    metricsTable
   })
 
-  const item = await getItemFromTable(t.context.dynamoClient, tableName, W3_METRICS_NAMES.STORE_ADD_ACCUM_SIZE)
+  const item = await getItemFromTable(t.context.dynamoClient, tableName, METRICS_NAMES.STORE_ADD_SIZE_TOTAL)
   t.truthy(item)
-  t.is(item?.name, W3_METRICS_NAMES.STORE_ADD_ACCUM_SIZE)
+  t.is(item?.name, METRICS_NAMES.STORE_ADD_SIZE_TOTAL)
   t.is(item?.value, car.size)
 })
 
@@ -78,7 +78,7 @@ test('handles batch of single invocations with multiple store/add attributes', a
     Array.from({ length: 10 }).map(() => randomCAR(128))
   )
 
-  const w3MetricsTable = createW3MetricsTable(REGION, tableName, {
+  const metricsTable = createMetricsTable(REGION, tableName, {
     endpoint: t.context.dbEndpoint
   })
 
@@ -99,13 +99,13 @@ test('handles batch of single invocations with multiple store/add attributes', a
   }]
 
   // @ts-expect-error
-  await updateAccumulatedSize(invocations, {
-    w3MetricsTable
+  await updateSizeTotal(invocations, {
+    metricsTable
   })
 
-  const item = await getItemFromTable(t.context.dynamoClient, tableName, W3_METRICS_NAMES.STORE_ADD_ACCUM_SIZE)
+  const item = await getItemFromTable(t.context.dynamoClient, tableName, METRICS_NAMES.STORE_ADD_SIZE_TOTAL)
   t.truthy(item)
-  t.is(item?.name, W3_METRICS_NAMES.STORE_ADD_ACCUM_SIZE)
+  t.is(item?.name, METRICS_NAMES.STORE_ADD_SIZE_TOTAL)
   t.is(item?.value, cars.reduce((acc, c) => acc + c.size, 0))
 })
 
@@ -131,7 +131,7 @@ async function createDynamouploadTable(dynamo) {
 
   await dynamo.send(new CreateTableCommand({
     TableName: tableName,
-    ...dynamoDBTableConfig(w3MetricsTableProps),
+    ...dynamoDBTableConfig(adminMetricsTableProps),
     ProvisionedThroughput: {
       ReadCapacityUnits: 1,
       WriteCapacityUnits: 1

ucan-invocation/types.ts

@@ -2,12 +2,12 @@ import type { DID, Link } from '@ucanto/interface'
 import { ToString, UnknownLink } from 'multiformats'
 import { Ability, Capability, Capabilities } from '@ucanto/interface'
 
-export interface W3MetricsTable {
-  incrementAccumulatedSize: (incrementTotalSize: Capability<Ability, `${string}:${string}`, unknown>[]) => Promise<void>
+export interface MetricsTable {
+  incrementStoreAddSizeTotal: (incrementSizeTotal: Capability<Ability, `${string}:${string}`, unknown>[]) => Promise<void>
 }
 
-export interface W3AccumulatedSizeCtx {
-  w3MetricsTable: W3MetricsTable
+export interface TotalSizeCtx {
+  metricsTable: MetricsTable
 }
 
 export interface UcanInvocation {