feat: restrict in flight psa requests with max quota

.env.tpl

@@ -42,4 +42,7 @@ S3_ENDPOINT = http://localhost:9095
 S3_REGION = test
 S3_ACCESS_KEY_ID = test
 S3_SECRET_ACCESS_KEY = test
-S3_BUCKET_NAME = test
+S3_BUCKET_NAME = test
+
+# PSA
+PSA_QUOTA = 100

packages/api/src/bindings.d.ts

@@ -75,6 +75,9 @@ export interface ServiceConfiguration {
 
   /** Mailchimp api key */
   MAILCHIMP_API_KEY: string
+
+  /** PSA quota with number of in flight requests possible */
+  PSA_QUOTA: number
 }
 
 export interface Ucan {
@@ -96,6 +99,7 @@ export interface AuthOptions {
   checkHasAccountRestriction?: boolean
   checkHasDeleteRestriction?: boolean
   checkHasPsaAccess?: boolean
+  checkHasPsaQuota?: boolean
 }
 
 export interface RouteContext {

packages/api/src/config.js

@@ -18,6 +18,8 @@ const CLUSTER_SERVICE_URLS = {
   IpfsCluster3: 'https://nft3.storage.ipfscluster.io/api/',
 }
 
+const DEFAULT_PSA_QUOTA = 100
+
 /** @type ServiceConfiguration|undefined */
 let _globalConfig
 
@@ -99,6 +101,7 @@ export function serviceConfigFromVariables(vars) {
     S3_ACCESS_KEY_ID: vars.S3_ACCESS_KEY_ID,
     S3_SECRET_ACCESS_KEY: vars.S3_SECRET_ACCESS_KEY,
     S3_BUCKET_NAME: vars.S3_BUCKET_NAME,
+    PSA_QUOTA: vars.PSA_QUOTA ? Number(vars.PSA_QUOTA) : DEFAULT_PSA_QUOTA,
     PRIVATE_KEY: vars.PRIVATE_KEY,
     // These are injected in esbuild
     // @ts-ignore
@@ -144,6 +147,7 @@ export function loadConfigVariables() {
     'S3_ACCESS_KEY_ID',
     'S3_SECRET_ACCESS_KEY',
     'S3_BUCKET_NAME',
+    'PSA_QUOTA',
   ]
 
   for (const name of required) {

packages/api/src/errors.js

@@ -197,6 +197,17 @@ export class ErrorPinningUnauthorized extends HTTPError {
 }
 ErrorPinningUnauthorized.CODE = 'ERROR_PINNING_UNAUTHORIZED'
 
+export class ErrorPinningQuotaExceeded extends HTTPError {
+  constructor(
+    msg = 'Pinning quota exceeded for this user, please wait for in flight pinning requests to end or delete failed ones.'
+  ) {
+    super(msg, 429)
+    this.name = 'PinningQuotaExceeded'
+    this.code = ErrorPinningQuotaExceeded.CODE
+  }
+}
+ErrorPinningQuotaExceeded.CODE = 'ERROR_PINNING_QUOTA_EXCEEDED'
+
 export class ErrorDeleteRestricted extends HTTPError {
   constructor(msg = 'Delete operations restricted.') {
     super(msg, 403)

packages/api/src/index.js

@@ -44,6 +44,7 @@ const r = new Router(getContext, {
 const checkHasAccountRestriction = true
 const checkHasDeleteRestriction = true
 const checkHasPsaAccess = true
+const checkHasPsaQuota = true
 const checkUcan = true
 
 // Monitoring
@@ -98,6 +99,7 @@ r.add(
   withAuth(withMode(pinsAdd, RW), {
     checkHasPsaAccess,
     checkHasAccountRestriction,
+    checkHasPsaQuota,
   }),
   [postCors]
 )
@@ -107,6 +109,7 @@ r.add(
   withAuth(withMode(pinsReplace, RW), {
     checkHasPsaAccess,
     checkHasAccountRestriction,
+    checkHasPsaQuota,
   }),
   [postCors]
 )

packages/api/src/middleware/auth.js

@@ -2,10 +2,14 @@ import {
   ErrorAccountRestricted,
   ErrorDeleteRestricted,
   ErrorPinningUnauthorized,
+  ErrorPinningQuotaExceeded,
 } from '../errors'
+import { getServiceConfig } from '../config.js'
 import { validate } from '../utils/auth'
 import { hasTag } from '../utils/utils'
 
+const { PSA_QUOTA } = getServiceConfig()
+
 /**
  *
  * @param {import('../bindings').Handler} handler
@@ -37,6 +41,16 @@ export function withAuth(handler, options) {
       throw new ErrorPinningUnauthorized()
     }
 
+    if (options?.checkHasPsaQuota) {
+      const countPendingPsaRequests = await auth.db.getPendingPsaRequestsCount(
+        auth.user.id
+      )
+
+      if (countPendingPsaRequests >= PSA_QUOTA) {
+        throw new ErrorPinningQuotaExceeded()
+      }
+    }
+
     return handler(event, { ...ctx, auth })
   }
 }

packages/api/src/utils/db-client.js

@@ -15,6 +15,7 @@ export const PIN_SERVICES = [
 ]
 /** @type {Array<definitions['pin']['status']>} */
 export const PIN_STATUSES = ['PinQueued', 'Pinning', 'Pinned', 'PinError']
+export const PIN_IN_FLIGHT_STATUSES = ['PinQueued', 'Pinning', 'PinError']
 
 export class DBClient {
   /**
@@ -635,6 +636,27 @@ export class DBClient {
 
     return stats
   }
+
+  /**
+   * @param {number} userId
+   */
+  async getPendingPsaRequestsCount(userId) {
+    const { error, count } = await this.client
+      .from('upload')
+      .select(this.uploadQuery, { count: 'exact', head: true })
+      .eq('user_id', userId)
+      .is('deleted_at', null)
+      .in('content.pin.status', PIN_IN_FLIGHT_STATUSES)
+      .in('type', ['Remote'])
+
+    if (error) {
+      throw new DBError(error)
+    }
+
+    console.log('count: ' + count)
+
+    return count || 0
+  }
 }
 
 export class DBError extends Error {

packages/api/test/pin-add.spec.js

@@ -17,7 +17,7 @@ describe('Pin add ', () => {
   /** @type{DBTestClient} */
   let client
 
-  before(async () => {
+  beforeEach(async () => {
     client = await createClientWithUser()
   })
 
@@ -229,4 +229,31 @@ describe('Pin add ', () => {
     assert.strictEqual(pin.meta.invalid, undefined)
     assert.strictEqual(pin.meta.valid, 'string')
   })
+
+  it('should restrict requests to quota', async () => {
+    const cids = [
+      'bafkreidyeivj7adnnac6ljvzj2e3rd5xdw3revw4da7mx2ckrstapoupoq',
+      'bafybeih74zqc6kamjpruyra4e4pblnwdpickrvk4hvturisbtveghflovq',
+    ]
+
+    await Promise.all(
+      cids.map(async (cid) => {
+        const res = await fetch('pins', {
+          method: 'POST',
+          headers: { Authorization: `Bearer ${client.token}` },
+          body: JSON.stringify({ cid }),
+        })
+        assert.strictEqual(res.status, 200)
+      })
+    )
+
+    const extraQuotaCid =
+      'bafkreihbjbbccwxn7hzv5hun5pxuswide7q3lhjvfbvmd7r3kf2sodybgi'
+    const res = await fetch('pins', {
+      method: 'POST',
+      headers: { Authorization: `Bearer ${client.token}` },
+      body: JSON.stringify({ cid: extraQuotaCid }),
+    })
+    assert.strictEqual(res.status, 429)
+  })
 })

packages/api/test/scripts/globals.js

@@ -24,3 +24,5 @@ globalThis.S3_REGION = 'test'
 globalThis.S3_ACCESS_KEY_ID = 'test'
 globalThis.S3_SECRET_ACCESS_KEY = 'test'
 globalThis.S3_BUCKET_NAME = 'test'
+
+globalThis.PSA_QUOTA = '2'