chore: reading repo variables

storacha · Oct 21, 2024 · b8275dd · b8275dd
1 parent 8279643
commit b8275dd
Show file tree

Hide file tree

Showing 3 changed files with 42 additions and 7 deletions.
diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml
@@ -25,3 +25,5 @@ jobs:
           AWS_ACCESS_KEY_ID: 'NOSUCH'
           AWS_SECRET_ACCESS_KEY: 'NOSUCH'
           STRIPE_TEST_SECRET_KEY: ${{ secrets.STRIPE_TEST_SECRET_KEY }}
+          STRIPE_BILLING_METER_ID: ${{ vars.STRIPE_BILLING_METER_ID }}
+          STRIPE_BILLING_METER_EVENT_NAME: ${{ vars.STRIPE_BILLING_METER_EVENT_NAME }}
diff --git a/billing/functions/egress-traffic-handler.js b/billing/functions/egress-traffic-handler.js
@@ -5,6 +5,7 @@ import { mustGetEnv } from '../../lib/env.js'
 import { createCustomerStore } from '../tables/customer.js'
 import Stripe from 'stripe'
 import { Config } from 'sst/node/config'
+import { DeleteMessageCommand, SQSClient } from '@aws-sdk/client-sqs'
 
 Sentry.AWSLambda.init({
     environment: process.env.SST_STAGE,
@@ -25,7 +26,7 @@ Sentry.AWSLambda.init({
 
 /**
  * AWS Lambda handler to process egress events from the egress traffic queue.
- * Each event is a JSON object with `customer`, `resource`, `bytes` and `timestamp`.
+ * Each event is a JSON object with `customer`, `resource`, `bytes` and `servedAt`.
  * The message is then deleted from the queue when successful.
  */
 export const handler = Sentry.AWSLambda.wrapHandler(
@@ -37,8 +38,8 @@ export const handler = Sentry.AWSLambda.wrapHandler(
         /** @type {CustomHandlerContext|undefined} */
         const customContext = context?.clientContext?.Custom
         const region = customContext?.region ?? mustGetEnv('AWS_REGION')
-        // const queueUrl = customContext?.egressTrafficQueueUrl ?? mustGetEnv('EGRESS_TRAFFIC_QUEUE_URL')
-        // const sqsClient = new SQSClient({ region })
+        const queueUrl = customContext?.egressTrafficQueueUrl ?? mustGetEnv('EGRESS_TRAFFIC_QUEUE_URL')
+        const sqsClient = new SQSClient({ region })
         const customerTable = customContext?.customerTable ?? mustGetEnv('CUSTOMER_TABLE_NAME')
         const customerStore = customContext?.customerStore ?? createCustomerStore({ region }, { tableName: customerTable })
 
@@ -64,10 +65,10 @@ export const handler = Sentry.AWSLambda.wrapHandler(
                  * SQS requires explicit acknowledgment that a message has been successfully processed.
                  * This is done by deleting the message from the queue using its ReceiptHandle
                  */
-                // await sqsClient.send(new DeleteMessageCommand({
-                //     QueueUrl: queueUrl,
-                //     ReceiptHandle: record.receiptHandle
-                // }))
+                await sqsClient.send(new DeleteMessageCommand({
+                    QueueUrl: queueUrl,
+                    ReceiptHandle: record.receiptHandle
+                }))
             } catch (error) {
                 console.error('Error processing egress event:', error)
             }
@@ -111,6 +112,7 @@ async function recordEgress(customerStore, stripe, billingMeterEventName, egress
         }
     }
 
+    // TODO (fforbeck): implement some retry logic in case rate limiting errors
     /** @type {import('stripe').Stripe.Billing.MeterEvent} */
     const meterEvent = await stripe.billing.meterEvents.create({
         event_name: billingMeterEventName,

diff --git a/stacks/billing-stack.js b/stacks/billing-stack.js
@@ -179,5 +179,36 @@ export function BillingStack ({ stack, app }) {
     CustomDomain: customDomain ? `https://${customDomain.domainName}` : 'Set BILLING_HOSTED_ZONE in env to deploy to a custom domain'
   })
 
+  // Lambda that handles egress traffic tracking
+  const egressTrafficQueueHandler = new Function(stack, 'egress-traffic-queue-handler', {
+    permissions: [customerTable],
+    handler: 'billing/functions/egress-traffic-queue.handler',
+    timeout: '15 minutes',
+    bind: [stripeSecretKey],
+    environment: {
+      AWS_REGION: stack.region,
+      CUSTOMER_TABLE_NAME: customerTable.tableName,
+      // TODO (fforbeck): make this a config based on the env: local, staging, prod
+      STRIPE_BILLING_METER_EVENT_NAME: 'test-gateway-egress-traffic'
+    }
+  })
+
+  // Queue for egress traffic tracking
+  const egressTrafficDLQ = new Queue(stack, 'egress-traffic-dlq', {
+    cdk: { queue: { retentionPeriod: Duration.days(14) } }
+  })
+  const egressTrafficQueue = new Queue(stack, 'egress-traffic-queue', {
+    consumer: {
+      function: egressTrafficQueueHandler,
+      deadLetterQueue: egressTrafficDLQ.cdk.queue,
+      cdk: { eventSource: { batchSize: 1 } }
+    },
+    cdk: { queue: { visibilityTimeout: Duration.seconds(60) } }
+  })
+
+  stack.addOutputs({
+    EgressTrafficQueueURL: egressTrafficQueue.queueUrl
+  })
+
   return { billingCron }
 }