fix(api): batch connection reorders, dispatch on failed recheck (#2507)

fluxer_api/src/api/connection/ConnectionRepository.ts

@@ -2,7 +2,7 @@
 
 import type {ConnectionType} from '@fluxer/constants/src/ConnectionConstants';
 import type {UserID} from '../BrandedTypes';
-import {deleteOneOrMany, fetchMany, fetchOne, upsertOne} from '../database/CassandraQueryExecution';
+import {BatchBuilder, deleteOneOrMany, fetchMany, fetchOne, upsertOne} from '../database/CassandraQueryExecution';
 import {Db, type DbOp} from '../database/CassandraTypes';
 import type {UserConnectionRow} from '../database/types/ConnectionTypes';
 import {UserConnections} from '../Tables';
@@ -109,6 +109,22 @@ export class ConnectionRepository extends IConnectionRepository {
 		}
 	}
 
+	async updateSortOrders(
+		userId: UserID,
+		entries: Array<{connectionType: ConnectionType; connectionId: string; sortOrder: number}>,
+	): Promise<void> {
+		const batch = new BatchBuilder();
+		for (const entry of entries) {
+			batch.addPrepared(
+				UserConnections.patchByPk(
+					{user_id: userId, connection_type: entry.connectionType, connection_id: entry.connectionId},
+					{sort_order: Db.set(entry.sortOrder)},
+				),
+			);
+		}
+		await batch.execute();
+	}
+
 	async delete(userId: UserID, connectionType: ConnectionType, connectionId: string): Promise<void> {
 		await deleteOneOrMany(
 			UserConnections.deleteByPk({user_id: userId, connection_type: connectionType, connection_id: connectionId}),

fluxer_api/src/api/connection/ConnectionService.ts

@@ -164,24 +164,24 @@ export class ConnectionService extends IConnectionService {
 		if (!updated) {
 			throw new ConnectionNotFoundError();
 		}
+		await this.dispatchConnectionsUpdate(userId);
 		if (!isValid) {
 			throw new ConnectionVerificationFailedError();
 		}
-		await this.dispatchConnectionsUpdate(userId);
 		return updated;
 	}
 
 	async reorderConnections(userId: UserID, connectionIds: Array<string>): Promise<void> {
 		const connections = await this.repository.findByUserId(userId);
-		for (let i = 0; i < connectionIds.length; i++) {
-			const connectionId = connectionIds[i];
-			const connection = connections.find((c) => c.connection_id === connectionId);
+		const byId = new Map(connections.map((connection) => [connection.connection_id, connection]));
+		const entries = new Map<string, {connectionType: ConnectionType; connectionId: string; sortOrder: number}>();
+		connectionIds.forEach((connectionId, sortOrder) => {
+			const connection = byId.get(connectionId);
 			if (connection) {
-				await this.repository.update(userId, connection.connection_type, connectionId, {
-					sort_order: i,
-				});
+				entries.set(connectionId, {connectionType: connection.connection_type, connectionId, sortOrder});
 			}
-		}
+		});
+		await this.repository.updateSortOrders(userId, Array.from(entries.values()));
 		await this.dispatchConnectionsUpdate(userId);
 	}
 

fluxer_api/src/api/connection/IConnectionRepository.ts

@@ -51,6 +51,11 @@ export abstract class IConnectionRepository {
 		params: UpdateConnectionParams,
 	): Promise<void>;
 
+	abstract updateSortOrders(
+		userId: UserID,
+		entries: Array<{connectionType: ConnectionType; connectionId: string; sortOrder: number}>,
+	): Promise<void>;
+
 	abstract delete(userId: UserID, connectionType: ConnectionType, connectionId: string): Promise<void>;
 
 	abstract count(userId: UserID): Promise<number>;

fluxer_api/src/api/connection/tests/ConnectionRecheckDispatch.test.ts

@@ -0,0 +1,108 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {APIErrorCodes} from '@fluxer/constants/src/ApiErrorCodes';
+import {ConnectionTypes, ConnectionVisibilityFlags} from '@fluxer/constants/src/ConnectionConstants';
+import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest';
+import {createUserID, type UserID} from '../../BrandedTypes';
+import type {IBlueskyOAuthService} from '../../bluesky/IBlueskyOAuthService';
+import {setCassandraQueryExecutorForTesting} from '../../database/CassandraQueryExecution';
+import type {UserConnectionRow} from '../../database/types/ConnectionTypes';
+import type {IGatewayService} from '../../infrastructure/IGatewayService';
+import {InMemoryCassandraQueryExecutor} from '../../test/InMemoryCassandraQueryExecutor';
+import {ConnectionRepository} from '../ConnectionRepository';
+import {ConnectionService} from '../ConnectionService';
+import {ConnectionVerificationFailedError} from '../errors/ConnectionVerificationFailedError';
+
+function connectionRow(userId: UserID, connectionId: string, verified: boolean): UserConnectionRow {
+	return {
+		user_id: userId,
+		connection_id: connectionId,
+		connection_type: ConnectionTypes.BLUESKY,
+		identifier: `did:plc:${connectionId}`,
+		name: `${connectionId}.bsky.social`,
+		verified,
+		visibility_flags: ConnectionVisibilityFlags.EVERYONE,
+		sort_order: 0,
+		verification_token: '',
+		verified_at: verified ? new Date(0) : null,
+		last_verified_at: verified ? new Date(0) : null,
+		created_at: new Date(0),
+		version: 1,
+	};
+}
+
+function failingBlueskyOAuthService(): IBlueskyOAuthService {
+	return {
+		restoreAndVerify: vi.fn().mockResolvedValue(null),
+	} as unknown as IBlueskyOAuthService;
+}
+
+describe('ConnectionService.verifyConnection failed recheck', () => {
+	let executor: InMemoryCassandraQueryExecutor;
+
+	beforeEach(() => {
+		executor = new InMemoryCassandraQueryExecutor();
+		setCassandraQueryExecutorForTesting(executor);
+	});
+
+	afterEach(() => {
+		executor.reset();
+		setCassandraQueryExecutorForTesting(null);
+	});
+
+	it('publishes the connections update before rejecting a recheck that unverified the connection', async () => {
+		const userId = createUserID(1n);
+		const repository = new ConnectionRepository();
+		await repository.create(connectionRow(userId, 'alice', true));
+		const dispatchPresence = vi.fn().mockResolvedValue(undefined);
+		const service = new ConnectionService(
+			repository,
+			{dispatchPresence} as unknown as IGatewayService,
+			failingBlueskyOAuthService(),
+		);
+
+		const error = await service.verifyConnection(userId, ConnectionTypes.BLUESKY, 'alice').then(
+			() => null,
+			(caught: unknown) => caught,
+		);
+
+		expect(error).toBeInstanceOf(ConnectionVerificationFailedError);
+		const response = (error as ConnectionVerificationFailedError).getResponse();
+		expect(response.status).toBe(403);
+		expect(await response.json()).toMatchObject({code: APIErrorCodes.CONNECTION_VERIFICATION_FAILED});
+
+		expect(dispatchPresence).toHaveBeenCalledTimes(1);
+		expect(dispatchPresence).toHaveBeenCalledWith({
+			userId,
+			event: 'USER_CONNECTIONS_UPDATE',
+			data: {connections: [expect.objectContaining({id: 'alice', verified: false})]},
+		});
+
+		const stored = await repository.findById(userId, ConnectionTypes.BLUESKY, 'alice');
+		expect(stored?.verified).toBe(false);
+		expect(stored?.verified_at).toBeNull();
+	});
+
+	it('publishes the connections update when the recheck fails against an already unverified connection', async () => {
+		const userId = createUserID(2n);
+		const repository = new ConnectionRepository();
+		await repository.create(connectionRow(userId, 'bob', false));
+		const dispatchPresence = vi.fn().mockResolvedValue(undefined);
+		const service = new ConnectionService(
+			repository,
+			{dispatchPresence} as unknown as IGatewayService,
+			failingBlueskyOAuthService(),
+		);
+
+		await expect(service.verifyConnection(userId, ConnectionTypes.BLUESKY, 'bob')).rejects.toBeInstanceOf(
+			ConnectionVerificationFailedError,
+		);
+
+		expect(dispatchPresence).toHaveBeenCalledTimes(1);
+		expect(dispatchPresence).toHaveBeenCalledWith({
+			userId,
+			event: 'USER_CONNECTIONS_UPDATE',
+			data: {connections: [expect.objectContaining({id: 'bob', verified: false})]},
+		});
+	});
+});

fluxer_api/src/api/connection/tests/ConnectionReorder.test.ts

@@ -0,0 +1,150 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {ConnectionTypes, ConnectionVisibilityFlags} from '@fluxer/constants/src/ConnectionConstants';
+import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest';
+import {createUserID, type UserID} from '../../BrandedTypes';
+import type {IBlueskyOAuthService} from '../../bluesky/IBlueskyOAuthService';
+import {
+	type CassandraQueryExecutorForTesting,
+	setCassandraQueryExecutorForTesting,
+} from '../../database/CassandraQueryExecution';
+import type {CassandraParams, KvQueryMeta, PreparedQuery} from '../../database/CassandraTypes';
+import type {UserConnectionRow} from '../../database/types/ConnectionTypes';
+import type {IGatewayService} from '../../infrastructure/IGatewayService';
+import {InMemoryCassandraQueryExecutor} from '../../test/InMemoryCassandraQueryExecutor';
+import {ConnectionRepository} from '../ConnectionRepository';
+import {ConnectionService} from '../ConnectionService';
+import type {IConnectionRepository} from '../IConnectionRepository';
+
+class BatchFaultInjectingExecutor implements CassandraQueryExecutorForTesting {
+	failNextBatch = false;
+	private readonly inner = new InMemoryCassandraQueryExecutor();
+
+	async executeQuery<T = Record<string, unknown>, P extends CassandraParams = CassandraParams>(
+		query: PreparedQuery<P>,
+	): Promise<Array<T>> {
+		return this.inner.executeQuery<T>(query);
+	}
+
+	async executeBatch(
+		queries: Array<{query: string; params: object; meta?: KvQueryMeta}>,
+		atomic?: boolean,
+	): Promise<void> {
+		if (this.failNextBatch) {
+			this.failNextBatch = false;
+			throw new Error('batch rejected');
+		}
+		await this.inner.executeBatch(queries, atomic);
+	}
+
+	reset(): void {
+		this.inner.reset();
+	}
+}
+
+function connectionRow(userId: UserID, connectionId: string, sortOrder: number): UserConnectionRow {
+	return {
+		user_id: userId,
+		connection_id: connectionId,
+		connection_type: ConnectionTypes.DOMAIN,
+		identifier: `${connectionId}.example`,
+		name: connectionId,
+		verified: true,
+		visibility_flags: ConnectionVisibilityFlags.EVERYONE,
+		sort_order: sortOrder,
+		verification_token: 'token',
+		verified_at: null,
+		last_verified_at: null,
+		created_at: new Date(0),
+		version: 1,
+	};
+}
+
+describe('ConnectionService.reorderConnections', () => {
+	it('writes every sort order through a single batched repository call', async () => {
+		const userId = createUserID(1n);
+		const updateSortOrders = vi.fn().mockResolvedValue(undefined);
+		const update = vi.fn().mockResolvedValue(undefined);
+		const repository = {
+			findByUserId: vi
+				.fn()
+				.mockResolvedValue([
+					connectionRow(userId, 'a', 0),
+					connectionRow(userId, 'b', 1),
+					connectionRow(userId, 'c', 2),
+				]),
+			update,
+			updateSortOrders,
+		} as unknown as IConnectionRepository;
+		const dispatchPresence = vi.fn().mockResolvedValue(undefined);
+		const service = new ConnectionService(
+			repository,
+			{dispatchPresence} as unknown as IGatewayService,
+			{} as unknown as IBlueskyOAuthService,
+		);
+
+		await service.reorderConnections(userId, ['c', 'missing', 'a', 'b']);
+
+		expect(update).not.toHaveBeenCalled();
+		expect(updateSortOrders).toHaveBeenCalledTimes(1);
+		expect(updateSortOrders).toHaveBeenCalledWith(userId, [
+			{connectionType: ConnectionTypes.DOMAIN, connectionId: 'c', sortOrder: 0},
+			{connectionType: ConnectionTypes.DOMAIN, connectionId: 'a', sortOrder: 2},
+			{connectionType: ConnectionTypes.DOMAIN, connectionId: 'b', sortOrder: 3},
+		]);
+		expect(dispatchPresence).toHaveBeenCalledTimes(1);
+	});
+});
+
+describe('ConnectionService.reorderConnections batch failure', () => {
+	let executor: BatchFaultInjectingExecutor;
+
+	beforeEach(() => {
+		executor = new BatchFaultInjectingExecutor();
+		setCassandraQueryExecutorForTesting(executor);
+	});
+
+	afterEach(() => {
+		executor.reset();
+		setCassandraQueryExecutorForTesting(null);
+	});
+
+	it('leaves every sort order untouched and publishes no dispatch when the batch fails', async () => {
+		const userId = createUserID(1n);
+		const repository = new ConnectionRepository();
+		for (const [index, connectionId] of ['a', 'b', 'c'].entries()) {
+			await repository.create(connectionRow(userId, connectionId, index));
+		}
+		const dispatchPresence = vi.fn().mockResolvedValue(undefined);
+		const service = new ConnectionService(
+			repository,
+			{dispatchPresence} as unknown as IGatewayService,
+			{} as unknown as IBlueskyOAuthService,
+		);
+
+		executor.failNextBatch = true;
+		await expect(service.reorderConnections(userId, ['c', 'b', 'a'])).rejects.toThrow('batch rejected');
+
+		const afterFailure = await repository.findByUserId(userId);
+		expect(new Map(afterFailure.map((row) => [row.connection_id, row.sort_order]))).toEqual(
+			new Map([
+				['a', 0],
+				['b', 1],
+				['c', 2],
+			]),
+		);
+		expect(dispatchPresence).not.toHaveBeenCalled();
+
+		await service.reorderConnections(userId, ['c', 'b', 'a']);
+
+		const afterRetry = await repository.findByUserId(userId);
+		expect(new Map(afterRetry.map((row) => [row.connection_id, row.sort_order]))).toEqual(
+			new Map([
+				['c', 0],
+				['b', 1],
+				['a', 2],
+			]),
+		);
+		expect(dispatchPresence).toHaveBeenCalledTimes(1);
+	});
+});

fluxer_api/src/api/test/InMemoryCassandraQueryExecutor.ts

@@ -203,7 +203,10 @@ export class InMemoryCassandraQueryExecutor implements CassandraQueryExecutorFor
 		return [];
 	}
 
-	async executeBatch(queries: Array<{query: string; params: object; meta?: KvQueryMeta}>): Promise<void> {
+	async executeBatch(
+		queries: Array<{query: string; params: object; meta?: KvQueryMeta}>,
+		_atomic?: boolean,
+	): Promise<void> {
 		for (const query of queries) {
 			await this.executeQuery({
 				cql: query.query,