From c72b1982cec6d73b0671eaf32e444d708dd30cd2 Mon Sep 17 00:00:00 2001 From: Artem Niehrieiev Date: Mon, 12 Oct 2026 12:14:15 +0000 Subject: [PATCH] Implement upsert functionality for user actions to handle concurrent requests and prevent duplicate entries --- ...user-action-custom-repository-extension.ts | 20 ++++ .../user-action.repository.interface.ts | 2 + ...sers-actions-and-mailing-users.use.case.ts | 34 ++++-- .../use-cases/create-user-action.use.case.ts | 14 +-- .../utils/build-new-user-action-entity.ts | 9 -- .../non-saas-user-action-e2e.test.ts | 102 ++++++++++++++++ ...actions-and-mailing-users.use.case.test.ts | 112 ++++++++++++++++++ 7 files changed, 261 insertions(+), 32 deletions(-) create mode 100644 backend/test/ava-tests/non-saas-tests/non-saas-user-action-e2e.test.ts create mode 100644 backend/test/ava-tests/unit-tests/check-users-actions-and-mailing-users.use.case.test.ts diff --git a/backend/src/entities/user-actions/repository/user-action-custom-repository-extension.ts b/backend/src/entities/user-actions/repository/user-action-custom-repository-extension.ts index f24f76881..7fe1e9a3e 100644 --- a/backend/src/entities/user-actions/repository/user-action-custom-repository-extension.ts +++ b/backend/src/entities/user-actions/repository/user-action-custom-repository-extension.ts @@ -7,6 +7,26 @@ export const userActionCustomRepositoryExtension = { return await this.save(userAction); }, + /** + * Atomically creates the user's action or overwrites its message. + * `user_action.userId` is UNIQUE (one-to-one with `user`); a read-then-insert races with itself when + * the same user sends two requests at once (e.g. two tabs opening "add connection") and the second + * insert fails with `duplicate key value violates unique constraint "REL_c025478b45e60017ed10c77f99"`. + * `INSERT … ON CONFLICT ("userId") DO UPDATE` makes the second request an update instead. + */ + async upsertUserAction(userId: string, message: string): Promise { + await this.createQueryBuilder() + .insert() + .into(UserActionEntity) + .values({ user: { id: userId }, message }) + .orUpdate(['message'], ['userId']) + .execute(); + return await this.createQueryBuilder('user_action') + .innerJoin('user_action.user', 'user') + .where('user.id = :userId', { userId }) + .getOne(); + }, + async findUserActionWithoutSentMail(userId: string): Promise { const actionQb = this.createQueryBuilder('user_action') .leftJoinAndSelect('user_action.user', 'user') diff --git a/backend/src/entities/user-actions/repository/user-action.repository.interface.ts b/backend/src/entities/user-actions/repository/user-action.repository.interface.ts index 0786f6830..1df63227a 100644 --- a/backend/src/entities/user-actions/repository/user-action.repository.interface.ts +++ b/backend/src/entities/user-actions/repository/user-action.repository.interface.ts @@ -3,6 +3,8 @@ import { UserActionEntity } from '../user-action.entity.js'; export interface IUserActionRepository { saveNewOrUpdatedUserAction(userAction: UserActionEntity): Promise; + upsertUserAction(userId: string, message: string): Promise; + findUserActionWithoutSentMail(userId: string): Promise; findAllNonFinishedActionsTwoWeeksOld(): Promise>; diff --git a/backend/src/entities/user-actions/use-cases/check-users-actions-and-mailing-users.use.case.ts b/backend/src/entities/user-actions/use-cases/check-users-actions-and-mailing-users.use.case.ts index 6c0df72f3..26f84e48e 100644 --- a/backend/src/entities/user-actions/use-cases/check-users-actions-and-mailing-users.use.case.ts +++ b/backend/src/entities/user-actions/use-cases/check-users-actions-and-mailing-users.use.case.ts @@ -30,8 +30,8 @@ export class CheckUsersActionsAndMailingUsersUseCase implements ICheckUsersActio for (const user of batch) { try { await queue.add(async () => { - await this.updateOrCreateActionForUser(user); - if (user.email) { + const shouldSendEmail = await this.updateOrCreateActionForUser(user); + if (shouldSendEmail && user.email) { emails.push(user.email.toLowerCase()); } }); @@ -54,30 +54,38 @@ export class CheckUsersActionsAndMailingUsersUseCase implements ICheckUsersActio } } - private async updateOrCreateActionForUser(u: { id: string; email: string }): Promise { - const userActionForUser = await this.findUserActionWithoutSentMail(u.id); + /** + * Marks the user's action as mailed, creating the action row when the user has none. + * Returns `true` when a reminder email should be sent to this user. + * + * `user_action.userId` is UNIQUE (one-to-one with `user`), so the lookup must ignore `mail_sent`: + * if a row exists it is updated in place — never inserted alongside — otherwise the insert fails with + * `duplicate key value violates unique constraint "REL_c025478b45e60017ed10c77f99"`. + */ + private async updateOrCreateActionForUser(u: { id: string; email: string }): Promise { + const userActionForUser = await this.findUserAction(u.id); if (userActionForUser) { - if (!userActionForUser.mail_sent) { - userActionForUser.mail_sent = true; - await this.userActionRepository.save(userActionForUser); - } else { - return; + if (userActionForUser.mail_sent) { + return false; } + userActionForUser.mail_sent = true; + await this.userActionRepository.save(userActionForUser); + return true; } const foundUser = await this.userRepository.findOne({ where: { id: u.id } }); if (!foundUser) { - return; + return false; } const newUserActionEntity = buildNewConnectionNotFinishedEmailSentAction(foundUser); await this.userActionRepository.save(newUserActionEntity); + return true; } - private async findUserActionWithoutSentMail(userId: string): Promise { + private async findUserAction(userId: string): Promise { const actionQb = this.userActionRepository .createQueryBuilder('user_action') .leftJoinAndSelect('user_action.user', 'user') - .andWhere('user.id = :user_id', { user_id: userId }) - .andWhere('user_action.mail_sent = :mail_sent', { mail_sent: false }); + .andWhere('user.id = :user_id', { user_id: userId }); return await actionQb.getOne(); } diff --git a/backend/src/entities/user-actions/use-cases/create-user-action.use.case.ts b/backend/src/entities/user-actions/use-cases/create-user-action.use.case.ts index d9c4ebbcc..80e33daf4 100644 --- a/backend/src/entities/user-actions/use-cases/create-user-action.use.case.ts +++ b/backend/src/entities/user-actions/use-cases/create-user-action.use.case.ts @@ -6,7 +6,6 @@ import { Messages } from '../../../exceptions/text/messages.js'; import { CreateUserActionDs } from '../application/data-sctructures/create-user-action.ds.js'; import { CreatedUserActionDs } from '../application/data-sctructures/created-user-action.ds.js'; import { buildCreatedUserActionDs } from '../utils/build-created-user-action-ds.js'; -import { buildNewUserActionEntity } from '../utils/build-new-user-action-entity.js'; import { ICreateUserAction } from './use-cases-interfaces.js'; @Injectable() @@ -23,7 +22,7 @@ export class CreateUserActionUseCase protected async implementation(actionData: CreateUserActionDs): Promise { const { message, userId } = actionData; - const foundUser = await this._dbContext.userRepository.findOneUserWithUserAction(userId); + const foundUser = await this._dbContext.userRepository.findOneUserById(userId); if (!foundUser) { throw new HttpException( { @@ -32,14 +31,9 @@ export class CreateUserActionUseCase HttpStatus.BAD_REQUEST, ); } - const userAction = foundUser.user_action; - if (userAction) { - userAction.message = message; - const updatedAction = await this._dbContext.userActionRepository.saveNewOrUpdatedUserAction(userAction); - return buildCreatedUserActionDs(updatedAction); - } - const newUserAction = buildNewUserActionEntity(actionData, foundUser); - const savedUserAction = await this._dbContext.userActionRepository.saveNewOrUpdatedUserAction(newUserAction); + // A user has at most one action row (UNIQUE userId). Upsert instead of read-then-insert so that + // concurrent requests for the same user cannot both take the insert branch. + const savedUserAction = await this._dbContext.userActionRepository.upsertUserAction(userId, message); return buildCreatedUserActionDs(savedUserAction); } } diff --git a/backend/src/entities/user-actions/utils/build-new-user-action-entity.ts b/backend/src/entities/user-actions/utils/build-new-user-action-entity.ts index 33df61f5b..0724b03fb 100644 --- a/backend/src/entities/user-actions/utils/build-new-user-action-entity.ts +++ b/backend/src/entities/user-actions/utils/build-new-user-action-entity.ts @@ -1,16 +1,7 @@ import { UserActionEnum } from '../../../enums/user-action.enum.js'; import { UserEntity } from '../../user/user.entity.js'; -import { CreateUserActionDs } from '../application/data-sctructures/create-user-action.ds.js'; import { UserActionEntity } from '../user-action.entity.js'; -export function buildNewUserActionEntity(actionData: CreateUserActionDs, user: UserEntity): UserActionEntity { - const { message } = actionData; - const newUserAction = new UserActionEntity(); - newUserAction.user = user; - newUserAction.message = message; - return newUserAction; -} - export function buildNewConnectionNotFinishedEmailSentAction(user: UserEntity): UserActionEntity { const newUserAction = new UserActionEntity(); newUserAction.user = user; diff --git a/backend/test/ava-tests/non-saas-tests/non-saas-user-action-e2e.test.ts b/backend/test/ava-tests/non-saas-tests/non-saas-user-action-e2e.test.ts new file mode 100644 index 000000000..e7131583e --- /dev/null +++ b/backend/test/ava-tests/non-saas-tests/non-saas-user-action-e2e.test.ts @@ -0,0 +1,102 @@ +import { INestApplication, ValidationPipe } from '@nestjs/common'; +import { Test } from '@nestjs/testing'; +import test from 'ava'; +import { ValidationError } from 'class-validator'; +import cookieParser from 'cookie-parser'; +import request from 'supertest'; +import { ApplicationModule } from '../../../src/app.module.js'; +import { WinstonLogger } from '../../../src/entities/logging/winston-logger.js'; +import { UserActionEnum } from '../../../src/enums/user-action.enum.js'; +import { AllExceptionsFilter } from '../../../src/exceptions/all-exceptions.filter.js'; +import { ValidationException } from '../../../src/exceptions/custom-exceptions/validation-exception.js'; +import { Cacher } from '../../../src/helpers/cache/cacher.js'; +import { DatabaseModule } from '../../../src/shared/database/database.module.js'; +import { DatabaseService } from '../../../src/shared/database/database.service.js'; +import { + createInitialTestUser, + registerUserAndReturnUserInfo, +} from '../../utils/register-user-and-return-user-info.js'; +import { setSaasEnvVariable } from '../../utils/set-saas-env-variable.js'; +import { TestUtils } from '../../utils/test.utils.js'; + +// `user_action.userId` is UNIQUE (one row per user). The frontend posts an action every time the +// "add connection" page opens, so the endpoint must be an upsert: a repeat (or concurrent) request for +// the same user updates the one row instead of failing with +// `duplicate key value violates unique constraint "REL_c025478b45e60017ed10c77f99"`. + +let app: INestApplication; +let currentTest: string; + +test.before(async () => { + setSaasEnvVariable(); + const moduleFixture = await Test.createTestingModule({ + imports: [ApplicationModule, DatabaseModule], + providers: [DatabaseService, TestUtils], + }).compile(); + app = moduleFixture.createNestApplication() as any; + + app.use(cookieParser()); + app.useGlobalFilters(new AllExceptionsFilter(app.get(WinstonLogger))); + app.useGlobalPipes( + new ValidationPipe({ + exceptionFactory(validationErrors: ValidationError[] = []) { + return new ValidationException(validationErrors); + }, + }), + ); + await app.init(); + await createInitialTestUser(app); + app.getHttpServer().listen(0); +}); + +test.after(async () => { + try { + await Cacher.clearAllCache(); + await app.close(); + } catch (e) { + console.error('After tests error ' + e); + } +}); + +function postAction(token: string, message: string) { + return request(app.getHttpServer()) + .post('/action') + .send({ message }) + .set('Cookie', token) + .set('Content-Type', 'application/json') + .set('Accept', 'application/json'); +} + +currentTest = 'POST /action'; + +test.serial( + `${currentTest} should create the user action on first call and update the same row afterwards`, + async (t) => { + const { token } = await registerUserAndReturnUserInfo(app); + + const first = await postAction(token, UserActionEnum.CONNECTION_CREATION_NOT_FINISHED); + t.is(first.status, 201, JSON.stringify(first.body)); + t.truthy(first.body.id); + t.is(first.body.message, UserActionEnum.CONNECTION_CREATION_NOT_FINISHED); + t.is(first.body.mail_sent, false); + + const second = await postAction(token, UserActionEnum.CONNECTION_CREATION_FINISHED); + t.is(second.status, 201, JSON.stringify(second.body)); + t.is(second.body.id, first.body.id, 'the existing row is updated, not duplicated'); + t.is(second.body.message, UserActionEnum.CONNECTION_CREATION_FINISHED); + }, +); + +test.serial(`${currentTest} should tolerate concurrent calls for the same user`, async (t) => { + const { token } = await registerUserAndReturnUserInfo(app); + + const responses = await Promise.all( + Array.from({ length: 8 }, () => postAction(token, UserActionEnum.CONNECTION_CREATION_NOT_FINISHED)), + ); + + for (const response of responses) { + t.is(response.status, 201, JSON.stringify(response.body)); + } + const ids = new Set(responses.map((response) => response.body.id)); + t.is(ids.size, 1, 'every concurrent request resolved to the single row for this user'); +}); diff --git a/backend/test/ava-tests/unit-tests/check-users-actions-and-mailing-users.use.case.test.ts b/backend/test/ava-tests/unit-tests/check-users-actions-and-mailing-users.use.case.test.ts new file mode 100644 index 000000000..c3e2a57ef --- /dev/null +++ b/backend/test/ava-tests/unit-tests/check-users-actions-and-mailing-users.use.case.test.ts @@ -0,0 +1,112 @@ +import test from 'ava'; +import { CheckUsersActionsAndMailingUsersUseCase } from '../../../src/entities/user-actions/use-cases/check-users-actions-and-mailing-users.use.case.js'; +import { UserActionEntity } from '../../../src/entities/user-actions/user-action.entity.js'; +import { UserActionEnum } from '../../../src/enums/user-action.enum.js'; + +// `user_action.userId` is UNIQUE (one-to-one with `user`), so the 08:00 reminder cron must update a user's +// existing action row in place. The original code marked the row as sent and then fell through to insert a +// second row for the same user, which failed with +// `duplicate key value violates unique constraint "REL_c025478b45e60017ed10c77f99"` and dropped the email. + +const USER = { id: 'a3f1c9d2-4b5e-4c7a-8d9e-0f1a2b3c4d5e', email: 'Person@Example.com' }; + +type Fixture = { + useCase: CheckUsersActionsAndMailingUsersUseCase; + saved: Array; +}; + +function makeUseCase( + actionsByUserId: Record never)>, + users: Array<{ id: string; email: string }> = [USER], +): Fixture { + const saved: Array = []; + const userActionRepository = { + save: async (entity: UserActionEntity) => { + saved.push(entity); + return entity; + }, + }; + const userRepository = { + findOne: async ({ where }: { where: { id: string } }) => users.find((u) => u.id === where.id) ?? null, + }; + const useCase = new CheckUsersActionsAndMailingUsersUseCase(userActionRepository as any, userRepository as any); + // The private lookups go through TypeORM query builders; replace them so the decision logic runs without a DB. + Object.assign(useCase, { + findDistinctUsersForProcessing: async () => users, + findUserAction: async (userId: string) => { + const entry = actionsByUserId[userId]; + return typeof entry === 'function' ? entry() : (entry ?? null); + }, + }); + return { useCase, saved }; +} + +function existingAction(mailSent: boolean): UserActionEntity { + const action = new UserActionEntity(); + action.id = 'existing-action-id'; + action.message = UserActionEnum.CONNECTION_CREATION_NOT_FINISHED; + action.mail_sent = mailSent; + return action; +} + +test('an existing unsent action is marked sent in place and the user is emailed', async (t) => { + const { useCase, saved } = makeUseCase({ [USER.id]: existingAction(false) }); + + const emails = await useCase.execute(); + + t.deepEqual(emails, ['person@example.com']); + t.is(saved.length, 1, 'exactly one save — no second row for the same user'); + t.is(saved[0].id, 'existing-action-id'); + t.true(saved[0].mail_sent); +}); + +test('an existing already-sent action is left alone and the user is not emailed again', async (t) => { + const { useCase, saved } = makeUseCase({ [USER.id]: existingAction(true) }); + + const emails = await useCase.execute(); + + t.deepEqual(emails, []); + t.is(saved.length, 0); +}); + +test('a user without an action gets one created as already mailed and is emailed', async (t) => { + const { useCase, saved } = makeUseCase({ [USER.id]: null }); + + const emails = await useCase.execute(); + + t.deepEqual(emails, ['person@example.com']); + t.is(saved.length, 1); + t.is(saved[0].id, undefined, 'a new row, not the existing one'); + t.is(saved[0].message, UserActionEnum.CONNECTION_CREATION_NOT_FINISHED); + t.true(saved[0].mail_sent); + t.is(saved[0].user.id, USER.id); +}); + +test('a user that no longer exists is skipped without an insert', async (t) => { + const { useCase, saved } = makeUseCase({ [USER.id]: null }, []); + Object.assign(useCase, { findDistinctUsersForProcessing: async () => [USER] }); + + const emails = await useCase.execute(); + + t.deepEqual(emails, []); + t.is(saved.length, 0); +}); + +test('a failure for one user does not block the others', async (t) => { + const other = { id: 'b4e2d0c3-5c6f-4d8b-9e0f-1a2b3c4d5e6f', email: 'other@example.com' }; + const { useCase, saved } = makeUseCase( + { + [USER.id]: () => { + throw new Error('boom'); + }, + [other.id]: null, + }, + [USER, other], + ); + + const emails = await useCase.execute(); + + t.deepEqual(emails, ['other@example.com']); + t.is(saved.length, 1); + t.is(saved[0].user.id, other.id); +});