feat(billing): use allocated bytes in store/add receipt

billing/lib/ucan-stream.js

@@ -16,13 +16,14 @@ export const findSpaceUsageDeltas = messages => {
     /** @type {number|undefined} */
     let size
     if (isReceiptForCapability(message, StoreCaps.add) && isStoreAddSuccess(message.out)) {
-      size = message.value.att[0].nb?.size
+      size = message.out.ok.allocated
     } else if (isReceiptForCapability(message, StoreCaps.remove) && isStoreRemoveSuccess(message.out)) {
       size = -message.out.ok.size
     }
 
-    // message is not a valid store/add or store/remove receipt
-    if (size == null) {
+    // Is message is a repeat store/add for the same shard or not a valid
+    // store/add or store/remove receipt?
+    if (size == 0 || size == null) {
       continue
     }
 

billing/package.json

@@ -13,7 +13,7 @@
     "@sentry/serverless": "^7.74.1",
     "@ucanto/interface": "^9.0.0",
     "@ucanto/server": "^9.0.1",
-    "@web3-storage/capabilities": "^11.3.1",
+    "@web3-storage/capabilities": "^13.0.0",
     "big.js": "^6.2.1",
     "multiformats": "^12.1.2",
     "p-retry": "^6.1.0",

billing/test/lib/ucan-stream.js

@@ -41,7 +41,7 @@ export const test = {
         aud: await randomDID(),
         cid: randomLink()
       },
-      out: { ok: { status: 'upload' } },
+      out: { ok: { status: 'upload', allocated: 138 } },
       ts: new Date()
     }, {
       type: 'receipt',
@@ -95,7 +95,7 @@ export const test = {
         aud: consumer.provider,
         cid: randomLink()
       },
-      out: { ok: { status: 'upload' } },
+      out: { ok: { status: 'upload', allocated: 138 } },
       ts: new Date(from.getTime() + 1)
     }, {
       type: 'receipt',
@@ -113,7 +113,7 @@ export const test = {
         aud: consumer.provider,
         cid: randomLink()
       },
-      out: { ok: { status: 'upload' } },
+      out: { ok: { status: 'upload', allocated: 1138 } },
       ts: new Date(from.getTime() + 2)
     }]
 
@@ -142,5 +142,49 @@ export const test = {
         d.delta === r.value.att[0].nb?.size
       )))
     }
+  },
+  'should filter non-allocating store/add messages': async (/** @type {import('entail').assert} */ assert, ctx) => {
+    const consumer = await randomConsumer()
+
+    await ctx.consumerStore.put(consumer)
+
+    const from = new Date()
+
+    /** @type {import('../../lib/api.js').UcanReceiptMessage<[import('@web3-storage/capabilities/types').StoreAdd]>[]} */
+    const receipts = [{
+      type: 'receipt',
+      carCid: randomLink(),
+      invocationCid: randomLink(),
+      value: {
+        att: [{
+          with: Schema.did({ method: 'key' }).from(consumer.consumer),
+          can: 'store/add',
+          nb: {
+            link: randomLink(),
+            size: 138
+          }
+        }],
+        aud: consumer.provider,
+        cid: randomLink()
+      },
+      // allocated: 0 indicates this shard was previously stored in this space
+      out: { ok: { status: 'upload', allocated: 0 } },
+      ts: new Date(from.getTime() + 1)
+    }]
+
+    const deltas = findSpaceUsageDeltas(receipts)
+
+    for (const d of deltas) {
+      const res = await storeSpaceUsageDelta(d, ctx)
+      assert.ok(res.ok)
+    }
+
+    const res = await ctx.spaceDiffStore.list({
+      provider: consumer.provider,
+      space: consumer.consumer,
+      from
+    }, { size: 1000 })
+    assert.ok(res.ok)
+    assert.equal(res.ok.results.length, 0)
   }
 }

upload-api/package.json

@@ -24,8 +24,8 @@
     "@ucanto/validator": "^9.0.1",
     "@web-std/fetch": "^4.1.0",
     "@web3-storage/access": "^18.0.5",
-    "@web3-storage/capabilities": "^12.1.0",
-    "@web3-storage/upload-api": "^7.3.5",
+    "@web3-storage/capabilities": "^13.0.0",
+    "@web3-storage/upload-api": "^8.0.0",
     "@web3-storage/w3infra-ucan-invocation": "*",
     "multiformats": "^12.1.2",
     "nanoid": "^5.0.2",

upload-api/tables/lib.js

@@ -0,0 +1,23 @@
+import { Failure } from '@ucanto/server'
+
+export class RecordNotFound extends Failure {
+  constructor () {
+    super()
+    this.name = /** @type {const} */ ('RecordNotFound')
+  }
+
+  describe () {
+    return 'record not found'
+  }
+}
+
+export class RecordKeyConflict extends Failure {
+  constructor () {
+    super()
+    this.name = /** @type {const} */ ('RecordKeyConflict')
+  }
+
+  describe () {
+    return 'record key conflict'
+  }
+}