fix(requests): serialize request creation per user

server/entity/MediaRequest.test.ts

@@ -0,0 +1,99 @@
+import assert from 'node:assert/strict';
+import { beforeEach, describe, it, mock } from 'node:test';
+
+import ExternalAPI from '@server/api/externalapi';
+import { MediaType } from '@server/constants/media';
+import { getRepository } from '@server/datasource';
+import {
+  DuplicateMediaRequestError,
+  MediaRequest,
+  QuotaRestrictedError,
+} from '@server/entity/MediaRequest';
+import { User } from '@server/entity/User';
+import { setupTestDb } from '@server/test/db';
+
+// get is a prototype method unlike getMovie, and replaces the cache lookup too
+const externalApiGetMock = mock.method(
+  ExternalAPI.prototype as unknown as {
+    get: (endpoint: string) => Promise<unknown>;
+  },
+  'get',
+  async (endpoint: string) => {
+    const movieId = Number(endpoint.replace('/movie/', ''));
+
+    if (!movieId) {
+      throw new Error(`Unstubbed external endpoint: ${endpoint}`);
+    }
+
+    return {
+      id: movieId,
+      external_ids: {},
+      // Skips getMovie's localized fallback call
+      videos: { results: [{ type: 'Trailer', key: 'trailer' }] },
+    };
+  }
+).mock;
+
+mock.method(MediaRequest, 'sendNotification', async () => undefined);
+
+setupTestDb();
+
+beforeEach(() => {
+  externalApiGetMock.resetCalls();
+});
+
+async function seedRequester(movieQuotaLimit: number): Promise<User> {
+  const userRepository = getRepository(User);
+
+  const requester = await userRepository.findOneOrFail({
+    where: { email: 'friend@seerr.dev' },
+  });
+  requester.movieQuotaLimit = movieQuotaLimit;
+
+  return userRepository.save(requester);
+}
+
+function requestMovies(mediaIds: number[], requester: User) {
+  return Promise.allSettled(
+    mediaIds.map((mediaId) =>
+      MediaRequest.request(
+        { mediaId, mediaType: MediaType.MOVIE, is4k: false },
+        requester
+      )
+    )
+  );
+}
+
+function rejections(results: PromiseSettledResult<MediaRequest>[]) {
+  return results.filter(
+    (result): result is PromiseRejectedResult => result.status === 'rejected'
+  );
+}
+
+describe('MediaRequest.request', () => {
+  it('rejects the second of two concurrent requests at the movie quota', async () => {
+    const requestRepository = getRepository(MediaRequest);
+    const requester = await seedRequester(1);
+
+    const results = await requestMovies([11111, 22222], requester);
+    const rejected = rejections(results);
+
+    assert.strictEqual(rejected.length, 1);
+    assert.ok(rejected[0].reason instanceof QuotaRestrictedError);
+    assert.strictEqual(await requestRepository.count(), 1);
+    assert.strictEqual(externalApiGetMock.callCount(), 1);
+  });
+
+  it('rejects a concurrent duplicate request for the same movie', async () => {
+    const requestRepository = getRepository(MediaRequest);
+    const requester = await seedRequester(5);
+
+    const results = await requestMovies([33333, 33333], requester);
+    const rejected = rejections(results);
+
+    assert.strictEqual(rejected.length, 1);
+    assert.ok(rejected[0].reason instanceof DuplicateMediaRequestError);
+    assert.strictEqual(await requestRepository.count(), 1);
+    assert.strictEqual(externalApiGetMock.callCount(), 2);
+  });
+});

server/entity/MediaRequest.ts

@@ -14,6 +14,7 @@ import { Permission } from '@server/lib/permissions';
 import { getSettings } from '@server/lib/settings';
 import logger from '@server/logger';
 import { DbAwareColumn, resolveDbType } from '@server/utils/DbColumnHelper';
+import requestLock from '@server/utils/requestLock';
 import { truncate } from 'lodash';
 import {
   AfterInsert,
@@ -48,6 +49,16 @@ export class MediaRequest {
     requestBody: MediaRequestBody,
     user: User,
     options: MediaRequestOptions = {}
+  ): Promise<MediaRequest> {
+    return requestLock.dispatch(requestBody.userId || user.id, () =>
+      MediaRequest.createRequest(requestBody, user, options)
+    );
+  }
+
+  private static async createRequest(
+    requestBody: MediaRequestBody,
+    user: User,
+    options: MediaRequestOptions
   ): Promise<MediaRequest> {
     const tmdb = new TheMovieDb();
     const mediaRepository = getRepository(Media);

server/utils/asyncLock.ts

@@ -37,14 +37,14 @@ class AsyncLock {
     setImmediate(() => this.ee.emit(key));
   };
 
-  public dispatch = async (
+  public dispatch = async <T>(
     key: string | number,
-    callback: () => Promise<void>
-  ) => {
+    callback: () => Promise<T>
+  ): Promise<T> => {
     const skey = String(key);
     await this.acquire(skey);
     try {
-      await callback();
+      return await callback();
     } finally {
       this.release(skey);
     }

server/utils/requestLock.ts

@@ -0,0 +1,7 @@
+import AsyncLock from '@server/utils/asyncLock';
+
+// Keyed on user id. Never dispatch from a subscriber or transaction as a waiter
+// would block while holding the save's connection.
+const requestLock = new AsyncLock();
+
+export default requestLock;