Implement upsert functionality for user actions to handle concurrent requests and prevent duplicate entries

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<UserActionEntity> {
+		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<UserActionEntity> {
 		const actionQb = this.createQueryBuilder('user_action')
 			.leftJoinAndSelect('user_action.user', 'user')

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<UserActionEntity>;
 
+	upsertUserAction(userId: string, message: string): Promise<UserActionEntity>;
+
 	findUserActionWithoutSentMail(userId: string): Promise<UserActionEntity>;
 
 	findAllNonFinishedActionsTwoWeeksOld(): Promise<Array<UserActionEntity>>;

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<void> {
-		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<boolean> {
+		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<UserActionEntity | null> {
+	private async findUserAction(userId: string): Promise<UserActionEntity | null> {
 		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();
 	}
 

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<CreatedUserActionDs> {
 		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);
 	}
 }

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;

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');
+});

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<UserActionEntity>;
+};
+
+function makeUseCase(
+	actionsByUserId: Record<string, UserActionEntity | null | (() => never)>,
+	users: Array<{ id: string; email: string }> = [USER],
+): Fixture {
+	const saved: Array<UserActionEntity> = [];
+	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);
+});