feat: add verifiable ts

carpark/package.json

@@ -6,8 +6,8 @@
     "test": "ava --verbose --timeout=60s **/*.test.js"
   },
   "dependencies": {
-    "@aws-sdk/client-eventbridge": "^3.216.0",
-    "@aws-sdk/client-sqs": "^3.213.0",
+    "@aws-sdk/client-eventbridge": "^3.226.0",
+    "@aws-sdk/client-sqs": "^3.226.0",
     "@sentry/serverless": "^7.22.0"
   },
   "devDependencies": {

package.json

@@ -42,6 +42,7 @@
     "rules": {
       "unicorn/prefer-number-properties": "off",
       "unicorn/no-null": "off",
+      "unicorn/no-array-callback-reference": "off",
       "unicorn/no-array-for-each": "off",
       "unicorn/no-await-expression-member": "off",
       "unicorn/prefer-set-has": "off",

replicator/package.json

@@ -6,7 +6,7 @@
     "test": "ava --verbose --timeout=60s **/*.test.js"
   },
   "dependencies": {
-    "@aws-sdk/client-s3": "^3.211.0",
+    "@aws-sdk/client-s3": "^3.226.0",
     "@sentry/serverless": "^7.22.0"
   },
   "devDependencies": {

satnav/package.json

@@ -6,8 +6,8 @@
     "test": "ava --verbose --timeout=60s **/*.test.js"
   },
   "dependencies": {
-    "@aws-sdk/client-s3": "^3.211.0",
-    "@aws-sdk/client-eventbridge": "^3.218.0",
+    "@aws-sdk/client-s3": "^3.226.0",
+    "@aws-sdk/client-eventbridge": "^3.226.0",
     "@sentry/serverless": "^7.22.0",
     "@ipld/car": "^5.0.1",
     "cardex": "^1.0.0",

stacks/ucan-invocation-stack.js

@@ -1,8 +1,10 @@
 import {
   Bucket,
   Function,
+  KinesisStream,
   use
 } from '@serverless-stack/resources'
+import { Duration } from 'aws-cdk-lib'
 
 import { BusStack } from './bus-stack.js'
 import { getConfig, setupSentry } from './config.js'
@@ -44,7 +46,20 @@ export function UcanInvocationStack({ stack, app }) {
     }
   })
 
+  // create a kinesis stream
+  const ucanStream = new KinesisStream(stack, 'ucan-stream', {
+    cdk: {
+      stream: {
+        retentionPeriod: Duration.days(365)
+      }
+    },
+    consumers: {
+      // consumer1: 'functions/consumer1.handler'
+    },
+  })
+
   return {
-    ucanBucket
+    ucanBucket,
+    ucanStream
   }
-}
+}

stacks/upload-api-stack.js

@@ -23,7 +23,7 @@ export function UploadApiStack({ stack, app }) {
   // Get references to constructs created in other stacks
   const { carparkBucket } = use(CarparkStack)
   const { storeTable, uploadTable } = use(UploadDbStack)
-  const { ucanBucket } = use(UcanInvocationStack)
+  const { ucanBucket, ucanStream } = use(UcanInvocationStack)
 
   // Setup API
   const customDomain = getCustomDomain(stack.stage, process.env.HOSTED_ZONE)
@@ -35,12 +35,13 @@ export function UploadApiStack({ stack, app }) {
     customDomain,
     defaults: {
       function: {
-        permissions: [storeTable, uploadTable, carparkBucket, ucanBucket],
+        permissions: [storeTable, uploadTable, carparkBucket, ucanBucket, ucanStream],
         environment: {
           STORE_TABLE_NAME: storeTable.tableName,
           STORE_BUCKET_NAME: carparkBucket.bucketName,
           UPLOAD_TABLE_NAME: uploadTable.tableName,
           UCAN_BUCKET_NAME: ucanBucket.bucketName,
+          UCAN_LOG_STREAM_NAME: ucanStream.streamName,
           NAME: pkg.name,
           VERSION: pkg.version,
           COMMIT: git.commmit,

ucan-invocation/package.json

@@ -7,7 +7,8 @@
   },
   "dependencies": {
     "@aws-sdk/client-eventbridge": "^3.218.0",
-    "@sentry/serverless": "^7.22.0"
+    "@sentry/serverless": "^7.22.0",
+    "uint8arrays": "^4.0.2"
   },
   "devDependencies": {
     "@serverless-stack/resources": "*",

upload-api/drand.js

@@ -0,0 +1,17 @@
+// @ts-ignore
+import Chain from 'drand-client/chain.js'
+
+const chainInfo = {
+  public_key: '868f005eb8e6e4ca0a47c8a77ceaa5309a47978a7c71bc5cce96366b5d7a569937c529eeda66c7293784a9402801af31',
+  period: 30,
+  // eslint-disable-next-line unicorn/numeric-separators-style
+  genesis_time: 1595431050,
+  hash: '8990e7a9aaed2ffed73dbd7092123d6f289930540d7651336225dc172e51b2ce'
+}
+
+/**
+ * @param {number} time
+ */
+export function roundAt (time) {
+  return Chain.roundAt(time, chainInfo.genesis_time * 1000, chainInfo.period * 1000)
+}

upload-api/functions/ucan-invocation-router.js

@@ -1,21 +1,25 @@
 import { DID } from '@ucanto/core'
+import { Kinesis } from '@aws-sdk/client-kinesis'
 import * as Sentry from '@sentry/serverless'
+import { fromString as uint8arrayFromString } from 'uint8arrays/from-string'
 
 import { createAccessClient } from '../access.js'
-import { persistUcanInvocation } from '../ucan-invocation.js'
+import { parseUcanInvocationRequest, persistUcanInvocation } from '../ucan-invocation.js'
 import { createCarStore } from '../buckets/car-store.js'
 import { createDudewhereStore } from '../buckets/dudewhere-store.js'
 import { createUcanStore } from '../buckets/ucan-store.js'
 import { createStoreTable } from '../tables/store.js'
 import { createUploadTable } from '../tables/upload.js'
 import { getServiceSigner } from '../config.js'
 import { createUcantoServer } from '../service/index.js'
+import { roundAt } from '../drand.js'
 
 Sentry.AWSLambda.init({
   dsn: process.env.SENTRY_DSN,
   tracesSampleRate: 1.0,
 })
 
+const kinesisClient = new Kinesis({})
 const AWS_REGION = process.env.AWS_REGION || 'us-west-2'
 
 // Specified in SST environment
@@ -40,6 +44,7 @@ async function ucanInvocationRouter (request) {
     STORE_BUCKET_NAME: storeBucketName = '',
     UPLOAD_TABLE_NAME: uploadTableName = '',
     UCAN_BUCKET_NAME: ucanBucketName = '',
+    UCAN_LOG_STREAM_NAME: ucanLogStreamName = '',
     // set for testing
     DYNAMO_DB_ENDPOINT: dbEndpoint,
     ACCESS_SERVICE_DID: accessServiceDID = '',
@@ -82,8 +87,30 @@ async function ucanInvocationRouter (request) {
     body: Buffer.from(request.body, 'base64'),
   })
 
-  // persist successful invocation handled
-  await persistUcanInvocation(request, ucanStoreBucket)
+  const ucanInvocation = await parseUcanInvocationRequest(request)
+
+  const [, round ] = await Promise.all([
+    // persist successful invocation handled
+    persistUcanInvocation(ucanInvocation, ucanStoreBucket),
+    // Get drand round for this moment in time
+    roundAt(Date.now())
+  ])
+
+  // Put invocation to UCAN stream
+  await kinesisClient.putRecord({
+    Data: uint8arrayFromString(JSON.stringify({
+      carCid: ucanInvocation.carCid,
+      value: ucanInvocation.value,
+      drand: {
+        round
+      }
+    })),
+    // https://docs.aws.amazon.com/streams/latest/dev/key-concepts.html
+    // A partition key is used to group data by shard within a stream.
+    // It is required, and now we are starting with one shard. We need to study best partition key
+    PartitionKey: 'key',
+    StreamName: ucanLogStreamName,
+  })
 
   return toLambdaSuccessResponse(response)
 }

upload-api/package.json

@@ -6,10 +6,12 @@
     "test": "ava --node-arguments='--experimental-fetch' --verbose --timeout=60s **/*.test.js"
   },
   "dependencies": {
-    "@aws-sdk/client-dynamodb": "^3.211.0",
-    "@aws-sdk/client-s3": "^3.211.0",
-    "@aws-sdk/s3-request-presigner": "^3.224.0",
-    "@aws-sdk/util-dynamodb": "^3.211.0",
+    "@aws-sdk/client-dynamodb": "^3.226.0",
+    "@aws-sdk/client-s3": "^3.226.0",
+    "@aws-sdk/client-kinesis": "^3.226.0",
+    "@aws-sdk/s3-request-presigner": "^3.226.0",
+    "@aws-sdk/util-dynamodb": "^3.226.0",
+    "@ipld/dag-ucan": "3.0.1",
     "@sentry/serverless": "^7.22.0",
     "@serverless-stack/node": "^1.18.2",
     "@ucanto/client": "^4.0.2",
@@ -20,8 +22,10 @@
     "@web-std/fetch": "^4.1.0",
     "@web3-storage/access": "^9.0.0",
     "@web3-storage/capabilities": "^2.0.0",
+    "drand-client": "^0.2.0",
     "multiformats": "^10.0.2",
-    "p-retry": "^5.1.2"
+    "p-retry": "^5.1.2",
+    "uint8arrays": "^4.0.2"
   },
   "devDependencies": {
     "@ipld/car": "^5.0.1",

upload-api/service/index.js

@@ -26,7 +26,7 @@ export function createServiceRouter (context) {
  * @param {import('@ucanto/interface').Signer} serviceSigner
  * @param {import('../service/types').UcantoServerContext} context
  */
- export async function createUcantoServer (serviceSigner, context) {
+export async function createUcantoServer (serviceSigner, context) {
   const server = Server.create({
     id: serviceSigner,
     encoder: CBOR,

upload-api/service/types.ts

@@ -7,6 +7,8 @@ import type {
 } from '@ucanto/interface'
 import type { API } from '@ucanto/server'
 
+import { ToString, UnknownLink } from 'multiformats'
+
 import {
   StoreAdd,
   StoreRemove,
@@ -130,6 +132,10 @@ export interface AccessClient {
   verifyInvocation: (invocation: Invocation) => Promise<boolean>
 }
 
+export interface LinkJSON<T extends UnknownLink = UnknownLink> {
+  '/': ToString<T>
+}
+
 // would be generated by sst, but requires `sst build` to be run, which calls out to aws; not great for CI
 declare module "@serverless-stack/node/config" {
   export interface SecretResources {

upload-api/test/service/ucan-invocation.test.js

@@ -7,9 +7,10 @@ import * as UCAN from '@ipld/dag-ucan'
 
 import { createSpace } from '../helpers/ucanto.js'
 import { createS3, createBucket } from '../helpers/resources.js'
+import { randomCAR } from '../helpers/random.js'
 
 import { createUcanStore } from '../../buckets/ucan-store.js'
-import { parseUcanInvocationRequest, persistUcanInvocation } from '../../ucan-invocation.js'
+import { parseUcanInvocationRequest, persistUcanInvocation, replaceAllLinkValues } from '../../ucan-invocation.js'
 
 test.before(async t => {
   const { client: s3Client, clientOpts: s3ClientOpts } = await createS3({ port: 9000 })
@@ -93,7 +94,8 @@ test('persists ucan invocation CAR file', async t => {
   ])
 
   // @ts-expect-error different type interface in AWS expected request
-  await persistUcanInvocation(request, ucanStore)
+  const ucanInvocationObject = await parseUcanInvocationRequest(request)
+  await persistUcanInvocation(ucanInvocationObject, ucanStore)
 
   const requestCar = await CAR.codec.decode(request.body)
   const requestCarRootCid = requestCar.roots[0].cid.toString()
@@ -126,6 +128,46 @@ test('persists ucan invocation CAR file', async t => {
   })
 })
 
+test('replace all link values as object and array', async t => {
+  const car = await randomCAR(128)
+  const otherCar = await randomCAR(40)
+
+  // invoke a upload/add with proof
+  const root = car.roots[0]
+  const shards = [car.cid, otherCar.cid].sort()
+
+  const att = [
+    {
+      nb: {
+        link: root,
+        size: car.size
+      },
+      can: 'store/add',
+      with: 'did:key:z6MkfTDbhRZz26kcDNmmehPxeujSkbXe8jqv5fLpKvtc3Wcv'
+    },
+    {
+      nb: {
+        root,
+        shards: [...shards]
+      },
+      can: 'upload/add',
+      with: 'did:key:z6MkfTDbhRZz26kcDNmmehPxeujSkbXe8jqv5fLpKvtc3Wcv'
+    },
+  ]
+
+  att.map(replaceAllLinkValues)
+
+  // Object with Link
+  // @ts-expect-error Property '/' does not exist on type 'Link<Partial<Model>
+  t.is(att[0].nb.link['/'], root.toString())
+  // @ts-expect-error Property '/' does not exist on type 'Link<Partial<Model>
+  t.is(att[1].nb.root['/'], root.toString())
+
+  // Array with Link
+  // @ts-expect-error Property '/' does not exist on type 'Link<Partial<Model>
+  t.deepEqual(att[1].nb.shards?.map(s => s['/']), shards.map(s => s.toString()))
+})
+
 /**
  * @param {import("@aws-sdk/client-s3").S3Client} s3Client
  */
@@ -136,3 +178,4 @@ async function prepareResources (s3Client) {
     bucketName
   }
 }
+