feat: ucan stream consumer provider add count

stacks/ucan-invocation-stack.js

@@ -123,6 +123,17 @@ export function UcanInvocationStack({ stack, app }) {
     handler: 'functions/metrics-upload-remove-total.consumer',
     deadLetterQueue: metricsUploadRemoveTotalDLQ.cdk.queue,
   })
+
+  // metrics provider/add count
+  const metricsProviderAddTotalDLQ = new Queue(stack, 'metrics-provider-add-total-dlq')
+  const metricsProviderAddTotalConsumer = new Function(stack, 'metrics-provider-add-total-consumer', {
+    environment: {
+      TABLE_NAME: adminMetricsTable.tableName
+    },
+    permissions: [adminMetricsTable],
+    handler: 'functions/metrics-provider-add-total.consumer',
+    deadLetterQueue: metricsProviderAddTotalDLQ.cdk.queue,
+  })
 
   // metrics per space
   const spaceMetricsDLQ = new Queue(stack, 'space-metrics-dlq')
@@ -258,6 +269,14 @@ export function UcanInvocationStack({ stack, app }) {
           }
         }
       },
+      metricsConsumerAddTotalConsumer: {
+        function: metricsProviderAddTotalConsumer,
+        cdk: {
+          eventSource: {
+            ...(getKinesisEventSourceConfig(stack))
+          }
+        }
+      },
       spaceMetricsUploadAddTotalConsumer: {
         function: spaceMetricsUploadAddTotalConsumer,
         cdk: {

ucan-invocation/constants.js

@@ -1,3 +1,6 @@
+import {
+  add as providerAdd,
+} from '@web3-storage/capabilities/provider'
 import { 
   add as storeAdd,
   remove as storeRemove
@@ -17,6 +20,7 @@ export const STORE_ADD = storeAdd.can
 export const STORE_REMOVE = storeRemove.can
 export const UPLOAD_ADD = uploadAdd.can
 export const UPLOAD_REMOVE = uploadRemove.can
+export const PROVIDER_ADD = providerAdd.can
 
 // Admin Metrics
 export const METRICS_NAMES = {
@@ -26,6 +30,7 @@ export const METRICS_NAMES = {
   STORE_ADD_SIZE_TOTAL: `${STORE_ADD}-size-total`,
   STORE_REMOVE_TOTAL: `${STORE_REMOVE}-total`,
   STORE_REMOVE_SIZE_TOTAL: `${STORE_REMOVE}-size-total`,
+  PROVIDER_ADD_TOTAL: `${PROVIDER_ADD}-total`,
 }
 
 // Spade Metrics

ucan-invocation/functions/metrics-provider-add-total.js

@@ -0,0 +1,47 @@
+import * as Sentry from '@sentry/serverless'
+
+import { hasOkReceipt } from '../utils/receipt.js'
+import { createMetricsTable } from '../tables/metrics.js'
+import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
+import { PROVIDER_ADD } from '../constants.js'
+
+Sentry.AWSLambda.init({
+  environment: process.env.SST_STAGE,
+  dsn: process.env.SENTRY_DSN,
+  tracesSampleRate: 1.0,
+})
+
+const AWS_REGION = process.env.AWS_REGION || 'us-west-2'
+
+/**
+ * @param {import('aws-lambda').KinesisStreamEvent} event
+ */
+async function handler(event) {
+  const ucanInvocations = parseKinesisEvent(event)
+
+  const {
+    TABLE_NAME: tableName = '',
+    // set for testing
+    DYNAMO_DB_ENDPOINT: dbEndpoint,
+  } = process.env
+
+  await updateProviderAddTotal(ucanInvocations, {
+    metricsTable: createMetricsTable(AWS_REGION, tableName, {
+      endpoint: dbEndpoint
+    })
+  })
+}
+
+/**
+ * @param {import('../types').UcanInvocation[]} ucanInvocations
+ * @param {import('../types').TotalSizeCtx} ctx
+ */
+export async function updateProviderAddTotal (ucanInvocations, ctx) {
+  const invocationsWithConsumerAdd = ucanInvocations.filter(
+    inv => inv.value.att.find(a => a.can === PROVIDER_ADD) && hasOkReceipt(inv)
+  ).flatMap(inv => inv.value.att)
+
+  await ctx.metricsTable.incrementProviderAddTotal(invocationsWithConsumerAdd)
+}
+
+export const consumer = Sentry.AWSLambda.wrapHandler(handler)

ucan-invocation/functions/metrics-upload-add-total.js

@@ -44,4 +44,4 @@ export async function updateUploadAddTotal (ucanInvocations, ctx) {
   await ctx.metricsTable.incrementUploadAddTotal(invocationsWithUploadAdd)
 }
 
-export const consumer = Sentry.AWSLambda.wrapHandler(handler)
+export const consumer = Sentry.AWSLambda.wrapHandler(handler)

ucan-invocation/package.json

@@ -9,7 +9,7 @@
     "@aws-sdk/client-dynamodb": "^3.226.0",
     "@aws-sdk/client-eventbridge": "^3.218.0",
     "@sentry/serverless": "^7.22.0",
-    "@web3-storage/capabilities": "^3.2.0",
+    "@web3-storage/capabilities": "https://gitpkg.now.sh/web3-storage/w3protocol/packages/capabilities?feat/auto-provider",
     "uint8arrays": "^4.0.2"
   },
   "devDependencies": {

ucan-invocation/tables/metrics.js

@@ -163,6 +163,28 @@ export function createMetricsTable (region, tableName, options = {}) {
         })
       })
 
+      await dynamoDb.send(updateCmd)
+    },
+    /**
+     * Increment total count from provider/add operations.
+     *
+     * @param {Capabilities} operationsInv
+     */
+    incrementProviderAddTotal: async (operationsInv) => {
+      const invTotalSize = operationsInv.length
+
+      const updateCmd = new UpdateItemCommand({
+        TableName: tableName,
+        UpdateExpression: `ADD #value :value`,
+          ExpressionAttributeNames: {'#value': 'value'},
+          ExpressionAttributeValues: {
+            ':value': { N: String(invTotalSize) },
+          },
+        Key: marshall({
+          name: METRICS_NAMES.PROVIDER_ADD_TOTAL
+        })
+      })
+
       await dynamoDb.send(updateCmd)
     }
   }

ucan-invocation/types.ts

@@ -9,6 +9,7 @@ export interface MetricsTable {
   incrementStoreRemoveSizeTotal: (incrementSizeTotal: Capability<Ability, `${string}:${string}`, unknown>[]) => Promise<void>
   incrementUploadAddTotal: (incrementSizeTotal: Capability<Ability, `${string}:${string}`, unknown>[]) => Promise<void>
   incrementUploadRemoveTotal: (incrementSizeTotal: Capability<Ability, `${string}:${string}`, unknown>[]) => Promise<void>
+  incrementProviderAddTotal: (incrementSizeTotal: Capability<Ability, `${string}:${string}`, unknown>[]) => Promise<void>
 }
 
 export interface CarStoreBucket {