feat: publish egress events to cf queue (#191)

## Overview

This PR refactors the egress tracking system to use asynchronous batch
processing instead of synchronous invocation execution. The freeway
gateway now serializes egress tracking invocations and sends them to a
Cloudflare Queue, where the
[egress-consumer](https://github.com/storacha/egress-consumer) processes
them in batches before forwarding to the upload-api in a CAR with
multiple invocations.

## Key Changes

### Freeway Gateway (Producer)

Modified the egress tracking middleware to serialize invocations instead
of executing them immediately:

- Removed the direct connection to upload-api from withEgressClient
- Invocations are now built, converted to IPLD view using
buildIPLDView(), and archived to bytes
- Serialized invocations are sent to EGRESS_QUEUE as message payloads
- Egress Tracking controlled using feature flags:
`FF_EGRESS_TRACKER_ENABLED` & `FF_EGRESS_TRACKER_ROLLOUT_PERCENTAGE`

## Flow

1. Gateway serves content and creates an egress record invocation
2. Invocation is archived to bytes and queued
3. Cloudflare Queue batches messages (configured batch size: 50,
timeout: 30s)
4. Egress-consumer receives batch, extracts invocations
5. Consumer builds CAR message with all invocations
6. Single POST to upload-api processes entire batch
7. Upload-api executes each invocation and records egress in
DynamoDB+Stripe

## Additional Changes
- Added missing `dev` packages
- Fixed type issues
- Fixed build issues
- Fixed tests
- Updated `package-lock.json`

Author: fforbeck

package-lock.json

@@ -22,8 +22,8 @@
   "main": "src/index.js",
   "types": "dist/src/index.d.ts",
   "scripts": {
-    "build": "esbuild --bundle src/index.js --format=esm --external:node:buffer --external:node:events --external:node:async_hooks --sourcemap --minify --outfile=dist/worker.mjs && npm run build:tsc",
-    "build:debug": "esbuild --bundle src/index.js --format=esm --external:node:buffer --external:node:events --external:node:async_hooks --outfile=dist/worker.mjs",
+    "build": "esbuild --bundle src/index.js --format=esm --external:node:buffer --external:node:events --external:node:async_hooks --external:cloudflare:workers --sourcemap --minify --outfile=dist/worker.mjs && npm run build:tsc",
+    "build:debug": "esbuild --bundle src/index.js --format=esm --external:node:buffer --external:node:events --external:node:async_hooks --external:cloudflare:workers --outfile=dist/worker.mjs",
     "build:tsc": "tsc --build",
     "dev": "npm run build:debug && miniflare dist/worker.mjs --watch --debug -m --r2-persist --global-async-io --global-timers",
     "lint": "standard",
@@ -36,12 +36,13 @@
     "test:unit:only": "npm run build:debug && mocha --experimental-vm-modules"
   },
   "dependencies": {
+    "@ipld/dag-pb": "^4.1.5",
     "@microlabs/otel-cf-workers": "^1.0.0-rc.48",
     "@opentelemetry/api": "^1.9.0",
     "@opentelemetry/sdk-trace-base": "^1.27.0",
     "@storacha/capabilities": "^1.10.0",
     "@storacha/indexing-service-client": "^2.6.6",
-    "@types/node": "^24.9.1",
+    "@types/node": "^24.10.1",
     "@ucanto/client": "^9.0.2",
     "@ucanto/core": "^10.4.5",
     "@ucanto/interface": "^11.0.1",
@@ -57,12 +58,16 @@
     "multiformats": "^13.0.1"
   },
   "devDependencies": {
+    "@storacha/blob-index": "^1.2.4",
     "@storacha/cli": "^1.6.2",
     "@storacha/client": "^1.8.2",
+    "@storacha/upload-client": "^1.3.6",
     "@types/chai": "^5.0.0",
     "@types/mocha": "^10.0.9",
+    "@types/node": "^24.10.1",
     "@types/node-fetch": "^2.6.11",
     "@types/sinon": "^17.0.3",
+    "@web3-storage/content-claims": "^5.2.4",
     "@web3-storage/public-bucket": "^1.4.0",
     "carstream": "^2.2.0",
     "chai": "^5.1.1",
@@ -71,6 +76,7 @@
     "miniflare": "^4.20251011.1",
     "mocha": "^10.7.3",
     "multipart-byte-range": "^3.0.1",
+    "sade": "^1.8.1",
     "sinon": "^19.0.2",
     "standard": "^17.1.2",
     "tree-kill": "^1.2.2",
@@ -81,5 +87,6 @@
     "ignore": [
       "*.ts"
     ]
-  }
+  },
+  "packageManager": "pnpm@10.14.0+sha512.ad27a79641b49c3e481a16a805baa71817a04bbe06a38d17e60e2eaee83f6a146c6a688125f5792e48dd5ba30e7da52a5cda4c3992b9ccf333f9ce223af84748"
 }

package.json

@@ -125,6 +125,7 @@ function config (env, _trigger) {
     }
   }
   return {
+    // @ts-expect-error - NoopSpanProcessor extends SpanProcessor, but ts doesn't recognize it
     spanProcessors: new NoopSpanProcessor(),
     service: { name: 'freeway' }
   }
@@ -154,8 +155,13 @@ async function initializeHandler (env) {
   return async (request, env, ctx) => {
     const response = await finalHandler(request, env, ctx)
     const cacheControl = response.headers.get('Cache-Control') ?? ''
-    response.headers.set('Cache-Control', cacheControl ? `${cacheControl}, no-transform` : 'no-transform')
-    return response
+    const newHeaders = new Headers(response.headers)
+    newHeaders.set('Cache-Control', cacheControl ? `${cacheControl}, no-transform` : 'no-transform')
+    return new Response(response.body, {
+      status: response.status,
+      statusText: response.statusText,
+      headers: newHeaders
+    })
   }
 }
 

src/index.js

@@ -1,9 +1,52 @@
 import { Verifier } from '@ucanto/principal'
-import { ok, access, Unauthorized } from '@ucanto/validator'
+import { ok, access, fail, Unauthorized } from '@ucanto/validator'
 import { HttpError } from '@web3-storage/gateway-lib/util'
 import * as serve from '../capabilities/serve.js'
 import { SpaceDID } from '@storacha/capabilities/utils'
 
+/**
+ * Extracts a SpaceDID string from various space object formats.
+ * Handles string DIDs, objects with .did() method, and Uint8Arrays.
+ *
+ * @param {any} space - The space object to extract DID from
+ * @returns {import('@storacha/capabilities/types').SpaceDID | undefined}
+ */
+function extractSpaceDID (space) {
+  if (!space) return undefined
+
+  try {
+    // Already a string DID
+    if (typeof space === 'string' && space.startsWith('did:')) {
+      return /** @type {import('@storacha/capabilities/types').SpaceDID} */ (space)
+    }
+
+    // Object with .did() method (most common case from indexing service)
+    if (typeof space === 'object' && typeof /** @type {any} */ (space).did === 'function') {
+      return /** @type {import('@storacha/capabilities/types').SpaceDID} */ (/** @type {any} */ (space).did())
+    }
+
+    // Uint8Array (fallback case)
+    if (ArrayBuffer.isView(space)) {
+      const spaceDID = SpaceDID.from(space)
+      return /** @type {import('@storacha/capabilities/types').SpaceDID} */ (spaceDID.toString())
+    }
+
+    // Last resort: try String() conversion
+    const spaceStr = String(space)
+    if (spaceStr.startsWith('did:')) {
+      return /** @type {import('@storacha/capabilities/types').SpaceDID} */ (spaceStr)
+    }
+
+    return undefined
+  } catch (error) {
+    // Log error in debug mode only
+    if (process.env.DEBUG) {
+      console.warn('Failed to extract space DID:', error, 'Raw space:', space)
+    }
+    return undefined
+  }
+}
+
 /**
  * @import * as Ucanto from '@ucanto/interface'
  * @import { IpfsUrlContext, Middleware } from '@web3-storage/gateway-lib'
@@ -43,51 +86,64 @@ export function withAuthorizedSpace (handler) {
     // Legacy behavior: Site results which have no Space attached are from
     // before we started authorizing serving content explicitly. For these, we
     // always serve the content, but only if the request has no authorization
-    // token.
+    // token AND there are no sites with space information available.
+    const sitesWithSpace = locRes.ok.site.filter((site) => site.space !== undefined)
+    const sitesWithoutSpace = locRes.ok.site.filter((site) => site.space === undefined)
     const shouldServeLegacy =
-      locRes.ok.site.some((site) => site.space === undefined) &&
+      sitesWithSpace.length === 0 &&
+      sitesWithoutSpace.length > 0 &&
       ctx.authToken === null
 
     if (shouldServeLegacy) {
       return handler(request, env, ctx)
     }
 
     // These Spaces all have the content we're to serve, if we're allowed to.
-    const spaces = locRes.ok.site
-      .map((site) => site.space)
-      .filter((s) => s !== undefined)
+    // Extract space DIDs from sites with space information
+    const spaces = sitesWithSpace
+      .map((site) => extractSpaceDID(site.space))
+      .filter((space) => space !== undefined)
 
     try {
       // First space to successfully authorize is the one we'll use.
       const { space: selectedSpace, delegationProofs } = await Promise.any(
         spaces.map(async (space) => {
-          const result = await authorize(SpaceDID.from(space), ctx)
-          if (result.error) throw result.error
+          const result = await authorize(SpaceDID.from(space.toString()), ctx)
+          if (result.error) {
+            throw result.error
+          }
           return result.ok
         })
       )
       return handler(request, env, {
         ...ctx,
-        space: SpaceDID.from(selectedSpace),
+        space: SpaceDID.from(selectedSpace.toString()),
         delegationProofs,
         locator: locator.scopeToSpaces([selectedSpace])
       })
     } catch (error) {
-      // If all Spaces failed to authorize, throw the first error.
-      if (
-        error instanceof AggregateError &&
-        error.errors.every((e) => e instanceof Unauthorized)
-      ) {
-        if (env.DEBUG === 'true') {
-          console.log(
-            [
-              'Authorization Failures:',
-              ...error.errors.map((e) => e.message)
-            ].join('\n\n')
-          )
-        }
+      // If all Spaces failed to authorize, return 404 (security through obscurity)
+      if (error instanceof AggregateError) {
+        // Check if all errors are authorization failures (not storage errors)
+        const isAuthFailure = error.errors.every((e) =>
+          e instanceof Unauthorized ||
+          (e.message && e.message.includes('not authorized to serve'))
+        )
 
-        throw new HttpError('Not Found', { status: 404, cause: error })
+        if (isAuthFailure) {
+          if (env.DEBUG === 'true') {
+            console.log(
+              [
+                'Authorization Failures:',
+                ...error.errors.map((e) => e.message)
+              ].join('\n\n')
+            )
+          }
+          // Don't reveal whether content exists in unauthorized spaces
+          throw new HttpError('Not Found', { status: 404, cause: error })
+        }
+        // For storage or other errors, throw the AggregateError as-is
+        throw error
       } else {
         throw error
       }
@@ -107,8 +163,17 @@ export function withAuthorizedSpace (handler) {
 const authorize = async (space, ctx) => {
   // Look up delegations that might authorize us to serve the content.
   const relevantDelegationsResult = await ctx.delegationsStorage.find(space)
-  if (relevantDelegationsResult.error) return relevantDelegationsResult
+  if (relevantDelegationsResult.error) {
+    return relevantDelegationsResult
+  }
+
   const delegationProofs = relevantDelegationsResult.ok
+
+  // If no delegations found, the server is not authorized to serve this content
+  if (!delegationProofs || delegationProofs.length === 0) {
+    return fail('The gateway is not authorized to serve this content.')
+  }
+
   // Create an invocation of the serve capability.
   const invocation = await serve.transportHttp
     .invoke({
@@ -129,6 +194,7 @@ const authorize = async (space, ctx) => {
     principal: Verifier,
     validateAuthorization: () => ok({})
   })
+
   if (accessResult.error) {
     return accessResult
   }

src/middleware/withAuthorizedSpace.js

@@ -88,7 +88,7 @@ function createStorage (env) {
       try {
         await env.CONTENT_SERVE_DELEGATIONS_STORE.put(
           `${space}:${delegation.cid.toString()}`,
-          value.ok.buffer,
+          /** @type {ArrayBuffer} */ (value.ok.buffer),
           options
         )
         return ok({})

src/middleware/withDelegationsStorage.js

@@ -4,6 +4,10 @@
  * @typedef {import('./withEgressTracker.types.js').Context} EgressTrackerContext
  */
 
+import { Space } from '@storacha/capabilities'
+import { SpaceDID } from '@storacha/capabilities/utils'
+import { DID } from '@ucanto/core'
+
 /**
  * The egress tracking handler must be enabled after the rate limiting, authorized space,
  * and egress client handlers, and before any handler that serves the response body.
@@ -18,14 +22,29 @@ export function withEgressTracker (handler) {
       return handler(req, env, ctx)
     }
 
+    // Check rollout percentage for gradual deployment
+    const rolloutPercentage = parseInt(env.FF_EGRESS_TRACKER_ROLLOUT_PERCENTAGE || '100')
+    const shouldTrack = Math.random() * 100 < rolloutPercentage
+    if (!shouldTrack) {
+      return handler(req, env, ctx)
+    }
+
     // If the space is not defined, it is a legacy request and we can't track egress
     const space = ctx.space
     if (!space) {
+      console.log('Egress tracking skipped: no space context available (legacy request)')
       return handler(req, env, ctx)
     }
+    console.log('Egress tracking enabled for space:', space)
 
-    if (!ctx.egressClient) {
-      console.error('EgressClient is not defined')
+    // Check if Cloudflare Queue is available for egress tracking
+    if (!env.EGRESS_QUEUE) {
+      console.error('EGRESS_QUEUE is not defined')
+      return handler(req, env, ctx)
+    }
+
+    if (!env.UPLOAD_SERVICE_DID) {
+      console.error('UPLOAD_SERVICE_DID is not defined')
       return handler(req, env, ctx)
     }
 
@@ -35,17 +54,44 @@ export function withEgressTracker (handler) {
     }
 
     const responseBody = response.body.pipeThrough(
-      createByteCountStream((totalBytesServed) => {
+      createByteCountStream(async (totalBytesServed) => {
         if (totalBytesServed > 0) {
-          // Non-blocking call to the accounting service to record egress
-          ctx.waitUntil(
-            ctx.egressClient.record(
-              /** @type {import('@ucanto/principal/ed25519').DIDKey} */(space),
-              ctx.dataCid,
-              totalBytesServed,
-              new Date()
+          try {
+            // Create UCAN invocation for egress record
+            const invocation = Space.egressRecord.invoke({
+              issuer: ctx.gatewayIdentity,
+              audience: DID.parse(env.UPLOAD_SERVICE_DID),
+              with: SpaceDID.from(space),
+              nb: {
+                resource: ctx.dataCid,
+                bytes: totalBytesServed,
+                servedAt: new Date().getTime()
+              },
+              expiration: Infinity,
+              nonce: Date.now().toString(),
+              proofs: ctx.delegationProofs
+            })
+
+            // Serialize and send to Cloudflare Queue
+            const delegation = await invocation.delegate()
+            const archiveResult = await delegation.archive()
+            if (archiveResult.error) {
+              console.error('Failed to serialize egress invocation:', archiveResult.error)
+              return
+            }
+            const serializedInvocation = archiveResult.ok
+
+            // Non-blocking call to queue the invocation
+            ctx.waitUntil(
+              env.EGRESS_QUEUE.send({
+                messageId: delegation.cid,
+                invocation: serializedInvocation,
+                timestamp: Date.now()
+              })
             )
-          )
+          } catch (error) {
+            console.error('Failed to create or queue egress invocation:', error)
+          }
         }
       })
     )

src/middleware/withEgressTracker.js

@@ -1,10 +1,13 @@
 import { IpfsUrlContext, Environment as MiddlewareEnvironment } from '@web3-storage/gateway-lib'
-import { EgressClientContext } from './withEgressClient.types.js'
-import { SpaceContext } from './withAuthorizedSpace.types.js'
+import { SpaceContext, DelegationProofsContext } from './withAuthorizedSpace.types.js'
+import { GatewayIdentityContext } from './withGatewayIdentity.types.js'
 
 export interface Environment extends MiddlewareEnvironment {
   FF_EGRESS_TRACKER_ENABLED: string
+  FF_EGRESS_TRACKER_ROLLOUT_PERCENTAGE?: string
+  EGRESS_QUEUE: Queue
+  UPLOAD_SERVICE_DID: string
 }
 
-export interface Context extends IpfsUrlContext, SpaceContext, EgressClientContext {
+export interface Context extends IpfsUrlContext, SpaceContext, GatewayIdentityContext, DelegationProofsContext {
 }

src/middleware/withEgressTracker.types.ts

@@ -30,6 +30,7 @@ export function withUcanInvocationHandler (handler) {
       headers: Object.fromEntries(request.headers)
     })
 
+    // @ts-expect-error - ByteView is compatible with BodyInit but TypeScript doesn't recognize it
     return new Response(body, { headers, status: status ?? 200 })
   }
 }

src/middleware/withUcanInvocationHandler.js

@@ -24,7 +24,7 @@ import { mockBucketService } from '../helpers/bucket.js'
 import { fromShardArchives } from '@storacha/blob-index/util'
 import { CAR_CODE } from '../../src/constants.js'
 import http from 'node:http'
-/** @import { Block, Position } from 'carstream' */
+/** @import { Block, Position } from 'carstream/api' */
 
 /**
  * @param {{ arrayBuffer: () => Promise<ArrayBuffer> }} a