feat: integration tests (#16)

package.json

@@ -9,13 +9,17 @@
     "dev": "sst dev",
     "build": "sst build",
     "test": "npm test -w packages/core",
+    "test-integration": "ava --verbose --serial --timeout=300s test/*.test.js",
     "deploy": "sst deploy",
     "remove": "sst remove",
     "console": "sst console",
     "typecheck": "tsc --build && npm run typecheck -w packages/core",
     "clean": "rm -rf node_modules pnpm-lock.yml packages/*/{pnpm-lock.yml,.next,out,coverage,.nyc_output,worker,dist,node_modules}"
   },
   "devDependencies": {
+    "@aws-sdk/client-dynamodb": "^3.398.0",
+    "@aws-sdk/client-s3": "^3.398.0",
+    "@aws-sdk/util-dynamodb": "^3.398.0",
     "@ipld/dag-ucan": "3.4.0",
     "@sentry/serverless": "^7.52.1",
     "@types/git-rev-sync": "^2.0.0",
@@ -26,11 +30,17 @@
     "@web3-storage/filecoin-api": "^1.4.3",
     "@web3-storage/filecoin-client": "1.3.0",
     "sst": "^2.8.3",
+    "ava": "^5.3.0",
     "aws-cdk-lib": "2.72.1",
     "constructs": "10.1.156",
+    "delay": "^6.0.0",
+    "dotenv": "^16.3.1",
     "git-rev-sync": "^3.0.2",
+    "p-retry": "^6.0.0",
+    "p-wait-for": "^5.0.2",
     "ts-node": "^10.9.1",
     "typescript": "^5.0.4",
+    "uint8arrays": "^4.0.6",
     "@tsconfig/node18": "^2.0.1"
   },
   "workspaces": [

packages/core/src/data/offer.ts

@@ -8,7 +8,7 @@ import { Deal } from './deal'
 
 // key in format of `YYYY-MM-DDTHH:MM:SS `${commitmentProof}``
 function createKey (deal: DealerMessageRecord) {
-  return `${(new Date()).toISOString()} ${deal.aggregate.toString()}.json`
+  return `${(new Date(deal.insertedAt)).toISOString()} ${deal.aggregate.toString()}.json`
 }
 
 export const encode = {

packages/core/src/queue/client.ts

@@ -4,6 +4,8 @@ import { Queue, Store } from '@web3-storage/filecoin-api/types'
 
 import { connectQueue, Target } from './index.js'
 
+const IMPLICIT_MESSAGE_GROUP_ID = '1'
+
 export function createQueueClient <Record> (conf: Target, context: QueueContext<Record>): Queue<Record> {
   const queueClient = connectQueue(conf)
   return {
@@ -31,11 +33,13 @@ export function createQueueClient <Record> (conf: Target, context: QueueContext<
         return { error }
       }
 
-      const messageGroupId = conf instanceof SQSClient ? undefined : options.messageGroupId
+      let messageGroupId = options.messageGroupId || IMPLICIT_MESSAGE_GROUP_ID
       const cmd = new SendMessageCommand({
         QueueUrl: context.queueUrl,
         MessageBody: encodedMessage,
-        MessageGroupId: messageGroupId
+        // If SQS Client was provided, we are in testing mode, and we cannot provide message group ID
+        // given resource used does not support FIFO Queues with it.
+        MessageGroupId: conf instanceof SQSClient ? undefined : messageGroupId
       })
 
       let r

stacks/DataStack.ts

@@ -22,10 +22,16 @@ export function DataStack({ stack }: StackContext) {
   /**
    * This table tracks the state of deals offered to a broker.
    */
-  const dealTable = new Table(stack, 'deal-store', {
+  const dealTableName = 'deal-store'
+  const dealTable = new Table(stack, dealTableName, {
     ...dealTableProps,
   })
 
+  stack.addOutputs({
+    OfferBucketName: offerBucketConfig.bucketName,
+    DealTableName: dealTableName,
+  })
+
   return {
     offerBucket,
     dealTable

test/api.test.js

@@ -0,0 +1,100 @@
+import { test } from './helpers/context.js'
+
+import git from 'git-rev-sync'
+import delay from 'delay'
+import { toString } from 'uint8arrays/to-string'
+import { Dealer } from '@web3-storage/filecoin-client'
+import { randomAggregate } from '@web3-storage/filecoin-api/test'
+
+import { getDealerClientConfig } from './helpers/dealer-client.js'
+import { pollTableItem } from './helpers/table.js'
+import { pollBucketItem } from './helpers/bucket.js'
+import {
+  getApiEndpoint,
+  getStage,
+  getDealStoreDynamoDb,
+  getOfferStoreBucketInfo
+} from './helpers/deployment.js'
+import pRetry from 'p-retry'
+
+test.before(t => {
+  t.context = {
+    apiEndpoint: getApiEndpoint(),
+    dealStoreDynamo: getDealStoreDynamoDb(),
+    offerStoreBucket: getOfferStoreBucketInfo()
+  }
+})
+
+test('GET /version', async t => {
+  const stage = getStage()
+  const response = await fetch(`${t.context.apiEndpoint}/version`)
+  t.is(response.status, 200)
+
+  const body = await response.json()
+  t.is(body.env, stage)
+  t.is(body.commit, git.long('.'))
+})
+
+test('POST /', async t => {
+  const {
+    invocationConfig,
+    connection
+  } = await getDealerClientConfig(new URL(t.context.apiEndpoint))
+
+  // Create a random aggregate
+  const label = 'label'
+  const { pieces, aggregate } = await randomAggregate(100, 128)
+  const offer = pieces.map((p) => p.link)
+
+  const res = await Dealer.dealQueue(
+    invocationConfig,
+    aggregate.link.link(),
+    offer,
+    invocationConfig.with,
+    label,
+    { connection }
+  )
+
+  t.truthy(res.out.ok)
+  t.falsy(res.out.error)
+  t.truthy(aggregate.link.link().equals(res.out.ok?.aggregate))
+
+  // wait for deal-store entry to exist given it is propagated with a queue message
+  await delay(5_000)
+
+  /** @type {import('../packages/core/src/data/deal.js').EncodedDeal | undefined} */
+  // @ts-expect-error does not automatically infer
+  const dealEntry = await pollTableItem(
+    t.context.dealStoreDynamo.client,
+    t.context.dealStoreDynamo.tableName,
+    { aggregate: res.out.ok?.aggregate?.toString() }
+  )
+  if (!dealEntry) {
+    throw new Error('deal store item was not found')
+  }
+  t.is(dealEntry.aggregate, aggregate.link.link().toString())
+  t.is(dealEntry.storefront, invocationConfig.with)
+  t.is(dealEntry.stat, 0)
+  t.truthy(dealEntry.insertedAt)
+  t.truthy(dealEntry.offer)
+
+  // Verify offer store
+  // remove bucket encoding
+  const bucketKey = dealEntry.offer.replace('s3://', '').replace(`${t.context.offerStoreBucket.bucket}/`, '')
+  console.log('try to get bucket item...', bucketKey)
+  const bucketItem = await pollBucketItem(
+    t.context.offerStoreBucket.client,
+    t.context.offerStoreBucket.bucket,
+    bucketKey
+  )
+  if (!bucketItem) {
+    throw new Error('offer store item was not found')
+  }
+  /** @type {import('../packages/core/src/data/offer.js').Offer} */
+  const encodedOffer = JSON.parse(toString(bucketItem))
+
+  t.is(encodedOffer.aggregate, aggregate.link.link().toString())
+  t.is(encodedOffer.tenant, invocationConfig.with)
+  t.truthy(encodedOffer.orderID)
+  t.deepEqual(encodedOffer.pieces, offer.map(o => o.toString()))
+})

test/helpers/bucket.js

@@ -0,0 +1,33 @@
+import { GetObjectCommand } from '@aws-sdk/client-s3'
+import pRetry from 'p-retry'
+
+/**
+ * @param {import('@aws-sdk/client-s3').S3Client} client
+ * @param {string} bucketName
+ * @param {string} key
+ */
+export async function pollBucketItem (client, bucketName, key) {
+  const cmd = new GetObjectCommand({
+    Bucket: bucketName,
+    Key: key
+  })
+
+  const response = await pRetry(async () => {
+    let r
+    try {
+      r = await client.send(cmd)
+    } catch (err) {
+      if (err?.$metadata?.httpStatusCode === 404) {
+        throw new Error('not found')
+      }
+    }
+
+    return r
+  }, {
+    retries: 100,
+    maxTimeout: 1000,
+    minTimeout: 1000
+  })
+
+  return await response?.Body?.transformToByteArray()
+}

test/helpers/context.js

@@ -0,0 +1,29 @@
+import anyTest from 'ava'
+import dotenv from 'dotenv'
+
+dotenv.config({
+  path: '.env.local'
+})
+
+/**
+ * @typedef {object} Dynamo
+ * @property {import('@aws-sdk/client-dynamodb').DynamoDBClient} client
+ * @property {string} endpoint
+ * @property {string} region
+ * @property {string} tableName
+ * 
+ * @typedef {object} Bucket
+ * @property {import('@aws-sdk/client-s3').S3Client} client
+ * @property {string} region
+ * @property {string} bucket
+ *
+ * @typedef {object} Context
+ * @property {string} apiEndpoint
+ * @property {Dynamo} dealStoreDynamo
+ * @property {Bucket} offerStoreBucket
+ *
+ * @typedef {import('ava').TestFn<Awaited<Context>>} TestContextFn
+ */
+
+// eslint-disable-next-line unicorn/prefer-export-from
+export const test  = /** @type {TestContextFn} */ (anyTest)