feat: ucan stream handles receipts

stacks/ucan-invocation-stack.js

@@ -189,8 +189,17 @@ export function UcanInvocationStack({ stack, app }) {
     deadLetterQueue: spaceMetricsDLQ.cdk.queue,
   })
 
+  // TODO: keep for historical content that we might want to process
+  new KinesisStream(stack, 'ucan-stream', {
+    cdk: {
+      stream: {
+        retentionPeriod: Duration.days(365)
+      }
+    },
+  })
+
   // create a kinesis stream
-  const ucanStream = new KinesisStream(stack, 'ucan-stream', {
+  const ucanStream = new KinesisStream(stack, 'ucan-stream-v2', {
     cdk: {
       stream: {
         retentionPeriod: Duration.days(365)

ucan-invocation/constants.js

@@ -7,6 +7,11 @@ import {
   remove as uploadRemove
 } from '@web3-storage/capabilities/upload'
 
+export const CONTENT_TYPE = {
+  WORKFLOW: 'application/invocations+car',
+  RECEIPT: 'application/receipt+dag-cbor'
+}
+
 // UCAN protocol
 export const STORE_ADD = storeAdd.can
 export const STORE_REMOVE = storeRemove.can

ucan-invocation/functions/metrics-store-add-size-total.js

@@ -2,7 +2,7 @@ import * as Sentry from '@sentry/serverless'
 
 import { createMetricsTable } from '../tables/metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { STORE_ADD } from '../constants.js'
+import { STORE_ADD, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -37,7 +37,7 @@ async function handler(event) {
  */
 export async function updateSizeTotal (ucanInvocations, ctx) {
   const invocationsWithStoreAdd = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === STORE_ADD)
+    inv => inv.value.att.find(a => a.can === STORE_ADD) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   await ctx.metricsTable.incrementStoreAddSizeTotal(invocationsWithStoreAdd)

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

@@ -2,7 +2,7 @@ import * as Sentry from '@sentry/serverless'
 
 import { createMetricsTable } from '../tables/metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { STORE_ADD } from '../constants.js'
+import { STORE_ADD, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -37,7 +37,7 @@ async function handler(event) {
  */
 export async function updateStoreAddTotal (ucanInvocations, ctx) {
   const invocationsWithStoreAdd = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === STORE_ADD)
+    inv => inv.value.att.find(a => a.can === STORE_ADD) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   await ctx.metricsTable.incrementStoreAddTotal(invocationsWithStoreAdd)

ucan-invocation/functions/metrics-store-remove-size-total.js

@@ -3,7 +3,7 @@ import * as Sentry from '@sentry/serverless'
 import { createCarStore } from '../buckets/car-store.js'
 import { createMetricsTable } from '../tables/metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { STORE_REMOVE } from '../constants.js'
+import { STORE_REMOVE, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -36,7 +36,7 @@ async function handler(event) {
  */
 export async function updateRemoveSizeTotal (ucanInvocations, ctx) {
   const invocationsWithStoreRemove = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === STORE_REMOVE)
+    inv => inv.value.att.find(a => a.can === STORE_REMOVE) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   // TODO: once we have receipts for store/remove, replace this

ucan-invocation/functions/metrics-store-remove-total.js

@@ -2,7 +2,7 @@ import * as Sentry from '@sentry/serverless'
 
 import { createMetricsTable } from '../tables/metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { STORE_REMOVE } from '../constants.js'
+import { STORE_REMOVE, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -37,7 +37,7 @@ async function handler(event) {
  */
 export async function updateStoreRemoveTotal (ucanInvocations, ctx) {
   const invocationsWithStoreRemove = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === STORE_REMOVE)
+    inv => inv.value.att.find(a => a.can === STORE_REMOVE) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   await ctx.metricsTable.incrementStoreRemoveTotal(invocationsWithStoreRemove)

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

@@ -2,7 +2,7 @@ import * as Sentry from '@sentry/serverless'
 
 import { createMetricsTable } from '../tables/metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { UPLOAD_ADD } from '../constants.js'
+import { UPLOAD_ADD, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -37,7 +37,7 @@ async function handler(event) {
  */
 export async function updateUploadAddTotal (ucanInvocations, ctx) {
   const invocationsWithUploadAdd = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === UPLOAD_ADD)
+    inv => inv.value.att.find(a => a.can === UPLOAD_ADD) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   await ctx.metricsTable.incrementUploadAddTotal(invocationsWithUploadAdd)

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

@@ -2,7 +2,7 @@ import * as Sentry from '@sentry/serverless'
 
 import { createMetricsTable } from '../tables/metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { UPLOAD_REMOVE } from '../constants.js'
+import { UPLOAD_REMOVE, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -37,7 +37,7 @@ async function handler(event) {
  */
 export async function updateUploadRemoveTotal (ucanInvocations, ctx) {
   const invocationsWithUploadRemove = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === UPLOAD_REMOVE)
+    inv => inv.value.att.find(a => a.can === UPLOAD_REMOVE) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   await ctx.metricsTable.incrementUploadRemoveTotal(invocationsWithUploadRemove)

ucan-invocation/functions/space-metrics-store-add-size-total.js

@@ -3,7 +3,7 @@ import * as Sentry from '@sentry/serverless'
 import { createCarStore } from '../buckets/car-store.js'
 import { createSpaceMetricsTable } from '../tables/space-metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { STORE_ADD } from '../constants.js'
+import { STORE_ADD, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -36,7 +36,7 @@ async function handler(event) {
  */
 export async function updateAddSizeTotal (ucanInvocations, ctx) {
   const invocationsWithStoreAdd = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === STORE_ADD)
+    inv => inv.value.att.find(a => a.can === STORE_ADD) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   await ctx.spaceMetricsTable.incrementStoreAddSizeTotal(invocationsWithStoreAdd)

ucan-invocation/functions/space-metrics-store-add-total.js

@@ -2,14 +2,14 @@ import * as Sentry from '@sentry/serverless'
 
 import { createSpaceMetricsTable } from '../tables/space-metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
+import { STORE_ADD, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
   dsn: process.env.SENTRY_DSN,
   tracesSampleRate: 1.0,
 })
 
-const STORE_ADD = 'store/add'
 const AWS_REGION = process.env.AWS_REGION || 'us-west-2'
 
 /**
@@ -43,7 +43,7 @@ async function handler(event) {
  */
 export async function updateStoreCount (ucanInvocations, ctx) {
   const invocationsWithStoreAdd = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === STORE_ADD)
+    inv => inv.value.att.find(a => a.can === STORE_ADD) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   await ctx.spaceMetricsTable.incrementStoreAddCount(invocationsWithStoreAdd)

ucan-invocation/functions/space-metrics-store-remove-size-total.js

@@ -3,7 +3,7 @@ import * as Sentry from '@sentry/serverless'
 import { createCarStore } from '../buckets/car-store.js'
 import { createSpaceMetricsTable } from '../tables/space-metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { STORE_REMOVE } from '../constants.js'
+import { STORE_REMOVE, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -36,7 +36,7 @@ async function handler(event) {
  */
 export async function updateRemoveSizeTotal (ucanInvocations, ctx) {
   const invocationsWithStoreRemove = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === STORE_REMOVE)
+    inv => inv.value.att.find(a => a.can === STORE_REMOVE) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   // TODO: once we have receipts for store/remove, replace this

ucan-invocation/functions/space-metrics-store-remove-total.js

@@ -2,7 +2,7 @@ import * as Sentry from '@sentry/serverless'
 
 import { createSpaceMetricsTable } from '../tables/space-metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { STORE_REMOVE } from '../constants.js'
+import { STORE_REMOVE, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -43,7 +43,7 @@ async function handler(event) {
  */
 export async function updateStoreCount (ucanInvocations, ctx) {
   const invocationsWithStoreRemove = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === STORE_REMOVE)
+    inv => inv.value.att.find(a => a.can === STORE_REMOVE) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   await ctx.spaceMetricsTable.incrementStoreRemoveCount(invocationsWithStoreRemove)

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

@@ -2,7 +2,7 @@ import * as Sentry from '@sentry/serverless'
 
 import { createSpaceMetricsTable } from '../tables/space-metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { UPLOAD_ADD } from '../constants.js'
+import { UPLOAD_ADD, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -43,7 +43,7 @@ async function handler(event) {
  */
 export async function updateUploadCount (ucanInvocations, ctx) {
   const invocationsWithUploadAdd = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === UPLOAD_ADD)
+    inv => inv.value.att.find(a => a.can === UPLOAD_ADD) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   await ctx.spaceMetricsTable.incrementUploadAddCount(invocationsWithUploadAdd)

ucan-invocation/functions/space-metrics-upload-remove-total.js

@@ -2,7 +2,7 @@ import * as Sentry from '@sentry/serverless'
 
 import { createSpaceMetricsTable } from '../tables/space-metrics.js'
 import { parseKinesisEvent } from '../utils/parse-kinesis-event.js'
-import { UPLOAD_REMOVE } from '../constants.js'
+import { UPLOAD_REMOVE, CONTENT_TYPE } from '../constants.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -37,7 +37,7 @@ async function handler(event) {
  */
 export async function updateUploadCount (ucanInvocations, ctx) {
   const invocationsWithUploadRemove = ucanInvocations.filter(
-    inv => inv.value.att.find(a => a.can === UPLOAD_REMOVE)
+    inv => inv.value.att.find(a => a.can === UPLOAD_REMOVE) && inv.type === CONTENT_TYPE.RECEIPT
   ).flatMap(inv => inv.value.att)
 
   await ctx.spaceMetricsTable.incrementUploadRemoveCount(invocationsWithUploadRemove)

ucan-invocation/test/functions/metrics-store-add-size-total.test.js

@@ -11,7 +11,7 @@ import { adminMetricsTableProps } from '../../tables/index.js'
 
 import { updateSizeTotal } from '../../functions/metrics-store-add-size-total.js'
 import { createMetricsTable } from '../../tables/metrics.js'
-import { METRICS_NAMES } from '../../constants.js'
+import { METRICS_NAMES, CONTENT_TYPE } from '../../constants.js'
 
 const REGION = 'us-west-2'
 
@@ -52,6 +52,7 @@ test('handles a batch of single invocation with store/add', async t => {
         aud: uploadService.did(),
         iss: alice.did()
     },
+    type: CONTENT_TYPE.RECEIPT,
     ts: Date.now()
   }]
 
@@ -95,6 +96,7 @@ test('handles batch of single invocations with multiple store/add attributes', a
       aud: uploadService.did(),
       iss: alice.did()
     },
+    type: CONTENT_TYPE.RECEIPT,
     ts: Date.now()
   }]
 
@@ -112,6 +114,91 @@ test('handles batch of single invocations with multiple store/add attributes', a
   t.is(item?.value, cars.reduce((acc, c) => acc + c.size, 0))
 })
 
+test('handles a batch of single invocation without store/add', async t => {
+  const { tableName } = await prepareResources(t.context.dynamoClient)
+  const uploadService = await Signer.generate()
+  const alice = await Signer.generate()
+  const { spaceDid } = await createSpace(alice)
+  const car = await randomCAR(128)
+
+  const metricsTable = createMetricsTable(REGION, tableName, {
+    endpoint: t.context.dbEndpoint
+  })
+
+  const invocations = [{
+    carCid: car.cid.toString(),
+    value: {
+        att: [
+          StoreCapabilities.remove.create({
+            with: spaceDid,
+            nb: {
+              link: car.cid,
+            }
+          })
+        ],
+        aud: uploadService.did(),
+        iss: alice.did()
+    },
+    type: CONTENT_TYPE.RECEIPT,
+    ts: Date.now()
+  }]
+
+  // @ts-expect-error
+  await updateSizeTotal(invocations, {
+    metricsTable
+  })
+
+  const item = await getItemFromTable(t.context.dynamoClient, tableName, {
+    name: METRICS_NAMES.STORE_ADD_SIZE_TOTAL
+  })
+  t.truthy(item)
+  t.is(item?.name, METRICS_NAMES.STORE_ADD_SIZE_TOTAL)
+  t.is(item?.value, 0)
+})
+
+test('handles a batch of single invocation without receipts', async t => {
+  const { tableName } = await prepareResources(t.context.dynamoClient)
+  const uploadService = await Signer.generate()
+  const alice = await Signer.generate()
+  const { spaceDid } = await createSpace(alice)
+  const car = await randomCAR(128)
+
+  const metricsTable = createMetricsTable(REGION, tableName, {
+    endpoint: t.context.dbEndpoint
+  })
+
+  const invocations = [{
+    carCid: car.cid.toString(),
+    value: {
+        att: [
+          StoreCapabilities.add.create({
+            with: spaceDid,
+            nb: {
+              link: car.cid,
+              size: car.size
+            }
+          })
+        ],
+        aud: uploadService.did(),
+        iss: alice.did()
+    },
+    type: CONTENT_TYPE.WORKFLOW,
+    ts: Date.now()
+  }]
+
+  // @ts-expect-error
+  await updateSizeTotal(invocations, {
+    metricsTable
+  })
+
+  const item = await getItemFromTable(t.context.dynamoClient, tableName, {
+    name: METRICS_NAMES.STORE_ADD_SIZE_TOTAL
+  })
+  t.truthy(item)
+  t.is(item?.name, METRICS_NAMES.STORE_ADD_SIZE_TOTAL)
+  t.is(item?.value, 0)
+})
+
 /**
  * @param {import('@aws-sdk/client-dynamodb').DynamoDBClient} dynamoClient
  */