diff --git a/fluxer_api/src/api/connection/ConnectionRepository.ts b/fluxer_api/src/api/connection/ConnectionRepository.ts index dbeab1891..c5a88c3a7 100644 --- a/fluxer_api/src/api/connection/ConnectionRepository.ts +++ b/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 { + 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 { await deleteOneOrMany( UserConnections.deleteByPk({user_id: userId, connection_type: connectionType, connection_id: connectionId}), diff --git a/fluxer_api/src/api/connection/ConnectionService.ts b/fluxer_api/src/api/connection/ConnectionService.ts index b4d647509..54c212fde 100644 --- a/fluxer_api/src/api/connection/ConnectionService.ts +++ b/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): Promise { 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(); + 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); } diff --git a/fluxer_api/src/api/connection/IConnectionRepository.ts b/fluxer_api/src/api/connection/IConnectionRepository.ts index d8361a7b5..1b3573ef1 100644 --- a/fluxer_api/src/api/connection/IConnectionRepository.ts +++ b/fluxer_api/src/api/connection/IConnectionRepository.ts @@ -51,6 +51,11 @@ export abstract class IConnectionRepository { params: UpdateConnectionParams, ): Promise; + abstract updateSortOrders( + userId: UserID, + entries: Array<{connectionType: ConnectionType; connectionId: string; sortOrder: number}>, + ): Promise; + abstract delete(userId: UserID, connectionType: ConnectionType, connectionId: string): Promise; abstract count(userId: UserID): Promise; diff --git a/fluxer_api/src/api/connection/tests/ConnectionRecheckDispatch.test.ts b/fluxer_api/src/api/connection/tests/ConnectionRecheckDispatch.test.ts new file mode 100644 index 000000000..6b4378167 --- /dev/null +++ b/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})]}, + }); + }); +}); diff --git a/fluxer_api/src/api/connection/tests/ConnectionReorder.test.ts b/fluxer_api/src/api/connection/tests/ConnectionReorder.test.ts new file mode 100644 index 000000000..0bfb43327 --- /dev/null +++ b/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, P extends CassandraParams = CassandraParams>( + query: PreparedQuery

, + ): Promise> { + return this.inner.executeQuery(query); + } + + async executeBatch( + queries: Array<{query: string; params: object; meta?: KvQueryMeta}>, + atomic?: boolean, + ): Promise { + 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); + }); +}); diff --git a/fluxer_api/src/api/test/InMemoryCassandraQueryExecutor.ts b/fluxer_api/src/api/test/InMemoryCassandraQueryExecutor.ts index 70eebac58..dd0bebdf5 100644 --- a/fluxer_api/src/api/test/InMemoryCassandraQueryExecutor.ts +++ b/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 { + async executeBatch( + queries: Array<{query: string; params: object; meta?: KvQueryMeta}>, + _atomic?: boolean, + ): Promise { for (const query of queries) { await this.executeQuery({ cql: query.query,